Conditional and dynamic workflows

Conditional and dynamic workflows in flytekit allow you to introduce logic that depends on data produced during execution. While standard workflows are compiled statically, these features provide mechanisms to handle branching and runtime graph generation.

Conditional Workflows

When you need to execute different tasks based on the output of a previous task or a workflow input, use the conditional function. Unlike standard Python if statements, which are evaluated at workflow compilation time, flytekit.conditional creates a BranchNode that the Flyte engine evaluates at runtime.

Defining Branches

The conditional function returns a ConditionalSection. You define logic by chaining .if_(), .then(), .elif_(), and .else_(). Every conditional block must conclude with either an .else_() or a .fail().

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("Value out of supported range")
    )

Compilation vs. Execution Semantics

The behavior of ConditionalSection changes depending on the context:

1. Compilation Mode: When the workflow is being compiled (e.g., for registration), ConditionalSection captures all branches into an IfElseBlock. The end_branch method in ConditionalSection computes the intersection of output variables across all branches to ensure the workflow remains type-safe.
2. Local Execution: When running locally, LocalExecutedConditionalSection evaluates the ComparisonExpression or ConjunctionExpression immediately. It uses ctx.execution_state.take_branch() to activate the selected path and short-circuits the others.
3. Nested Conditionals: If a branch is skipped during local execution (e.g., in a nested condition where the outer branch was false), SkippedConditionalSection ensures that tasks within that branch are not executed.

Constraints and Requirements

• Workflow Context Only: The conditional function checks the FlyteContextManager. If called outside a workflow, it raises an AssertionError.
• Logical Operators: Standard Python and, or, and not do not work on Flyte promises. You must use bitwise operators & (and) and | (or) for conjunctions, and methods like .is_true() for boolean promises.
• Output Consistency: All branches in a conditional must return the same type. ConditionalSection.compute_output_vars identifies the common set of outputs across all registered cases.

Dynamic Workflows

Dynamic workflows, defined with the @dynamic decorator, are used when the structure of the workflow (the number of nodes or their dependencies) depends on runtime data.

A @dynamic function is modeled as a task but behaves like a workflow. When the Flyte engine executes a dynamic task, it runs the function body to produce a new workflow graph, which is then executed as a subworkflow.

Using Runtime Data for Graph Generation

Unlike standard workflows, dynamic workflows allow you to use the values of inputs (like the length of a list) to control the generation of the graph.

from typing import List
from flytekit import task, dynamic

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

@dynamic
def my_dynamic_subwf(items: List[int]) -> List[int]:
    results = []
    for i in items:
        # In a standard @workflow, this loop would fail because 
        # 'items' is a Promise, not a literal list.
        results.append(process_item(item=i))
    return results

When to Use Dynamic vs. Conditional

• Use conditional when you have a fixed set of possible paths and the decision depends on a simple comparison of values.
• 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 to perform complex logic to determine the next steps.

Performance Considerations

Dynamic workflows introduce overhead because the Flyte engine must compile the generated subworkflow at runtime. As noted in flytekit/core/dynamic_workflow_task.py, it is recommended to keep dynamic workflows to under 50 tasks. For large-scale parallel processing of identical items, consider using map_task instead.