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 logic that isn't strictly linear, they operate at different stages of the Flyte lifecycle and solve distinct problems.

Conditional Branches

Conditional branches allow you to execute different paths in a workflow based on the values of task outputs or workflow inputs. Unlike standard Python if statements, which are evaluated during workflow compilation, Flytekit conditionals are evaluated at runtime by the Flyte engine.

Using the Conditional API

To create a conditional branch, use the conditional function from the flytekit package. This function starts a ConditionalSection that supports .if_(), .elif_(), and .else_() methods.

from flytekit import task, workflow, conditional

@task
def success_task() -> str:
    return "Success"

@task
def failure_task() -> str:
    return "Failure"

@workflow
def my_conditional_wf(val: int) -> str:
    return (
        conditional("check_value")
        .if_(val > 10)
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Constraints on Expressions

Because these conditions are evaluated by the Flyte engine (Propeller), you cannot use arbitrary Python logic. The expressions must use Promise objects and specific operators that Flytekit can translate into a BranchNode.

1. Bitwise Operators for Logic: Use & for AND and | for OR. Standard Python and and or will not work because they attempt to evaluate the truthiness of the Promise object immediately.
2. Comparison Operators: Use standard comparisons like ==, !=, <, <=, >, >=.
3. No Unary Truthiness: You cannot write .if_(my_input). You must use explicit comparisons, such as .if_(my_input == True) or the .is_true() helper on boolean promises.

# Correct usage of conjunctions and comparisons
v = (
    conditional("complex_logic")
    .if_((val >= 0) & (val <= 10))
    .then(task_a())
    .elif_((val > 10) | (val < 0))
    .then(task_b())
    .else_()
    .fail("Unexpected value")
)

Internal Implementation: Compilation vs. Execution

The ConditionalSection class handles the transition between these states:

• Compilation Mode: When you define the workflow, ConditionalSection captures each branch into a Case object. When the section ends (via end_branch), Flytekit compiles these into a BranchNode (defined in flytekit.models.core.workflow). This node contains the IfElseBlock that the Flyte engine uses to route execution.
• Local Execution: When running the workflow locally, Flytekit uses LocalExecutedConditionalSection. It evaluates the expressions using c.expr.eval() and "takes" the first branch that evaluates to true, effectively short-circuiting the other branches to mimic runtime behavior.

Dynamic Workflows

Dynamic workflows are used when the structure of the workflow (the number of nodes or the specific tasks to run) depends on data that is only available at runtime. While a conditional chooses between pre-defined paths, a @dynamic task generates a new workflow graph on the fly.

Defining Dynamic Logic

A dynamic workflow is defined using the @dynamic decorator. Inside the function, you can use native Python control flow like for loops and if statements on the input values.

from flytekit import dynamic, task

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

@dynamic
def my_dynamic_task(count: int) -> list[int]:
    results = []
    # Native Python loop is allowed here because this runs at execution time
    for i in range(count):
        results.append(process_item(item=i))
    return results

How Dynamic Workflows Work

Internally, a @dynamic task is treated as a single Task node during the initial workflow compilation. However, when that node executes:

1. The function body runs locally on a worker.
2. Instead of returning simple values, it returns a collection of Promise objects (from calling other tasks).
3. Flytekit captures these calls and compiles them into a subworkflow.
4. The worker returns this subworkflow back to the Flyte engine, which then executes the generated nodes.

When to Use Which

Feature	Use Case	Evaluation Time	Logic Constraints
Conditional	Choosing between a few fixed paths based on a result.	Runtime (Engine)	Limited to ComparisonExpression and ConjunctionExpression.
Dynamic	Generating a variable number of tasks (e.g., processing a list of unknown length).	Runtime (Worker)	Full Python flexibility (loops, recursion, etc.).

Warning: Dynamic workflows should be used judiciously. Because they generate a new workflow graph at runtime, they can lead to very large graphs if used with large loops. For processing thousands of identical items, prefer map_task.