Conditional and dynamic workflows

Conditional and dynamic workflows allow you to control the execution flow of a Flyte workflow based on data generated at runtime. While both mechanisms enable branching, they differ significantly in how they are compiled and when they are evaluated.

Conditional Workflows

Conditional workflows use the conditional function to define logical branches within a @workflow. These branches are evaluated by the Flyte engine (Propeller) at runtime, but the entire structure of the conditional must be known at compile time.

Defining Conditions

The conditional API uses a fluent interface: if_, elif_, and else_. Each branch must terminate with a .then() call (to return a value or execute a task) or a .fail() call (to terminate the workflow with an error).

from flytekit import workflow, conditional, task

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

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

@workflow
def my_conditional_wf(val: int) -> str:
    return (
        conditional("value_check")
        .if_(val > 10)
        .then(success_task())
        .elif_(val < 0)
        .fail("Value cannot be negative")
        .else_()
        .then(failure_task())
    )

Expression Constraints

Flytekit conditions operate on Promise objects (the outputs of tasks or workflow inputs). Because these values are not available during Python's compilation phase, you cannot use standard Python logical operators like if x:, and, or or.

Instead, flytekit provides specific methods and bitwise operator overrides on the Promise class (found in flytekit/core/promise.py):

• Comparisons: Use standard operators: ==, !=, <, <=, >, >=.
• Conjunctions: Use & for AND and | for OR.
• Boolean Helpers: Use .is_true(), .is_false(), or .is_none().

# Valid expression using conjunction and boolean helper
conditional("complex_check").if_((val > 0) & (val < 100) | flag.is_false()).then(...)

Internal Implementation

When you define a conditional block, flytekit creates a ConditionalSection (in flytekit/core/condition.py).

• In Compilation Mode, it records all branches into a BranchNode which is added to the workflow graph. The engine then evaluates the ComparisonExpression or ConjunctionExpression at runtime to decide which node to execute.
• In Local Execution Mode, the LocalExecutedConditionalSection immediately evaluates the expressions using the actual Python values and executes only the selected branch.

Dynamic Workflows

Dynamic workflows, defined with the @dynamic decorator, allow you to generate a workflow graph at runtime based on the inputs provided to a task. Unlike standard workflows, the body of a @dynamic function is executed at runtime, allowing you to use native Python logic (like loops and standard if statements) to build a subworkflow.

When to use Dynamic Workflows

Use @dynamic when the structure of your workflow depends on runtime data, such as:

• Processing a variable number of files discovered at runtime.
• Implementing recursive algorithms like Merge Sort.
• Chaining tasks where the number of steps is determined by an input value.

from flytekit import dynamic, task
import typing

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

@dynamic
def my_dynamic_wf(count: int) -> typing.List[int]:
    results = []
    # Native Python loops and range() are allowed here
    for i in range(count):
        results.append(process_item(item=i))
    return results

Execution Semantics

A @dynamic function is modeled as a task. When it runs:

1. The Flyte engine executes the function body.
2. The function returns a set of Promise objects representing a new workflow graph.
3. The engine compiles this new graph into a subworkflow and executes it.

Comparison: Conditional vs. Dynamic

Feature	Conditional (conditional)	Dynamic (@dynamic)
Evaluation Time	Engine-side at runtime	Task-side at runtime
Graph Structure	Fixed at compile time	Generated at runtime
Python Logic	Restricted (no for, if x:)	Full Python support
Overhead	Low (simple branch evaluation)	Higher (requires task execution + subworkflow compilation)
Use Case	Simple branching based on task outputs	Complex, data-dependent graph generation

Limitations and Gotchas

• Conditional Unary Expressions: Flytekit does not support unary expressions like if_(val). You must use if_(val.is_true()) or if_(val == True).
• Dynamic Node Limits: Because dynamic workflows generate new graphs, they can be abused. It is recommended to keep the number of nodes generated by a dynamic workflow under 50. For large-scale parallel processing, use map_task instead.
• Workflow Context: The conditional function can only be used inside a function decorated with @workflow or @dynamic. Attempting to use it elsewhere will result in a context error.