Conditional and dynamic workflows

Flytekit provides two primary mechanisms for introducing control flow into your pipelines: Conditional Branches and Dynamic Workflows. While both allow for non-linear execution, they operate at different stages of the workflow lifecycle and serve distinct purposes.

Conditional Branches

Conditional branches in flytekit allow you to execute specific tasks or subworkflows based on the values of inputs or task outputs. These conditions are evaluated by the Flyte engine (Propeller) at runtime, but the structure of the branches must be known at compile time.

Using the conditional function

To create a conditional branch, use the conditional function from flytekit. This function creates a ConditionalSection that supports a fluent API for defining if, elif, and else logic.

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_()
        .then(my_input)
    )

Comparison and Conjunction Expressions

Conditions in flytekit are not standard Python bool expressions. Because they are evaluated by the Flyte engine, they must be constructed using ComparisonExpression or ConjunctionExpression objects.

• Supported Operators: Use standard comparison operators: <, <=, >, >=, ==, !=.
• Logical Operators: Use & (AND) and | (OR). Standard Python and and or keywords will not work because they cannot be overridden to return the expression trees required by flytekit.
• Unary Expressions: You cannot use a raw promise as a condition (e.g., .if_(my_promise)). You must compare it to a value (e.g., .if_(my_promise == True)).

The Case class in flytekit.core.condition enforces these rules during workflow construction, raising an AssertionError if it receives a raw boolean or an unsupported expression type.

Branching Semantics and Constraints

1. Exhaustiveness: Every conditional block must end with an .else_() or a .fail(). A "dangling if" is not permitted because the workflow must guarantee an output or a terminal state for all possible input values.
2. Output Consistency: The ConditionalSection.compute_output_vars method determines the return type of the entire conditional block. It calculates the intersection of output variables across all branches. If your branches return different sets of outputs, only the common subset is available to subsequent nodes.
3. The .fail() method: You can terminate a workflow execution with an error message if a specific branch is reached using Case.fail("error message").

Internal Implementation: Compilation vs. Local Execution

The behavior of ConditionalSection changes based on the context:

• Compilation/Remote Execution: When you register a workflow, flytekit builds a BranchNode. This node contains an IfElseBlock that the Flyte engine uses to decide which path to take.
• Local Execution: When running locally, flytekit uses LocalExecutedConditionalSection. It evaluates the expressions immediately using c.expr.eval() and "takes" the first branch that evaluates to true, skipping the execution of other branches to mimic remote behavior.

Dynamic Workflows

While conditional branches have a fixed structure defined at compile time, Dynamic Workflows allow you to define the workflow structure itself at runtime based on data.

The @dynamic Decorator

A dynamic workflow is defined using the @dynamic decorator. Internally, a dynamic workflow is a hybrid: it is modeled as a Task on the backend, but when it executes, it returns a compiled WorkflowSpec that the engine then executes as a subworkflow.

from flytekit import dynamic, task

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

@dynamic
def my_dynamic_subwf(a: int) -> list[int]:
    s = []
    # Unlike a standard @workflow, you can use Python control flow 
    # like 'for' loops and 'if' statements on input values here.
    for i in range(a):
        s.append(t1(a=i))
    return s

When to use Dynamic Workflows vs. Conditionals

Feature	Conditional Branch (conditional)	Dynamic Workflow (@dynamic)
Evaluation Time	Evaluated by the engine at runtime.	Evaluated by a worker node at runtime.
Structure	Fixed at compile time.	Generated at runtime.
Python Logic	Limited to ComparisonExpression.	Full Python power (loops, recursion).
Visibility	All possible branches visible in the UI.	Sub-graph only visible after the dynamic task runs.
Overhead	Low (engine-level switch).	Higher (requires running a task to generate the graph).

Caveats for Dynamic Workflows

As noted in flytekit/core/dynamic_workflow_task.py, dynamic workflows should be used judiciously:

• Scale: Avoid generating thousands of nodes in a dynamic workflow. If you are performing identical operations on a large list, use a map_task instead.
• Complexity: The workflow generated by a @dynamic task must still be compiled and processed by Flyte. Keep the number of generated tasks under 50 for optimal performance.