Workflow composition and nodes

Workflows are graph builders

When a workflow function calls a task, the call does not produce the task's native Python result during Flyte compilation. It produces a Promise that identifies a future node output. Flytekit uses those promises to build the graph and later to bind the workflow's declared outputs. The workflow decorator documents this explicitly: the function body is evaluated at serialization time to express the workflow structure, and is not evaluated again when the workflow runs on Flyte.

Use ordinary task and workflow calls to express data flow:

@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

Here x is passed as the input binding for the second add_5 node, simple_wf() is composed as an entity, and the conditional produces the second returned value. Inside the workflow body, do not use the result as if it were a native integer: operations such as range(task_output) and truth-value testing are not valid for the compilation-time Promise. Use Flyte-supported promise operations and workflow conditionals instead.

From a decorated function to a workflow definition

workflow accepts failure_policy, interruptible, on_failure, docs, pickle_untyped, and default_options. Its defaults are FAIL_IMMEDIATELY, False, None, None, False, and None, respectively. For example:

@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. Its constructor derives the native interface from the Python callable and passes the workflow metadata, metadata defaults, failure handler, documentation, and launch-plan options to WorkflowBase (workflow.py).

Compilation is the graph-building pass:

1. PythonFunctionWorkflow.compile creates a CompilationState and calls construct_input_promises for every declared workflow input.
2. It invokes the original workflow function once with those input promises, collecting nodes from the compilation state.
3. A task, subworkflow, or launch-plan call is routed through flyte_entity_call_handler to create_and_link_node while compilation is active.
4. create_and_link_node converts each keyword argument with binding_from_python_std, discovers referenced upstream nodes, constructs a Node, and adds it to the compilation state.
5. The returned Promise objects become workflow output bindings. PythonFunctionWorkflow.compile stores the nodes in _nodes and the bindings in _output_bindings, which are the information needed to create the serialized workflow template.

The central linker creates node IDs from the compilation prefix and node count when no explicit ID is supplied:

upstream_nodes = list(set([n for n in nodes if n.id != _common_constants.GLOBAL_INPUT_NODE_ID]))
node_id = node_id or (
    f"{ctx.compilation_state.prefix}n{len(ctx.compilation_state.nodes)}"
    if add_node_to_compilation_state and ctx.compilation_state
    else node_id
)

flytekit_node = Node(
    id=node_id,
    metadata=entity.construct_node_metadata(),
    bindings=sorted(bindings, key=lambda b: b.var),
    upstream_nodes=upstream_nodes,
    flyte_entity=entity,
)
ctx.compilation_state.add_node(flytekit_node)

Compilation is cached by PythonFunctionWorkflow.compiled; after the first successful call, compile returns without reevaluating the function. Create a fresh workflow entity if you need compilation to reflect a changed function body or different compilation behavior.

Nodes, bindings, and dependencies

A Node in node.py is the in-memory graph record. Its constructor stores a DNS-normalized ID, NodeMetadata, input bindings, upstream_nodes, and the Flyte entity that will run. The public properties expose these parts as id/name, metadata, bindings, upstream_nodes, and flyte_entity. A node with no ID is rejected with ValueError.

Data dependencies are inferred from promise references. If a binding contains the output of an earlier task, binding_from_python_std returns that referenced node along with the binding; create_and_link_node removes the global input node and records the remaining nodes as upstream_nodes. The resulting relationship is therefore represented twice: the node has a literal binding describing which value it consumes, and an upstream-node list describing which graph nodes must precede it.

A void task returns a VoidPromise; it has no usable output binding. Passing a VoidPromise as a downstream input is invalid. For tasks that do have outputs, the linker returns one Promise per declared output, or a tuple/single wrapper according to the entity interface.

Explicit ordering for side effects

Promise data flow is normally enough to establish ordering. For side-effect-only entities that do not consume one another's outputs, use create_node and an explicit edge:

t1_node = create_node(t1)
t2_node = create_node(t2)
t2_node.runs_before(t1_node)
# OR
t2_node >> t1_node

Node.runs_before appends the left node to the other node's upstream list if it is not already present. It returns None. Node.__rshift__ calls the same method and returns the right-hand node, so t2_node >> t1_node can be chained. This adds a graph edge; it does not execute either entity.

The manual-node path also exposes outputs as promises. For example, create_node is used for an input-bound node and then for a downstream task:

t3_node = create_node(t3, in1=some_int)
t4_node = create_node(t4)
t5(in1=t4_node.o0)

This is an important distinction: Node.outputs raises unless the node was created through create_node, which attaches output promises as attributes such as o0 and in an outputs mapping. Nodes made internally by create_and_link_node or the conditional compiler should be consumed through the Promise returned by the entity call, not by assuming that the internal Node has .outputs.

Subworkflows and conditionals

A decorated workflow is itself a workflow entity, so calling simple_wf() in another workflow follows the same linker path as calling a task. Launch plans and reference entities likewise use create_and_link_node while a workflow is compiling.

conditional is workflow-only. During compilation its branch promises are converted into bindings; the conditional compiler gathers their referenced nodes, creates a synthetic Node, and adds that node to the active CompilationState. If a conditional is returned as a workflow output, it must end with else_(); otherwise PythonFunctionWorkflow.compile raises an assertion while binding outputs.

Collections and maps can be carried through bindings, but output typing still comes from the declared Python interface. A one-output workflow treats a tuple specially so a list-valued output is not accidentally iterated while constructing its binding. Multiple outputs must be returned as a tuple with exactly the declared length. A declared output with no return value, or a return shape that does not match the interface, fails during compilation rather than during Flyte execution.

Customizing a node with overrides

Apply node-level execution metadata to the promise returned by an entity call:

node_output = t1(a=1).with_overrides(
    node_name="first-step",
    timeout=60,
    retries=2,
    interruptible=True,
    container_image="example/image:tag",
)

The promise wrappers forward with_overrides to the referenced Node; create_node(...).with_overrides(...) is the equivalent manual-node form. Node.with_overrides mutates the existing node. It can change the DNS-normalized node_name, aliases, requests and limits (or a combined resources value), timeout, retries, interruptibility, cache metadata, image, accelerator, shared memory, pod template, and compatible task configuration.

Resource forms are mutually exclusive: resources cannot be combined with requests or limits. Supplying requests without limits emits a warning and clamps requests to the original limits. Resource and metadata override values cannot be promises; assert_no_promises_in_resources and assert_not_promise enforce this because these are compile-time node settings.

Cache overrides retain compatibility behavior. cache=True without a cache version constructs a default Cache, while a Cache object used in an override must have a version. Combining a Cache object with deprecated cache parameters such as cache_serialize or cache_version raises ValueError. task_config overrides are beta and must have the same type as the entity's existing task configuration.

Programmatic composition with ImperativeWorkflow

Use ImperativeWorkflow when the graph must be assembled by application code rather than by evaluating a decorated function. The equivalent of the decorated example above is:

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 extends the workflow interface and creates a Promise whose NodeOutput points to GLOBAL_START_NODE. add_entity enters the workflow's compilation state and calls create_node; all entity inputs must be bound. It also removes consumed input promises from _unbound_inputs. add_workflow_output converts the supplied promise (or promise collection, when a Python type is supplied) into a binding and extends the workflow interface.

The function-based equivalent uses a NamedTuple only when the output needs a particular name:

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

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

Call ready() before local execution. It requires at least one node and rejects every workflow input that was declared but never consumed. ImperativeWorkflow.execute then walks compilation_state.nodes in insertion order, resolves each node's bindings with get_promise_map, invokes the entity, stores its outputs, and resolves the final workflow output bindings. Consequently, add entities in topological order for local execution; the imperative executor does not perform a separate graph sort.

The imperative API rejects duplicate workflow inputs and outputs. create_node also rejects positional arguments: pass task and workflow inputs by keyword. The linker validates missing and extra inputs, rejects untyped inputs, rejects tuple-valued inputs (often indicating that multiple outputs were passed as one argument), and does not support mutable list, dictionary, or set defaults in this node-creation path.

Local execution, serialization, and integrations

The graph is built during compilation, but local execution has a separate path. PythonFunctionWorkflow.execute delegates to the original function. WorkflowBase.local_execute translates native arguments into Flyte literals, executes through the workflow interface, and repackages results according to the declared interface. The platform serialization path uses the compiled _nodes and _output_bindings; serialization settings such as project, domain, version, and image configuration are supplied by the surrounding Flyte context rather than read from workflow-module environment variables.

Two integrations reuse this same model:

• Dynamic tasks cache a PythonFunctionWorkflow, compile it inside a dynamic context, and serialize the resulting workflow definition. An empty dynamic body returns literal outputs; otherwise the dynamic task produces a dynamic job specification containing the compiled template nodes.
• Eager task conversion creates an ImperativeWorkflow, adds each task input, adds the eager task entity, exposes each node output as a workflow output, and installs an EagerFailureHandlerTask with add_on_failure_handler.

Failure handlers are compiled separately. PythonFunctionWorkflow._validate_add_on_failure_handler compiles the handler with a separate prefix and stores the resulting node as _failure_node, rather than adding it to the main node list. The handler interface must contain all workflow inputs; any additional handler inputs must be optional, or Flytekit raises FlyteFailureNodeInputMismatchException. A workflow can use explicit ordering in its main graph alongside a failure handler:

@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

The result is an executable workflow definition whose nodes carry entity metadata and input bindings, whose upstream lists encode data and explicit ordering, and whose output bindings identify the values returned by the workflow interface.