Conditional and dynamic workflows

Flytekit provides two primary mechanisms for introducing non-linear logic into your pipelines: Conditional Workflows and Dynamic Workflows. While both allow for branching and decision-making, they operate at different stages of the Flyte lifecycle and serve distinct purposes.

Conditional Workflows

Conditional workflows allow you to define branching logic that is evaluated by the Flyte engine at runtime based on the outputs of previous tasks. Unlike standard Python if statements, which are evaluated during workflow compilation, the conditional construct creates a BranchNode in the workflow graph.

Basic Usage

To create a conditional branch, use the conditional function from flytekit. Every conditional block must start with an if_ and end with either an else_ or a fail.

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 workflow, task, conditional

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

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

@workflow
def my_conditional_wf(should_succeed: bool) -> str:
    # should_succeed is a Promise here.
    # The .is_true() method is provided by flytekit.core.promise.Promise.
    return (
        conditional("check_result")
        .if_(should_succeed.is_true())
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Comparison and Conjunction Operators

Flytekit supports standard comparison operators (<, <=, >, >=, ==, !=) and conjunctions (& for AND, | for OR) within if_ and elif_ expressions. These operators are implemented on the Promise class in flytekit/core/promise.py.

[!WARNING] You must use & and | instead of the Python keywords and and or. Using and/or will result in a ValueError because Flytekit needs to capture the expression tree rather than evaluating it immediately.

@workflow
def complex_condition_wf(val: float) -> str:
    return (
        conditional("range_check")
        .if_((val >= 0.1) & (val < 1.0))
        .then(success_task())
        .elif_((val >= 1.0) & (val < 10.0))
        .then(success_task())
        .else_()
        .fail("Value out of supported range")
    )

Implementation Details

Internally, flytekit.core.condition.ConditionalSection manages the state of the branch.

• During compilation, it builds an IfElseBlock which is then wrapped in a BranchNode.
• During local execution, LocalExecutedConditionalSection eagerly evaluates the expressions. If a branch is taken, it calls ctx.execution_state.take_branch() to ensure only the relevant tasks are executed locally.

The Case.then() method is responsible for capturing the output Promise of the branch. Flytekit attempts to merge these outputs using compute_output_vars(), ensuring that the conditional block returns a consistent interface regardless of which branch is executed.

Dynamic Workflows

Dynamic workflows are used when the structure of the workflow (the number of tasks or their dependencies) depends on data that is only available at runtime. A dynamic workflow is defined using the @dynamic decorator.

When to use Dynamic Workflows

Use @dynamic when you need to:

1. Iterate over a dynamic list: For example, running a task for every file found in an S3 bucket.
2. Use runtime values in Python control flow: Unlike standard workflows, dynamic workflows allow you to use inputs in native Python for loops and if statements because the function body is executed 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:
        # items is a realized list here, not a Promise
        results.append(process_item(item=i))
    return results

Execution Semantics

A @dynamic function is modeled as a Task during the initial workflow compilation. However, when the Flyte engine executes this task, it runs the function body to produce a new workflow graph (a subworkflow) based on the actual input data. This subworkflow is then compiled and executed by the engine.

Comparison: Conditional vs. Dynamic

Feature	Conditional (conditional)	Dynamic (@dynamic)
Evaluation Time	Evaluated by the Flyte engine at runtime.	Evaluated by running Python code at runtime to generate a graph.
Graph Structure	Fixed at compile time (all branches are known).	Determined at runtime (nodes are generated dynamically).
Python Control Flow	Not allowed (must use .if_().then()).	Allowed (standard if, for, while).
Overhead	Low (simple branch evaluation).	Higher (requires running a task to generate the subworkflow).
Use Case	Simple branching based on task outputs.	Complex logic, variable number of tasks, or data-dependent graphs.

Nested Conditionals

Flytekit supports nesting conditional blocks. When a branch is not taken during local execution, SkippedConditionalSection ensures that nested tasks are not triggered, preventing unnecessary local execution of skipped branches.

v = (
    conditional("outer")
    .if_(my_input > 0.1)
    .then(
        conditional("inner")
        .if_(my_input < 0.5)
        .then(task_a(n=my_input))
        .else_()
        .then(task_b(n=my_input))
    )
    .else_()
    .then(task_c(n=my_input))
)