Conditional and dynamic workflows

Choosing between branches and dynamic workflows

Use conditional(...) when the workflow graph has a fixed set of arms and the branch predicate can be represented as a Flyte comparison or conjunction. Use @dynamic when a runtime-native input determines the workflow structure itself—for example, how many task calls a loop creates. The distinction is visible in flytekit's execution model: ordinary workflow functions run at compilation time (with local execution as an exception), while a dynamic function is modeled as a task and its body runs at execution time to generate a workflow.

A conditional therefore gives you one compiled branch node whose alternatives are known while the workflow is being built. A dynamic workflow first runs its function with native values, compiles the task calls it creates into a workflow, and returns that generated workflow to Flyte as a DynamicJobSpec. Choose the latter when range(a), runtime data, or another Python operation must determine the number or arrangement of nodes.

Conditional branches

Use the fluent conditional API

Create conditions inside a @workflow and finish the chain with an explicit .else_() arm. The source-embedded local execution example uses a Boolean input and returns the selected task result:

@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

For a compiled workflow, the same expression can compare an integer input with a literal and use task outputs as branch values. This example is embedded in workflow.py:

@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

conditional("bool") returns a Condition-backed section. .if_(...) and .elif_(...) create Case objects; each .then(...) records that arm's output. Intermediate .then(...) calls return the Condition, so you can continue with .elif_(...) or .else_(). The final .else_().then(...) returns the conditional's output promise (or a void result), which is why the result can be assigned to e or returned directly from a workflow.

Conditions are deliberately narrower than Python control flow. Case accepts only ComparisonExpression and ConjunctionExpression values. Use comparison operators (<, <=, >, >=, ==, !=) and Flyte's & or | conjunction operators; do not use Python and, or, is, or not. An already-evaluated bool, a raw Promise such as if_(x), and arbitrary expression objects are rejected by Case.

For example, the conditional module's nested example uses supported comparisons and &:

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

Here my_input, double, and square are the typed workflow values and tasks used by the module's documented example. A final arm is required: to_ifelse_block indexes the last case and requires at least two cases, so a dangling .if_(...).then(...) is not a complete conditional. Use .fail("...") on the final arm when falling through should be an error rather than a task output. In compiled execution that error becomes an Error in the model IfElseBlock; in local execution selecting it raises ValueError.

What compilation creates

conditional first inspects FlyteContextManager.current_context(). With a compilation state it constructs ConditionalSection; otherwise, with local execution state, it constructs LocalExecutedConditionalSection (or SkippedConditionalSection for an already skipped nested branch). Outside either context it raises AssertionError("Branches can only be invoked within a workflow context!").

The compile-time path is a graph transformation:

ConditionalSection.__init__ records the name, creates its Condition, and pushes a context marked as being inside a conditional section.
Each Condition._if, elif_, or else_ registers a Case. Case.then stores its Promise or tuple of promises and identifies the producing node when possible.
Intermediate arms return the Condition. When the final arm completes, ConditionalSection.end_branch pops the conditional context and calls to_branch_node.
to_ifelse_block converts the cases into model IfBlock/IfElseBlock objects. It uses the final case as else_node when it has an output, or creates an Error when that case called .fail(...).
merge_promises removes duplicate node/output references and rewrites promise variable names using create_branch_node_promise_var. end_branch turns those promises into Binding objects and upstream-node references.
A BranchNode containing the model IfElseBlock is placed in a graph Node. The returned promises point at that generated node's outputs.

The resulting shape is therefore not a Python if executed during static workflow construction:

workflow inputs/promises
        │
        ▼
  IfElseBlock in BranchNode
   ├── case task graph
   ├── elif task graph(s)
   └── else task graph or Error
        │
        ▼
 generated graph Node → conditional output promise(s)

Output compatibility across arms

ConditionalSection.compute_output_vars computes the common output-variable names across all cases. It starts with the first non-void output and intersects that set with each subsequent arm. Consequently, variables absent from one arm are discarded. If an arm has neither an output promise nor an error, or returns a VoidPromise, the whole conditional is treated as void. With no common outputs, _compute_outputs returns a VoidPromise; otherwise it creates promises backed by the generated branch node.

Keep the branch result type and output shape compatible with the workflow interface. Single promises, tuples, and NamedTuple-like results receive special handling in Case.then; for a NamedTuple output, flytekit searches its fields for a promise that identifies the producing node. A branch that produces no output can still be represented, but it cannot provide a usable value to a workflow output.

Local conditional execution

Calling the same workflow locally does not build and select a remote branch node in the same way. conditional returns LocalExecutedConditionalSection. Its start_branch evaluates c.expr.eval() for each candidate (or selects the last, expression-less case), calls ExecutionState.take_branch(), and remembers the selected Case. Each end_branch calls branch_complete; after the final arm it returns the selected arm's local values. A selected .fail(...) raises ValueError, and an unresolved selected case raises an assertion.

For a nested conditional inside an arm that evaluated false, FlyteContext.Builder.enter_conditional_section propagates BranchEvalMode.BRANCH_SKIPPED, and conditional returns SkippedConditionalSection. Local node creation checks that mode and does not execute entities in the skipped branch. The skipped section still records the cases so it can preserve the output shape; at the end it returns a VoidPromise or promises whose values are None. Do not rely on task bodies running, or on real values being available, in such nested skipped code.

The context for a conditional is manually pushed when the section starts and popped only after its final case. Because the fluent API is not a Python context manager, an exception while constructing or evaluating a conditional can leave that nested context temporarily present; FlyteContextManager.with_context contains cleanup for leaked conditional contexts.

Dynamic workflows

Use @dynamic when runtime values must be ordinary Python values while the generated graph is assembled. The decorator is defined as a partial of task.task with PythonFunctionTask.ExecutionBehavior.DYNAMIC, so the backend models the dynamic function as a task even though its body creates a workflow.

The module's documented example uses an integer directly in range and appends each generated task output:

@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

That use of range(a) is specifically supported because dynamic execution receives native input values. A normal workflow cannot use its workflow input as a native Python value while statically constructing its graph. Dynamic functions can also construct ordinary task dependencies:

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

Here t1 and t2 are the typed tasks referenced by the module documentation; the second call consumes the first call's output, so the generated workflow contains that dependency chain.

Runtime compilation and local execution

PythonFunctionTask.execute dispatches a task with ExecutionBehavior.DYNAMIC to dynamic_execute. In production task execution, dynamic_execute calls compile_into_workflow. That method creates or reuses a compilation state with a d prefix, installs ExecutionState.Mode.DYNAMIC_TASK_EXECUTION, executes the dynamic function under that context, serializes the generated PythonFunctionWorkflow, and returns a DynamicJobSpec containing its tasks, nodes, outputs, and subworkflows. If the generated workflow has no nodes, it returns a literal map of its strict outputs instead.

For local execution, dynamic_execute installs ExecutionState.Mode.LOCAL_DYNAMIC_TASK_EXECUTION, executes the generated workflow with native keyword arguments, and translates the resulting native values into a LiteralMap. This is distinct from conditional local execution: the dynamic body is run to construct and execute a generated workflow, whereas a local conditional evaluates already-built branch expressions and skips the unselected branch.

Dynamic-workflow constraints

Keep generated dynamic workflows under roughly fifty tasks; the module documentation warns that a loop can otherwise produce a workflow much larger than a manually written one and recommends map tasks for large-scale identical runs. PythonFunctionTask.compile_into_workflow raises ValueError("Reference tasks are currently unsupported within dynamic tasks") when it encounters a ReferenceTask.

If a dynamic function uses a launch plan, declare it in node_dependency_hints so it can be registered before runtime use. The source-embedded example is:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

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

The task documentation notes that launch plans must already be registered on FlyteAdmin before they can be run; tasks and workflows do not have that requirement.

Validation checklist

Create conditional(...) only inside an active workflow compilation or execution context.
Use only comparisons and Flyte conjunctions (& and |); never pass a raw Promise or a Python boolean expression produced by and/or.
Close every conditional with .else_(), using .fail(...) when the final arm should error.
Make branch outputs compatible: flytekit retains only output variables common to every arm and treats void arms as void conditionals.
Expect local nested skipped branches to yield placeholders, including None, rather than execute their tasks.
Choose @dynamic instead of conditional when runtime-native values determine task count or graph shape; keep the generated workflow small and avoid reference tasks.

Choosing between branches and dynamic workflows​

Conditional branches​

Use the fluent conditional API​

What compilation creates​

Output compatibility across arms​

Local conditional execution​

Dynamic workflows​

Runtime compilation and local execution​

Dynamic-workflow constraints​

Validation checklist​