Conditional and dynamic workflows

Conditional and dynamic workflows in flytekit allow you to introduce logic that depends on the results of previous tasks. While both enable branching and decision-making, they operate at different stages of the workflow lifecycle and have distinct constraints.

Conditional Workflows

When you need to branch based on the output of a task or a workflow input, use the conditional function. Unlike standard Python if statements, which are evaluated at compilation time (when the workflow is first defined), flytekit.conditional creates a BranchNode that is evaluated by the Flyte engine at execution time.

Basic Syntax

The conditional function uses a fluent API to define branches. Every conditional block must be complete, meaning it must end with an .else_() or a .fail() call.

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: Python builtin 'bool' has no member 'is_true'

from flytekit import task, workflow, conditional

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

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

@workflow
def my_workflow(status: bool) -> str:
    # status is a Promise here, not a Python bool.
    # We use .is_true() which is a method on the Promise class.
    return (
        conditional("check-status")
        .if_(status.is_true())
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Supported Expressions

Conditions in flytekit are not standard Python booleans; they are ComparisonExpression or ConjunctionExpression objects defined in flytekit.core.promise. These are created by using operators on Promise objects (the outputs of tasks).

• Comparisons: ==, !=, <, <=, >, >=
• Conjunctions: & (AND), | (OR)
• Boolean Methods: For boolean inputs or outputs, the Promise class provides .is_true(), .is_false(), and .is_none().

[!WARNING] You cannot use Python's native and, or, or not keywords within a conditional block. These will raise a ValueError (from Promise.__bool__) because they attempt to evaluate the Promise immediately rather than building a graph expression.

Complex Branching and Nesting

You can chain multiple conditions using .elif_() and even nest conditional blocks within a .then() clause.

@workflow
def complex_workflow(val: float) -> str:
    return (
        conditional("range-check")
        .if_((val >= 0.0) & (val <= 0.5))
        .then(
            conditional("inner-check")
            .if_(val < 0.25)
            .then(task_a(val=val))
            .else_()
            .then(task_b(val=val))
        )
        .elif_(val > 0.5)
        .then(task_c(val=val))
        .else_()
        .fail("Value is out of range")
    )

Internal Implementation

When conditional(name) is called, flytekit checks the current FlyteContext.

• During compilation, it returns a ConditionalSection which records the branches into an IfElseBlock.
• During local execution, it returns a LocalExecutedConditionalSection. This implementation evaluates the expressions immediately using ComparisonExpression.eval() to decide which branch to "take," ensuring that only the tasks in the active branch are executed locally.

The Case.then() method (found in condition.py) captures the Promise returned by the task call. If the task returns multiple values, the ConditionalSection computes the intersection of output variables across all branches to ensure the workflow remains type-safe regardless of which path is taken.

Dynamic Workflows

While conditional handles simple branching, @dynamic workflows allow you to generate the workflow structure itself at execution time. A dynamic workflow is a hybrid: it is modeled as a task, but its body is executed to produce a subworkflow.

When to use @dynamic

Use @dynamic when the number of tasks or the specific dependencies between them cannot be determined until you have the results of a previous task. A common scenario is processing a list of items where the list size is only known at runtime.

from flytekit import dynamic, task

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

@dynamic
def my_dynamic_wf(items: list[int]) -> list[int]:
    results = []
    for i in items:
        # Inside @dynamic, you can use native Python loops and logic
        # on the materialized values of inputs.
        results.append(process_item(item=i))
    return results

Key Differences

Feature	conditional	@dynamic
Evaluation Time	Execution time (by Flyte engine)	Execution time (by running the function)
Python Logic	Limited to flytekit expressions	Full Python logic (loops, if statements)
Structure	Static (all branches known at compile time)	Dynamic (graph is built at runtime)
Inputs	Operates on Promise objects	Operates on materialized values

Constraints and Performance

Because @dynamic workflows generate a new workflow graph at runtime, they incur overhead. The Flyte engine must compile the generated subworkflow before executing it.

1. Scale: Keep dynamic workflows to a reasonable number of nodes. For massive parallelism, prefer map_task.
2. Context: Inside a @dynamic function, inputs are materialized. This means you can use them in range(), as dictionary keys, or in standard Python if statements, which is impossible in a standard @workflow.
3. Return Values: A dynamic workflow must return Promise objects (outputs of tasks called within it) or simple values that flytekit can wrap. Unlike standard workflows, dynamic workflows can return a variable number of outputs.