Workflow composition and nodes

The workflow graph is built before execution

When a Flyte workflow runs on the platform, its Python body is not re-evaluated to perform the work. The workflow decorator's implementation in flytekit/core/workflow.py states that the body is evaluated at serialization time (compile time), because the body expresses the workflow structure. During that evaluation, calls to Flyte tasks and workflows produce Promise objects. Those promises carry references to node outputs, so using one as the input to another creates data-flow edges in the DAG.

Write the workflow in the usual Python style, but pass task and workflow inputs by name:

from typing import NamedTuple
from flytekit import task, workflow


@task(enable_deck=True)
def t1(a: int) -> 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

This is the pattern used by tests/flytekit/integration/remote/workflows/basic/basic_workflow.py. x and y are not the integer and string returned by a running Python call while the workflow is being compiled. They are promise-backed references to the outputs of the t1 node; t2(a=y, b=b) therefore records that t2 consumes t1's second output and the workflow input b.

The decorator supports both bare and configured forms. Its implementation signature is:

def workflow(
    _workflow_function=None,
    failure_policy=None,
    interruptible: bool = False,
    on_failure=None,
    docs=None,
    pickle_untyped: bool = False,
    default_options=None,
):
    ...

The concrete decorated object is a PythonFunctionWorkflow. It retains the original function, compiles it into nodes and output bindings, and exposes local execution separately. WorkflowBase owns the workflow's nodes, inputs, output bindings, and failure configuration. Consequently, a local call can produce ordinary Python values, while serialization produces the executable Flyte workflow definition.

Promise data flow

A task's return annotation determines the shape of the promises exposed during compilation. Named-tuple outputs may be unpacked, or accessed by field name:

from typing import NamedTuple
from flytekit import task, workflow


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


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

In tests/flytekit/unit/core/test_composition.py, a named output is also used as a field:

@workflow
def my_subwf(a: int) -> nt:
    x = t1(a=a)
    y = t2(a=a, b=x.sub_int)
    return nt(y.sub_int)

The same mechanism connects a workflow to a nested workflow. Calling my_subwf(a=x) inside another decorated workflow creates a sub-workflow node whose input is the promise x. The serializer (get_serializable_workflow in flytekit/tools/translator.py) recursively places such workflows in WorkflowSpec.sub_workflows.

A workflow's outputs must be derived from task or sub-workflow outputs. Returning a raw Python value from a workflow that declares outputs does not create an output binding; workflow compilation validates the returned values and raises for missing or invalid workflow outputs.

How implicit calls become Node objects

In the decorated form, you do not instantiate Node directly. The call path uses the promise call machinery (flyte_entity_call_handler and create_and_link_node in flytekit/core/promise.py). During compilation it creates a Node, registers it in the active CompilationState, and returns Promise or VoidPromise objects to the workflow function.

A Node in flytekit/core/node.py stores the pieces needed for serialization:

• id, normalized with _dnsify();
• metadata, including node-level execution metadata;
• bindings, which describe how inputs are supplied;
• upstream_nodes, which contain explicit ordering dependencies; and
• flyte_entity, the underlying task, workflow, or launch plan.

The translator turns these fields into a Flyte IDL node. get_serializable_node emits the node's input bindings and upstream node IDs, then selects the appropriate task, workflow, branch, or array node representation. Node overrides such as resources, extended resources, container image, and pod template are applied while constructing the serialized task-node overrides.

Explicit ordering and node handles

Data flow normally supplies ordering automatically: if t2 consumes a promise from t1, Flyte knows that t1 must run first. Use create_node from flytekit/core/node_creation.py when you need a node handle, particularly for void tasks whose calls do not provide a useful value to pass onward.

from flytekit.core.node_creation import create_node
from flytekit.core.task import task
from flytekit.core.workflow import workflow


@task
def t2():
    ...


@task
def t3():
    ...


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

The equivalent operator form appears immediately afterward in tests/flytekit/unit/core/test_node_creation.py:

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

Node.runs_before() appends the left node to the right node's upstream_nodes list if it is not already there. Node.__rshift__() calls runs_before() and returns the right node, which permits chained expressions. The reverse shift operator is not implemented. You can also chain the results of ordinary task calls when you only need ordering:

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


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

For a node created with create_node(), output promises are available through attributes such as t1_node.o0 and through the node's output mapping, for example node.outputs["o0"]. The Node.outputs property deliberately raises an assertion if the node was not created through create_node(), so use that API when dictionary-style output access or a node reference is required.

Per-node execution overrides

A task call returns a node-like promise object that supports with_overrides(). Apply overrides to the particular invocation rather than changing the task declaration:

@workflow
def my_wf(a: str) -> str:
    s = t1(a=a).with_overrides(timeout=timeout)
    s1 = t1(a=s).with_overrides()  # keeps task defaults
    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. Node.with_overrides() returns the same node after changing its fields. The supported node-level settings include node_name, output aliases, requests, limits, or the resources shorthand, timeout, retries, interruptible, cache, task_config, container_image, accelerator, shared_memory, and pod_template. The tests exercise these variants:

# Examples from tests/flytekit/unit/core/test_node_creation.py:
t1(a=a).with_overrides(retries=3)
t1(a=a).with_overrides(interruptible=True)
t1(a=a).with_overrides(cache=Cache(version="foo", serialize=True))
t1(a=a).with_overrides(name="display-name", node_name="dns-node-id")
t1(a=a).with_overrides(container_image="my-image:latest")
t1(a=a).with_overrides(accelerator=A100.partition_1g_5gb)
t1(a=a).with_overrides(shared_memory="128Mi")

Values used for overrides must be known during compilation. Node.with_overrides() explicitly checks values such as node_name, timeout, retries, interruptible, resources, container image, accelerator, shared memory, and pod template with assert_not_promise; do not pass a task output promise as an override.

For resources, choose either resources=Resources(...) or requests=/limits=. The implementation rejects combining the shorthand with either explicit form. If only requests are supplied, it logs that requests are clamped to the original limits. Integer timeouts are interpreted as seconds; a datetime.timedelta is accepted directly. When overriding with a Cache object, specify its version or Node._override_node_metadata() raises ValueError("must specify cache version when overriding").

Node IDs are DNS-normalized both at construction and for node_name overrides. For example, an underscore in a supplied node name is converted by _dnsify() to the corresponding hyphenated form.

Failure policy and failure handlers

Set the workflow-level failure policy when runnable nodes should be allowed to finish after a failure. The workflow tests use the non-default policy together with an interruptibility default:

from flytekit.core.workflow import WorkflowFailurePolicy, workflow


@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 policy when no policy is supplied. FAIL_AFTER_EXECUTABLE_NODES_COMPLETE is serialized into the workflow metadata and allows executable runnable nodes to complete before the workflow fails.

Attach cleanup behavior with on_failure. The handler can accept the workflow's inputs and an optional err: FlyteError. The failure-handler test uses a task with the workflow's name input and the optional error:

@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

WorkflowBase records the failure entity and failure node, and workflow compilation validates the handler interface. A handler with extra required inputs does not match the workflow interface; the supported additional error input is optional FlyteError.

Programmatic composition with Workflow

Use ImperativeWorkflow (commonly imported as Workflow) when you want to construct and register nodes through method calls rather than evaluating a decorated function. Declare inputs with add_workflow_input(), pass those promises into add_entity(), and declare the result with add_workflow_output():

from flytekit.core.workflow import ImperativeWorkflow as Workflow


@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 is the complete construction pattern from tests/flytekit/unit/core/test_imperative.py. add_entity() creates and registers each node immediately in the persistent compilation state. ImperativeWorkflow also provides add_task(), add_subwf(), and add_launch_plan() convenience methods for adding those entity types. Call ready() to validate that the workflow has nodes and that declared inputs have been consumed before compilation/execution.

The equivalent decorated form in the same test makes the difference clear:

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


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

Imperative workflows are also used by plugins. The OpenAI batch workflow in plugins/flytekit-openai/flytekitplugins/openai/batch/workflow.py wires node outputs explicitly and applies per-node resources:

wf = Workflow(name=f"openai-batch-{name}")
wf.add_workflow_input("jsonl_in", Iterator[JSON])

node_1 = wf.add_entity(upload_jsonl_file_task_obj, jsonl_in=wf.inputs["jsonl_in"])
node_2 = wf.add_entity(batch_endpoint_task_obj, input_file_id=node_1.outputs["result"])
node_3 = wf.add_entity(download_json_files_task_obj, batch_endpoint_result=node_2.outputs["result"])

node_1.with_overrides(requests=Resources(mem=file_upload_mem), limits=Resources(mem=file_upload_mem))
node_3.with_overrides(requests=Resources(mem=file_download_mem), limits=Resources(mem=file_download_mem))

wf.add_workflow_output("batch_output", node_3.outputs["result"], BatchResult)

The AWS SageMaker inference plugin demonstrates the other kind of edge: it uses nodes[-1] >> node to impose sequential execution even when the next node's inputs do not carry the preceding node's output.

Referencing an already-registered workflow

Use reference_workflow when the workflow already exists on Flyte and you want to describe its interface locally:

from flytekit.core.workflow import reference_workflow


@reference_workflow(project="project", domain="domain", name="registered-workflow", version="version")
def remote_wf(a: int) -> str:
    ...

ReferenceWorkflow records the project, domain, name, and version without making a network call; the function signature supplies the interface. Flytekit does not serialize a ReferenceWorkflow as an embedded sub-workflow node. The translator raises ValueError for that case; invoke the registered workflow through a LaunchPlan reference instead.

Composition constraints to keep visible

• Use keyword arguments for task, workflow, and launch-plan calls inside a decorated workflow. Positional arguments are rejected with Flytekit's keyword-only input assertion.
• Treat task results as promises while compiling. Python control flow that needs a real result, such as if task_output > 5 or range(task_output), cannot operate on that promise; use Flytekit's conditional() API for workflow branching.
• Keep overrides static. A promise can connect a node's data input, but it cannot supply timeout, retries, resources, container_image, node_name, or the other override values checked by Node.
• Supply a cache version when using with_overrides(cache=Cache(...)).
• Prefer create_node() for void-task ordering and whenever you need node.outputs["..."] or a node handle. For ordinary value flow, calling the task directly and passing its returned promise is the normal decorated-workflow form.
• Compilation is represented by CompilationState.nodes; the translator consumes those nodes to create workflow and node protobuf models. Once a PythonFunctionWorkflow has compiled, its compiled flag remains set and subsequent compilation calls do not re-run the workflow function.