Skip to main content

Conditional and dynamic workflows

Choosing between conditional and dynamic workflows​

Use a conditional when the set of branches and the tasks in those branches can be represented while flytekit compiles the workflow. Use a dynamic workflow when the workflow body must inspect native input values at execution time—for example, to use an input in Python's range() and create a variable number of task calls. These are separate mechanisms: conditionals are implemented in condition.py, while dynamic workflows are created by dynamic_workflow_task.py and executed by PythonFunctionTask.

Conditional workflows: a typed functional expression​

A conditional is an expression whose result is the result of the selected branch. Assign or return that result, and finish the chain with else_():

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

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

This is the example embedded in workflow.py, with the predicate expressed using the supported comparison operator. The source example uses a.is_true() (and a type-ignore comment because the workflow input is represented specially while compiling); a == True is the equivalent supported comparison form. Both branches return bool, matching the workflow's declared -> bool output.

This example is a conditional expression whose result is assigned or returned. .then(...) completes the current Case and returns the conditional chain for a subsequent .elif_() or .else_().

A conditional can also consume upstream task and workflow outputs. The following workflow.py example returns the conditional result alongside another task output:

@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

assert my_wf_example(a=1) == (6, 16)

Here a == 5 is a comparison involving a workflow input, while the branch tasks consume d and z, which are outputs of earlier nodes. The conditional therefore behaves as another node dependency when my_wf_example is compiled.

For more than two cases, continue with elif_() before the terminal else_():

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 comes from the conditional() documentation in condition.py. An inner conditional can itself be the value passed to an outer .then(...); .fail(...) is a terminal branch with no successful output.

What compilation produces​

conditional(name) inspects the active FlyteContext. If ctx.compilation_state exists, it returns a ConditionalSection; if the context has local execution state, it returns a local-execution implementation instead. Calling it without either state raises:

AssertionError("Branches can only be invoked within a workflow context!")

ConditionalSection.__init__ records the name and cases, creates its Condition fluent helper, and pushes a context produced by enter_conditional_section(). The fluent operations then follow this path:

conditional("name")
  -> Condition._if / Condition.elif_ / Condition.else_
  -> Case
  -> Case.then / Case.fail
  -> ConditionalSection.end_branch

For every non-final branch, end_branch() returns the same Condition chain, so another case can be registered. On the final branch, it pops the conditional context, calls to_branch_node(), and creates a Node whose flyte_entity is a BranchNode. BranchNode is a small wrapper containing the conditional name and the serialized _core_wf.IfElseBlock; its public name property exposes that name.

to_ifelse_block() requires at least two cases. It converts each comparison or conjunction with transform_to_boolexpr(), then builds _core_wf.IfBlock entries and an _core_wf.IfElseBlock. A final Case.then(...) supplies the else_node; a final Case.fail(...) supplies an Error instead. Thus a dangling if_() is not a compilable conditional—to_ifelse_block() raises At least an if/else is required. Dangling If is not allowed.

Promise operands become branch-node operands using the producing node and output name. transform_to_operand() creates a name in the form node_id.var and returns the original promise as a dependency. merge_promises() removes duplicate (node_id, var) dependencies and renames the retained promises with that same branch-node variable convention. ConditionalSection.end_branch() uses those dependencies to create Binding objects and attach upstream nodes to the generated Node.

Branch outputs and completion​

Every branch must provide a compatible result for the expression you assign or return. ConditionalSection.compute_output_vars() computes the intersection of promise variable names across the cases. If a branch has no output, returns a VoidPromise, or otherwise makes the conditional void, the method returns None; an empty intersection is also represented as a VoidPromise. When common variables exist, _compute_outputs() exposes promises for those variables as the conditional's result.

This common-output rule matters especially for multiple outputs and named tuples: the conditional can expose only the output variables present across the branches. Case.then() records the branch promise and, for a non-ready compiled promise, records the producing node. For a named-tuple-like result, it searches its fields for the first field backed by a node.

Close the conditional before returning it from a workflow. Workflow output binding explicitly rejects an unfinished ConditionalSection and reports that a conditional must end with an else_() clause. If you use .fail(err), compilation stores the message in the branch's serialized Error; it is not a successful typed output.

Valid predicates and rejected forms​

Use comparison operators (<, <=, >, >=, ==, and !=) or combine supported expressions with & and |:

conditional("range") \
    .if_((my_input > 0.1) & (my_input < 1.0)) \
    .then(double(n=my_input)) \
    .else_() \
    .then(square(n=my_input))

Case rejects an already evaluated Python bool, a bare Promise, and expression types other than ComparisonExpression and ConjunctionExpression. In particular, do not write Python and, or, is, or not for the predicate: those operations can evaluate immediately rather than producing the Flytekit expression objects that transform_to_boolexpr() serializes. A bare input or output promise is likewise invalid; compare it to a value or use a supported boolean-expression method such as a.is_true().

Local execution semantics​

Calling the same conditional inside local workflow execution does not construct the compiled branch node first. conditional() selects LocalExecutedConditionalSection. Its start_branch() evaluates the case expression with expr.eval() until a case is selected. The selected case calls take_branch(), and later cases are not executed. Each branch completion calls branch_complete(); after the final case, the conditional context is popped and the selected branch's local values are returned while preserving the common output interface.

For nested conditionals inside a branch that has already been skipped, the factory selects SkippedConditionalSection when the execution state's branch_eval_mode is BRANCH_SKIPPED. It still records the cases, but its final result uses None-valued promises for common outputs (or a VoidPromise) rather than executing the nested branch tasks. The surrounding node-creation logic also disallows manual node creation while a branch is skipped.

Failure has different observable behavior in the two modes. In compilation, Case.fail("message") becomes the Error on the _core_wf.IfElseBlock. In local execution, if the failed case is selected, LocalExecutedConditionalSection.end_branch() raises ValueError with that recorded error. A selected case that has neither an output nor an error raises an assertion instead.

Dynamic workflows: generate the graph at execution time​

Choose @dynamic when Python must receive native input values and use them to build the task graph. dynamic_workflow_task.py defines dynamic as a partial application of task.task with PythonFunctionTask.ExecutionBehavior.DYNAMIC:

@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

The range(a) call is specifically documented as valid for a dynamic workflow because its function body runs at execution time with native inputs. A dynamic body can also express task dependencies:

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

The implementation distinction is visible in PythonFunctionTask.compile_into_workflow(). For dynamic execution it creates an execution state with ExecutionState.Mode.DYNAMIC_TASK_EXECUTION, compiles the generated workflow under that context, serializes it, gathers its task templates and nodes, and returns a _dynamic_job.DynamicJobSpec. During local execution, dynamic_execute() uses LOCAL_DYNAMIC_TASK_EXECUTION and executes the generated workflow with the raw kwargs.

The resulting models differ from a conditional in both time and shape:

conditional:
  compile the fixed cases and predicates
  -> one BranchNode containing an IfElseBlock

@dynamic:
  execute the decorated function with native inputs
  -> compile the task calls created by that execution
  -> return a DynamicJobSpec for the generated workflow

The dynamic-workflow documentation warns that loops can generate very large workflows and recommends keeping dynamic workflows under fifty tasks; for large-scale identical runs it points to the upcoming map task. Use a conditional for a fixed, typed if/elif/else graph, and use @dynamic when execution-time Python control flow determines which task calls exist.