Skip to main content

Workflow composition and nodes

Define a workflow as a graph of Flyte entities​

When you write a @workflow function, its Python body is not the code Flyte runs for every workflow execution. The body is evaluated while flytekit compiles or serializes the workflow, and the calls to tasks, subworkflows, and launch plans become the nodes and edges of a DAG. Use the decorator as the normal entry point:

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

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

The workflow function in flytekit/core/workflow.py accepts either the function directly (@workflow) or workflow options (@workflow(interruptible=True, ...)). It constructs a PythonFunctionWorkflow, whose name is derived from the decorated function and whose interface is derived from its annotations. PythonFunctionWorkflow is a WorkflowBase and stores the compiled Node objects in _nodes and workflow output bindings in _output_bindings.

Compilation turns task calls into nodes​

Inside a workflow body, task outputs are not ordinary Python values. They are Promise objects that describe an output of a node. A later task call consumes that promise, so flytekit can infer the data-flow edge:

@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

This example is from tests/flytekit/unit/core/test_workflows.py. The x promise is bound to the input of the node for the second add_5 call. The workflow itself returns promises produced by prior nodes; PythonFunctionWorkflow.compile() converts those returned values into literal Binding objects for the workflow outputs.

The compilation path is lazy. WorkflowBase.__call__() invokes compile() before dispatching the entity call handler, and serialization also accesses compiled workflow data. PythonFunctionWorkflow.compile() only runs its body when compiled is false:

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)

construct_input_promises() represents each workflow input as a Promise whose NodeOutput points to GLOBAL_START_NODE. When a Flyte entity is called during compilation, flyte_entity_call_handler() routes the call through create_and_link_node() in flytekit/core/promise.py. That function creates the input bindings and associates promise inputs with their upstream nodes. The resulting Node records:

  • a DNS-compatible ID, normally assigned from compilation order (n0, n1, ...),
  • metadata describing the node,
  • bindings connecting entity inputs to literals or upstream outputs,
  • upstream_nodes for dependencies, and
  • the flyte_entity being invoked (a task, workflow, or launch plan).

Node.__init__() rejects a missing ID and applies _dnsify() to the ID. Therefore, with_overrides(node_name="my_task") changes the node ID to a DNS-compatible form such as my-task; it does not change the metadata display name, which is controlled by the name override.

Compilation is idempotent: after the first compilation, subsequent calls reuse the stored nodes and bindings. This means the structure of a decorated workflow is fixed after its first compilation.

Connect task outputs and workflow inputs​

Pass a task's output promise as a keyword argument to another entity to create data flow. Multiple outputs preserve their declared names, including named-tuple outputs:

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

@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

The first t1 call creates a node with two output promises. Passing x into the second call links the second node to the first. Returning y and v makes them workflow-level output bindings. The test serializes this workflow and verifies that interruptible is present in template.metadata_defaults and that the failure policy is represented in template.metadata.

A workflow's input promises are also valid task inputs. At compile time they point to GLOBAL_START_NODE; during local execution, WorkflowBase.local_execute() translates the supplied Python values into literals, wraps them in promises, and executes the compiled graph with those values.

Produce an executable workflow definition​

Use get_serializable() with SerializationSettings to turn a workflow into Flyte's serialized model for registration or inspection:

serialization_settings = flytekit.configuration.SerializationSettings(
    project="test_proj",
    domain="test_domain",
    version="abc",
    image_config=ImageConfig(Image(name="name", fqn="image", tag="name")),
    env={},
)
wf_spec = get_serializable(OrderedDict(), serialization_settings, my_wf)

The get_serializable() calls in tests/flytekit/unit/core/test_node_creation.py verify the resulting WorkflowSpec contains the compiled nodes and workflow outputs. The translator recursively serializes node entities; a nested workflow is represented as a workflow node and its definition is included among the serialized subworkflows.

Add ordering when there is no data dependency​

Data flow is normally enough to establish execution order. For side-effecting tasks that do not consume one another's outputs, use runs_before() or the >> operator:

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

The equivalent method form is t3_node.runs_before(t2_node). Node.__rshift__() calls runs_before() and returns the right-hand node, which permits chains such as c >> t >> d. The operator also works on the Promise or VoidPromise returned by a task call:

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

Node.runs_before() appends the left node to the other node's upstream_nodes when it is not already present. This adds an ordering dependency without creating an input binding between the tasks. The node-creation tests serialize empty_wf2 and verify that the node with ID n0 has n1 as its upstream node.

A failure-cleanup workflow uses the same mechanism:

@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

Here the workflow-level on_failure handler is separate from the normal c → t → d ordering. During compilation, PythonFunctionWorkflow._validate_add_on_failure_handler() checks that the handler accepts the workflow inputs (and optional additional inputs), compiles it, and stores its node as _failure_node. If the workflow call raises an exception, WorkflowBase.__call__() supplies a FlyteError as err when the handler declares that input.

Create nodes explicitly when you need node handles​

Call create_node() from flytekit/core/node_creation.py when you need explicit ordering or dictionary-style output access. It accepts only keyword inputs; positional arguments raise FlyteAssertion.

@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() calls the entity, obtains the last node from the active CompilationState, and attaches the resulting promises to that node. A node with outputs exposes both attributes such as t1_node.o0 and dictionary entries such as t1_node.outputs["o0"]. Node.outputs is intentionally unavailable for nodes created implicitly by direct task calls; for those calls, use the returned promise or named-tuple attributes.

The dictionary form is useful when output names are computed or when building an imperative workflow:

wb = ImperativeWorkflow(name="my.workflow.a")
wf_in1 = wb.add_workflow_input("in1", typing.List[int])
t1_node = wb.add_entity(t1, a=2)
t2_node = wb.add_entity(t2, a=t1_node.outputs["o0"], b=wf_in1)
wb.add_workflow_output("from_n0t2", t2_node.outputs["o0"])

In local execution, create_node() calls the entity and returns local results, while its compilation path attaches promises to the node. This difference is why code inside the workflow body must be written in terms of Flyte entities and promises rather than assuming task outputs are immediately available Python values.

Override one node without changing the task​

Call with_overrides() on a task result, Node, or void result to change configuration for that occurrence in the graph:

@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

Node.with_overrides() returns self, so the overridden node/promise can be assigned, passed to another task, or returned directly. Supported node-level settings include node_name, name, aliases, requests, limits, resources, timeout, retries, interruptible, cache, task_config, container_image, accelerator, shared_memory, and pod_template.

The timeout override distinguishes omission from an explicit None through Node.TIMEOUT_OVERRIDE_SENTINEL: an integer is interpreted as seconds, a datetime.timedelta is used directly, and None resets the timeout to an empty duration. Resource overrides cannot combine resources with requests or limits; resources is converted to request and limit models by ResourceSpec.from_multiple_resource(). The method also validates promise-valued settings such as retries, interruptibility, container images, and resource values.

For cache overrides, pass either a boolean or a Cache object. A Cache used in an override must have a version; deprecated cache parameters cannot be combined with a Cache object. These checks happen while the node metadata is being changed, before serialization.

Compose subworkflows​

Calling a decorated workflow from another workflow creates a workflow node just as calling a task does:

@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()
    return x, d

The subworkflow's outputs are promises and can be passed onward or returned. During serialization, the parent node refers to the subworkflow, and the serialized WorkflowSpec contains the subworkflow definition. Imperative workflows use the same composition model through add_subwf(); the imperative tests verify that the parent template has a workflow_node.sub_workflow_ref and one serialized subworkflow.

A ReferenceWorkflow, created with @reference_workflow(project=..., domain=..., name=..., version=...), points to an already registered workflow and does not make a network call. It cannot be used as a subworkflow inside another workflow; use a reference launch plan for that composition case.

Build the graph programmatically​

Use ImperativeWorkflow, exported by flytekit as Workflow, when application code—not a decorated function body—needs to assemble the graph:

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"

add_workflow_input() creates a workflow-level promise, add_entity() adds a node, and add_workflow_output() plays the role of a function return. ImperativeWorkflow keeps its own CompilationState; entities must be added in topological order. Its execute() iterates through the stored nodes, resolves each node's bindings with get_promise_map(), calls the node's entity, and records its outputs for later nodes. The resulting workflow serializes to the same backend graph shape as the equivalent function-based workflow.

Configure workflow-level behavior​

Set workflow execution policy and defaults on the decorator:

@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.FAIL_IMMEDIATELY is the default and makes the workflow enter a failed state when a component node fails. FAIL_AFTER_EXECUTABLE_NODES_COMPLETE allows remaining runnable nodes to finish. WorkflowMetadata validates the policy and maps it to the Flyte model (0 for immediate failure and 1 for the after-executable-nodes policy). WorkflowMetadataDefaults validates that interruptible is exactly True or False and serializes it as the workflow's task default.

The decorator also accepts on_failure, docs, pickle_untyped, and default_options. pickle_untyped=True bypasses type checking, but the workflow API documents it as not recommended for general use. Workflow docstrings are parsed into documentation; the first line becomes the short description, and Sphinx-style parameter and return tags contribute interface descriptions.

Composition gotchas​

  • Do not treat the workflow body as runtime task code. workflow() explicitly documents that the body is evaluated at serialization/compile time and is not evaluated again when the workflow runs on Flyte. Call Flyte entities there; ordinary Python statements execute during compilation.
  • Promises are placeholders for graph values. A task output cannot be used as an ordinary Python value for operations such as range(a) or a Python comparison. Use Flyte-supported conditional composition for branching, as in conditional("bool").if_(...).then(...).else_().then(...).
  • Compilation happens once. PythonFunctionWorkflow.compiled guards compile(), so changing assumptions in the function after its first call does not rebuild the node list.
  • Mutable task defaults are rejected during node creation. create_and_link_node() detects list, dictionary, and set defaults as unsupported mutable defaults. Use a None default and create the collection in the task body instead.
  • Reserve the failure-node ID. efn is the reserved default failure-node ID; using it for a user node causes serialization to raise ValueError.
  • Use the right output access form. Implicit task calls return promises directly, while create_node() populates node.outputs and output attributes. create_node() also requires keyword arguments.
  • Distinguish node ID from display name. node_name changes the DNS-ified ID used for graph identity. name changes node metadata; it does not replace the ID.
  • Reference workflows are not nested workflows. A ReferenceWorkflow identifies an existing registered workflow but cannot be embedded as a subworkflow; choose a reference launch plan when that is required.