Conditional and dynamic workflows

Flytekit provides two distinct mechanisms for implementing non-linear workflow logic: conditional branches and dynamic workflows. These serve different purposes and have different execution semantics.

When to Use Conditional Branches vs. Dynamic Workflows

Conditional branches are compiled into the workflow graph and executed by the Flyte engine. They are ideal for simple if/elif/else logic that can be expressed as a fixed graph structure. The entire workflow graph, including all branches, is known at serialization time.

Dynamic workflows are tasks that generate a new workflow at execution time. They allow Python-native control flow (like loops, conditionals, and dynamic task generation) that cannot be determined until runtime. The generated workflow is then submitted to the Flyte engine as a subworkflow.

Use conditionals when:

You have a fixed set of branches determined at design time
You need type-safe, compiled workflow graphs
Your logic fits the if/elif/else pattern

Use dynamic workflows when:

You need to generate tasks dynamically based on input data (e.g., a loop over a list of files)
You need Python-native control flow that isn't expressible as a static graph
The workflow structure depends on runtime data

The `conditional()` Function

Conditional branches are created using the conditional() function, which returns a ConditionalSection object. This function can only be called within a workflow context.

from flytekit import workflow, task
from flytekit.core.condition import conditional

@task
def task_a(x: int) -> int:
    return x + 1

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

@workflow
def my_workflow(x: int) -> int:
    # Start a conditional section with a unique name
    result = conditional("my_branch").if_(x > 5).then(task_a(x=x)).else_().then(task_b(x=x))
    return result

The conditional() function behaves differently depending on the execution context:

During compilation (serialization), it returns a ConditionalSection that builds the workflow graph
During local execution, it returns a LocalExecutedConditionalSection that evaluates conditions at runtime
When a parent branch is skipped, it returns a SkippedConditionalSection that skips all nested branches

Fluent API for Branches

The ConditionalSection provides a fluent API for defining branches:

.if_(expression) - Start the first branch with a condition
.elif_(expression) - Add additional branches with conditions
.else_() - Add a catch-all branch (required if no else is provided, the last branch will be treated as else)
.then(entity) - Specify the task or workflow to execute for this branch
.fail(message) - Specify an error for this branch (used when no valid output is possible)

Each branch must return a value, and all branches must return compatible types. The conditional expression returns the output of the selected branch.

from flytekit import workflow, task
from flytekit.core.condition import conditional

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

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

@workflow
def conditional_example(my_input: float) -> int:
    v = (
        conditional("fractions")
        .if_((my_input > 0.1) & (my_input < 1.0))
        .then(
            conditional("inner_fractions")
            .if_(my_input < 0.5)
            .then(double(n=my_input))
            .elif_((my_input > 0.5) & (my_input < 0.7))
            .then(square(n=my_input))
            .else_()
            .fail("Only <0.7 allowed")
        )
        .elif_((my_input > 1.0) & (my_input < 10.0))
        .then(square(n=my_input))
        .else_()
        .then(double(n=my_input))
    )
    return v

Expression Types and Operators

Conditional expressions support comparison and conjunction operators:

Comparison operators:

> (greater than)
< (less than)
>= (greater than or equal)
<= (less than or equal)
== (equal)
!= (not equal)

Conjunction operators:

& (AND) - both conditions must be true
| (OR) - at least one condition must be true

Important notes about expressions:

Expressions must use Flyte Promises (task outputs), not raw Python values
Use bitwise operators & and | instead of logical operators and and or
Parentheses are required around conjunctions due to operator precedence

# Correct usage
conditional("test").if_((x > 5) & (y < 10)).then(task_a())

# Incorrect - will raise ValueError
# if_(x > 5 and y < 10)  # Python's 'and' tries to evaluate the Promise

Output Handling and Type Safety

All branches in a conditional must return compatible outputs. Flytekit computes the intersection of output variables across all branches to ensure type safety.

@task
def task_a() -> int:
    return 1

@task
def task_b() -> str:
    return "hello"

@workflow
def invalid_conditional(x: int) -> int:  # This will fail
    # task_b returns str, but workflow expects int
    return conditional("test").if_(x > 5).then(task_a()).else_().then(task_b())

If branches return different outputs, the conditional will only return the common subset. If there are no common outputs, it returns a VoidPromise:

@task
def task_a() -> int:
    return 1

@task
def task_b() -> str:
    return "hello"

@workflow
def void_conditional(x: int):
    # No common outputs, returns VoidPromise
    conditional("test").if_(x > 5).then(task_a()).else_().then(task_b())

Local Execution Behavior

During local execution, LocalExecutedConditionalSection evaluates conditions at runtime using the .eval() method on expressions. It selects the first matching branch and skips subsequent branches:

# In LocalExecutedConditionalSection.start_branch():
if self._selected_case is None:
    if c.expr is None or c.expr.eval() or last_case:
        ctx.execution_state.take_branch()
        self._selected_case = added_case

The BranchEvalMode enum controls branch evaluation:

BRANCH_ACTIVE - The current branch is being evaluated
BRANCH_SKIPPED - Branches should be skipped (used for nested conditionals when parent branch is false)

Advanced: Variable Naming and Promise Merging

When compiling conditionals, Flytekit generates unique variable names for branch conditions to avoid collisions. The create_branch_node_promise_var(node_id, var) function combines the node ID and variable name (e.g., n1.o0).

The merge_promises() function deduplicates promises by (node_id, var) and renames duplicates:

def merge_promises(*args: Optional[Promise]) -> typing.List[Promise]:
    node_vars: typing.Set[typing.Tuple[str, str]] = set()
    merged_promises: typing.List[Promise] = []
    for p in args:
        if p is not None and p.ref:
            node_var = (p.ref.node_id, p.ref.var)
            if node_var not in node_vars:
                new_p = p.with_var(create_branch_node_promise_var(p.ref.node_id, p.ref.var))
                merged_promises.append(new_p)
                node_vars.add(node_var)
    return merged_promises

This ensures that even if multiple nodes produce outputs with the same variable name, the compiled workflow has globally unique references.

Gotchas and Best Practices

Branches require at least two cases: A single .if_() without .else_() raises an AssertionError. Always provide an else branch or use .fail().
Output compatibility: All branches must return compatible types. The conditional's output is the intersection of all branch outputs.
Local vs. compiled semantics: Local execution evaluates expressions at runtime, while compiled execution builds a static graph. Test workflows locally before serialization.
Dynamic workflows for complex control flow: If your logic requires loops, dynamic task generation, or complex control flow that doesn't fit if/elif/else, use @dynamic tasks instead.
Variable name collisions: In compiled workflows, variable names may be transformed (e.g., o0 becomes n1.o0). Don't rely on original variable names in branch conditions.
Nested conditionals: Nested conditionals are supported but can become complex. Consider using dynamic workflows for deeply nested logic.

Dynamic Workflows

Dynamic workflows are created using the @dynamic decorator. They are tasks that generate a workflow at execution time:

from flytekit import dynamic, task

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

@dynamic
def my_dynamic_workflow(n: int) -> typing.List[str]:
    results = []
    for i in range(n):  # Python-native loop - not possible in static workflows
        results.append(process_item(item=i))
    return results

Key differences from conditionals:

Dynamic workflows run at execution time, not compilation time
They can use Python-native control flow (loops, conditionals, etc.)
The generated workflow is submitted as a subworkflow
They're limited to ~50 tasks for performance reasons

For large-scale identical runs, use map tasks instead of dynamic workflows with loops.

When to Use Conditional Branches vs. Dynamic Workflows​

The conditional() Function​

Fluent API for Branches​

Expression Types and Operators​

Output Handling and Type Safety​

Local Execution Behavior​

Advanced: Variable Naming and Promise Merging​

Gotchas and Best Practices​

Dynamic Workflows​