Workflow composition and nodes

Compile a workflow into a node graph

A Flyte workflow function is not the code that runs on a Flyte worker. Its body is evaluated while flytekit compiles or serializes the workflow so that task calls can be turned into a DAG. At that point, values such as a and t1's outputs are represented by promises, not by the results of executing Python functions. The resulting nodes and output bindings are what Flyte registers.

Use the @workflow decorator

Call tasks with keyword arguments, pass one task's output to the next task, and return the outputs that should become workflow outputs:

import typing
from flytekit import task, workflow


@task(enable_deck=True)
def t1(a: int) -> typing.NamedTuple("OutputsBC", t1_int_output=int, c=str):
    return a + 2, "world"


@task
def t2(a: str, b: str) -> str:
    return b + a


@workflow
def my_wf(a: int, b: str) -> (int, str):
    x, y = t1(a=a)
    d = t2(a=y, b=b)
    return x, d

This example is from tests/flytekit/integration/remote/workflows/basic/basic_workflow.py. The two outputs of t1 are unpacked and the string output y is bound to t2.a; the workflow input b is bound to t2.b. The same workflow can be invoked locally with keyword arguments, for example my_wf(a=50, b="hello"). Positional arguments are not supported for task and workflow calls in this composition API.

You can also configure the workflow at decoration time:

from flytekit.core.workflow import WorkflowFailurePolicy


@workflow(
    interruptible=True,
    failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE,
)
def wf(a: int) -> typing.Tuple[str, str]:
    x, y = t1(a=a)
    _, v = t1(a=x)
    return y, v

The workflow decorator accepts failure_policy, interruptible, on_failure, docs, pickle_untyped, and default_options. Its implementation in flytekit/core/workflow.py constructs a PythonFunctionWorkflow.

What PythonFunctionWorkflow.compile() does

PythonFunctionWorkflow stores the original function as _workflow_function, derives its interface with transform_function_to_interface, and starts with compiled = False. Its compile() method is lazy and cached:

1. It creates input promises with construct_input_promises for the workflow interface.
2. It calls the original workflow function with those promises.
3. Calls to tasks and nested Flyte entities add nodes to the active CompilationState.
4. It converts the returned promise values into output bindings.
5. It stores the collected nodes in _nodes and the bindings in _output_bindings.

The compiled flag is set before the function body is evaluated and subsequent calls to compile() return without rebuilding the graph. Accessing workflow nodes or output bindings, serializing the workflow, and invoking it in compilation mode therefore use the same compiled definition. For local execution, PythonFunctionWorkflow.execute() calls the original function, while the decorator's documented platform behavior is that the body is not evaluated again when the workflow runs on Flyte.

Keep non-Flyte computation and side effects out of a workflow body when you expect them to happen on a worker. The body is used to express structure at compile/serialization time; put worker-side behavior in a @task.

Nodes connect entities and dependencies

Each task, sub-workflow, launch plan, or other supported Flyte entity call made during compilation produces a Node. flytekit.core.promise.create_and_link_node is the central path used when a Flyte entity is invoked: it creates the node, derives bindings from the arguments, detects upstream promises, adds the node to the compilation state, and returns output promises.

flytekit.core.node.Node stores the pieces needed for the eventual workflow model:

• id is normalized with _dnsify; automatically created nodes commonly use names such as n0 and n1.
• metadata is a NodeMetadata containing node settings such as timeout, retries, cacheability, and interruptibility.
• bindings contains the input connections resolved from workflow inputs, literals, and upstream outputs.
• upstream_nodes contains explicit graph predecessors.
• flyte_entity is the task, workflow, launch plan, or other entity executed by the node.

The graph has two related forms:

workflow input promise ──> t1 Node ──> t1 output promise ──> t2 Node
                              └────────────── workflow output binding

Passing a promise as a task argument creates a binding and gives the consumer node a data dependency. A literal argument, such as t1(a=1), is instead represented as literal binding data. Workflow inputs are associated with GLOBAL_START_NODE, the sentinel node whose ID is the empty string. The translator skips that start node when it serializes the actual workflow nodes.

Task outputs can be named outputs. In the example above, x and y correspond to the fields in OutputsBC; returning x, d causes PythonFunctionWorkflow.compile() to create the workflow's output bindings from those promises. A tuple-valued multi-output result must be unpacked when that is what the task interface declares—passing the entire tuple as one task input is rejected by flytekit.

Add dependencies that are not data bindings

A data edge is not always enough. For side-effect tasks that have no input or output to connect, use create_node() from flytekit.core.node_creation and add an ordering edge explicitly:

from flytekit import task, workflow
from flytekit.core.node_creation import create_node


@task
def t2():
    ...


@task
def t3():
    ...


@workflow
def empty_wf():
    t2_node = create_node(t2)
    t3_node = create_node(t3)
    t3_node >> t2_node

runs_before() expresses the same relationship:

@workflow
def empty_wf2():
    t2_node = create_node(t2)
    t3_node = create_node(t3)
    t3_node.runs_before(t2_node)

Node.__rshift__() calls runs_before() and returns the right-hand node, so t1_node >> t2_node >> t3_node can be chained. runs_before() appends the left node to the other node's upstream_nodes only if it is not already present.

create_node() accepts only keyword inputs. In compilation mode it returns the actual Node and populates its output accessors. For a task with outputs, use either an attribute such as o0 or the output dictionary:

@workflow
def my_wf(a: str) -> (str, typing.List[str]):
    t1_node = create_node(t1, a=a)
    dyn_node = create_node(my_subwf, a=3)
    return t1_node.o0, dyn_node.o0

The implementation initializes node._outputs, then maps each interface output name to both a node attribute and node.outputs[name]. The outputs property is intentionally unavailable on arbitrary Node instances; accessing it on a node created by an ordinary direct task call raises an assertion. In local execution, create_node() executes the entity and returns its results rather than a compilation node.

Customize an individual node with with_overrides()

Apply an override to the node returned by a task or explicit node creation at the call site:

from flytekit import Cache


@workflow
def my_wf(a: str) -> str:
    s = t1(a=a).with_overrides(timeout=timeout)
    s1 = t1(a=s).with_overrides()
    s2 = t2(a=s1).with_overrides(timeout=timeout)
    return t2(a=s2).with_overrides(
        retries=3,
        interruptible=True,
        cache=Cache(version="foo", serialize=True),
    )

The override method is implemented by Node.with_overrides() in flytekit/core/node.py. It returns the same node after applying supported node-level settings, including:

• node_name, which is DNS-normalized;
• name, which changes the metadata display name;
• timeout as integer seconds or datetime.timedelta;
• retries and interruptible;
• requests and limits, or a combined resources specification;
• container_image, accelerator, shared_memory, and pod_template;
• cache, including a Cache object.

Use one resource form at a time. resources= cannot be combined with requests= or limits=. A Cache object used for an override must have a version; cache=Cache(version="foo", serialize=True) is valid, while a cache object without a version raises ValueError. Override values must be concrete: Node.with_overrides() calls assert_not_promise() for values such as node names, retries, interruptibility, container images, accelerators, and cache settings, so an upstream task output cannot be used as an override argument.

Compose sub-workflows

A decorated workflow is itself a Flyte entity, so a parent workflow can call it like a task and pass its promise outputs onward:

@workflow
def my_subwf(a: int = 42) -> (str, str):
    x, y = t1(a=a)
    u, v = t1(a=x)
    return y, v


@workflow
def parent_wf(a: int) -> (int, str, str):
    x, y = t1(a=a)
    u, v = my_subwf(a=x)
    return x, u, v

The parent contains a workflow node for my_subwf; its returned values are still promises during compilation and can be unpacked or passed to later entities. During serialization, get_serializable_workflow() recursively serializes workflow nodes and collects their templates in sub_workflows. A ReferenceWorkflow cannot be serialized as a sub-workflow because flytekit would need to retrieve its template from Flyte Admin; the translator raises an error instructing you to use a reference launch plan instead.

Failure policy and failure handlers

Select how Flyte treats independent executable nodes after a failure with WorkflowFailurePolicy:

• FAIL_IMMEDIATELY is the default policy.
• FAIL_AFTER_EXECUTABLE_NODES_COMPLETE allows nodes that do not depend on the failed node to continue.

The enum is translated into workflow metadata during serialization. The test in tests/flytekit/unit/core/test_workflows.py verifies that FAIL_AFTER_EXECUTABLE_NODES_COMPLETE is serialized as the workflow's on_failure metadata value.

Register a task or workflow as an on_failure handler when you need a failure node:

from flytekit.types.error.error import FlyteError


@task
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name} due to {err}")


@workflow(on_failure=clean_up)
def wf(name: str = "flyteorg"):
    c = create_cluster(name=name)
    t = t1(a=1, b="2")
    d = delete_cluster(name=name)
    c >> t >> d

PythonFunctionWorkflow._validate_add_on_failure_handler() compiles the handler in a separate compilation state and stores the resulting node as _failure_node. The handler must accept all workflow inputs; any additional handler inputs must be optional. The handler compilation must produce exactly one task or workflow node. During serialization, the failure node is serialized separately from the main node list.

Build workflows programmatically with ImperativeWorkflow

Use ImperativeWorkflow—also exported as flytekit.Workflow—when you want to construct the graph through an API rather than by evaluating a Python function body:

from flytekit.core.workflow import ImperativeWorkflow as Workflow


wb = Workflow(name="my_workflow")
wb.add_workflow_input("in1", str)
node = wb.add_entity(t1, a=wb.inputs["in1"])
wb.add_entity(t2)
wb.add_workflow_output("from_n0t1", node.outputs["o0"])

assert wb(in1="hello") == "hello world"

This example comes from tests/flytekit/unit/core/test_imperative.py. add_workflow_input() creates a typed workflow input promise, add_entity() adds a task or other Flyte entity and returns a node, and add_workflow_output() binds a named output. The imperative API also provides add_task(), add_subwf(), add_launch_plan(), and add_on_failure_handler(). Its compilation state accumulates nodes as you add them, and ready() validates that the workflow has nodes and that declared inputs used by the workflow have been bound.

A sub-workflow is added through the same interface:

wb2 = ImperativeWorkflow(name="parent.imperative")
p_in1 = wb2.add_workflow_input("p_in1", str)
p_node0 = wb2.add_subwf(wb, in1=p_in1)
wb2.add_workflow_output("parent_wf_output", p_node0.from_n0t1, str)

The functional equivalent uses a decorated function and returns its task output:

nt = typing.NamedTuple("wf_output", [("from_n0t1", str)])


@workflow
def my_workflow(in1: str) -> nt:
    x = t1(a=in1)
    t2()
    return nt(x)

Both forms produce nodes, bindings, interfaces, and workflow outputs that can be serialized.

Serialize the executable definition

flytekit.tools.translator.get_serializable_workflow() converts a WorkflowBase into an Admin WorkflowSpec. It iterates over entity.nodes, skips GLOBAL_INPUT_NODE_ID, rejects the reserved failure-node ID, and calls get_serializable() for each remaining node. It also recursively collects sub-workflow templates and serializes output bindings and the failure node into the workflow specification.

For a compiled function workflow, the relevant data flow is:

Python workflow function
        │ compile()
        ▼
CompilationState.nodes + output bindings
        │ get_serializable_workflow()
        ▼
admin_workflow_models.WorkflowSpec

The integration tests verify the resulting specification: a workflow built with create_node() has two serialized nodes and two outputs, while the imperative example has two nodes, one interface input, and one interface output.

Composition gotchas

Situation	flytekit behavior
Code in a workflow body	Evaluated during compilation/serialization; it is not evaluated again on the Flyte platform when the workflow runs.
Task or sub-workflow results in the body	Promise objects during compilation, not Python values. Use Flyte entities and conditional() rather than applying ordinary Python control flow to promises.
Calling a task or workflow with positional arguments	Rejected; pass inputs by keyword.
Multi-output task result	Unpack the named outputs before binding them individually.
node.outputs on a directly-called task result	Not available; use create_node() when you need the Node, oN attributes, or output dictionary.
Recompiling a PythonFunctionWorkflow	compiled caches the first compilation; later compile() calls are no-ops.
Reference workflow as a sub-workflow	Serialization raises an error; use a reference launch plan instead.