Skip to main content

Workflow composition and nodes

A Flyte workflow is a directed acyclic graph (DAG) where each node is a task, sub-workflow, or launch plan invocation, and edges are either data-flow connections (one node's output becomes another's input) or explicit ordering constraints. The @workflow decorator and the Node class are the two pieces that make this machinery work.

The @workflow Decorator​

The simplest workflow looks like a regular Python function:

from flytekit import task, workflow

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

@workflow
def simple_wf() -> int:
    return add_5(a=1)

Flytekit transforms this decorated function into a PythonFunctionWorkflow instance. The function body runs at compile time (during serialization or on first access to .nodes / .output_bindings), not when the workflow executes on the platform. During that single compile-time execution, every task or sub-workflow call returns a Promise rather than a real Python value, and those Promises wire up the DAG edges automatically.

Here is a more complete example that demonstrates the range of things you can do inside a workflow body:

import typing
from flytekit import task, workflow, conditional

@workflow
def my_wf_example(a: int) -> typing.Tuple[int, int]:
    """example

    Workflows can have inputs and return outputs of simple or complex types.

    :param a: input a
    :return: outputs
    """
    x = add_5(a=a)

    # Outputs of a prior task become inputs to the next.
    z = add_5(a=x)

    # You can call other workflows from within this workflow.
    d = simple_wf()

    # Conditionals branch on primitive Promise values.
    e = conditional("bool").if_(a == 5).then(add_5(a=d)).else_().then(add_5(a=z))

    return x, e

Compile-Time vs. Runtime​

Because the function body runs at compile time, every object flowing between task calls is a Promise, not a real value. Attempting to call range(), len(), if promise:, or any other Python operation that needs to inspect the actual value will fail at compile time. The one safe operation on a Promise is passing it as an argument to another task or sub-workflow call.

PythonFunctionWorkflow.compile() is called lazily — the first time .nodes or .output_bindings is accessed, or when __call__ is invoked outside eager mode. A compiled flag on the instance prevents double-compilation. For local testing, __call__ dispatches to execute(**kwargs), which simply calls the original Python function with real values.

Decorator Parameters​

The @workflow decorator accepts several parameters that control execution behavior:

from flytekit 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 (default False): propagates to WorkflowMetadataDefaults, which Flyte hands down to constituent tasks as their default interruptible setting.
  • failure_policy: a WorkflowFailurePolicy enum value. FAIL_IMMEDIATELY (the default) stops execution as soon as any node fails; FAIL_AFTER_EXECUTABLE_NODES_COMPLETE lets any remaining independent nodes finish first.
  • on_failure: a task or workflow to invoke when the workflow fails (covered in the Failure Handling section).
  • docs: an optional Documentation object for workflow-level docs surfaced in the Flyte UI.
  • pickle_untyped (default False): bypasses type checking during workflow construction. Not recommended for general use.
  • default_options: an optional Options object for the workflow's default launch plan (currently only labels and annotations are supported).

How Nodes Are Created and Connected​

Calling a task inside @workflow does two things: it creates a Node object (auto-assigned the next available ID: n0, n1, n2, …) and it returns a Promise (or a tuple of Promises for multi-output tasks). Passing that Promise as an argument to the next task creates a data-flow edge — the binding on the downstream node points back to the upstream node's output.

Multi-Output Tasks and NamedTuples​

When a task returns a NamedTuple, destructuring works as expected at compile time:

@task
def t1(a: int) -> typing.NamedTuple("OutputsBC", [("t1_int_output", int), ("c", str)]):
    return a + 2, "world-" + str(a + 2)

@workflow
def wf(a: int) -> typing.Tuple[str, str]:
    x, y = t1(a=a)   # x and y are Promises
    _, v = t1(a=x)   # x flows as input to the second call
    return y, v

Sub-Workflow Calls​

Call another @workflow the same way you call a task. Access named outputs by attribute:

nt = typing.NamedTuple("SingleNamedOutput", [("named1", int)])

@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)   # named1 is a Promise from subwf's output

The out.named1 attribute access is safe here because out is a special Promise-like object that knows the names of the sub-workflow's declared outputs.

Explicit Node Creation and Ordering​

When you call a task as result = my_task(a=x), flytekit creates a Node internally but only gives you back the Promise. If you need the Node object itself — to access its full .outputs dictionary or to declare a pure ordering constraint — use create_node() from flytekit.core.node_creation:

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

create_node() accepts only keyword arguments — positional args raise FlyteAssertion. Under compilation, it calls the entity to register the node and then attaches each output as a named attribute on the returned Node (.o0, .o1, or the declared NamedTuple field names). Accessing .outputs on a node that was not created with create_node() raises AssertionError.

Data-Independent Ordering with >> and runs_before()​

Sometimes two nodes must run in a specific order but don't share data. The Node class supports this with runs_before(other) and the >> operator:

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

@workflow
def empty_wf2():
    t2_node = create_node(t2)
    t3_node = create_node(t3)
    t3_node >> t2_node            # same as above

The >> operator also works directly on Promise return values, avoiding the need for create_node() when you just want ordering:

@workflow
def wf(x: int) -> str:
    a = task_a(x=x)
    b = task_b(x=x)
    c = task_c(x=x)
    a >> b          # task_b waits for task_a
    c >> a          # task_a waits for task_c
    return b

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

Internally, runs_before appends self to other._upstream_nodes. During serialization, the translator reads upstream_nodes to emit upstream_node_ids in the workflow spec.

Node Overrides with with_overrides()​

Every task call inside @workflow returns something you can chain .with_overrides() on. The method mutates the underlying Node and returns self, so it can be chained or the result stored:

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

After serialization, node.metadata.name is "foo" and node.id is "t-1" — node_name is passed through _dnsify(), which converts underscores to hyphens and strips other non-DNS characters.

Resource Overrides​

Override CPU, memory, or storage requests per node:

from flytekit import Resources

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

resources= is a shorthand that accepts a single Resources object covering both requests and limits. Using resources= together with requests= or limits= raises ValueError.

Timeout and Retry Overrides​

import datetime

@workflow
def my_wf(a: str) -> str:
    s = t1(a=a).with_overrides(timeout=datetime.timedelta(minutes=30))
    s2 = t1(a=s).with_overrides(timeout=60)          # integer = seconds
    s3 = t1(a=s2).with_overrides(timeout=None)        # clear the timeout
    return t1(a=s3).with_overrides(retries=3)

Caching Overrides​

Enabling caching on a node via with_overrides() requires an explicit version string. Two equivalent forms are supported:

from flytekit.core.cache import Cache

@workflow
def my_wf(a: str) -> str:
    # Legacy form
    return t1(a=a).with_overrides(cache=True, cache_version="foo", cache_serialize=True)

@workflow
def my_wf_cache_policy(a: str) -> str:
    # Cache object form (preferred)
    return t1(a=a).with_overrides(cache=Cache(version="foo", serialize=True))

Omitting the version when cache=True raises ValueError.

Infrastructure Overrides​

with_overrides() also handles container-level settings:

@workflow
def my_wf(a: str) -> str:
    return t1(a=a).with_overrides(
        container_image="my-registry/my-image:v2",
        interruptible=True,
    )

accelerator=, shared_memory=, and pod_template= are also accepted for GPU-accelerated and custom-pod workloads.

ImperativeWorkflow for Programmatic DAG Construction​

When you need to build a workflow dynamically — for example, looping over a list of dates to create backfill nodes — the @workflow decorator is not the right tool. ImperativeWorkflow (exported as Workflow from flytekit) provides a builder API instead:

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

# Create the workflow with a name; this takes the place of the function name.
wb = Workflow(name="my_workflow")

# Declare workflow-level inputs.
wb.add_workflow_input("in1", str)

# Add entities as nodes, binding inputs to workflow inputs or prior node outputs.
node = wb.add_entity(t1, a=wb.inputs["in1"])
wb.add_entity(t2)

# Declare workflow outputs, pointing at node outputs.
wb.add_workflow_output("from_n0t1", node.outputs["o0"])

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)

add_entity() has convenience aliases: add_task(), add_subwf(), and add_launch_plan(). Each returns the Node created for that entity, allowing you to access node.outputs[...] or call node.with_overrides(...) directly.

Building Workflows in a Loop​

The real power of ImperativeWorkflow is dynamic DAG construction. The flytekit/remote/backfill.py module uses it to generate backfill workflows from a scheduled launch plan:

wf = ImperativeWorkflow(name=f"backfill-{for_lp.name}", failure_policy=failure_policy)

prev_node = None
while True:
    next_start_date = date_iter.get_next()
    if next_start_date >= end_date:
        break
    inputs = {input_name: next_start_date}
    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:
        if prev_node:
            prev_node.runs_before(next_node)  # sequential execution
    prev_node = next_node

wf.ready() returns True when the workflow has at least one node and all declared inputs are bound.

Failure Handling​

WorkflowFailurePolicy​

WorkflowFailurePolicy is an enum with two values:

  • FAIL_IMMEDIATELY (default): the workflow fails as soon as any single node fails.
  • FAIL_AFTER_EXECUTABLE_NODES_COMPLETE: remaining nodes that are not blocked by the failed node continue to run before the workflow is marked as failed.
@workflow(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

on_failure Handlers​

Pass a task or sub-workflow to on_failure= to run cleanup logic when the workflow fails. The handler must accept all of the workflow's inputs. Any additional parameters must be Optional; the special err: typing.Optional[FlyteError] parameter receives information about the failed node:

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

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

@task
def t1(a: int, b: str): ...

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

@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, PythonFunctionWorkflow validates that the failure handler's inputs are a superset of the workflow's inputs and that any extra parameters are Optional; a mismatch raises FlyteFailureNodeInputMismatchException. The failure node is assigned the special ID DEFAULT_FAILURE_NODE_ID and is excluded from the main node list.

For ImperativeWorkflow, call add_on_failure_handler(entity) instead:

wb = Workflow(name="cleanup_demo")
wb.add_workflow_input("name", str)
...
wb.add_on_failure_handler(clean_up)

Reference Workflows​

When you need to call a workflow that is already registered on the Flyte platform — perhaps in a different project — use @reference_workflow. The decorator wraps a function whose type annotations define the interface; the body is ignored:

from flytekit import reference_workflow

@reference_workflow(project="proj", domain="development", name="wf_name", version="abc")
def ref_wf1(a: int) -> typing.Tuple[str, str]:
    ...

ref_wf1 can then be called from another @workflow body just like a locally defined sub-workflow, with the same Promise-passing semantics. The resulting ReferenceWorkflow object uses only the declared interface for type checking; no local implementation is needed.

Common Pitfalls​

Workflow body is compile-time code. The function body of @workflow runs once during serialization. Promises are not real Python values — if promise_value: raises ValueError (Promise overrides __bool__), and range(promise_value) will fail. Use conditional() for branching on runtime values.

create_node() requires a compilation context. Calling it outside a @workflow or @dynamic body raises RuntimeError. Inside @workflow, all arguments must be keyword arguments; positional args raise FlyteAssertion.

node.outputs only works after create_node(). If you get a node implicitly from a task call (result = my_task(a=x)), accessing result.outputs raises AssertionError. Use node = create_node(my_task, a=x) when you need .outputs.

Cache version is mandatory in with_overrides. Calling .with_overrides(cache=True) without cache_version= or .with_overrides(cache=Cache(...)) without version= raises ValueError.

resources= and requests=/limits= are mutually exclusive. Passing both in the same with_overrides() call raises ValueError.

node_name is DNS-ified. with_overrides(node_name="my_task") stores "my-task" as the node ID — underscores become hyphens. The serialized workflow spec will reflect the transformed name.