Conditional and dynamic workflows

When building pipelines in flytekit, you often need to alter execution flow based on upstream outputs. You might need to pick between two alternate tasks based on a threshold check, or you might need to inspect the length of a dataset at runtime and spawn an arbitrary number of subtasks in a loop. Flytekit provides two distinct mechanisms for these scenarios: conditional sections (conditional) and dynamic workflows (@dynamic).

Choosing the wrong mechanism leads to common errors: attempting to use Python's native if statements on unresolved Flyte Promise objects inside @workflow causes compilation failures, while overusing @dynamic for simple binary branching incurs unnecessary container execution overhead.

+---------------------------------------------------------------------------------------+
|                                    flytekit Branching                                 |
+---------------------------------------------------------------------------------------+
|                                                                                       |
|  Static Conditionals (conditional)                Dynamic Workflows (@dynamic)        |
|  - Graph structure fixed at registration time    - Subworkflow graph built at runtime|
|  - Control plane evaluates conditions             - Runs inside a task container      |
|  - No Python value inspection (works on Promise)  - Materializes native Python values |
|  - Evaluates branches without running extra tasks - Can loop, fan-out, and recurse    |
|                                                                                       |
+---------------------------------------------------------------------------------------+

---

Static Conditionals

When you write a standard @workflow, task invocations do not return Python values—they return Promise instances representing outputs that will only exist at execution time. Standard Python control flow (if promise:, and, or, not) cannot evaluate a Promise. The conditional function solves this by constructing backend BranchNode entities evaluated directly by the Flyte control plane (FlytePropeller) during execution.

Basic Branching and Syntax Rules

To create a conditional branch, call conditional(name) inside a workflow and chain .if_(), optional .elif_(), and a required terminal .else_() clause.

from flytekit import task, workflow, conditional

@task
def success_step(val: int) -> str:
    return f"Value {val} is within range"

@task
def high_step(val: int) -> str:
    return f"Value {val} is too high"

@task
def low_step(val: int) -> str:
    return f"Value {val} is too low"

@workflow
def threshold_workflow(val: int) -> str:
    return (
        conditional("check_threshold")
        .if_((val >= 10) & (val <= 100))
        .then(success_step(val=val))
        .elif_(val > 100)
        .then(high_step(val=val))
        .else_()
        .then(low_step(val=val))
    )

Static conditions must conform to these rules enforced by Case in flytekit/core/condition.py:

1. Bitwise logical operators only: Python's and, or, and not cannot be overloaded to produce expression trees. You must use & (conjunction) and | (disjunction), with each comparison enclosed in parentheses. For example, write (x > 0) & (x < 10) rather than x > 0 and x < 10.
2. No unary promises: You cannot pass a raw boolean promise directly into if_() or elif_(). Use explicit comparison or promise helper methods like ready_flag.is_true(), ready_flag == True, or optional_output.is_none() rather than bare if_(ready_flag).
3. Required terminal clause: Every conditional chain must conclude with an .else_() branch. That final clause must call either .then(...) to return an output or .fail("error message") to fail the workflow execution node explicitly.

@workflow
def validated_workflow(score: float) -> str:
    return (
        conditional("score_validation")
        .if_(score >= 0.7)
        .then(success_step(val=int(score * 100)))
        .else_()
        .fail("Score did not meet minimum threshold of 0.7")
    )

Output Matching Across Branches

The ConditionalSection.compute_output_vars method calculates the intersection of output variables across all branches. Every .then() branch must resolve to identical output variable names and matching types:

@task
def path_a() -> str:
    return "result_a"

@task
def path_b() -> str:
    return "result_b"

@workflow
def multi_output_wf(flag: bool) -> str:
    return (
        conditional("multi_branch")
        .if_(flag == True)
        .then(path_a())
        .else_()
        .then(path_b())
    )

If one branch returns a VoidPromise (a task with no return value) while another returns a promise, compute_output_vars() returns None, defaulting the entire conditional block to a VoidPromise.

Internal Compilation and Local Execution

The factory function conditional(name) in flytekit/core/condition.py inspects the active FlyteContext and instantiates one of three ConditionalSection variants:

• ConditionalSection: Used during workflow compilation. It records Case objects and translates them into an IfElseBlock model containing IfBlock structures and boolean expression trees (ComparisonExpression, ConjunctionExpression). It attaches a Node containing the resulting BranchNode to the workflow's CompilationState.
• LocalExecutedConditionalSection: Used during local in-memory execution (e.g., calling threshold_workflow(val=50) directly in unit tests or local scripts). It evaluates expressions on the fly using expr.eval(), marks chosen paths with ExecutionState.take_branch(), runs only the selected task branch, and extracts its return value through _compute_outputs.
• SkippedConditionalSection: Used when evaluating nested conditionals inside an enclosing branch that was already evaluated to False during local execution. It suppresses local task execution inside inactive branches and yields dummy void bindings.

---

Dynamic Workflows

When your workflow topology depends on runtime data—such as dynamically looping over an input array of variable size, recursive algorithms, or inspecting task outputs inside Python for loops—static conditionals and workflows are insufficient.

The @dynamic decorator (defined in flytekit/core/dynamic_workflow_task.py) solves this.

import typing
from flytekit import dynamic, task, workflow

@task
def process_item(item: int) -> int:
    return item * 2

@task
def summarize_results(results: typing.List[int]) -> int:
    return sum(results)

@dynamic
def dynamic_fanout(count: int) -> typing.List[int]:
    outputs = []
    # Native Python control flow using the runtime input value:
    for i in range(count):
        outputs.append(process_item(item=i))
    return outputs

@workflow
def dynamic_pipeline(count: int) -> int:
    processed = dynamic_fanout(count=count)
    return summarize_results(results=processed)

Execution Model: Hybrid Task-Workflow

A dynamic workflow is modeled on the backend as a task, but behaves like a workflow compiler at execution time:

1. Compilation at Registration: At registration time, the enclosing @workflow treats @dynamic like a single task node. The interior graph of the dynamic workflow is not compiled into the primary workflow spec.
2. Container Execution at Runtime: When the dynamic workflow node executes, Flyte launches a task container with actual, materialized input values (e.g., count=5).
3. Subworkflow Generation: Inside the container, the decorated Python function executes. Unlike a standard @workflow, inputs are actual Python variables (integers, strings, lists), allowing native Python constructs (for, while, if val > 10:).
4. DynamicJobSpec Submission: As tasks are invoked inside the dynamic function, flytekit captures them into a dynamic subworkflow definition (DynamicJobSpec). The container returns this graph to FlytePropeller without executing the child tasks locally.
5. Control-Plane Scheduling: FlytePropeller receives the generated subworkflow DAG and schedules each generated task node across the cluster.

---

Choosing Between Conditional, Dynamic, and Map Tasks

Select the appropriate pattern based on how your pipeline determines its execution structure:

Requirement	Recommended Construct	Execution Model
Predetermined branching based on input parameters or upstream task outputs	conditional(name)	Evaluated by FlytePropeller control plane; skips unselected branches with zero container overhead.
Graph structure depends on runtime values (dynamic loops, data-dependent DAG shapes)	@dynamic	Launches a task container to compile a runtime subworkflow sent back to the engine.
Running one identical task over thousands of elements in a homogeneous list	map_task	High-throughput parallel task execution without compiling large subworkflow graphs.

Scale Considerations

Because dynamic workflow functions compile into a full subworkflow graph that FlytePropeller tracks in its etcd state engine, keep dynamic workflow subgraphs under 50 generated tasks. If you need to fan out to hundreds or thousands of elements, use map_task rather than generating thousands of task nodes inside a @dynamic loop.