Skip to main content

Workflow composition and nodes

Workflows are compiled DAGs​

A Flyte workflow function is not the code that Flyte executes as ordinary Python. The function body is evaluated once while flytekit compiles or serializes the workflow: task and workflow calls are turned into DAG nodes, and their return values are Promise objects that refer to node outputs. The workflow() docstring in flytekit/core/workflow.py states that the body is evaluated at serialization time and is not evaluated again when the workflow runs on Flyte (local execution is the exception).

That distinction matters when composing tasks. Use task outputs to express data flow, and use conditional() rather than a normal Python decision based on a task output. A task result such as t1(a=3) is not a concrete integer during compilation, so Python code cannot inspect it as if the task had already run.

Declarative composition with @workflow​

Use the workflow decorator from flytekit to turn a Python function into a PythonFunctionWorkflow. The function's arguments become workflow inputs, calls in its body become nodes, and the returned values become workflow outputs. Calls inside a workflow must use keyword arguments.

This is the basic composition pattern used in tests/flytekit/integration/remote/workflows/basic/basic_workflow.py:

@task(enable_deck=True)
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

Here a and b are workflow-level inputs. x and y are promises for the two outputs of t1; passing y to t2 creates a data dependency. The return x, d statement supplies the workflow's output bindings. When a workflow has ordinary output annotations, flytekit names outputs o0, o1, and so on; use a typing.NamedTuple return annotation when you need explicit output names.

A workflow can also call another workflow. In the workflow example in flytekit/core/workflow.py, simple_wf() is called like a task and its result is passed into a conditional branch:

@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 nested workflow call becomes a workflow node in the serialized graph. The conditional becomes a branch node; it is not an ordinary Python if evaluated from a task result.

How a task call becomes a node​

Node in flytekit/core/node.py is the in-memory DAG entry for a task, sub-workflow, launch plan, or other Flyte entity. Its constructor stores:

  • an ID, normalized through _dnsify();
  • NodeMetadata, including name, timeout, retries, interruptibility, and cache settings;
  • bindings, which connect the entity's inputs to literals or upstream outputs;
  • upstream_nodes, which represent explicit ordering dependencies; and
  • flyte_entity, the task or workflow represented by the node.

During compilation, the call path is effectively:

workflow body
  -> Flyte entity call handler
  -> create_and_link_node() in flytekit/core/promise.py
  -> Binding objects and upstream-node discovery
  -> Node appended to CompilationState.nodes
  -> Promise objects referring to NodeOutput values

create_and_link_node() iterates over the entity's typed inputs and calls binding_from_python_std() for each argument. When an argument is a Promise, binding conversion records the referenced NodeOutput and adds that output's node to the upstream-node set. Literals are converted into literal binding data instead. The resulting bindings and dependencies are then used to construct the node and its output promises.

Node IDs are generated in creation order (n0, n1, n2, ...), with the compilation state's prefix included for nested compilation. Node.__init__() normalizes the supplied ID with _dnsify(). If you need a readable ID, use a node override:

@workflow
def my_wf(a: str) -> str:
    s = t1(a=a).with_overrides(timeout=timeout)
    s1 = t1(a=s).with_overrides()
    s2 = t2(a=s1).with_overrides(timeout=timeout)
    s3 = t2(a=s2).with_overrides()
    return s3

This example is from tests/flytekit/unit/core/test_node_creation.py; the returned value remains usable as the next task's input because node-level overrides mutate the node associated with the promise.

Promises, bindings, and workflow inputs​

A Promise is a reference to a future value, not the value itself. A promise points to a NodeOutput, which contains the producing node and output variable name. When flytekit binds a promise as an input, binding_data_from_python_std() returns binding data containing that NodeOutput reference. This is the data-flow edge that connects nodes.

Workflow inputs use the same mechanism. flytekit/core/workflow.py creates one module-level GLOBAL_START_NODE with the empty global-input ID. construct_input_promises() creates each input promise as follows:

Promise(var=input_name, val=NodeOutput(node=GLOBAL_START_NODE, var=input_name))

Consequently, a workflow input is the output of the conceptual start node, and a first task consuming that input gets a binding from GLOBAL_START_NODE to the task input. The start node is not emitted as an ordinary serialized node: get_serializable_workflow() in flytekit/tools/translator.py skips nodes whose ID is GLOBAL_INPUT_NODE_ID.

Compilation is lazy for function workflows. PythonFunctionWorkflow.compile() establishes a CompilationState, constructs input promises, evaluates the function body, and resolves the returned values into bindings. The compiled node list and output bindings are retained, so subsequent compilation requests do not rebuild the graph. Local execution follows a different path: execute(**kwargs) calls the wrapped Python function, while imperative local execution resolves bindings through get_promise() and get_promise_map().

Explicit node handles with create_node()​

Ordinary task calls are the concise form. Use create_node() from flytekit/core/node_creation.py when you need the actual Node handle—for example, to order tasks that have no outputs or to address an output by name. It accepts Flyte tasks, launch plans, workflows, and supported remote entities, and rejects positional input arguments.

@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

In compilation mode, create_node() invokes the entity, retrieves the node most recently added to the active CompilationState, and populates node._outputs. Each output is available both as an attribute and in the outputs dictionary, for example t1_node.o0 or t1_node.outputs["o0"]. For a no-output entity, create_node() returns the node handle rather than a usable output promise.

Use explicit ordering when data flow does not express the dependency:

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

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

Node.__rshift__() calls runs_before() and returns the right-hand node, so t3_node >> t2_node means “run t3_node before t2_node,” and chaining works. Internally, runs_before() appends self to other._upstream_nodes; it modifies the other node's dependency list, not the calling node's list. The serialized node therefore contains the upstream node ID even when it has no input binding.

The node output name outputs is reserved for the output dictionary. create_node() raises a FlyteAssertion if an entity has an output whose name would collide with an existing node attribute, so avoid naming a task output outputs.

Node-level overrides​

Apply execution metadata to the node created by a task or workflow call with with_overrides(). Node.with_overrides() supports node names and aliases, timeout, retries, interruptibility, cache settings, resources, container image, accelerator, shared memory, pod templates, and task configuration. It returns the same node, so it can be used inline as in the earlier chaining example.

For example, the backfill implementation in flytekit/remote/backfill.py names each generated node and applies retry and timeout values:

next_node = next_node.with_overrides(
    name=f"b-{next_start_date}", retries=per_node_retries, timeout=per_node_timeout
)

Resource overrides can use requests and limits, or the resources argument. Node.with_overrides() rejects using resources together with requests or limits; it also validates that resource values are flytekit.Resources. Override values must be compile-time values: flytekit calls assert_not_promise() for scalar override fields and rejects promises inside resource specifications. For example, a workflow input cannot be used as a memory limit or timeout.

Failure policy and failure handlers​

Select workflow failure behavior with WorkflowFailurePolicy:

  • FAIL_IMMEDIATELY is the default and moves the workflow into a failed state when a component node fails.
  • FAIL_AFTER_EXECUTABLE_NODES_COMPLETE allows remaining runnable nodes to finish before the workflow fails.

Pass the policy to the decorator or to ImperativeWorkflow. A failure handler is supplied with on_failure:

@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

This pattern is from tests/flytekit/unit/core/test_workflows.py. WorkflowBase models the handler as a failure node rather than as part of the normal node list. During validation, the handler interface must contain all workflow inputs; any additional handler inputs must be optional. An optional err: FlyteError input is therefore valid. Imperative workflows expose the corresponding add_on_failure_handler(entity) method.

Programmatic composition with Workflow​

Use Workflow—the public alias for ImperativeWorkflow—when the graph is assembled by application code rather than by walking one decorated function. The imperative API follows the same node, promise, and binding model:

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

This exact pattern appears in tests/flytekit/unit/core/test_imperative.py. add_workflow_input() returns a promise whose node is GLOBAL_START_NODE; add_entity() creates and returns a node; and add_workflow_output() converts the selected promise into a workflow output binding. Convenience methods add_task(), add_subwf(), and add_launch_plan() delegate to add_entity().

The equivalent decorated workflow is:

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

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

Imperative construction is useful when the number or arrangement of nodes is determined programmatically. flytekit/remote/backfill.py creates launch-plan nodes in a date-range loop and, when parallel execution is disabled, links each node to the previous one with runs_before(). The OpenAI batch plugin similarly uses add_entity() to connect upload, batch, and download nodes, then applies resource overrides to selected nodes.

ImperativeWorkflow maintains its own CompilationState and tracks declared-but-unused inputs in _unbound_inputs. ready() returns false unless the workflow has at least one node and every declared input has been consumed. Duplicate input names are rejected by add_workflow_input(). This check also runs before imperative local execution, so declaring an input without passing its promise to an entity leaves the workflow unready.

From nodes to an executable workflow definition​

Serialization is handled by get_serializable_workflow() in flytekit/tools/translator.py. It iterates over entity.nodes, skips the global input node, recursively serializes each remaining node, and builds a WorkflowTemplate containing the workflow ID, metadata, typed interface, serialized nodes, and entity.output_bindings. Nested WorkflowBase entities are recursively added as sub-workflow templates. Branch nodes and failure nodes receive separate handling so their nested workflows are included in the resulting WorkflowSpec.

The resulting structure can be summarized as:

WorkflowBase
  ├── interface: workflow inputs and outputs
  ├── nodes: Node objects
  │     ├── bindings: data edges from Promise/NodeOutput or literals
  │     └── upstream_nodes: explicit ordering edges
  ├── output_bindings: selected node outputs exposed by the workflow
  └── failure_node / nested workflows when configured
        -> WorkflowTemplate / WorkflowSpec

A reference workflow is an exception: get_serializable_workflow() raises ValueError for a ReferenceWorkflow used as a sub-workflow and directs you to use a reference launch plan instead.

Composition checks to keep in mind

  • The decorated workflow body is evaluated during compilation, so ordinary Python loops and conditions run then; they do not dynamically execute on Flyte. Use Flyte entities and conditional() to express runtime graph behavior.
  • Pass task and workflow inputs by keyword. create_node() and node creation reject positional arguments.
  • Use promises for task-to-task data flow, but not for with_overrides() values or resources.
  • a >> b adds a to b's upstream list; runs_before() has the same behavior.
  • Every imperative workflow input must be consumed, and a workflow must contain at least one node to be ready.
  • Reference sub-workflows cannot be serialized; use reference launch plans.
  • outputs is reserved on explicit node handles, so do not use it as a task output name.