Skip to main content

Workflows and workflow composition

Two workflow authoring surfaces​

When you need a workflow whose graph is expressed by ordinary Python syntax, use @workflow. When the graph must be assembled programmatically, use Workflow (the public alias for ImperativeWorkflow). Both produce a WorkflowBase, so tasks, sub-workflows, local execution, and serialization use the same workflow contract.

Function-based workflows​

The workflow decorator creates a PythonFunctionWorkflow. Its interface is derived from the decorated function, and calls to Flyte tasks inside the function body provide the data-flow edges:

import typing

from flytekit import task, workflow


@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

This is a graph-building function, not a backend runtime function. workflow documents that the function body is evaluated at serialization time (compile time), and is not evaluated again when the workflow runs on Flyte. During that evaluation, calls such as t1(a=a) produce promises that can be passed to later tasks or returned as workflow outputs.

The PythonFunctionWorkflow.compile() method constructs input promises, invokes the original function once, collects the compilation-state nodes, validates an optional failure handler, and turns the returned promises into output bindings. Compilation is cached by the compiled flag. PythonFunctionWorkflow.execute() is deliberately small: it calls the original function so that WorkflowBase.local_execute() can use the same local execution path as other workflow implementations.

Imperative workflows​

Use the public Workflow alias when you want to name inputs and outputs explicitly and add entities one at a time:

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

assert wb(in1="hello") == "hello world"

flytekit.__init__ defines Workflow as ImperativeWorkflow. The imperative constructor starts with an empty Interface, creates a CompilationState, and defaults to WorkflowFailurePolicy.FAIL_IMMEDIATELY and interruptible=False. add_workflow_input() extends both the native and typed interfaces and returns a Promise associated with the global start node. add_entity() uses create_node() under the workflow's compilation state; add_task(), add_launch_plan(), and add_subwf() are convenience wrappers around it.

The two styles can describe the same graph. In the imperative version, the output name is supplied to add_workflow_output(). In a function workflow, the return annotation supplies the output interface; a NamedTuple is useful when you need stable output names:

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


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

The WorkflowBase contract: interfaces, promises, and nodes​

WorkflowBase is the shared runtime and serialization contract rather than an end-user workflow to instantiate directly. Its constructor stores workflow metadata, metadata defaults, the native Interface, an optional failure handler, documentation, and default launch-plan options. It immediately converts the native interface with transform_interface_to_typed_interface() and initializes the input promises, unbound-input set, node list, output-binding list, and failure-node slot. It also appends every instance to FlyteEntities.entities, the registry scanned by Flyte's serialization and CLI tooling.

The two interface properties serve different layers:

  • python_interface is the native interface derived from Python annotations and defaults.
  • interface is the typed Flyte interface used for literal types and serialization.

Function workflows receive their native interface from transform_function_to_interface(). Imperative workflows begin with Interface() and update it as add_workflow_input() and add_workflow_output() are called.

A workflow's graph is exposed through nodes and output_bindings. Those properties call compile() before returning their values, so inspecting the graph can trigger compilation. WorkflowBase.__call__() also compiles before dispatch unless the current execution state is eager execution. It fills in declared default inputs, accepts positional arguments for direct workflow calls, rejects more positional arguments than the interface declares, and dispatches through flyte_entity_call_handler.

At compile time, create_node() accepts a PythonTask, LaunchPlan, WorkflowBase, or supported remote entity. When a callable entity is invoked while a compilation state is active, create_node() takes the newly created node and attaches its promises to the node using each output name. This is why a composed node can be addressed as either node.outputs["name"] or through a named output attribute.

The base class itself does not implement a graph: WorkflowBase.execute() raises NotImplementedError and WorkflowBase.compile() is a no-op. PythonFunctionWorkflow and ImperativeWorkflow supply those operations.

Compilation and workflow composition​

A function workflow's task calls become nodes while PythonFunctionWorkflow.compile() runs the function with input promises. An imperative workflow creates nodes incrementally. In both cases, the resulting model is a data-flow graph whose output bindings refer to node outputs rather than to already-computed Python values.

You can compose function workflows exactly as you compose tasks. The nested workflow is recognized as a WorkflowBase, and create_node() attaches the nested workflow's output promises to the parent node:

@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

For imperative composition, use add_subwf() for another workflow and add_launch_plan() for a launch plan. Each is added as one parent node:

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)

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_workflow() encounters a nested WorkflowBase, it serializes the node and recursively adds the nested template to WorkflowSpec.sub_workflows. It also serializes a workflow's separate failure node. A ReferenceWorkflow is the exception in this path: the translator rejects a reference workflow used as a sub-workflow because populating its template would require a network call, and the error directs you to use a reference launch plan instead.

Local execution and workflow outputs​

A direct local call such as wb(in1="hello") or parent_wf(a=3) follows WorkflowBase.local_execute(). That method deliberately keeps the local data-flow representation in terms of promises:

  1. Incoming native values are translated to Flyte literals using the workflow's typed and native input interfaces.
  2. Each literal is wrapped in a Promise.
  3. The concrete workflow is compiled and executed.
  4. Coroutine results are awaited with asyncio.run().
  5. Returned values are checked against the declared output interface, translated back to literals, and repackaged as promises named after the workflow outputs.

ImperativeWorkflow.execute() assumes entities were added in topological order. It starts an intermediate output map with the global start-node inputs, resolves each node's bindings from that map, calls the node's entity, stores its promises, and finally resolves the workflow output bindings. ready() rejects an imperative workflow with no nodes or with declared inputs that have not been consumed.

The local path handles these output shapes:

  • A single output is returned as one value/promise.
  • Multiple outputs are returned as a tuple.
  • A one-element NamedTuple is recognized using output_tuple_name and is normalized specially.
  • A no-output workflow must return None or a VoidPromise.

For example, the imperative API can construct a collection output, but it needs the collection's Python type because the type cannot be inferred from a list or dictionary of promises:

@task
def t3(a: typing.List[str]) -> str:
    return ",".join(a)


wf = Workflow(name="my.imperative.workflow.example")
wf.add_workflow_input("in1", str)
node_t1 = wf.add_entity(t1, a=wf.inputs["in1"])
wf.add_workflow_output("output_from_t1", node_t1.outputs["o0"])

wf_in2 = wf.add_workflow_input("in2", str)
node_t3 = wf.add_entity(t3, a=[wf.inputs["in1"], wf_in2])

wf.add_workflow_output(
    "output_list",
    [node_t1.outputs["o0"], node_t3.outputs["o0"]],
    python_type=typing.List[str],
)

For a scalar promise, ImperativeWorkflow.add_workflow_output() infers the Python type from the referenced node output. For a list or dictionary of promises, pass python_type; otherwise it raises FlyteValidationException.

Keep the declared interface and the function return aligned. Returning a task output from a workflow with no declared outputs raises FlyteValueException; conversely, a workflow declared with outputs cannot finish with None or a VoidPromise. A function workflow's output bindings also validate tuple shape and output count during compilation.

Failure handlers​

Attach a failure handler with on_failure on a function workflow, or with add_on_failure_handler() on an imperative workflow. A handler receives the workflow's inputs. It may also declare an optional input named err to receive a FlyteError:

@task
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name} due to {err}")
    print("This is err:", str(err))


@task
def create_cluster(name: str):
    print(f"Creating cluster: {name}")


@task
def delete_cluster(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name}")
    print(err)


@task
def t1(a: int, b: str):
    print(f"{a} {b}")
    raise ValueError(error_message)


@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

When WorkflowBase.__call__() catches an exception, it checks whether the handler's interface contains the literal parameter name err. If so, it supplies FlyteError(failed_node_id=..., message=str(exc)), calls the handler with the workflow inputs, and then re-raises the original exception. Local failure handling therefore runs the handler for its observable effects; it does not turn a failed workflow into a successful result.

The handler's interface is checked during compilation. Every workflow input must be accepted by the handler. Additional handler inputs are allowed only when their types are optional. For example, a handler with a and optional b is valid for a workflow with only a, while a handler that omits one of the workflow's inputs raises FlyteFailureNodeInputMismatchException:

@task()
def t1(a: int) -> int:
    return a + 3


@workflow(on_failure=t1)
def wf1(a: int = 3, b: str = "hello"):
    t1(a=a)


@task()
def t2(a: int, b: typing.Optional[int] = None) -> int:
    return a + 3


@workflow(on_failure=t2)
def wf2(a: int = 3):
    t2(a=a)

PythonFunctionWorkflow compiles a function-workflow failure handler in a separate compilation state and requires exactly one resulting task or workflow node. ImperativeWorkflow.add_on_failure_handler() similarly creates the node, removes it from the main compilation-state node list, stores it in _failure_node, and assigns the reserved failure-node ID. Consequently, the failure node is not part of the normal graph node list. The translator emits it separately as WorkflowTemplate.failure_node, while recursively including a nested failure workflow when necessary.

Failure policy and interruptibility​

A failure handler answers “what should run after failure?” The workflow failure policy answers “when does the workflow enter its failed state after a node fails?” Configure the policy when declaring the 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 has exactly two values:

  • FAIL_IMMEDIATELY causes the workflow execution to fail when a component node fails. This is the default for the decorator and ImperativeWorkflow.
  • FAIL_AFTER_EXECUTABLE_NODES_COMPLETE allows remaining runnable nodes to finish after a component node failure.

The policy is stored in WorkflowMetadata. WorkflowMetadata.to_flyte_model() maps immediate failure to model value 0 and the after-executable-nodes policy to 1. interruptible is separate: WorkflowMetadataDefaults validates that it is a strict boolean, serializes it as metadata defaults, and WorkflowBase.construct_node_metadata() uses it when constructing node metadata. The serialization test verifies both settings:

wf_spec = get_serializable(OrderedDict(), serialization_settings, wf)
assert wf_spec.template.metadata_defaults.interruptible
assert wf_spec.template.metadata.on_failure == 1

Practical edge cases​

  • Do not instantiate WorkflowBase as a complete workflow. Its execute() is NotImplementedError and its compile() does nothing. Use @workflow, Workflow/ImperativeWorkflow, or a specialized subclass.

  • Add imperative entities in dependency order. ImperativeWorkflow.execute() walks compilation_state.nodes sequentially, and ready() requires all declared workflow inputs to have been consumed.

  • Use keyword arguments with add_entity(). create_node() rejects positional input arguments. This is distinct from direct workflow calls through WorkflowBase.__call__(), which support positional arguments subject to the interface's argument-count check.

  • Treat task results in a function body as data-flow values. The workflow decorator evaluates the body during compilation; it does not execute arbitrary non-Flyte entities on the backend. Express graph-dependent behavior with Flyte's data-flow constructs rather than relying on native Python control flow over a task result.

  • Match void and non-void interfaces. A no-output workflow may call a no-output task without returning it, but a declared output requires a matching return value. A returned value from an interface with zero outputs raises FlyteValueException.

  • Give collection outputs an explicit type in imperative workflows. add_workflow_output() can infer a scalar promise's type, but requires python_type for a list or dictionary of promises.

  • Expect serialization settings when producing a WorkflowSpec. The translator uses SerializationSettings.project, .domain, and .version to construct the workflow identifier. The repository's serialization tests provide those settings explicitly:

    serialization_settings = flytekit.configuration.SerializationSettings(
        project="project",
        domain="domain",
        version="version",
        env=None,
        image_config=ImageConfig(default_image=default_img, images=[default_img]),
    )

  • Account for workflow registration. Constructing any WorkflowBase appends it to FlyteEntities.entities; serialization helpers and CLI discovery inspect that registry.

  • Use a reference launch plan for a reference sub-workflow. The translator explicitly rejects a ReferenceWorkflow nested in a serializable workflow because it cannot populate that referenced workflow template without contacting Flyte Admin.