Conditional and dynamic workflows

Flytekit provides two primary mechanisms for controlling execution flow: Conditional Workflows for static, ahead-of-time compiled branching, and Dynamic Workflows for runtime-determined graph structures.

Conditional Workflows

Conditional workflows allow you to define branching logic that is evaluated by the Flyte engine at execution time. Unlike standard Python if statements, which are evaluated during workflow compilation, Flytekit conditionals are serialized into the workflow graph as a BranchNode.

The Fluent API

You create a conditional using the conditional(name) factory. This returns a ConditionalSection that supports a fluent interface for building if, elif, and else branches.

from flytekit import task, workflow, conditional

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

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

@workflow
def my_conditional_wf(val: int) -> str:
    return (
        conditional("value_check")
        .if_(val > 10)
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Expression Constraints

Flytekit conditionals do not support arbitrary Python booleans. Because the logic must be serialized, you must use Flyte-specific ComparisonExpression or ConjunctionExpression objects. These are created by using standard comparison operators (<, <=, >, >=, ==, !=) and bitwise conjunctions (& for AND, | for OR) on Flyte promises.

The Case class (flytekit/core/condition.py) explicitly rejects:

• Raw Python bool values.
• Bare Promise objects (e.g., if_(my_promise)).
• Python logical operators and, or, not, and is.

Compilation and Serialization

When a workflow is compiled, the ConditionalSection tracks every branch. The to_ifelse_block helper converts these into a core IfElseBlock model.

1. Variable Mapping: Promises used in expressions are mapped to unique IDs using create_branch_node_promise_var(node_id, var). This generates a name like node1.o0 to prevent collisions between different nodes that might share output names.
2. Node Creation: The entire conditional block is encapsulated in a single workflow Node with flyte_entity set to a BranchNode.
3. Output Merging: The compute_output_vars method calculates the intersection of output variables across all branches. Every branch must return the same set of output variables (same types and names) for the conditional to have a valid return value.

Local Execution Semantics

During local execution, Flytekit uses LocalExecutedConditionalSection. Instead of serializing the graph, it evaluates c.expr.eval() for each case.

• It selects the first branch that evaluates to True.
• It calls ctx.execution_state.take_branch() to mark the active path.
• If a branch is skipped (e.g., in nested conditionals where the parent branch was false), SkippedConditionalSection ensures that tasks within that branch are not executed, returning None placeholders instead.

Dynamic Workflows

Dynamic workflows are used when the structure of the workflow depends on runtime data that is not available at compilation time, such as the length of an input list.

The @dynamic Decorator

A dynamic workflow is defined using the @dynamic decorator. Internally, this is a PythonFunctionTask with ExecutionBehavior.DYNAMIC.

from typing import List
from flytekit import task, dynamic

@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:
        # Native Python control flow is allowed here
        results.append(process_item(item=i))
    return results

Execution Lifecycle

Unlike a standard @workflow, the body of a @dynamic function is not compiled into a static graph ahead of time. Instead:

1. The dynamic task is executed like a normal task.
2. The function body runs, using native Python values (e.g., you can iterate over a list using for i in items).
3. The PythonFunctionTask.compile_into_workflow() method captures the resulting nodes and returns a DynamicJobSpec.
4. Flyte treats this spec as a subworkflow and executes it.

Operational Considerations

• Scale: Dynamic workflows are powerful but should be used judiciously. The Flytekit source recommends keeping dynamic graphs under 50 tasks. For larger, identical workloads, use Map Tasks.
• Dependency Hints: Because dynamic workflows generate their graph at runtime, static discovery cannot always see dependencies like Launch Plans. Use the node_dependency_hints parameter in the @dynamic decorator to explicitly declare these:

@dynamic(node_dependency_hints=[my_launch_plan])
def dynamic_subwf():
    return [my_launch_plan(val=i) for i in range(10)]

• Constraints: Reference tasks are currently unsupported inside dynamic tasks. compile_into_workflow will raise a ValueError if it encounters a ReferenceTask.

Comparison Summary

Feature	Conditional Workflow (conditional)	Dynamic Workflow (@dynamic)
Evaluation Time	Execution time (by Flyte Engine)	Execution time (by Task Runner)
Graph Structure	Static / Fixed at compile time	Dynamic / Generated at runtime
Control Flow	Fluent API (if_, elif_, else_)	Native Python (if, for, while)
Input Usage	Promises (cannot be iterated)	Native values (can be iterated)
Use Case	Simple branching based on task outputs	Variable-length loops or complex runtime logic