Workflow composition and nodes

When you write a Flyte workflow, the function body is not the code Flyte runs for every workflow execution. The @workflow decorator uses the body to declare a directed acyclic graph (DAG): calls to Flyte tasks and other workflows become graph nodes, and the values returned by those calls are Promise objects that describe data flowing between nodes. The decorator’s docstring in workflow.py explicitly calls workflow functions declarative and says that their bodies are evaluated at serialization time (local execution is a separate path).

Compose a workflow with task outputs

Use ordinary task calls to express data dependencies, call a workflow as a subworkflow, and return the outputs that form the workflow interface:

@task
def add_5(a: int) -> int:
    a = a + 5
    return a


@workflow
def simple_wf() -> int:
    return add_5(a=1)


@workflow
def my_wf_example(a: int) -> typing.Tuple[int, int]:
    x = add_5(a=a)
    z = add_5(a=x)
    d = simple_wf()
    e = conditional("bool").if_(a == 5).then(add_5(a=d)).else_().then(add_5(a=z))
    return x, e

This is the composition pattern used in workflow.py: z depends on x, the call to simple_wf() is a subworkflow node, and the conditional selects between two task branches. The returned x and e become the two workflow outputs. During compilation, x, z, and d are not native integers; they are promise-backed references to node outputs. Consequently, use Flyte-supported promise operations—such as selecting an output field or indexing a supported value—rather than treating a promise as an iterable or using it as a Python truth value. For example, the promise examples in promise.py use o["a"][0] to traverse an output.

A task call inside a compiling workflow follows the ordinary task compilation path in PythonTask.compile, which delegates to create_and_link_node. The compiler does not invoke the task’s Python body to obtain a result. Instead, it creates a Node and returns a Promise, a named tuple of promises, or a VoidPromise corresponding to the task interface.

How calls become graph nodes

create_and_link_node in promise.py is the central link between data flow and graph structure. Its processing is:

It verifies that a compilation state exists when the node is being added to the workflow.
It transforms the entity interface into a typed interface and validates inputs, including required inputs, defaults, type annotations, and extra arguments.
It converts each input into a literal binding. If an input contains a promise, the binding refers to the promise’s NodeOutput.
It collects the referenced nodes as upstream nodes, excluding the global input node.
It assigns an ID using the compilation prefix and current node count, such as n0 and n1.
It constructs a Node, adds it to CompilationState, and creates one Promise per declared output.

The resulting in-memory vertex is the Node class from node.py. Its constructor stores the DNS-normalized ID, NodeMetadata, literal bindings, upstream_nodes, and the original Flyte entity in _flyte_entity. These fields provide both sides of composition:

input promise or native input
        │
        ▼
Node.bindings ──► Node.flyte_entity
        ▲                 │
        │                 ▼
Node.upstream_nodes   Promise(NodeOutput(node, output_name))

The compiler rejects a tuple-valued input in create_and_link_node. In Flyte workflow code, a tuple commonly means that a multi-output result was passed without selecting one output, so select the named output or tuple field before passing it to another task.

Node IDs are generated from CompilationState.prefix and the number of nodes already present. A node created with id=None is rejected by Node, while IDs supplied through Node.with_overrides(node_name=...) are DNS-normalized.

Compile the decorated workflow into its definition

workflow supports both @workflow and @workflow(...). Its public options are failure_policy, interruptible, on_failure, docs, pickle_untyped, and default_options:

@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 decorator creates a PythonFunctionWorkflow. That class derives its interface from the decorated function’s signature and holds the original function for local execution. Its compile method is the graph-building pass:

It creates input promises with construct_input_promises for every declared input.
It applies any keyword arguments passed to compile as static native values; the method documents this as closure-like compilation input.
It runs the workflow function once inside a fresh CompilationState.
It gathers the nodes created by task, workflow, launch-plan, conditional, and related entity calls.
It converts the returned promises or native values into output bindings with binding_from_python_std.
It stores the result in _nodes and _output_bindings.

WorkflowBase.nodes and WorkflowBase.output_bindings trigger compilation before returning those stored values. The serialized workflow machinery can then consume those nodes and bindings. The source’s serialization example passes a workflow to get_serializable with SerializationSettings; the resulting model contains the workflow template, including its interface and nodes. The compilation itself does not supply project, domain, version, or image settings—those are provided later through serialization settings.

Compilation is guarded by PythonFunctionWorkflow.compiled. After the first call, later calls to compile return immediately, so static keyword arguments affect only the first compilation. This also means that changing a closure-like value after compilation does not rebuild the graph.

Match the workflow interface to its outputs

The decorated function’s annotations define the workflow interface. PythonFunctionWorkflow.compile validates the returned shape while creating output bindings:

A void workflow must return None or a VoidPromise.
A one-output workflow may return one value; a one-element named tuple receives special handling so the single value is not mistaken for multiple outputs.
A workflow with multiple outputs must return a tuple whose length matches the declared output names.
A returned, unfinished ConditionalSection is rejected; conditional composition must complete with else_().

The output binding is a reference to a prior node output when the function returns a task or workflow promise. Native values can also be bound when the compiler is given static values or when the binding conversion supports the declared type. The output bindings saved on WorkflowBase are the workflow-level links that make the graph’s results available to callers.

Local execution uses values wrapped in promises

Local execution does not change the workflow’s composition model into ordinary Python task calls. WorkflowBase.local_execute translates incoming native arguments into Flyte literals, wraps them in Promise objects, calls execute, and then repackages the results under the workflow’s declared output names. For a PythonFunctionWorkflow, execute delegates to the original function; the task calls made by that function use local execution behavior instead of adding serialized nodes.

This distinction explains why the same workflow function can build a graph during serialization and still run locally. In compilation mode, a task call returns a promise pointing to a NodeOutput. In local execution, the promises hold literal-backed values, and local_execute validates void, single-output, named-tuple, and multiple-output shapes before returning the workflow result.

For values that are only known at execution time and must control Python execution, use a dynamic workflow path rather than ordinary compile-time branching. PythonFunctionTask creates and caches a PythonFunctionWorkflow for dynamic execution, calls its compile(**kwargs) when producing a dynamic workflow specification, and calls its execute(**kwargs) for local dynamic execution.

Explicit dependencies and node overrides

Data flow automatically supplies upstream dependencies when one task consumes another task’s promise. It does not express ordering between two side-effect-only tasks that have no input or output connection. For that case, use create_node and the Node sequencing API:

t1_node = create_node(t1)
t2_node = create_node(t2)

t2_node.runs_before(t1_node)
# OR
t2_node >> t1_node

The example is the actual create_node docstring pattern in node_creation.py. Node.runs_before appends the source node to the other node’s upstream list if it is not already present. Node.__rshift__ calls the same method and returns the downstream node, so chaining is possible. The reverse shift direction is not implemented.

create_node accepts only keyword inputs and only Flyte entities: PythonTask, WorkflowBase, LaunchPlan, or RemoteEntity. In compilation mode it invokes the entity through the active compilation context, retrieves the newly appended node, and attaches output promises to that node. A multi-output node can therefore be consumed as follows:

t4_node = create_node(t4)
t5(in1=t4_node.o0)

The same promises are available by name through t4_node.outputs["o0"]. Node.outputs is intentionally available only for nodes produced through create_node; arbitrary Node instances raise an assertion when that property is accessed. Normal task calls expose their results as promises or named output fields instead.

create_node has separate compilation and local-execution branches. In local execution it calls the entity and normalizes the result into the entity’s output tuple shape. Remote entities are rejected for local execution, and manual node creation is rejected in a skipped conditional branch.

Override a composed node

Apply node-specific settings to the promise returned by a task or to a manually created node:

t3_node = create_node(t3, in1=some_int).with_overrides(...)

Node.with_overrides mutates the node. It supports a node name and aliases, requests and limits or a combined resources value, timeout, retries, interruptibility, task configuration, container image, accelerator, cache, shared memory, and pod template. The method DNS-normalizes node_name and delegates timeout, retry, interruptibility, and cache changes to _override_node_metadata.

Important validation behavior is implemented directly by Node:

Do not combine resources with requests or limits.
Resource override values cannot contain promises. Supplying requests without limits emits a warning and clamps requests to the original limits.
timeout may be an integer number of seconds or a datetime.timedelta; timeout=None resets it, while the sentinel means “leave it unchanged.”
Node names, retries, interruptibility, cache fields, container images, accelerators, shared memory, and pod templates must be compile-time values rather than promises.
A Cache used as an override must specify a version. Deprecated cache arguments cannot be combined with a Cache object.

The override is applied to the node metadata or, for an ArrayNode, to its sub-node metadata. This is why overrides belong after the task or explicit node call: they modify the already-created graph vertex rather than changing the Python task definition.

Programmatic composition with an imperative workflow

When the graph is assembled by an application rather than by evaluating a decorated function, use ImperativeWorkflow (shown as Workflow in the embedded example). Declare inputs, add entities in dependency order, and register outputs explicitly:

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"])

add_workflow_input creates a promise whose NodeOutput points to GLOBAL_START_NODE and records the input as unbound until an entity consumes it. add_entity temporarily installs the workflow’s CompilationState, calls create_node, and removes consumed input promises from the unbound set. add_task, add_launch_plan, and add_subwf are convenience methods that delegate to add_entity.

add_workflow_output turns a promise into a workflow output binding. It can infer the Python type from a single promise; list and dictionary collections require an explicit python_type. Duplicate input and output names are rejected. ready() requires at least one node and requires every declared workflow input to have been consumed.

For local execution, ImperativeWorkflow.execute walks compilation_state.nodes in their stored order. It resolves each node’s bindings from the output cache, invokes the entity, records its outputs, and finally resolves the workflow output bindings. The implementation therefore requires entities to be added in topological order for local execution. The imperative example is equivalent to this decorated workflow:

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

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

The imperative form makes output names explicit through add_workflow_output; the decorated form derives them from the return annotation, including named-tuple fields.

Conditionals, array nodes, and nested workflows

Not every graph vertex is a plain task node. ConditionalSection.end_branch converts a completed conditional into a BranchNode, gathers its promise bindings and upstream nodes, constructs a Node whose flyte_entity is that branch object, and adds it to the active compilation state. A conditional must be created inside a workflow context and must finish with else_() before its output can be bound.

Array-node composition uses the same node-linking machinery with a deliberate distinction. ArrayNode first calls create_and_link_node for the mapped entity with add_node_to_compilation_state=False, stores the resulting sub-node bindings, and then links the parent array node. The underlying task binding is retained, but the subnode is not added as a separate top-level workflow node.

A workflow call inside another workflow is handled as another Flyte entity call and becomes a workflow node. The canonical simple_wf() call above therefore participates in the containing workflow’s graph, while its own task node remains in its nested workflow definition. Launch plans and reference entities use the same create_and_link_node path, so their calls also contribute bindings, upstream references, and executable nodes.

Workflow-level operational behavior

Set workflow-wide metadata at the decorator boundary, while using Node.with_overrides for a particular graph node:

failure_policy defaults to WorkflowFailurePolicy.FAIL_IMMEDIATELY; the other accepted value is FAIL_AFTER_EXECUTABLE_NODES_COMPLETE.
interruptible defaults to False and is stored in WorkflowMetadataDefaults; it must be an actual boolean.
default_options is retained by WorkflowBase. When a default launch plan is created, its labels and annotations are copied from the workflow’s default options, and workflow signature defaults become launch-plan saved inputs.
docs supplies workflow documentation. If it is omitted, WorkflowBase populates documentation fields from the parsed function docstring when available.

Attach a failure handler with on_failure when the failure action is itself a Flyte task or workflow:

@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 compiles the failure handler separately and requires the handler interface to include all workflow inputs; any additional handler inputs must be optional. At execution time, WorkflowBase.__call__ supplies a FlyteError under the input name err when the handler declares that input. The failure node is validated and stored separately rather than becoming part of the main workflow node list.

The resulting model is therefore assembled from concrete pieces: PythonFunctionWorkflow or ImperativeWorkflow owns the interface, nodes, and output bindings; Node records entity, bindings, and dependencies; promises point to node outputs; and serialization converts the stored workflow structure into the Flyte workflow definition.

Compose a workflow with task outputs​

How calls become graph nodes​

Compile the decorated workflow into its definition​

Match the workflow interface to its outputs​

Local execution uses values wrapped in promises​

Explicit dependencies and node overrides​

Override a composed node​

Programmatic composition with an imperative workflow​

Conditionals, array nodes, and nested workflows​

Workflow-level operational behavior​