Skip to main content

Workflow composition and nodes

Building a workflow graph from task calls​

When a workflow calls a task, the task call is not treated as an ordinary Python function call while Flyte is compiling the workflow. Flytekit walks the typed workflow function, records each invocation as a graph node, and turns the values returned by those invocations into dataflow references. The smallest useful composition is therefore a typed task whose output is passed to another task or returned from a workflow:

from typing import NamedTuple

from flytekit import task, workflow


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


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


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

This is the composition pattern exercised by tests/flytekit/unit/core/test_composition.py. In local execution, my_wf(a=5, b="hello ") returns (7, "world-9", "world-11"). During compilation, however, x, y, u, and v are promises representing declared task or workflow outputs. Passing x to t1(a=x) creates a data dependency; returning x, u, and v creates workflow output bindings.

The public workflow decorator supports both bare and configured forms. It creates a PythonFunctionWorkflow and preserves the decorated function's wrapper metadata. Workflow-level options include failure_policy, interruptible, on_failure, docs, pickle_untyped, and default_options; the default failure policy is WorkflowFailurePolicy.FAIL_IMMEDIATELY.

from flytekit import WorkflowFailurePolicy, workflow


@workflow(
    interruptible=True,
    failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE,
)
def wf(a: int) -> (str, str):
    x, y = t1(a=a)
    _, v = t1(a=x)
    return y, v

PythonFunctionWorkflow.__init__ in flytekit/core/workflow.py extracts the function name and transforms the Python signature and docstring into an interface. Its compile() method then creates input promises, invokes the original function once inside a CompilationState, collects the generated nodes, and converts the returned promises or collections into output bindings. The resulting nodes and bindings are stored on WorkflowBase as _nodes and _output_bindings.

The compilation flow is:

workflow inputs
    -> Promise objects connected to GLOBAL_START_NODE
    -> task/workflow calls
    -> Node objects + output Promise objects
    -> workflow output bindings
    -> serialized workflow template

CompilationState in flytekit/core/context_manager.py is the accumulator for the middle of this flow. It stores the compilation prefix, mode, resolver, and mutable nodes list; add_node() appends each generated Node. promise.create_and_link_node translates task-call arguments into literal bindings, discovers upstream nodes referenced by input promises, constructs the node, adds it to the active compilation state, and creates a promise for each declared output.

Promise values are graph edges​

Do not use a compiled task result as if it were a native Python value. A task result is a Promise (or a named tuple containing promises), so ordinary operations such as range(a) or truth-value testing are not valid when a is a workflow input promise. Use task calls and Flyte's conditional composition for graph-level decisions instead. Flytekit's workflow source includes this dataflow example:

@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 workflow function body is evaluated to discover the DAG during compilation. Flytekit's workflow documentation warns against calling arbitrary non-Flyte entities in that body. After compilation, the graph—not a replay of arbitrary Python statements—is what is translated and executed.

Workflow output shape is checked against the annotated interface. PythonFunctionWorkflow.compile() treats a one-output interface specially so that a list can be one output rather than a tuple of outputs; multiple outputs must be returned as a tuple of exactly the declared length. A void workflow must return None or a VoidPromise. A one-element NamedTuple is also handled specially using the interface's output_tuple_name.

Nested workflows, launch plans, and specialized nodes​

A decorated workflow can be composed exactly like a task: pass workflow inputs or earlier promises as keyword arguments and return its output promises. In the example above, my_subwf(a=x) becomes a workflow node in my_wf; it is not inlined into the parent's node list at serialization time. The translator serializes that node with a WorkflowNode(sub_workflow_ref=...) and serializes the subworkflow separately.

The same graph representation accommodates launch plans. The imperative API below adds a launch-plan node to wb3:

wb3 = ImperativeWorkflow(name="parent.imperative")
p_in1 = wb3.add_workflow_input("p_in1", str)
p_node0 = wb3.add_launch_plan(lp, in1=p_in1)
wb3.add_workflow_output("parent_wf_output", p_node0.from_n0t1, str)

When get_serializable_node() encounters a LaunchPlan, it removes bindings supplied by the launch plan's fixed inputs and emits a workflow node containing launchplan_ref. For a nested WorkflowBase, it emits sub_workflow_ref. The same translator function handles PythonTask, ArrayNode, ArrayNodeMapTask, and BranchNode, so task maps, array nodes, and conditionals participate in composition through the same core Node abstraction.

Nodes and explicit dependencies​

flytekit.core.node.Node is Flytekit's in-memory representation of one entity invocation. It stores:

  • a DNS-normalized node ID;
  • NodeMetadata;
  • literal input bindings;
  • explicit upstream_nodes; and
  • the invoked flyte_entity, such as a task, workflow, launch plan, branch, or array entity.

The normal user-facing path is to call a task and use its returned promise. Use create_node() when you need the node itself—for example, to order tasks that have no data dependency, access outputs by string key, or apply a node override before returning it.

from flytekit.core.node_creation import create_node


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

create_node() accepts only keyword inputs. In compilation it invokes the entity in the active compilation context, obtains the newly appended core node, and stores its output promises on the node. For an entity with outputs, use the generated output attributes or the outputs mapping:

@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

Node.outputs is available only for nodes created by create_node(); ordinary Node instances do not necessarily have an output map and accessing it raises an assertion. create_node() also requires a compilation or local workflow context and rejects positional input arguments.

runs_before() and >> add an ordering edge without passing data:

t3_node.runs_before(t2_node)
# Equivalent, and chainable:
t3_node >> t2_node

Node.runs_before() appends the source node to the other node's upstream_nodes list if it is not already present. __rshift__ returns the right-hand node, which is why c >> t >> d can express a chain. This is an execution-order edge, not a value edge: use a task output promise when the downstream task consumes data. The serialized node therefore receives upstream IDs; the node IDs in the test for the example above are n0 and n1, with n0.upstream_node_ids containing n1.

Per-node overrides​

Apply most per-invocation settings to the result of a task call. Promise objects forward with_overrides() to the node that produced them, so this is the common form:

@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

The corresponding implementation is Node.with_overrides() in flytekit/core/node.py. It mutates the node and supports node_name, aliases, requests, limits, resources, timeout, retries, interruptible, name, task_config, container_image, accelerator, cache, shared_memory, and pod_template. node_name changes the DNS-normalized node ID, while name changes metadata name. The translator carries resource, extended-resource, image, and pod-template overrides into TaskNodeOverrides.

For example, the timeout accepts integer seconds or a datetime.timedelta; None clears it. Resource settings must be static Flytekit values:

from datetime import timedelta

from flytekit import Resources


result = t1(a=a).with_overrides(
    node_name="preprocess-v2",
    name="Preprocess data",
    timeout=timedelta(minutes=5),
    retries=2,
    interruptible=True,
    resources=Resources(cpu="2", mem="4Gi"),
)

resources= cannot be combined with requests= or limits=. requests and limits must be flytekit.Resources; if only requests are supplied, Flytekit logs a warning that requests are clamped to the original limits. Promise-valued resource, retry, interruptibility, image, pod-template, and related override values are rejected. A Cache used in an override must specify a cache version, and deprecated cache arguments cannot be combined with a Cache object.

Producing an executable workflow definition​

Compilation produces Flytekit's internal graph; serialization turns it into a registrable workflow model. The tested path supplies project, domain, version, image configuration, and environment through SerializationSettings:

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)

flytekit.tools.translator.get_serializable_node() first recursively serializes each explicit upstream node and records its ID, bindings, metadata, and upstream_node_ids. It then chooses the node payload based on Node.flyte_entity: task_node for a PythonTask, workflow_node with sub_workflow_ref for a nested workflow, workflow_node with launchplan_ref for a launch plan, branch_node for a conditional, and array_node for array entities. The resulting workflow template contains executable Flyte model nodes and workflow output bindings rather than Python promise objects.

Programmatic composition with Workflow​

If a function decorator is not the right shape for your builder, use ImperativeWorkflow. Flytekit exports it as Workflow from the top-level package:

from flytekit import Workflow, task


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

ImperativeWorkflow.add_entity() enters a compilation context, creates a node, and removes consumed workflow-input promises from its unbound-input set. add_task, add_subwf, and add_launch_plan are specialized additions. add_workflow_output() is the imperative equivalent of a function return; list and dictionary outputs require an explicit python_type.

Nested imperative composition uses the node's named output:

wb2 = ImperativeWorkflow(name="parent.imperative")
p_in1 = wb2.add_workflow_input("p_in1", str)
p_node0 = wb2.add_subwf(wb, in1=p_in1)
wb2.add_workflow_output("parent_wf_output", p_node0.from_n0t1, str)

ready() requires at least one node and no unconsumed workflow inputs. For local execution, ImperativeWorkflow.execute() walks its compiled nodes in insertion/topological order, resolves bindings with get_promise_map(), executes each entity, caches outputs, and then resolves workflow output bindings. Consequently, imperative entities must be added in an order suitable for that local walk.

Failure nodes and execution behavior​

A workflow failure handler is compiled as a separate node, not inserted into the main node list. Its interface must include every workflow input; any additional handler input must be optional. The implementation convention used by local execution is an optional input named err:

@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

During compilation, PythonFunctionWorkflow._validate_add_on_failure_handler() compiles the handler in a separate CompilationState and requires exactly one resulting node. During local calls, WorkflowBase.__call__ catches an exception, creates a FlyteError containing the failure-node ID and exception message when the handler has an err input, invokes the handler, and re-raises the original exception. Serialization stores the resulting failure node separately as failure_node.

The same WorkflowBase.__call__ fills declared Python defaults, validates positional-argument count, compiles outside eager execution, and dispatches through flyte_entity_call_handler. local_execute() translates native inputs into Flyte literals, executes the compiled workflow, handles coroutine and void results, validates output arity, and recreates workflow-level promises using the declared output names. This is why the same composed workflow can be locally executed while its compilation-time values remain promises.