Skip to main content

Workflow composition and nodes

The @workflow decorator in flytekit is the primary way to define a Flyte workflow. It takes a plain Python function and produces a PythonFunctionWorkflow object that can be run locally, compiled to a DAG, and registered with a Flyte cluster. Understanding how compilation works — and how the resulting Node objects connect — is key to using flytekit effectively.

The @workflow Decorator and Compilation​

Here's the simplest possible workflow, taken from the flytekit README:

from flytekit import task, workflow

@task(cache=True, cache_version="1", retries=3)
def sum(x: int, y: int) -> int:
    return x + y

@task(cache=True, cache_version="1", retries=3)
def square(z: int) -> int:
    return z * z

@workflow
def my_workflow(x: int, y: int) -> int:
    return sum(x=square(z=x), y=square(z=y))

When Python imports this module, @workflow creates a PythonFunctionWorkflow instance and stores the original function as self._workflow_function. The function body is not run at import time. It's run later, during compilation, to discover the DAG structure.

Compilation happens the first time you access .nodes, .output_bindings, or call the workflow. Inside PythonFunctionWorkflow.compile() (in flytekit/core/workflow.py), flytekit:

  1. Calls construct_input_promises() to seed each declared input as a Promise pointing at GLOBAL_START_NODE.
  2. Runs self._workflow_function(**input_kwargs) inside a CompilationState context — a context-local accumulator for nodes.
  3. Each task or workflow call in the body invokes create_and_link_node() (from flytekit/core/promise.py), which creates a Node and registers it with the CompilationState.
  4. After the function returns, compile() calls binding_from_python_std() on the return value to produce output_bindings.
  5. Sets self.compiled = True so subsequent calls skip recompilation.

The critical implication: the workflow function body is run once to build the DAG, not to compute anything. Task calls during compilation return Promise objects, not real values. Code like if t1_output > 5: raises an error because t1_output is a Promise, not an integer. Similarly, range(t1_output) fails for the same reason.

@workflow
def bad_wf(n: int) -> int:
    result = my_task(a=n)
    # This would fail at compile time — result is a Promise, not an int!
    for i in range(result):
        ...

Nodes and Promises: The Two Core Abstractions​

Every entity call inside a workflow body produces two things: a Node and one or more Promises.

Nodes​

A Node (defined in flytekit/core/node.py) represents one vertex in the DAG. It holds:

  • id — auto-generated string like "n0", "n1" (DNS-ified); overridable via with_overrides(node_name=...)
  • metadata — a NodeMetadata carrying the display name, timeout, retries, and cache settings
  • bindings — a List[Binding] describing each input to this node, where that input comes from
  • upstream_nodes — a List[Node] for explicit dependency ordering (separate from data flow)
  • flyte_entity — the underlying PythonTask, WorkflowBase, or LaunchPlan

Promises​

When a task is called during compilation, create_and_link_node() returns a Promise (or a named tuple of Promise objects). Each Promise wraps a NodeOutput(node=<Node>, var="output_name"), pointing to a specific output variable on a specific upstream node.

When you pass that Promise as input to a downstream task, binding_from_python_std() inspects it, finds the upstream NodeOutput, and stores a Binding(var="input_name", binding=BindingData(promise=NodeOutput(...))) on the downstream Node. This is how data flow is encoded in the DAG: not by copying values, but by recording references.

Workflow inputs: GLOBAL_START_NODE​

Workflow inputs are handled by construct_input_promises(), which creates a Promise for every declared input:

def construct_input_promises(inputs: List[str]) -> Dict[str, Promise]:
    return {
        input_name: Promise(var=input_name, val=NodeOutput(node=GLOBAL_START_NODE, var=input_name))
        for input_name in inputs
    }

GLOBAL_START_NODE is a module-level singleton Node with id="-" that serves as the notional source of all workflow inputs. During serialization, get_serializable_workflow() filters out nodes with id == GLOBAL_INPUT_NODE_ID so this sentinel never appears in the compiled template.

Connecting Sub-workflows​

Calling another @workflow inside your workflow body is straightforward — the called workflow becomes a sub-workflow node:

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

@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)      # x is a Promise from the first call
    d = simple_wf()     # simple_wf becomes a WorkflowNode in the compiled DAG
    e = conditional("bool").if_(a == 5).then(add_5(a=d)).else_().then(add_5(a=z))
    return x, e

Named tuple outputs from tasks or sub-workflows can be unpacked directly or accessed by attribute:

@workflow
def wf(b: int) -> nt:
    out = subwf(a=b)        # out is a named tuple of Promises
    return t1(a=out.named1) # access individual outputs by name

Workflow Decorator Parameters​

@workflow accepts several keyword arguments that customize behavior. The two most used are failure_policy and interruptible:

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

interruptible=True propagates to WorkflowMetadataDefaults and marks all tasks in the workflow as runnable on preemptible (spot) instances by default. failure_policy controls what WorkflowMetadata.on_failure is set to:

  • WorkflowFailurePolicy.FAIL_IMMEDIATELY (default) — the workflow fails as soon as any node fails.
  • WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE — remaining runnable nodes continue executing; the workflow is marked failed only after all are done.

Failure Handlers​

Pass on_failure= to register a task or workflow that runs when the workflow fails. The failure handler receives the same inputs as the workflow (plus an optional FlyteError parameter for extra error context):

from flytekit.types.error import FlyteError

@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

The failure handler's inputs must be a superset of the workflow's inputs. Any extra parameters the handler declares beyond the workflow's inputs must be Optional; otherwise flytekit raises FlyteFailureNodeInputMismatchException. During compilation, PythonFunctionWorkflow._validate_add_on_failure_handler() enforces this check and builds a separate _failure_node that is excluded from the main nodes list.

Explicit Node Creation with create_node()​

Calling a task inside a workflow body implicitly creates a node and returns a Promise. That's enough when tasks produce outputs that feed downstream tasks. But two situations require the explicit create_node() function from flytekit/core/node_creation.py:

1. Tasks with no outputs. When a task returns nothing, calling it produces a VoidPromise. You can't assign ordering through data flow, so you need a Node reference for the >> operator:

from flytekit.core.node_creation import create_node

@workflow
def empty_wf():
    t2_node = create_node(t2)
    t3_node = create_node(t3)
    t3_node >> t2_node   # t3 must complete before t2 starts

2. Accessing outputs by string key. create_node() sets node._outputs to a dict, enabling both attribute access (node.output_name) and dict access (node.outputs["output_name"]). Without create_node(), the .outputs property raises AssertionError.

@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  # positional output names

Dependency Ordering with >>​

The >> operator (defined as Node.__rshift__) calls runs_before(), which appends self to other._upstream_nodes. This adds upstream_node_ids to the serialized node without creating any data binding. You can chain it:

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

Because task calls inside a workflow return Promise objects, >> is also defined on Promise — flytekit delegates to the underlying node automatically, so you don't always need create_node() just to express ordering.

Per-Node Overrides with with_overrides()​

Node.with_overrides() lets you customize node-level settings after the node is created, without changing the task definition. Call it on the return value of a task call (which is a Promise that delegates to its node) or on a Node from create_node():

@workflow
def my_wf(a: str) -> str:
    s = t1(a=a).with_overrides(timeout=datetime.timedelta(seconds=100))
    s2 = t2(a=s).with_overrides(retries=3)
    return s2

Override resource requests and limits independently:

@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

The full set of overridable properties includes: node_name (display ID), aliases (output variable aliasing), requests, limits, resources (combined shorthand), timeout (int seconds or timedelta), retries, interruptible, name (display name), task_config, container_image, accelerator, cache (bool or Cache object), shared_memory, and pod_template.

A few constraints to keep in mind:

  • resources= cannot be combined with requests= or limits= in the same call.
  • When overriding cache=Cache(...), the Cache object must have version set — cache.version cannot be None.
  • timeout=None explicitly clears a previously set timeout (sets it to timedelta()), while omitting timeout entirely leaves the existing value unchanged. This distinction is implemented via the Node.TIMEOUT_OVERRIDE_SENTINEL sentinel object.

Programmatic Workflow Construction with ImperativeWorkflow​

@workflow requires you to express the DAG as a Python function body. ImperativeWorkflow (also aliased as Workflow when imported from flytekit.core.workflow) lets you build the same DAG programmatically — useful for generated pipelines, backfill workflows, or dynamic code-generation scenarios.

from flytekit.core.workflow import ImperativeWorkflow as Workflow

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 equivalent 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 from the moment of construction. Each call to add_entity() (or its convenience aliases add_task(), add_subwf(), add_launch_plan()) runs create_node() in the context of that state, building up the node list incrementally. Sub-workflows and launch plans work the same way:

wb2 = ImperativeWorkflow(name="parent.imperative")
p_in1 = wb2.add_workflow_input("p_in1", str)
p_node0 = wb2.add_subwf(wb, in1=p_in1)          # wb becomes a WorkflowNode
wb2.add_workflow_output("parent_wf_output", p_node0.from_n0t1, str)

add_workflow_input() returns a Promise pointing at GLOBAL_START_NODE and tracks it as unbound until it's consumed by add_entity(). The ready() method returns True only when the node list is non-empty and no declared inputs remain unbound.

A real-world example: flytekit's backfill generator (flytekit/remote/backfill.py) uses ImperativeWorkflow to build a sequence of launch plan nodes, using runs_before() to enforce serial execution:

wf = ImperativeWorkflow(name=f"backfill-{for_lp.name}", failure_policy=failure_policy)
prev_node = None
while True:
    next_start_date = date_iter.get_next()
    ...
    next_node = wf.add_launch_plan(for_lp, **inputs)
    next_node = next_node.with_overrides(
        name=f"b-{next_start_date}", retries=per_node_retries, timeout=per_node_timeout
    )
    if not parallel and prev_node:
        prev_node.runs_before(next_node)
    prev_node = next_node

From DAG to Serialized Template​

When get_serializable() (from flytekit/tools/translator.py) is called on a workflow, it accesses workflow.nodes and workflow.output_bindings — triggering compile() on a PythonFunctionWorkflow if it hasn't been called yet. Then get_serializable_node() converts each Node to a workflow_model.Node, carrying:

  • A task_node, workflow_node, or branch_node discriminator (determined by the type of node.flyte_entity)
  • The bindings list (input connections)
  • upstream_node_ids (from node.upstream_nodes)
  • Node metadata (name, timeout, retries, cache settings)

The resulting WorkflowTemplate is what gets registered with the Flyte platform. The workflow function itself is not run again on the cluster — only each task's execute() method runs remotely.