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 have distinct constraints.

Conditional Workflows

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

Basic Syntax

To create a conditional branch, use the conditional function from flytekit. It follows a fluent API pattern: .if_().then().elif_().then().else_().then().

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")
    )

Implementation Details

When you call conditional("name"), flytekit initializes a ConditionalSection. Internally, this pushes a new context via FlyteContextManager.push_context(ctx.enter_conditional_section().build()).

• Compilation: During workflow compilation, flytekit records every branch. The end_branch method in ConditionalSection eventually calls to_branch_node, which converts the section into a BranchNode containing an IfElseBlock.
• Local Execution: When running locally, flytekit uses LocalExecutedConditionalSection. It evaluates the expressions immediately using c.expr.eval() and uses ctx.execution_state.take_branch() to execute only the selected path.
• Output Consistency: The compute_output_vars method ensures that all branches return compatible types. The output of the entire conditional block is a Promise representing the union of the outputs from all possible branches.

Constraints and Gotchas

1. Workflow Context Only: The conditional function can only be used inside a @workflow. Using it elsewhere will raise an AssertionError.
2. No Unary Expressions: You cannot use if_(my_promise). You must use explicit comparisons like if_(my_promise == True) or if_(my_promise.is_true()).
3. Bitwise Operators: Use & (AND) and | (OR) for compound conditions. Standard Python and/or keywords will not work because they attempt to eagerly evaluate the Promise objects.
4. Mandatory Else/Fail: Every conditional block must terminate with either an .else_().then(...) or an .else_().fail("message"). Dangling if statements are not supported.

Dynamic Workflows

Dynamic workflows are defined using the @dynamic decorator. A dynamic workflow is technically a task that, when executed, returns a new workflow graph (a subworkflow) to the Flyte engine.

When to use Dynamic Workflows

Use @dynamic when the structure of your workflow depends on the value of a task's output. For example, if you need to run a task for every file in a directory, but the number of files is only known at runtime.

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 = []
    for i in range(a):
        # In a @dynamic task, you can use native Python logic like loops
        # and range() on inputs, which is forbidden in a @workflow.
        s.append(t1(a=i))
    return s

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

How it Works Internally

The @dynamic decorator is a partial application of the @task decorator with execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC.

1. Task Execution: The Flyte engine runs the dynamic function as a normal task.
2. Graph Generation: Instead of returning data, the function body runs and produces a set of task executions (nodes).
3. Subworkflow Submission: Flytekit captures these nodes and returns a WorkflowTemplate to the engine.
4. Engine Resumption: The Flyte engine then executes this generated subworkflow before continuing with the rest of the parent workflow.

Comparison: Conditional vs. Dynamic

Feature	Conditional (conditional)	Dynamic (@dynamic)
Evaluation Time	Evaluated by Flyte Propeller at runtime.	Evaluated by a worker node during task execution.
Graph Structure	Fixed at compile time; all branches are visible in the UI.	Generated at runtime; the graph expands dynamically.
Python Logic	Limited to comparison/conjunction on Promises.	Full Python power (loops, recursion, native types).
Overhead	Very low; just a branch node in the graph.	Higher; requires starting a task to generate the subworkflow.

Use Conditional Workflows for simple branching based on task outputs. Use Dynamic Workflows when you need to programmatically generate a complex graph based on runtime data.