Skip to main content

Workflow composition and nodes

A decorated Flyte workflow does not run its task calls as ordinary Python calls when Flyte compiles it. Instead, @workflow produces a PythonFunctionWorkflow; compilation evaluates the function body with input Promise objects, records Node objects, and converts the returned promises into workflow output bindings. This distinction matters when writing the body: a task result during compilation is a promise, not the task's native Python value.

Compose a workflow with task outputs​

Use ordinary task and workflow calls to express data flow. flytekit/core/workflow.py contains this complete composition example:

@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 and z are promises connected by data flow, simple_wf() is a sub-workflow node, and the conditional contributes the second returned value. Return values must come from prior node outputs; the workflow compiler binds the returned values to the declared interface.

Do not use a promise as though it were the eventual native value while the graph is being compiled. For example, the workflow module warns against using a task result in Python operations such as range(result) or truth-value testing. Put such computation in a task, or express the choice with conditional, as in the example above.

How a task call becomes a Node​

flytekit.core.promise.create_and_link_node() is the common path used when a task, workflow, launch plan, or other Flyte entity is called while compiling. Its behavior is:

  1. It obtains the entity's Python interface and transforms it into a typed interface.
  2. For every declared input, it creates a literal Binding through binding_from_python_std(). A Python value becomes a literal binding; a Promise contributes a reference to its producing node.
  3. It derives upstream_nodes from those referenced promises, excluding the global input node.
  4. It generates an id such as n0 (with the compilation prefix when one is present), constructs Node, and adds it to the active CompilationState.
  5. It creates one Promise per output, each pointing to a NodeOutput for the new node. A no-output entity instead returns a VoidPromise.

The essential construction is visible in promise.py:

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

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

Node in flytekit.core.node DNS-normalizes its id and stores the metadata, bindings, upstream nodes, and original Flyte entity. Its outputs property is intentionally not available on every Node; it raises an assertion unless outputs were attached by create_node(). In a normal decorated workflow, use the promises returned by task calls (x = add_5(a=a)), rather than trying to access a node output map.

Compile the function body into an executable definition​

PythonFunctionWorkflow.compile() performs graph construction once. It sets compiled = True, creates input promises associated with GLOBAL_START_NODE, and evaluates the original function inside a new CompilationState:

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)

It then compiles and validates the optional failure handler, converts the returned values to bindings, and stores the result in self._nodes and self._output_bindings. A second call to compile() returns immediately because compilation is treated as a one-time graph construction operation. execute() is separate: it invokes the original Python function for local execution.

Workflow output shape is checked during compilation. A multi-output interface requires a tuple of exactly the declared length. A one-output interface accepts a single value, with special handling for a one-element named tuple. A conditional returned as a workflow output must have an else_() branch; otherwise compilation raises. A zero-output workflow must produce no output.

Add ordering that data flow does not express​

Passing one task's promise to another automatically supplies a data dependency. It does not express ordering between independent side-effect tasks. Use Node.runs_before() or the right-shift operator for that separate control dependency:

c >> t >> d

The failure-handler example in workflow.py uses this form for create_cluster, a task that may fail, and delete_cluster. Internally, node_a >> node_b calls node_a.runs_before(node_b), appends node_a to node_b._upstream_nodes if it is not already present, and returns node_b, which permits chaining. The operation changes dependency metadata on the other node; it does not reverse the operator direction.

For explicit graph construction, use create_node() from flytekit.core.node_creation. Its docstring shows the complete pattern:

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)
t3_node = create_node(t3, in1=some_int).with_overrides(...)

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

create_node() accepts only keyword inputs and rejects positional arguments. During compilation it invokes the entity, takes the newly created node from the compilation state, and attaches its output promises both as attributes (node.o0) and in node.outputs (node.outputs["o0"]). For a no-output entity it returns the node directly. This output map is the explicit-node API; it is not a general property of arbitrary Node instances.

Override node metadata and resources​

Node.with_overrides() mutates the node and returns the same node, so apply it to the node or unresolved output promise you want to configure:

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

The method supports DNS-normalized node_name, aliases, requests and limits, a combined resources specification, timeout, retries, interruptible, cache, task_config, container_image, accelerator, shared_memory, and pod_template. The mapped-task example applies a resource override to the output promise returned by a mapped task:

@task
def my_mappable_task(a: int) -> typing.Optional[str]:
    return str(a)


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

These settings are static node metadata. Node rejects promises for names, timeout metadata, retry and cache settings, images, accelerators, shared memory, pod templates, and resource entries. Choose either resources or requests/limits; combining them raises ValueError. Supplying requests without limits logs a warning and clamps requests to the original limits. A timeout may be an integer number of seconds or a datetime.timedelta; None resets it to an empty duration.

Cache overrides have an additional validation rule: a Cache object must contain a cache version. Deprecated cache parameters cannot be combined with a Cache object. cache=True without a version creates a default Cache policy using the supplied legacy serialization and ignored-input options when present. A task_config override is marked beta in the source and must have the same type as the underlying entity's task configuration.

Build the graph imperatively​

Use ImperativeWorkflow when you need to register inputs, entities, and outputs programmatically rather than by evaluating a decorated function. The source's equivalent examples are:

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 corresponding function 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_input() creates a Promise whose NodeOutput points at GLOBAL_START_NODE and records that promise as unbound. add_entity() calls create_node() under the imperative workflow's compilation state and removes any consumed input promises from the unbound set. add_workflow_output() creates a typed binding and extends the workflow interface. It can infer the type from a single promise; for a list or dictionary of promises, pass python_type explicitly.

Before local execution, ready() requires at least one node and no unbound workflow inputs. Imperative local execution walks compilation_state.nodes in insertion order, resolves each node's bindings with get_promise_map(), invokes the entity, caches its outputs, and resolves the workflow output bindings. Consequently, declare imperative entities in topological order: adding >> records an explicit dependency but does not reorder that insertion-order executor.

Workflow-level metadata and failure handling​

Configure workflow defaults at the decorator boundary when the behavior belongs to the workflow rather than one node:

@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

WorkflowFailurePolicy accepts FAIL_IMMEDIATELY or FAIL_AFTER_EXECUTABLE_NODES_COMPLETE; the default is FAIL_IMMEDIATELY. interruptible defaults to False and is validated as a strict boolean. The decorator creates WorkflowMetadata and WorkflowMetadataDefaults, constructs PythonFunctionWorkflow, and uses update_wrapper so the resulting callable retains the original function metadata. The serialized template receives the workflow metadata and defaults.

Pass on_failure to attach a task or workflow for failure handling:

@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

During compilation, _validate_add_on_failure_handler() compiles the handler separately. The handler must accept all workflow inputs; any extra handler inputs must be optional, and compilation must produce exactly one task or workflow node. The failure node is stored separately from the main node list, so cleanup is not part of the ordinary workflow node sequence. In the source's implementation and examples, the additional failure value uses the input name err.

Local execution and integrated node producers​

WorkflowBase.local_execute() translates native inputs through the workflow interface, executes the graph, validates the output shape, and repackages results as literal-backed promises. For imperative workflows, the execution loop described above calls each underlying entity with resolved native values. For a PythonFunctionWorkflow, execute() invokes the original function, which makes local execution different from platform compilation even though both use the workflow interface.

Conditionals, launch plans, and sub-workflows participate in composition through the same node model. A conditional materializes a regular Node carrying the branch entity and its unresolved bindings. A launch plan's __call__() merges saved inputs with current keyword arguments and uses create_and_link_node() during compilation. Calling a sub-workflow from a decorated workflow therefore creates a graph node and returns promises just like calling a task.

Eager-task integration uses the imperative API internally. PythonFunctionTask.get_as_workflow() creates an ImperativeWorkflow, adds each task input with add_workflow_input(), calls add_entity(self, **input_kwargs), binds every node.outputs[output_name] with add_workflow_output(), and adds an EagerFailureHandlerTask cleanup handler. This is the same explicit input → node → output structure exposed for user-built imperative workflows.