Conditional and dynamic workflows

When you need to control the execution path of a workflow based on data produced at runtime, flytekit provides two primary mechanisms: Conditional Sections for static branching and Dynamic Workflows for runtime graph generation.

Static Conditional Branching

Static conditionals allow you to define branches that are visible in the Flyte console and compiled into the workflow graph at registration time. You use the conditional function to create these branches.

Defining Conditions

A conditional block in flytekit is a fluent API that must always follow a specific structure: it starts with if_, can have multiple elif_ branches, and must terminate with either an else_() or a fail().

from flytekit import task, workflow, conditional

@task
def double(n: float) -> float:
    return n * 2.0

@task
def square(n: float) -> float:
    return n * n

@workflow
def my_workflow(my_input: float) -> float:
    return (
        conditional("fractions")
        .if_((my_input > 0.1) & (my_input < 1.0))
        .then(double(n=my_input))
        .elif_((my_input >= 1.0) & (my_input < 10.0))
        .then(square(n=my_input))
        .else_()
        .fail("Input out of range")
    )

Expression Rules and Limitations

Flytekit expressions are not standard Python boolean expressions. Because the workflow is compiled before the data exists, flytekit uses the ComparisonExpression and ConjunctionExpression classes to represent logic.

1. Use Bitwise Operators: You must use & (AND) and | (OR) instead of the Python keywords and and or.
2. Parentheses are Mandatory: Due to Python operator precedence, you must wrap each comparison in parentheses (e.g., (a > 5) & (b < 10)).
3. No Unary Truth Testing: You cannot use if_(promise). You must explicitly compare against a value, such as if_(promise.is_true()) or if_(promise == True).
4. Primitive Types Only: Comparisons are only supported for primitive types (integers, floats, strings, booleans).

How it Works Internally

When you call conditional(name), flytekit checks the FlyteContext.

• During Compilation: It creates a ConditionalSection and pushes a new context via FlyteContextManager.push_context. Each .then() call captures the output Promise of the task. When .else_() or .fail() is called, end_branch is triggered. It pops the context and uses to_branch_node to serialize the logic into a BranchNode (a wrapper for the Flyte model IfElseBlock).
• During Local Execution: It uses LocalExecutedConditionalSection. It evaluates the ComparisonExpression using .eval() against the actual local values. It then calls ExecutionState.take_branch() to ensure only the selected task runs, while other branches are marked as BRANCH_SKIPPED.

Output Consistency

All branches in a conditional section must return the same output interface. The ConditionalSection.compute_output_vars method calculates the intersection of all variable names returned by the .then() calls across all branches. If a variable is missing from one branch, it will not be available in the final output of the conditional block.

Dynamic Workflows

Use the @dynamic decorator when the structure of your workflow depends on runtime values that cannot be expressed with simple static comparisons—such as the length of a list or a complex dictionary structure.

Runtime Graph Generation

A dynamic workflow is a hybrid: it is treated as a task by the parent workflow, but when it executes, it returns a DynamicJobSpec which Flyte then executes as a sub-workflow.

from typing import List
from flytekit import task, dynamic

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

@dynamic
def my_dynamic_subwf(a: int) -> List[str]:
    s = []
    # Native Python control flow is allowed here because 
    # 'a' is a concrete value at runtime.
    for i in range(a):
        s.append(t1(a=i))
    return s

Key Differences from Static Conditionals

Feature	Static conditional	@dynamic Workflow
Visibility	Entire graph visible at registration.	Graph is generated at runtime.
Control Flow	Must use conditional(...).if_()	Uses standard Python if, for, while.
Inputs	Operates on Promise objects.	Operates on native Python values.
Performance	Low overhead; compiled once.	Higher overhead; requires a runtime "compilation" step.

Implementation Details

When a @dynamic task runs, flytekit enters a special execution mode. The python_function_task.py implementation creates a new CompilationState with a prefix (usually "d") to track the nodes generated during the function's execution.

The resulting DynamicJobSpec contains:

• tasks: The definitions of tasks invoked inside the dynamic body.
• nodes: The execution graph generated by those invocations.
• outputs: The bindings for the final results.

Best Practices and Constraints

1. Scale Limits: Dynamic workflows should generally not exceed a few hundred nodes. If you need to process thousands of items, use a map_task instead.
2. Node Dependency Hints: If your dynamic workflow calls a LaunchPlan, you must provide it in the node_dependency_hints parameter of the @dynamic decorator so that Flyte knows about the dependency at registration time.
3. No Reference Tasks: Reference tasks (tasks defined in other projects/registrations) are not supported inside dynamic tasks.