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, they differ fundamentally in when the logic is evaluated and how the workflow graph is constructed.

Conditional Branches

Use conditional when you have a fixed set of possible execution paths that depend on the value of a task output or workflow input. In flytekit, a conditional block is compiled into a BranchNode within the workflow graph. This means all possible branches are known at compile time, but only one is executed at runtime.

Basic Usage

The conditional function in flytekit/core/condition.py is the entry point for creating these branches. It follows a functional style where the entire block returns a Promise representing the output of the selected branch.

from flytekit import task, workflow, conditional

@task
def double(n: float) -> float:
    return n * 2.0

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

@workflow
def my_workflow(val: float) -> float:
    return (
        conditional("compare-val")
        .if_(val > 10.0)
        .then(double(n=val))
        .else_()
        .then(square(n=val))
    )

Supported Expressions

Flytekit conditionals do not support standard Python logical operators like and, or, or not because these would evaluate the Promise objects immediately during compilation. Instead, you must use bitwise operators:

• & for AND
• | for OR
• Comparison operators: <, <=, >, >=, ==, !=

The Case class in flytekit/core/condition.py explicitly validates these expressions. If you pass a raw boolean or a Promise directly to if_(), flytekit raises an AssertionError.

# Valid conjunction usage within a conditional block
# .if_((val > 0.1) & (val < 1.0))

# Invalid: will raise AssertionError if used in .if_()
# .if_(val > 0.1 and val < 1.0) 

Handling Failures

You can explicitly fail a workflow branch using the .fail() method on a Case. This is useful for validating inputs or ensuring that unexpected states stop execution.

from flytekit import task, workflow, conditional

@task
def double(n: float) -> float:
    return n * 2.0

@workflow
def my_workflow(val: float) -> float:
    return (
        conditional("validate")
        .if_(val < 0.0)
        .fail("Value must be non-negative")
        .else_()
        .then(double(n=val))
    )

Internal Implementation

When you call conditional(name), flytekit creates a ConditionalSection.

• During compilation, it tracks each Case (if, elif, else) and the tasks invoked within them. When the block ends, ConditionalSection.end_branch calls to_branch_node to transform the section into a BranchNode for the Flyte backend.
• During local execution, LocalExecutedConditionalSection evaluates the expressions immediately. It uses ctx.execution_state.take_branch() to mark which path is active and skips the execution of tasks in other branches.

Dynamic Workflows

Use @dynamic when the structure of your workflow (the number of tasks or their dependencies) depends on runtime data. A dynamic workflow is a hybrid: it is modeled as a task but, when executed, it generates a new workflow graph based on its inputs.

When to use Dynamic vs. Conditional

Feature	Conditional (conditional)	Dynamic (@dynamic)
Graph Structure	Fixed at compile time.	Determined at runtime.
Logic Evaluation	Evaluated by the Flyte engine.	Evaluated in a Python container.
Python Logic	Restricted to &, `	`, and comparisons.
Inputs	Can only use Promise objects.	Can use inputs as native Python values.

Dynamic Workflow Example

The @dynamic decorator (defined in flytekit/core/dynamic_workflow_task.py) allows you to use inputs like native Python variables. For example, you can use a workflow input to determine the length of a loop:

from flytekit import task, dynamic, workflow
import typing

@task
def t1(a: int) -> str:
    return str(a)

@dynamic
def my_dynamic_subwf(a: int) -> typing.List[str]:
    s = []
    # 'a' is an int here, not a Promise, so we can use range()
    for i in range(a):
        s.append(t1(a=i))
    return s

@workflow
def wf(n: int) -> typing.List[str]:
    return my_dynamic_subwf(a=n)

Execution Semantics

A dynamic task runs in two phases:

1. Generation: The function body executes in a task container. Instead of returning values, it returns Promise objects from the tasks it invokes. Flytekit captures these invocations and compiles them into a subworkflow.
2. Execution: The Flyte engine receives the generated subworkflow and executes it.

Because the generation phase runs in a container, you have the full power of Python. However, you should keep dynamic workflows relatively small (under 500 nodes) to avoid excessive overhead during the generation and compilation phase at runtime. For massive parallelization of the same task, use map_task instead.