Workflow composition and nodes

Compose a workflow from task calls

When you write a decorated workflow, the task calls in its body become the workflow graph. For example, x below is not the integer returned by add_5 while flytekit is compiling the workflow; it is a Promise referring to a future node output.

@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

The z = add_5(a=x) call establishes a data dependency: the second task consumes the first task's output. Calling simple_wf() composes a workflow as another node-capable entity, and the conditional contributes its selected branch output. Use task and workflow outputs as values passed to other entities or returned from the workflow; do not apply ordinary Python operations that require a native value. The workflow source explicitly notes that an output such as t1() -> int is a Promise during scanning, so an expression such as range(a) is not valid when a is a task output.

The workflow function in workflow.py turns the decorated callable into a PythonFunctionWorkflow. It captures the callable's Python and typed interfaces and docstring, and accepts workflow-level options such as failure_policy, interruptible, on_failure, docs, pickle_untyped, and default_options. WorkflowFailurePolicy.FAIL_IMMEDIATELY is the default policy; WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE is the other accepted value. interruptible defaults to False and is validated as a boolean.

For example, decorator options are represented in the compiled template:

@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 decorated function is evaluated to construct the graph during compilation or serialization. PythonFunctionWorkflow.execute() delegates to the original callable for local execution, but the callable's platform representation is the compiled collection of nodes and bindings rather than the normal runtime execution of its Python body.

How calls become nodes

PythonFunctionWorkflow.compile() is the composition path that produces the executable workflow definition. It is idempotent: if self.compiled is already true, it returns without rebuilding the graph. Otherwise, it:

1. Gets the current FlyteContext and installs a CompilationState with a task resolver.
2. Calls construct_input_promises() for every declared workflow input. These promises reference the global start node.
3. Invokes the original workflow function with those promises, plus any static values supplied to compile(**kwargs).
4. Collects the nodes added to the compilation state and registers nested task entities with its resolver.
5. Validates and compiles an optional failure handler.
6. Converts the workflow's returned promises, collections, or literals into output Binding objects.

The central part of the source is:

with FlyteContextManager.with_context(
    ctx.with_compilation_state(CompilationState(prefix=prefix, task_resolver=resolver))
) as comp_ctx:
    # Construct the default input promise bindings, but then override with the provided inputs, if any
    input_kwargs = construct_input_promises([k for k in self.interface.inputs.keys()])
    input_kwargs.update(kwargs)
    workflow_outputs = self._workflow_function(**input_kwargs)
    all_nodes.extend(comp_ctx.compilation_state.nodes)

Task calls use the shared linker in promise.py. PythonTask.compile() delegates to create_and_link_node(), which resolves native values and promises into literal input bindings, finds the referenced upstream nodes, and constructs a Node:

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,
)

if add_node_to_compilation_state and ctx.compilation_state:
    ctx.compilation_state.add_node(flytekit_node)

The linker normally generates IDs using the compilation prefix and node count, such as prefix + "n" + index. It removes the global input node from the stored upstream list. For an entity with outputs, it creates a Promise for each output, with a NodeOutput pointing to the new node; an output-less entity produces a VoidPromise instead.

Launch plans use the same mechanism. When a compilation context exists, LaunchPlan.__call__ merges saved and supplied inputs and calls create_and_link_node(). Outside compilation, it forwards to the underlying workflow. Conditional sections and array-node map tasks also participate in this node construction path: a completed conditional becomes a branch Node, while an array node creates a hidden subnode for deriving bindings and then creates the actual array node.

Understand the Node graph record

Node in node.py stores the information needed to create the Flyte node definition later:

• a DNS-normalized ID;
• NodeMetadata;
• literal input bindings;
• upstream_nodes; and
• the flyte_entity, such as a task, workflow, launch plan, or map-task entity.

The flyte_entity property returns the stored entity. run_entity unwraps MapPythonTask to its run_task and ArrayNodeMapTask to its python_function_task, which lets callers reach the executable task behind those wrappers. metadata, bindings, and upstream_nodes expose the corresponding graph data.

Data flow creates upstream edges automatically. If a binding contains a promise produced by node n0, create_and_link_node() includes n0 in the new node's upstream_nodes. This is why passing x to the next task in z = add_5(a=x) both creates an input binding and orders the nodes.

Add an edge without passing data

Use an explicit edge when tasks have no data dependency but must run in a particular order—for example, side-effect tasks that create and delete an external resource. Node.runs_before(other) appends the receiver to other._upstream_nodes if it is not already there. The right-shift operator is the chainable equivalent and returns the downstream node:

c >> t >> d

The complete failure-handling example in workflow.py uses this form:

@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 same relationship can be written with the method form:

t2_node.runs_before(t1_node)

Promise and VoidPromise also forward >> to their referenced nodes when both operands have node references. This allows an output promise or a no-output task promise to participate in explicit ordering.

Return workflow outputs with the declared shape

After evaluating the body, PythonFunctionWorkflow.compile() binds the returned values to the declared interface. A one-output workflow is handled specially so a list or map remains one collection rather than being interpreted as multiple workflow outputs. A one-output named tuple is accepted only when the Python interface declares the corresponding tuple convention.

For multiple outputs, the return value must be a tuple with exactly the same length as the declared output interface. Each returned element is converted with binding_from_python_std(). A conditional returned as an output must be complete: compilation raises if a ConditionalSection has not ended with else_().

The source performs these checks before saving the compiled artifacts:

elif len(output_names) > 1:
    if not isinstance(workflow_outputs, tuple):
        raise AssertionError("The Workflow specification indicates multiple return values, received only one")
    if len(output_names) != len(workflow_outputs):
        raise ValueError(f"Length mismatch {len(output_names)} vs {len(workflow_outputs)}")
    for i, out in enumerate(output_names):
        if isinstance(workflow_outputs[i], ConditionalSection):
            raise AssertionError("A Conditional block (if-else) should always end with an `else_()` clause")

Once binding is complete, the workflow stores self._nodes and self._output_bindings. These are the artifacts used to create the workflow template and serialize it. An output-less workflow must return None or a VoidPromise; returning a value is invalid.

Choose explicit node construction for imperative workflows

Use ImperativeWorkflow when you need to register inputs, entities, and outputs explicitly rather than have a decorated function body scanned. Its add_workflow_input() creates a promise tied to the global start node. add_entity(), add_task(), add_launch_plan(), and add_subwf() call node_creation.create_node(), consume the input promises they use, and return a Node.

The source presents this imperative definition:

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

The equivalent decorated form is:

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

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

add_workflow_output() creates a typed literal binding from a promise, list, or dictionary and updates the workflow interface. For a single promise, it can infer the type. For list or dictionary promise outputs, provide a python_type beginning with the container type, such as List[int].

Imperative workflows differ from decorated workflows in execution order. ready() requires at least one node and requires every declared input to have been consumed. execute() walks nodes in insertion order, resolves bindings from its intermediate node-output cache, invokes each entity, and then resolves workflow output bindings. It does not reorder nodes based on dependencies, so add imperative nodes in topological order.

create_node() outputs are explicit

create_node() is the public manual-construction helper used by imperative composition and eager-workflow conversion. It is also useful when you need an output reference for a manually created node:

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

The helper's docstring also shows explicit ordering and overrides:

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

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

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

This output access is intentionally not universal. Ordinary nodes produced by the Promise linker do not populate Node._outputs; accessing node.outputs on one raises an assertion. Node.outputs is available only for a node created through create_node(), where the helper attaches output attributes and the output dictionary. In a normal decorated workflow, keep and pass the Promise returned by the task or workflow call instead.

Override an individual node

Call with_overrides() on a Node when one composition site needs different node settings. The method mutates the same node and returns self, so it can be chained. Its surface includes node_name, aliases, requests, limits, resources, timeout, retries, interruptible, name, task_config, container_image, accelerator, cache, shared_memory, and pod_template.

A mapped-task call in legacy_map_task.py applies a resource override at the call site:

@workflow
def my_wf(x: typing.List[int]) -> typing.List[typing.Optional[str]]:
    return map_task(
        my_mappable_task,
        metadata=TaskMetadata(retries=1),
        concurrency=10,
        min_success_ratio=0.75,
    )(a=x).with_overrides(requests=Resources(cpu="10M"))

The implementation validates static configuration. resources cannot be combined with requests or limits; requests and limits must be Resources objects. A request-only override logs that requests are clamped to the original limits. Promise values are rejected in static resource and override fields. Node names are DNS-normalized, aliases must be a dict[str, str], and a task_config override must have the same type as the entity's existing task configuration.

Timeout accepts an integer number of seconds or a datetime.timedelta; None clears the timeout, while other values raise ValueError. Retry count, interruptibility, and cache settings are written into the node metadata. A Cache object used for an override must have a cache version, and deprecated cache parameters cannot be combined with a Cache object. For an ArrayNodeMapTask, metadata overrides are applied to its sub-node metadata.

Failure handlers and execution contexts

Attach a failure handler with on_failure when the cleanup operation should be represented with the workflow. The handler must accept all workflow inputs, and any additional inputs must be optional. The optional parameter named err receives the failure error during failure handling; the source's validation and invocation path uses that literal interface key.

In the example above, clean_up is attached with @workflow(on_failure=clean_up). During compilation, PythonFunctionWorkflow.compile() calls _validate_add_on_failure_handler() in the compilation context and requires the handler to produce exactly one task or workflow node. During local execution, WorkflowBase.__call__ invokes the handler if execution raises.

Local execution and compilation intentionally use different values. WorkflowBase.local_execute() translates native inputs to Flyte literals, wraps them in promises, executes the original workflow, and repackages the results under the declared output names. Compilation instead uses input promises tied to GLOBAL_START_NODE and records graph bindings. Consequently, code that works with native values in the local execution path is not necessarily valid in the compilation scan if it performs unsupported operations on promises.

From compiled nodes to a published workflow

Accessing PythonFunctionWorkflow.nodes or output_bindings triggers compilation, after which the stored Node list and binding list can be serialized into the workflow template. get_serializable() consumes that compiled workflow representation together with SerializationSettings; the workflow's metadata and defaults are translated into Flyte IDL models. For example, the interruptible=True decorator option is carried through workflow metadata defaults, while FAIL_AFTER_EXECUTABLE_NODES_COMPLETE is represented by the serialized workflow failure policy.

A LaunchPlan can wrap a WorkflowBase and participate in another compiled workflow through the same Promise/node linker. For an already-registered workflow, reference_workflow() creates a ReferenceWorkflow from the annotated interface and the supplied project, domain, name, and version. That object points to the registered workflow; it does not execute a local workflow body or contact Admin while creating the reference.