Conditional and dynamic workflows

Conditional logic and dynamic execution allow Flyte workflows to adapt to data at runtime. While both enable branching, they operate at different stages of the workflow lifecycle: conditional blocks are evaluated by the Flyte engine (Propeller) during workflow execution, whereas @dynamic workflows generate a new execution graph on the fly based on input data.

Conditional Workflows

Use conditional when you need to choose between a fixed set of predefined tasks or subworkflows based on the output of a previous task. Because the Flyte engine sees all possible branches at compile time, it can provide full visibility into the potential execution paths.

Defining Branches

The conditional function in flytekit.core.condition is the entry point for building these blocks. It uses a fluent API to define if_, elif_, and else_ branches. Each branch must terminate with either .then() to return a value or .fail() to abort the execution.

from flytekit import task, workflow, conditional

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

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

@workflow
def my_workflow(my_input: float) -> float:
    return (
        conditional("fractions")
        .if_((my_input > 0.1) & (my_input < 1.0))
        .then(double(n=my_input))
        .elif_((my_input >= 1.0) & (my_input < 10.0))
        .then(square(n=my_input))
        .else_()
        .fail("Input out of range")
    )

Expression Constraints

Expressions in conditional blocks are not standard Python booleans. They are ComparisonExpression or ConjunctionExpression objects that Flyte evaluates at runtime.

1. Bitwise Operators: Use & (AND) and | (OR) instead of and and or.
2. Boolean Promises: You cannot use a boolean Promise directly in an if_ statement. Use the .is_true() or .is_false() methods.
3. No Unary Expressions: if_(x) is invalid if x is a promise; it must be a comparison like if_(x == True).

# Correct usage for boolean inputs/outputs
return conditional("bool_check").if_(a.is_true()).then(t1()).else_().then(t2())

Internal Implementation

When a workflow is compiled, ConditionalSection (found in flytekit/core/condition.py) captures the branches into a BranchNode.

• Compilation: ConditionalSection.end_branch() computes the intersection of output variables across all branches using compute_output_vars(). This ensures that regardless of which branch is taken, the workflow maintains a consistent interface.
• Local Execution: LocalExecutedConditionalSection eagerly evaluates the expressions. It uses ctx.execution_state.take_branch() to track which path was followed during a local Python run.

Dynamic Workflows

Use the @dynamic decorator when the structure of your workflow (the number of tasks or their dependencies) depends on runtime data, such as processing a variable number of files discovered by a previous task.

A dynamic workflow is a hybrid: it is modeled as a task in the parent workflow but behaves like a workflow when executed.

from flytekit import task, dynamic, workflow
from typing import List

@task
def process_sample(s: int) -> int:
    return s * 2

@dynamic
def my_dynamic_subwf(a: int) -> List[int]:
    s = []
    # Unlike standard workflows, you can use Python iterables and range() here
    for i in range(a):
        s.append(process_sample(s=i))
    return s

@workflow
def parent_wf(n: int) -> List[int]:
    return my_dynamic_subwf(a=n)

Key Differences from Standard Workflows

1. Input Materialization: In a standard @workflow, inputs are Promise objects and cannot be used in Python control flow (like if x > 5: or for i in range(x):). In a @dynamic task, inputs are materialized into actual Python values, allowing full use of Python's native syntax.
2. Graph Generation: When the Flyte engine executes a dynamic task, it runs the function body to produce a WorkflowTemplate. This template is then submitted back to the engine to be executed as a subworkflow.
3. Performance: Because dynamic workflows require the engine to compile a new graph at runtime, they incur more overhead than conditional blocks. Flytekit recommends keeping dynamic workflows to under 50 tasks to avoid excessive pressure on the scheduler.

Comparison Summary

Feature	conditional	@dynamic
Evaluation Time	Runtime (by Flyte Propeller)	Runtime (by executing Python code)
Visibility	All branches visible in UI before execution	Graph is only visible after the dynamic task runs
Python Control Flow	Restricted (must use conditional API)	Full (native if, for, while allowed)
Use Case	Simple branching between known tasks	Variable number of tasks or complex logic