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 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
code_task = CodeTask(...)
workflow.add_task(code_task)
# Define conditionals
workflow.If(code_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 recommend using the .add_task() approach 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 conditional 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

While add_task() technically accepts conditionals, it’s recommended to use workflow.If() instead, as it provides cleaner syntax and better integrates with the workflow structure. The .If() method handles the conditional setup internally.

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.

Advanced Conditional Features

Conditional ID and Description

By default, conditionals are assigned auto-generated IDs. You can optionally specify a custom ID for better traceability:

workflow.If(
    code_task.out() == "expected_result",
    id="quality_check_condition"  # Optional custom ID
).Then(
    proceed_task := ActionTask(...)
).Else(
    remediate_task := ActionTask(...)
)

You can also provide a description for the conditional:

workflow.If(condition).Then(...)  # description defaults to "Conditional"

Conditional Batching

Conditionals can be assigned to specific batches when working with multi-batch workflows:

from linq.task import If

# Assign a conditional to batch 2
conditional = workflow.If(
    sensor_data.out() > 100,
    id="batch_2_check"
).Then(
    high_value_task := ActionTask(...)
).Else(
    normal_task := ActionTask(...)
)

# Manually set batch (if not already handled by SDK assignment)
conditional.batch = 2

Using ReferenceData in Conditionals

In addition to task outputs, you can reference stored reference data in conditional comparisons. This is useful for comparing workflow results against baseline data or configuration values:

from linq.task import ReferenceData

# Reference stored reference data in a conditional
workflow.If(
    analysis_result.out() > ReferenceData(key="threshold_baseline")
).Then(
    alert_task := ActionTask(...)
).Else(
    normal_processing := ActionTask(...)
)

The ReferenceData key corresponds to data stored in your system that can be referenced during workflow execution.

Multi-Output Task References

When a task produces multiple outputs, use the output_name parameter to reference a specific output:

from linq.task import TaskOutputReference

# Task with multiple outputs
analysis_task = CodeTask(...)

# Reference specific output in conditional (first output)
workflow.If(
    analysis_task.out(output_name="quality_score") > 85
).Then(
    pass_task := ActionTask(...)
).Else(
    retest_task := ActionTask(...)
)

# Alternative: use numeric index (0-based)
first_output = analysis_task.out()  # First output, output_name can be None
second_output = analysis_task.out(output_name="secondary_result")

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)