Conditional and dynamic workflows

Flytekit provides two primary mechanisms for introducing control flow into your workflows: static conditionals and dynamic workflows. While both allow your code to make decisions based on runtime data, they differ significantly in how they are compiled, executed, and represented in the Flyte backend.

Static Conditionals

Static conditionals are used when you have a fixed set of potential execution paths that depend on the value of a task output or workflow input. They are defined using the conditional() function from the flytekit package.

Basic Usage

A conditional block must always be defined within a @workflow or @dynamic context. It follows a fluent API pattern: if_().then(), optional elif_().then(), and a mandatory else_().then() or else_().fail().

from flytekit import task, workflow, conditional

@task
def add_five(x: int) -> int:
    return x + 5

@task
def double(x: int) -> int:
    return x * 2

@workflow
def my_conditional_wf(a: int) -> int:
    return (
        conditional("value-check")
        .if_(a > 10)
        .then(add_five(x=a))
        .else_()
        .then(double(x=a))
    )

Expression Rules and Operators

Flytekit conditionals do not support standard Python logical operators (and, or, not) or bare boolean truth testing (e.g., if_(my_promise)). Instead, you must use bitwise operators and explicit comparison methods provided by the Promise class in flytekit/core/promise.py.

• Comparisons: Use standard operators: ==, !=, <, <=, >, >=.
• Conjunctions: Use & for AND and | for OR. Always parenthesize individual comparisons.
• Boolean Promises: If a task returns a bool, use .is_true(), .is_false(), or .is_none().

# Valid compound expression
.if_((a > 0) & (a < 10))

# Valid boolean check
.if_(my_bool_input.is_true())

Compilation and Execution Semantics

When you define a conditional block, Flytekit's ConditionalSection (found in flytekit/core/condition.py) manages the state:

1. Compilation Mode: During workflow registration, the entire block is compiled into a single BranchNode. This node contains an IfElseBlock that encapsulates all branches. The backend evaluates the predicates at runtime to decide which branch to execute.
2. Local Execution: When running locally, LocalExecutedConditionalSection evaluates the expressions immediately. It selects the matching branch and executes only the tasks within that branch.
3. Output Consistency: Every branch in a conditional must return the same type. Flytekit's compute_output_vars method intersects the output variable names across all cases to ensure a consistent interface for downstream nodes.

Nested Conditionals and Failures

You can nest conditionals by passing another conditional() expression to a .then() call. You can also explicitly fail a workflow execution using .fail().

v = (
    conditional("outer")
    .if_(a > 0)
    .then(
        conditional("inner")
        .if_(a < 5)
        .then(t1(x=a))
        .else_()
        .fail("Value too high for inner branch")
    )
    .else_()
    .then(t2(x=a))
)

Dynamic Workflows

Dynamic workflows are used when the structure of the workflow (the number of tasks or their dependencies) can only be determined at runtime based on input data.

Defining Dynamic Workflows

Use the @dynamic decorator. A dynamic workflow is technically a task that, when executed, returns a DynamicJobSpec—essentially a mini-workflow generated on the fly.

import typing
from flytekit import task, dynamic

@task
def process_item(item: int) -> str:
    return f"processed-{item}"

@dynamic
def my_dynamic_wf(n: int) -> typing.List[str]:
    results = []
    for i in range(n):
        results.append(process_item(item=i))
    return results

When to use @dynamic vs conditional()

Feature	conditional()	@dynamic
Graph Structure	Static (fixed branches)	Dynamic (runtime-generated)
Logic	Limited to Promise comparisons	Full Python (loops, recursion, etc.)
Overhead	Low (evaluated by Flyte engine)	Higher (requires a task execution to "compile")
Visibility	All branches visible in UI	Only the generated graph is visible after execution

Node Dependency Hints

Because dynamic workflows are compiled at runtime, the Flyte platform might not know about certain dependencies (like Launch Plans) ahead of time. You can use node_dependency_hints in the @dynamic decorator to inform the system about these entities.

@dynamic(node_dependency_hints=[my_launch_plan])
def dynamic_subwf():
    return [my_launch_plan] * 5

Critical Constraints

1. Context Requirement: conditional() can only be called inside a @workflow or @dynamic function. Calling it at the module level will raise an AssertionError.
2. Mandatory Else: Every conditional must end with an else_() clause. If omitted, the workflow will fail to compile with a message stating that it "should always end with an else_() clause".
3. No Side Effects: In local execution, branches that are not selected are skipped using SkippedConditionalSection. Do not rely on side effects (like printing or global state changes) inside conditional branches, as they may not behave as expected during local testing.
4. Dynamic Scale: While @dynamic is powerful, avoid generating massive graphs (e.g., thousands of tasks). For large-scale identical operations, prefer map_task.