Skip to main content

Workflow composition and nodes

Building a workflow graph with @workflow​

When one task consumes another task's result, write the workflow as a normal Python function and connect the calls with their returned values. In tests/flytekit/integration/remote/workflows/basic/basic_workflow.py, the workflow returns two values while sending t1's string output to t2:

@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

The workflow decorator creates a PythonFunctionWorkflow. It supports both @workflow and @workflow(...); its wrapper constructs WorkflowMetadata with FAIL_IMMEDIATELY unless another failure_policy is supplied, creates WorkflowMetadataDefaults from interruptible, captures the function docstring, and passes through on_failure, docs, pickle_untyped, and default_options. update_wrapper preserves the decorated function's identity (flytekit/core/workflow.py).

Compilation uses promises as graph placeholders​

A decorated workflow's body is evaluated while Flyte builds the graph, not only when a task eventually runs. PythonFunctionWorkflow.compile() enters a CompilationState, replaces workflow inputs with promises associated with GLOBAL_START_NODE, and evaluates task and workflow calls. Those calls create Node objects and return output Promise objects. Compilation records the nodes and converts returned promises into workflow output Binding objects. The compiled flag makes this operation idempotent; workflow calls, workflow properties, and serialization can request compilation without rebuilding the graph.

Consequently, treat the workflow body as graph construction code. A task output during compilation is a promise rather than a native int or str; use Flyte-supported promise operations and conditionals rather than ordinary Python operations that require an actual value. Local invocation is a separate execution path, described below.

A promise-valued argument becomes a binding, and its source node becomes an upstream dependency. This also works through collections and maps: promise handling recursively discovers promises nested in lists and dictionaries, so one node can consume several upstream outputs through a single collection argument. The output side has matching shape checks: multiple workflow outputs are normally returned as a tuple, and the returned shape must agree with the annotated workflow interface.

Task results behave like typed output collections. You can unpack multiple outputs, select an individual named output, pass that output to another task, or pass it into a sub-workflow. A sub-workflow is called in the same way as a task:

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

@workflow
def my_subwf(a: int = 42) -> (str, str):
    x, y = t1(a=a)
    u, v = t1(a=x)
    return y, v

@workflow
def parent_wf(a: int) -> (int, str, str):
    x, y = t1(a=a)
    u, v = my_subwf(a=x)
    return x, u, v

Here my_subwf(a=x) creates a workflow node in parent_wf; the nested workflow's own nodes are serialized recursively later.

Explicit nodes and ordering​

Automatic data dependencies are sufficient when one entity consumes another entity's output. Use create_node when you need a concrete node handle—for example, when tasks have no inputs or outputs, when you need explicit ordering, or when you want to apply a node override during composition.

create_node accepts only keyword inputs. Its implementation in flytekit/core/node_creation.py rejects positional arguments and is intended for use inside workflows and dynamic tasks. In compilation mode it calls the Flyte entity, takes the node most recently added to the compilation state, and populates that node's output map. The resulting outputs are also exposed as attributes:

@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

r, x = my_wf(a="hello")
assert r == "hello world"
assert x == ["0 world", "1 world", "2 world"]

For named outputs, use the generated attribute or the outputs mapping. Node.outputs is deliberately unavailable on an arbitrary Node; it raises an assertion unless the node was produced by create_node. The explicit form is useful when output names are dynamic or when you want to combine outputs inside a collection:

@workflow
def my_wf(a: int, b: str) -> (str, str):
    t1_node = create_node(t1, a=a).with_overrides(aliases={"t1_str_output": "foo"})
    t1_nt_node = create_node(t1_nt, a=a)
    t2_node = create_node(t2, a=[t1_node.t1_str_output, t1_nt_node.t1_str_output, b])
    return t1_node.t1_str_output, t2_node.o0

Node is the graph vertex in flytekit/core/node.py. Its constructor stores a DNS-normalized ID, node metadata, literal bindings, upstream nodes, and the underlying Flyte entity. Promise bindings normally populate upstream_nodes automatically. For nodes that do not exchange data, add an ordering edge explicitly:

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

Node.runs_before(other) appends self to other._upstream_nodes if it is not already there. Node.__rshift__ delegates to runs_before and returns other, so a >> b >> c can be chained. The operation mutates the other node's upstream list; it is not a symmetric graph operation. The same ordering is visible in the serialized node: the test for empty_wf2 verifies that the downstream node contains the upstream node ID.

No-output entities return a VoidPromise. It supports ordering and override pass-through for local compatibility, but it cannot be used as a data value. This is why explicit ordering is the relevant composition mechanism for side-effect-only tasks.

Per-node overrides​

Apply runtime and presentation settings to the node returned by a task call or to an explicit node from create_node:

node = create_node(t1, a=a).with_overrides(
    aliases={"t1_str_output": "foo"},
    retries=2,
    timeout=60,
    interruptible=True,
    container_image="my-image:tag",
)

Node.with_overrides() mutates the node and returns the same node. It supports DNS-normalized node_name, output aliases, requests and limits, the combined resources form, timeout, retries, interruptible, task_config, container_image, accelerator and shared-memory settings, cache, and pod_template. The implementation stores resource and extended-resource overrides on the node and updates metadata through _override_node_metadata (flytekit/core/node.py).

Keep overrides static. assert_not_promise rejects promise-valued values for metadata and settings such as retries, interruptibility, cache versions, container images, accelerators, shared memory, and pod templates. Resource values are converted to Flyte resource models and checked by assert_no_promises_in_resources. timeout must be None, an integer number of seconds, or a datetime.timedelta. Do not combine resources with requests or limits; with_overrides raises a ValueError when those forms are mixed. Supplying requests without limits emits a warning and clamps requests to the original limits.

For caching, pass a Cache object with a version when overriding a node. A cache object without a version raises an error, and combining it with deprecated cache_serialize or cache_version arguments also raises. task_config must retain the original task configuration type. These checks happen while the graph is being composed, before serialization.

Programmatic composition with Workflow​

Use the imperative API when the graph is assembled by application code rather than by evaluating a decorated function. Workflow is the public alias for ImperativeWorkflow. Declare inputs, add entities, and bind outputs explicitly:

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 promise from the global input node and tracks the input as unbound. add_entity (or the task- and sub-workflow-specific helpers) creates a node in an incremental compilation state and marks consumed inputs bound. add_workflow_output creates the typed workflow output binding. ready() requires at least one node and rejects a graph with unbound inputs; local execution raises Workflow not ready before resolving bindings when those conditions are not met.

The imperative executor walks nodes in insertion order. It resolves workflow-input promises, prior node outputs, scalar literals, collections, and maps through the promise helpers construct_input_promises, get_promise, and get_promise_map. Therefore, add entities in topological order: a later node can consume an earlier node's output, but the imperative local executor does not reorder nodes for you.

The imperative graph is equivalent to the decorated form for serialization and nesting. The test suite adds wb as a sub-workflow using add_subwf, and also creates a LaunchPlan from it. The decorated equivalent is a normal task call followed by a return:

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

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

Local execution and executable definitions​

Calling a workflow locally does not turn task results into graph nodes. WorkflowBase.local_execute converts native inputs into literal-backed promises, invokes the workflow, handles synchronous or asynchronous results, validates void, single-output, and multiple-output shapes, converts output literals, and rewraps results using the workflow's declared output names. This path explains why the same workflow can be called in a test with native values while compilation sees promises.

Serialization asks the workflow to compile first, then flytekit.tools.translator.get_serializable_workflow converts the in-memory graph into an executable Flyte model. Its flow is:

WorkflowBase.nodes
    -> get_serializable_node for each Node
    -> upstream_node_ids + input bindings + task/workflow node payload
    -> WorkflowTemplate(nodes, outputs, metadata, interface)

The translator skips the global input node, serializes each remaining node, and recursively serializes nested WorkflowBase entities. It builds the workflow identifier from SerializationSettings.project, domain, and version, then creates a WorkflowTemplate containing the workflow metadata, metadata defaults, typed interface, serialized nodes, output bindings, and optional failure node. Nested workflow templates are returned in the WorkflowSpec.sub_workflows collection. SerializationSettings.image_config supplies the image context used while task nodes are serialized.

A workflow with explicit ordering therefore produces a node definition whose upstream_node_ids reflect runs_before/>>, while a data-connected workflow produces input bindings and corresponding upstream IDs from its promises. The resulting WorkflowSpec is the executable definition consumed by Flyte registration and execution.

Workflow-level policies and failure handlers​

Set workflow metadata at the decorator boundary when the whole graph needs a different policy:

@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

WorkflowMetadata supports FAIL_IMMEDIATELY and FAIL_AFTER_EXECUTABLE_NODES_COMPLETE; WorkflowMetadataDefaults requires interruptible to be a real boolean. The test serializes the example and verifies both metadata_defaults.interruptible and the translated metadata.on_failure value.

Attach on_failure to @workflow for a failure task or workflow. During compilation, Flyte checks that the handler accepts all workflow inputs and that additional handler inputs are optional. The compiled failure node is kept separately, removed from the ordinary node list, and serialized through the template's failure_node field. A node cannot use the reserved failure-node ID.

Reference workflows are different from local nested workflows: ReferenceWorkflow records a remote workflow identity and expected interface without making an Admin network call. The serializer rejects a reference workflow used directly as a sub-workflow and recommends a reference launch plan instead.