Skip to main content

Workflow composition and nodes

The Conceptual Model: A DAG Built in Python​

A Flyte workflow is a Python function that, when decorated with @workflow, becomes an executable DAG definition. The crucial thing to understand is when that function runs: the body of a @workflow function executes once at serialization/compile time, not when remote tasks run. During that single pass, every task call returns a Promise object representing a future value — not the actual computed result. Flytekit records how those Promises flow between tasks to build a graph of Node objects, which is then serialized into a WorkflowSpec protobuf for registration with Flyte.

This means the workflow function is a graph description language embedded in Python. Python control flow (for, if) that runs at definition time is fine; Python operations on task outputs (like range(task_result)) will fail because those outputs are Promise objects at compile time.

The @workflow Decorator​

The @workflow decorator in flytekit.core.workflow is an overloaded function — you can use it bare (@workflow) or with arguments (@workflow(interruptible=True)). Either way, it wraps your function into a PythonFunctionWorkflow instance.

from flytekit import task, workflow
import typing

@task
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

The decorator signature is:

def workflow(
    _workflow_function: Optional[Callable] = None,
    failure_policy: Optional[WorkflowFailurePolicy] = None,
    interruptible: bool = False,
    on_failure: Optional[Union[WorkflowBase, Task]] = None,
    docs: Optional[Documentation] = None,
    pickle_untyped: bool = False,
    default_options: Optional[Options] = None,
)

Internally, workflow creates two metadata containers: a WorkflowMetadata holding the failure_policy, and a WorkflowMetadataDefaults holding interruptible. Both are thin dataclasses that serialize into the Flyte IDL model. Then it constructs a PythonFunctionWorkflow, passing both containers along with the original callable and docstring.

PythonFunctionWorkflow inherits from WorkflowBase and ClassStorageTaskResolver. At construction time it calls transform_function_to_interface to derive the typed Interface from the function's Python annotations. The compiled flag starts as False — compilation is deferred until the workflow is first called or serialized.

Compile-Time Execution and Promises​

When the workflow is invoked (locally or during serialization), PythonFunctionWorkflow.compile() is called. This is what actually runs the function body:

# from flytekit/core/workflow.py
def compile(self, **kwargs):
    if self.compiled:
        return
    self.compiled = True
    # ...
    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)

construct_input_promises creates a Promise for each declared workflow input, each backed by GLOBAL_START_NODE — a sentinel Node with id '' (the GLOBAL_INPUT_NODE_ID constant). These Promise objects become the "values" that flow into the first task calls.

Every time a task or sub-workflow is called inside the workflow body, flyte_entity_call_handler is triggered. In compilation mode it calls create_and_link_node, which creates a new Node, attaches the input Binding objects (linking upstream Promise outputs to downstream task inputs), registers the node with the active CompilationState, and returns the task's output as one or more new Promise objects. Those new Promises can then be fed to subsequent tasks.

The docstring of workflow makes this explicit:

Unlike a task, the function body of a workflow is evaluated at serialization-time (aka compile-time)... it is not evaluated again when the workflow runs on Flyte.

Practically, this means code like the following will raise at compile time:

@workflow
def broken_wf(a: int):
    result = my_task(a=a)
    for i in range(result):  # ERROR: result is a Promise, not an int
        ...

Nodes — The Unit of Execution​

Node (in flytekit.core.node) is the object that represents a single callable within the workflow DAG. Each node carries:

  • id — Auto-assigned as n0, n1, etc. (or with a prefix for sub-workflows). IDs are normalized through _dnsify(), so uppercase letters become lowercase with a dash prefix, underscores become dashes, and non-alphanumeric characters are stripped.
  • metadata — A NodeMetadata model holding name, timeout, retries, interruptible, and cache settings.
  • bindings — A list of Binding objects describing which upstream Promise outputs feed into this node's inputs.
  • upstream_nodes — A list of Node objects that must complete before this node runs, independent of data flow.
  • flyte_entity — The actual task, sub-workflow, or launch plan wrapped by this node.

Nodes are created implicitly when you call a task inside a workflow. You rarely construct them directly.

Connecting Inputs and Outputs​

Data flows between tasks through Promise objects. When t1(a=a) is called in a workflow, it returns a Promise (or a named tuple of Promises for multi-output tasks). Passing that Promise to the next task's argument creates a Binding on the downstream node connecting it to t1's output.

Multi-output tasks use Python tuple unpacking:

@workflow
def my_wf(a: int) -> typing.Tuple[str, str]:
    x, y = t1(a=a)   # x and y are Promise objects
    _, v = t1(a=x)   # pipe t1's first output into the next call
    return y, v

The workflow's return values are also captured as Binding objects — compile() maps each returned Promise to the corresponding output name to build _output_bindings.

When you need a Node object directly (rather than just the Promise), use create_node() from flytekit.core.node_creation. This is the explicit node creation API, and it sets up the outputs property on the returned node so you can access output values by string key:

from flytekit.core.node_creation import create_node

@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

Note that outputs is only available on nodes created via create_node(). Calling .outputs on a node that wasn't created that way raises AssertionError: Cannot use outputs with all Nodes.

Explicit Dependency Ordering with >> and runs_before​

When two tasks need to run in order but don't share data, you can declare an ordering constraint directly. Both node_a >> node_b (via Node.__rshift__) and node_a.runs_before(node_b) do exactly the same thing — they append node_a to node_b._upstream_nodes:

@workflow
def empty_wf():
    t2_node = create_node(t2)
    t3_node = create_node(t3)
    t3_node.runs_before(t2_node)  # t3 runs before t2

@workflow
def empty_wf2():
    t2_node = create_node(t2)
    t3_node = create_node(t3)
    t3_node >> t2_node            # identical effect

The >> operator can also be applied directly to task call results (which return a VoidPromise for void tasks), and it chains:

@workflow
def wf1(x: int):
    task_a(x=x) >> task_b(x=x) >> task_c(x=x)

In the serialized WorkflowSpec, these ordering dependencies appear as upstream_node_ids on the node protobuf, distinct from input bindings.

with_overrides() — Per-Node Configuration​

After calling a task in a workflow, you can customize the resulting node's execution settings by chaining .with_overrides() on the return value. This mutates the node in place and returns it, so chaining is possible.

Resources:

@workflow
def my_wf(a: typing.List[str]) -> typing.List[str]:
    mappy = map_task(t1)
    map_node = mappy(a=a).with_overrides(
        requests=Resources(cpu="1", mem="100", ephemeral_storage="500Mi")
    )
    return map_node

If you specify requests without limits, flytekit logs a warning: "Requests overridden on node ... without specifying limits. Requests are clamped to original limits."

You can also use the combined resources parameter instead of separate requests/limits, but the two styles cannot be mixed in the same call.

Timeout and retries:

@workflow
def my_wf(a: str) -> str:
    s = t1(a=a).with_overrides(timeout=datetime.timedelta(minutes=30), retries=3)
    return s

timeout accepts a datetime.timedelta, an integer number of seconds, or None (to clear the timeout). Any other type raises ValueError.

Naming:

@workflow
def my_wf(a: str) -> str:
    return t1(a=a).with_overrides(name="foo", node_name="t_1")

name sets the display name in the node metadata; node_name sets the node's actual ID (which is passed through _dnsify(), so "t_1" becomes "t-1" in the serialized output).

Interruptible and cache:

t1(a=a).with_overrides(interruptible=True)

t1(a=a).with_overrides(
    cache=Cache(version="1.0", serialize=True),
)

When overriding cache, a version is always required — passing cache=True without a version raises ValueError: must specify cache version when overriding.

Accelerators and pod templates:

from flytekit.extras.accelerators import A100
t1(a=a).with_overrides(accelerator=A100, container_image="my-registry/my-image:latest")

task_config override is marked beta — flytekit logs a warning when you use it, and the new config must be the same type as the original.

The full signature of with_overrides:

def with_overrides(
    self,
    node_name: Optional[str] = None,
    aliases: Optional[Dict[str, str]] = None,
    requests: Optional[Resources] = None,
    limits: Optional[Resources] = None,
    timeout: Optional[Union[int, datetime.timedelta]] = ...,
    retries: Optional[int] = None,
    interruptible: Optional[bool] = None,
    name: Optional[str] = None,
    task_config: Optional[Any] = None,
    container_image: Optional[str] = None,
    accelerator: Optional[BaseAccelerator] = None,
    cache: Optional[Union[bool, Cache]] = None,
    shared_memory: Optional[Union[Literal[True], str]] = None,
    pod_template: Optional[PodTemplate] = None,
    resources: Optional[Resources] = None,
)

create_node() — Explicit Node Creation​

Direct task calls inside a workflow return Promises, not Nodes. When you need the Node object — to access outputs by name, to apply runs_before, or to handle a void task that returns nothing useful — use create_node() from flytekit.core.node_creation:

from flytekit.core.node_creation import create_node

@workflow
def my_wf(a: int, b: str) -> (str, typing.List[str], int):
    subwf_node = create_node(my_subwf, a=a)
    t2_node = create_node(t2, a=b, b=b)
    subwf_node.runs_before(t2_node)
    return t2_node.o0, subwf_node.o0, subwf_node.o1

create_node() behaves identically to a direct task call during compilation — it calls the entity and collects the resulting Node into CompilationState. The difference is that it returns the Node directly rather than a tuple of Promises, and it sets up the outputs mapping on that node so node.outputs['o0'] works.

In local execution mode, create_node() runs the entity immediately and returns the result, bypassing the Promise machinery.

All Flyte entity calls (both direct and via create_node) accept only keyword arguments. Positional arguments raise a FlyteAssertion.

Sub-Workflows​

Calling one @workflow function inside another is fully supported. The callee becomes a workflow node in the parent's DAG, with its outputs accessible as named-tuple attributes:

@workflow
def subwf(a: int) -> nt:
    return t1(a=a)

@workflow
def wf(b: int) -> nt:
    out = subwf(a=b)
    return t1(a=out.named1)

The serializer (get_serializable_workflow in flytekit.tools.translator) recognizes sub-workflow nodes and recursively extracts them into sub_workflows on the WorkflowSpec. One important constraint: reference sub-workflows are not supported. If you need to call a remotely-registered workflow by reference, use a reference_launch_plan instead — the serializer raises ValueError with the message "Reference sub-workflows are currently unsupported. Use reference launch plans instead." if you attempt otherwise.

Imperative (Programmatic) Workflow Construction​

When the decorator pattern doesn't fit — for example, when building workflows dynamically from configuration — ImperativeWorkflow (exported from flytekit as Workflow) provides an equivalent API:

from flytekit.core.workflow import ImperativeWorkflow as Workflow
from flytekit import task

@task
def t1(a: str) -> str:
    return a + " world"

@task
def t2():
    print("side effect")

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 is semantically identical to:

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

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

ImperativeWorkflow maintains its own CompilationState. Workflow inputs are exposed via .inputs['name'], returning Promise objects you can pass to add_entity. Outputs are declared explicitly with add_workflow_output.

The API provides dedicated methods for different entity types:

  • add_entity(entity, **kwargs) — tasks and workflows
  • add_task(task, **kwargs) — alias for add_entity
  • add_subwf(subworkflow, **kwargs) — explicit sub-workflow addition
  • add_launch_plan(lp, **kwargs) — launch plan nodes

The ready() method checks that at least one node exists and all declared inputs are bound. If an input is declared with add_workflow_input but never used in an add_entity call, ready() returns False and local execution raises FlyteValidationException.

Mixed collection inputs work too. Promise objects can be placed inside lists or dicts, and flytekit's binding machinery resolves them individually:

wb = ImperativeWorkflow(name="my.workflow.a")
wb.add_workflow_input("in1", int)
wb.add_workflow_input("in2", int)
node = wb.add_entity(t1, a=[wb.inputs["in1"], wb.inputs["in2"]])
wb.add_workflow_output("from_n0t1", node.outputs["o0"])

An imperative workflow can itself be used as a sub-workflow inside another:

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)

Failure Handling​

Pass on_failure to @workflow to run a cleanup task or workflow whenever any node in the DAG raises an exception:

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

@task
def create_cluster(name: str): ...

@task
def delete_cluster(name: str): ...

@task
def t1(a: int, b: str):
    raise ValueError("Fail!")

@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

When t1 raises, clean_up is invoked with the workflow's original inputs (name="flyteorg") plus a FlyteError object describing which node failed and the error message.

The failure handler's signature must satisfy a constraint checked in WorkflowBase._validate_add_on_failure_handler: all workflow inputs must be a subset of the failure handler's inputs, and any extra parameters the handler accepts beyond the workflow inputs must be Optional. Violating this raises FlyteFailureNodeInputMismatchException.

Internally, the failure handler is compiled into a special Node with id DEFAULT_FAILURE_NODE_ID ("efn"). This ID is reserved — assigning it to a regular node via with_overrides(node_name="efn") will cause serialization errors.

ImperativeWorkflow also supports failure handlers via the same on_failure constructor argument.

WorkflowFailurePolicy​

By default, a workflow halts immediately when any node fails. Override this with WorkflowFailurePolicy:

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

WorkflowFailurePolicy is an Enum with two values:

  • FAIL_IMMEDIATELY (default) — the workflow enters a failed state as soon as any node fails.
  • FAIL_AFTER_EXECUTABLE_NODES_COMPLETE — runnable nodes that are not blocked by the failed node are allowed to finish before the workflow is marked failed.

The enum values map to WorkflowMetadata.OnFailurePolicy in the Flyte IDL. After serialization, FAIL_IMMEDIATELY serializes to 0 and FAIL_AFTER_EXECUTABLE_NODES_COMPLETE to 1 in the template.metadata.on_failure field.

The interruptible=True argument flows into WorkflowMetadataDefaults and is inherited by all tasks in the workflow unless overridden at the node level with with_overrides(interruptible=False).

From Compiled DAG to WorkflowSpec​

After compile() completes, the PythonFunctionWorkflow holds:

  • _nodes — the ordered list of Node objects
  • _output_bindings — Binding objects linking node outputs to the workflow's declared outputs

When you call get_serializable(OrderedDict(), serialization_settings, my_wf) (from flytekit.tools.translator), it calls get_serializable_workflow, which:

  1. Accesses wf.nodes (triggering compilation if not yet done)
  2. Converts each Node into a protobuf node, recursively serializing any embedded sub-workflows
  3. Builds a WorkflowSpec containing the WorkflowTemplate and a list of sub_workflows

Sub-workflows discovered during this traversal are extracted into the sub_workflows field of the outermost WorkflowSpec, enabling the Flyte platform to register them as a unit.

The serialization step is also where any PythonAutoContainerTask defined inside a workflow body gets registered onto the workflow object itself (via ClassStorageTaskResolver), so the workflow can serve as its own task resolver during remote execution.