Conditional and dynamic workflows

A @workflow function in flytekit is not executed as ordinary Python — it runs once at serialization time to build a graph of nodes. That means a plain Python if inside a workflow can't work: the condition hasn't been computed yet, so Python would evaluate it immediately and bake in a single path. Flytekit solves this with two complementary mechanisms: conditional sections (conditional(...)) build a real if/else node into the workflow graph, while dynamic workflows (@dynamic) defer graph construction to task execution time and let you use native Python freely.

Conditional sections: the fluent API

conditional(name) returns a conditional section and is used in a ternary-like, functional style — the value of the expression is the output of whichever branch ran:

from flytekit import task, workflow, conditional

@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())

Locally, wf() returns True and wf(a=False) returns False. Remotely, the same code compiles into a single branch node in the graph — both branch tasks are registered, but only one executes. Note the condition: inside the workflow, a is a promise, and comparing it (or calling a promise method like a.is_true(), as flytekit's own docs do) produces a lazy expression flytekit can serialize — you cannot pass the bare promise itself.

A conditional can consume promises from earlier nodes and be returned alongside other task outputs:

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

@workflow
def simple_wf() -> int:
    return add_5(a=1)

@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

Branches chain with .elif_(...) and terminate with .else_(). .fail(message) marks the final branch as a backend failure — it doesn't raise at compile time; it records an Error for the serialized block, and during local execution the selected failure surfaces as a ValueError:

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

This nested example also shows conjunction expressions: & combines two comparisons into a ConjunctionExpression.

What expressions are allowed

Only flytekit's lazy expression objects are valid branch conditions — ComparisonExpression and ConjunctionExpression, produced by promise comparisons (<, <=, >, >=, ==, !=) and the &/| operators. The Case constructor in flytekit/core/condition.py explicitly rejects everything else:

An already-evaluated Python bool (e.g. from using and/or/not/is, which evaluate immediately) — rejected with a message that only Comparison and Conjunction expressions are supported.
A raw unary promise: conditional("x").if_(x) is rejected; compare the promise to a literal instead, or use a promise method such as a.is_true() for booleans.

# OK: comparison expression, lazily evaluated
e = conditional("c").if_(a == 5).then(x).else_().then(y)

# OK: boolean input via is_true()
e = conditional("bool").if_(a.is_true()).then(x).else_().then(y)

# NOT OK: an eagerly evaluated Python bool sneaks in through `and`
#   if a and b:
#       ...      # `a and b` evaluated immediately — never reaches the graph

# NOT OK: raw unary promise
#   conditional("x").if_(a)   # AssertionError: unary expressions not supported

.else_() is always terminal — it creates the case with last_case=True, so you cannot append another .elif_() after it.

How compilation works

Which class conditional() returns depends on the active FlyteContext:

ctx = FlyteContextManager.current_context()

if ctx.compilation_state:
    return ConditionalSection(name)
elif ctx.execution_state:
    if ctx.execution_state.is_local_execution():
        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!")

Three implementations of ConditionalSection exist in flytekit/core/condition.py, one per mode:

ConditionalSection (compilation). Calling conditional(name) pushes a conditional context onto FlyteContextManager. Each .if_()/.elif_()/.else_() calls start_branch(), which appends a Case to the section; .then(task_call) records the branch's output promise (and the producing node) and calls end_branch(). Intermediate branches return the Condition object so chaining continues. On the final case, end_branch() pops the context, calls to_branch_node(), and builds a workflow Node whose flyte_entity is a BranchNode wrapping a backend IfElseBlock. Bindings are created for the branch node's upstream promises.
LocalExecutedConditionalSection (local execution, e.g. wf() on your machine). start_branch() evaluates the expression immediately — the first case whose expr.eval() is True, or the last_case fallback — and calls ctx.execution_state.take_branch() so only the selected branch's tasks actually run. On the final case, branch_complete() is called and the selected case's output is returned.
SkippedConditionalSection (nested conditionals inside a skipped local branch). The section still records the branch shape, but returns placeholder promises with None values, so tasks inside a skipped branch are never executed locally.

Serialization from cases to the backend model happens in to_ifelse_block():

if len(cs.cases) < 2:
    raise AssertionError("At least an if/else is required. Dangling If is not allowed")

The first case becomes the if, middle cases become elif blocks, and the last case is either the else_node or — if it used .fail() — an Error on the _core_wf.IfElseBlock. Promise operands referenced inside conditions are transformed into backend operands named node_id.output_var, and merge_promises() deduplicates them so the branch node carries exactly the bindings it needs.

Output shape must be consistent

compute_output_vars() computes the intersection of output variable names across all cases — the conditional's result is only the outputs every branch shares. If any branch is void (a VoidPromise, or no output and no error), the entire conditional is treated as void. Keep branch return types aligned, or your conditional will silently lose outputs.

Rules of the road

conditional() only works inside a workflow context; elsewhere it raises the AssertionError above.
Returning an unfinished conditional from a workflow (no terminal .else_()) is rejected during output binding.
conditional() is intentionally not a context manager — a failed compile can leak a conditional context, and outer context cleanup handles the error case.
Eager workflows do not support these conditionals at all; use a plain Python if there.

Dynamic workflows: building the graph at runtime

Conditionals describe a fixed if/else structure at compile time. When the graph's shape depends on runtime values — say, N tasks where N is an input — use @dynamic instead:

@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

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

The key semantic difference: a workflow function's body runs at compilation time and cannot treat its inputs as native values, but a dynamic workflow is modeled on the backend as a task whose function body runs at execution time to produce a workflow, which is then handed back to the Flyte engine and run as a subworkflow. That's why range(a) — illegal in a @workflow — is fine here: a is a real Python int by then.

The flytekit docs for dynamic workflows carry an explicit caveat: it's rare to hand-write a workflow with thousands of nodes, but a loop can get there easily, so keep dynamic workflows under roughly fifty tasks and prefer a map task for large-scale identical runs.

Choosing between the two

	Conditional (`conditional(...)`)	Dynamic (`@dynamic`)
Graph structure	Fixed at compile time (compiled `IfElseBlock` / `BranchNode`)	Generated at execution time as a subworkflow
Branch condition	Lazy `ComparisonExpression`/`ConjunctionExpression` on promises	Any native Python logic, since inputs are real values
Loops over runtime counts	Not supported	Supported (`range(a)`)
Backend model	Branch node with `IfElseBlock`	A task that emits a subworkflow
Caveats	Needs at least `if_` + `else_`; all branches share common output vars	Keep generated workflows small (≤ ~50 tasks)

Use conditionals when the set of tasks is known and only the path varies; use @dynamic when the workflow's shape itself depends on runtime data. And if you're writing an eager workflow, neither applies — plain Python control flow already works there.

Conditional sections: the fluent API​

What expressions are allowed​

How compilation works​

Output shape must be consistent​

Rules of the road​

Dynamic workflows: building the graph at runtime​

Choosing between the two​