Workflow Branching

Overview

Workflow branching refers to a workflow that splits into two or more parallel or alternative paths based on certain runtime conditions or decisions.

This section covers the implementation of conditional logic and handling non-deterministic outcomes for workflow branching, including adapting workflows based on dynamic conditions and results.

Conditionals

A conditional is a SDK construct that allows a workflow to execute different tasks based on whether a specified runtime condition is met.

Conditionals make workflows dynamic at runtime using If/ElseIf/Else blocks. The initial plan assumes an expected outcome which is the first outcome of the If block, but if the actual outcome differs, the scheduler generates a new plan to match the result.

from linq.task import (
    ActionTask,
    CodeTask,
    ConditionalSourceTask,
    If,
    LabwareSource,
)

# Define your workflow
workflow = Workflow(...)
# Define workflow tasks
workflow.add_task(CodeTask(...))
# Define conditionals
workflow.If(task.out() == True, id="condition_1").Then(
    do_something := ActionTask(...)
).Else(
    do_something_else := ActionTask(...)

)
# Define tasks that follow conditionals
workflow.add_task(
    ActionTask(
        ...
        labware_sources=[
            LabwareSource(
                ...
                source_task=ConditionalSourceTask(tasks=[do_something, do_something_else]
                ),
            )
        ],
    )
)

Tip

Using conditionals requires tasks to be added to the workflow object with the .If method as opposed to defining tasks in a task list. We recommed using the .add_task() apporach throughout your workflow.

Tip

Make sure the first outcome in your conditional is the expected outcome to minimise replanning.

The following comparison operators are available:

== and “!=
<, <=, > and >=
in, not in

"in" and "not in" can also be expressed as .equals_any() or .equals_none().

Examples:

workflow.If(roll_the_dice.out() > 3)

workflow.If(roll_the_dice.out() != 5)

workflow.If(roll_the_dice.out().equals_any([1, 6]))
# Is the same as:
workflow.If(roll_the_dice.out() in [1, 6])

The left hand side on a condtional comparison always has to reference the output of a task or a piece of reference data. Static data can only be used on the right-hand side of the comparison.

Multiple comparisons can be chained in a conditional using the ElseIf block. No comparison is needed when using the Else block.

workflow.If(roll_the_dice.out().equals_any([1, 6]))
.Then(
    ....
)
.ElseIf(roll_the_dice.out() == 2)
.Then(
    ...
)
.Else(
    ...
)

Attention

.add_task() does not support conditonals, do not do workflow.add_task(IF(.....)). Instead define the condtional directly on the workflow class as seen above.

Multiple comparisons can not be chained within the same block. Operators like AND and OR are not supported. To achieve this behaviour a nested conditional should be used.

workflow.If(roll_the_dice.out().equals_any([1, 6]))
.Then(
    If(coin_flip.out() == 'Tails')
    .Then(
        ....
    )
)

A ConditionalSourceTask is used to specify that labware comes from a conditional source task when defining labware sources. These are necessary because when a source task is a conditional, the actual outcome is not known before runtime, therefore you need to provide all the options that may evaluate to true.

workflow.add_task(
    ActionTask(
        ...
        labware_sources=[
            LabwareSource(
                labware=qc_plate,
                source_task=ConditionalSourceTask(
                    tasks=[
                        conditional_If_result,
                        conditional_ElseIf_result,
                        conditional_Else_result
                    ]
                ),
            )
        ],
    )
)

Attention

Using .forward() on conditional tasks is not supported yet.

For a full implementation tutorial for conditionals, see the Conditionals Tutorial.

class linq.task.If(*, id: str = ..., is_expected: bool = True, comparison: linq.task.Comparison | Literal['Else'], description: str = 'Conditional', next: 'Conditional | None' = None, _prev: 'Conditional | None' = None)