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 (when the workflow graph is built), Flytekit's conditional constructs are preserved as BranchNode entities in the workflow graph.

Using the conditional Function

When you need to choose between different tasks based on the output of a previous task or a workflow input, use the conditional function from flytekit.core.condition.

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("check_value")
        .if_(val > 10)
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Expression Constraints

The expressions used in .if_() and .elif_() must be built using Flyte-compatible operators. Standard Python logical operators like and, or, and not will not work because they immediately evaluate to a boolean, whereas Flyte needs to capture the expression itself.

• Comparisons: Use standard operators like ==, !=, <, <=, >, >=.
• Conjunctions: Use & (AND) and | (OR) for combining expressions.
• Booleans: For boolean inputs or task outputs, which are represented as Promise objects within a workflow, use the .is_true() or .is_false() methods.

from flytekit import task, workflow, conditional

@task
def is_even(val: int) -> bool:
    return val % 2 == 0

@workflow
def complex_condition_wf(a: int) -> str:
    even_promise = is_even(val=a)
    return (
        conditional("complex_check")
        .if_((a < 10) & (even_promise.is_true()))
        .then(success_task())
        .elif_(a == 0)
        .then(success_task())
        .else_()
        .fail("Condition not met")
    )

Internal Implementation

When you call conditional("name"), Flytekit creates a ConditionalSection. As you chain .if_(), .then(), and .else_(), it populates a list of Case objects.

• Compilation: In ConditionalSection.end_branch, Flytekit converts the accumulated cases into a BranchNode containing an IfElseBlock. This node is then added to the workflow's compilation_state.
• Local Execution: During local runs, Flytekit uses LocalExecutedConditionalSection. It evaluates the expressions against local values and uses ctx.execution_state.take_branch() to execute only the selected path, mimicking the backend's behavior.

Dynamic Workflows

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

The @dynamic Decorator

A dynamic workflow is essentially a hybrid: it is modeled as a task in the parent workflow, but when it executes, it generates a workflow graph based on its inputs.

import typing
from flytekit import task, dynamic

@task
def process_item(item: int) -> int:
    return item * 2

@dynamic
def my_dynamic_wf(n: int) -> typing.List[int]:
    results = []
    for i in range(n):
        # In a @dynamic task, you can use native Python logic like loops
        # and access the value of 'n' directly.
        results.append(process_item(item=i))
    return results

Key Differences from Standard Workflows

Feature	@workflow	@dynamic
Evaluation Time	Compilation time (before execution)	Runtime (during task execution)
Python Logic	Limited to Flyte constructs (e.g., conditional)	Full Python support (loops, if statements)
Input Access	Inputs are Promise objects (cannot be iterated)	Inputs are materialized values
Graph Structure	Static and fixed	Dynamic and data-dependent

When to Use Which

1. Use conditional when you have a fixed set of possible paths and the decision depends on a simple comparison of task outputs. It is more efficient because the Flyte engine understands the entire branching structure upfront.
2. Use @dynamic when the number of tasks to run is unknown until runtime (e.g., processing every file in a directory) or when you need complex Python logic to determine the workflow structure.

Implementation Detail: ExecutionBehavior

The @dynamic decorator is a partial application of the standard @task decorator with a specific execution mode defined in flytekit.core.dynamic_workflow_task:

# From flytekit/core/dynamic_workflow_task.py
dynamic = functools.partial(task.task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)

When the Flyte engine executes a dynamic task, it expects the task to return a compiled workflow closure. Flytekit handles this by intercepting the execution, running the function to build a local workflow, and then serializing that workflow back to the engine to be executed as a subworkflow.