Skip to main content

Conditional and dynamic workflows

When you need a workflow to take different actions depending on the value of an input or intermediate result, Flyte offers two complementary mechanisms: conditional() branches for compile-time branching on primitive comparisons, and @dynamic workflows for runtime workflow generation with full Python control flow. They solve different problems and have different semantics — picking the wrong one leads to errors or workflows that can't compile.

Conditional Branches​

Basic pattern​

Inside a @workflow function, call conditional(name) to build a ternary-style if/else chain that returns the output of the taken branch. The shape is:

conditional("name").if_(expr).then(...).elif_(expr).then(...).else_().then(...)

Each .if_()/.elif_() takes a comparison or conjunction expression; .then() takes the output of a task call (or another conditional). .else_() is required — a dangling if without an else raises AssertionError('At least an if/else is required. Dangling If is not allowed') from to_ifelse_block. Instead of .then(...), the else or any branch can call .fail(msg) to raise an error.

Here is a real example from flytekit/core/workflow.py:

@task
def add_5(a: int) -> int:
    a = a + 5
    return a

@workflow
def my_wf_example(a: int) -> typing.Tuple[int, int]:
    x = add_5(a=a)
    z = add_5(a=x)
    d = simple_wf()
    e = conditional("bool").if_(a == 5).then(add_5(a=d)).else_().then(add_5(a=z))
    return x, e

The variable e holds the output promise of whichever branch is taken. Because the conditional is a functional expression, you can assign its result and feed it to downstream tasks or return it from the workflow.

Building condition expressions​

You never construct ComparisonExpression or ConjunctionExpression directly. Flytekit's Promise class overrides the comparison dunders (__eq__, __ne__, __gt__, __ge__, __lt__, __le__) so that applying Python comparison operators to a Promise — such as a workflow input or a task output — returns a ComparisonExpression object rather than a bool:

# from flytekit/core/promise.py
def __eq__(self, other) -> ComparisonExpression:
    return ComparisonExpression(self, ComparisonOps.EQ, other)

def __gt__(self, other) -> ComparisonExpression:
    return ComparisonExpression(self, ComparisonOps.GT, other)

For boolean promises, use the helper methods is_true(), is_false(), is_none(), or is_(v). These are defined on Promise in flytekit/core/promise.py and return ComparisonExpression objects:

# from flytekit/core/promise.py
def is_(self, v: bool) -> ComparisonExpression:
    return ComparisonExpression(self, ComparisonOps.EQ, v)

def is_true(self) -> ComparisonExpression:
    return self.is_(True)

def is_false(self) -> ComparisonExpression:
    return self.is_(False)

A workflow that branches on a boolean input looks like this (note that inside a workflow, inputs are Promise objects at runtime despite being typed as bool — hence the # type: ignore):

@task
def t() -> bool:
    return True

@task
def f() -> bool:
    return False

@workflow
def wf(a: bool = True) -> bool:
    return conditional("bool").if_(a == True).then(t()).else_().then(f())  # type: ignore

assert wf() is True
assert wf(a=False) is False

In the source, this example uses a.is_true() — which is equivalent to a == True — as a more readable shorthand for boolean comparisons.

Combine comparisons with & (AND) and | (OR). The ComparisonExpression.__and__ and __or__ methods return ConjunctionExpression objects:

# (my_input > 0.1) & (my_input < 1.0)  →  ConjunctionExpression(AND)
# (my_input > 0.5) | (my_input < 0.2)   →  ConjunctionExpression(OR)

Critical gotcha: use & / |, not and / or​

Python's and and or keywords invoke __bool__ on the operands. Flytekit deliberately raises ValueError from __bool__ on both ComparisonExpression and ConjunctionExpression to prevent silent evaluation:

# from flytekit/core/promise.py
def __bool__(self):
    raise ValueError(
        "Cannot perform truth value testing,"
        " This is a limitation in python. For Logical `and\or` use `&\|` (bitwise) instead."
        f" Expr {self}"
    )

This is a Python language limitation (PEP-335) — and/or cannot be overridden to return objects. Write (a > 0.1) & (a < 1.0), never (a > 0.1 and a < 1.0).

No unary boolean conditions​

You cannot pass a bare Promise to if_(). The Case constructor explicitly rejects it:

if isinstance(expr, Promise):
    raise AssertionError(
        "Flytekit does not support unary expressions of the form `if_(x) - where x is an"
        " input value or output of a previous node."
        f" Received var {expr} in condition {cs.name}.{stmt}"
    )

Use a == True or a.is_true() instead of if_(a).

Only primitive values can be compared​

ComparisonExpression.__init__ raises ValueError('Only primitive values can be used in comparison') if a ready Promise contains a non-primitive scalar. Comparisons are limited to integers, floats, strings, and booleans.

Nested conditionals and fail()​

Conditionals can nest and can terminate a branch with .fail(msg) instead of .then(...). This is the canonical example from the conditional() docstring in flytekit/core/condition.py:

v = (
    conditional("fractions")
    .if_((my_input > 0.1) & (my_input < 1.0))
    .then(
        conditional("inner_fractions")
        .if_(my_input < 0.5)
        .then(double(n=my_input))
        .elif_((my_input > 0.5) & (my_input < 0.7))
        .then(square(n=my_input))
        .else_()
        .fail("Only <0.7 allowed")
    )
    .elif_((my_input > 1.0) & (my_input < 10.0))
    .then(square(n=my_input))
    .else_()
    .then(double(n=my_input))
)

The inner conditional("inner_fractions") is itself the argument to the outer .then(...), so the output of the selected inner branch becomes the output of the outer case.

Output compatibility across branches​

All branches must return compatible outputs. ConditionalSection.compute_output_vars() computes the intersection of output variable names across every registered Case. If any case returns None or a VoidPromise (e.g., a task with no outputs), the entire conditional becomes void — compute_output_vars returns None. If branches return mismatched output names, the intersection shrinks silently to an empty set, also producing a VoidPromise. Design all branches to return the same output names and types.

Compilation vs. Execution Semantics​

The single conditional() function returns one of three ConditionalSection subclasses depending on the current FlyteContext:

# from flytekit/core/condition.py
def conditional(name: str) -> ConditionalSection:
    ctx = FlyteContextManager.current_context()
    if ctx.compilation_state:
        return ConditionalSection(name)
    elif ctx.execution_state:
        if ctx.execution_state.is_local_execution():
            from flytekit.core.context_manager import BranchEvalMode
            if ctx.execution_state.branch_eval_mode == BranchEvalMode.BRANCH_SKIPPED:
                return SkippedConditionalSection(name)
            return LocalExecutedConditionalSection(name)
    raise AssertionError("Branches can only be invoked within a workflow context!")

Compilation mode (registration time)​

When ctx.compilation_state is set — during PythonFunctionWorkflow.compile() — conditional() returns a base ConditionalSection. This class does not evaluate expressions. Every if_(), elif_(), and else_() registers a Case via start_branch. After each .then() or .fail(), end_branch() is called; on the last case it pops the conditional context, builds a BranchNode wrapping an IfElseBlock (via to_branch_node), wraps it in a Node, and adds it to compilation_state.nodes. The compiled BranchNode is what gets serialized into the Flyte workflow spec.

Local execution mode​

During local workflow execution, LocalExecutedConditionalSection takes over. Its start_branch actually calls c.expr.eval() — which in turn calls Promise.eval() and ComparisonExpression.eval() to resolve primitive values — and marks the first matching branch as active:

# from flytekit/core/condition.py
def start_branch(self, c: Case, last_case: bool = False) -> Case:
    added_case = super().start_branch(c, last_case)
    ctx = FlyteContextManager.current_context()
    if self._selected_case is None:
        if c.expr is None or c.expr.eval() or last_case:
            ctx.execution_state.take_branch()
            self._selected_case = added_case
    return added_case

ConjunctionExpression.eval() short-circuits: for AND it returns False as soon as the left side is false; for OR it returns True as soon as the left side is true.

Branch selection is controlled by BranchEvalMode (defined in flytekit/core/context_manager.py): BRANCH_ACTIVE means the next .then() body should run; BRANCH_SKIPPED means it should not. take_branch() sets the mode to BRANCH_ACTIVE; branch_complete() sets it to BRANCH_SKIPPED. Other parts of flytekit — node_creation.py, promise.py, and reference_entity.py — check this mode to decide whether to execute or skip node creation inside conditional branches.

Skipped nested conditionals​

When local execution is in BRANCH_SKIPPED mode — meaning the parent branch was not taken — conditional() returns SkippedConditionalSection. Its end_branch never evaluates anything. On the last case it returns either a VoidPromise or Promise objects with val=None as placeholders, so downstream code in the skipped branch sees dummy promises without erroring.

Context stack management​

Every ConditionalSection.__init__ pushes a new conditional context onto the stack via FlyteContextManager.push_context(ctx.enter_conditional_section().build()). Because conditionals use a fluent method-chaining API rather than a with block (for ergonomics), a failed conditional can leak context. FlyteContextManager.with_context includes a cleanup loop that pops leaked contexts to handle this trade-off.

Dynamic Workflows​

When conditional is not enough​

Conditionals only branch on primitive comparisons known at compile time — you cannot write for i in range(a) or if a > 5 with native Python control flow in a regular @workflow, because workflow functions run at compilation time and their inputs are Promise objects, not resolved Python values.

@dynamic solves this. It is defined as a partial of the @task decorator with execution_mode=DYNAMIC:

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

A dynamic task is modeled on the backend as a task, but at execution time its function body runs to produce a workflow. The resulting DynamicJobSpec is passed back to the Flyte engine and run as a subworkflow. Crucially, the function body receives its inputs as native Python values, so you can loop, branch, and index on them:

@dynamic
def my_dynamic_subwf(a: int) -> (typing.List[str], int):
    s = []
    for i in range(a):
        s.append(t1(a=i))
    return s, 5

Calling range(a) on a workflow input is impossible in a @workflow but valid in a @dynamic because the function executes at runtime when a is a concrete integer.

Task-to-task dependencies​

Dynamic workflows can express dependencies between tasks the same way regular workflows do:

@dynamic
def my_dynamic_subwf(a: int, b: int) -> int:
    x = t1(a=a)
    return t2(b=b, x=x)

Execution modes​

PythonFunctionTask.ExecutionBehavior is an enum with DEFAULT = 1 and DYNAMIC = 2 (in flytekit/core/python_function_task.py). At runtime, pre_execute compiles the function body into a PythonFunctionWorkflow and serializes it into a DynamicJobSpec. In real remote execution (DYNAMIC_TASK_EXECUTION), the task produces the spec and stops — the Flyte engine runs the subworkflow. In local execution (LOCAL_DYNAMIC_TASK_EXECUTION), flytekit actually executes the compiled subworkflow inline so you get real outputs.

Gotchas and constraints​

Keep dynamic workflows small. The module docstring warns: "Please keep dynamic workflows to under fifty tasks. For large-scale identical runs, we recommend the upcoming map task." Every node in the compiled subworkflow is processed like any other workflow node, so a loop producing thousands of tasks creates a massive spec.

Reference tasks are not supported inside dynamic tasks — pre_execute raises ValueError('Reference tasks are currently unsupported within dynamic tasks').

If your dynamic task references launch plans, those launch plans must be pre-registered on FlyteAdmin. Pass them via node_dependency_hints so registration picks them up:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    return [launchplan0] * 10

Conditionals vs. dynamic workflows​

The two features address different problems:

Aspectconditional()@dynamic
When the body runsCompilation time (branch structure is static)Execution time (workflow is generated per run)
Control flowBranch on primitive comparisons onlyFull Python control flow — loops, conditionals, indexing
Input accessInputs are Promise objects; cannot use range() or native ifInputs are native Python values
Backend modelBranchNode with IfElseBlock in the workflow specA task that emits a DynamicJobSpec subworkflow
ScaleAny number of branches, each a normal nodeKeep under ~50 tasks

Use conditional() when the set of possible branches is fixed and depends only on primitive comparisons. Use @dynamic when the workflow structure itself varies per run — for example, the number of tasks depends on an input value, or the branching logic requires native Python operations on resolved data.