Workflow: Orchestrating Agents with Flexibility

Workflow is a powerful system in Synthora that can be used to orchestrate agents to solve various problems, allowing users to define their own workflows with a high level of flexibility.

In this tutorial, we will explore how to use workflow to manage and automate tasks efficiently. We will start with basic concepts and gradually move to more advanced features. Thanks to the flexibility of workflow, the overall process is simple and can be easily customized.

Now, if you are ready, let’s start!

Prerequisites

You are probably excited to start already, but before that, there are some prerequisites you need to prepare. (Sorry!)

Install Synthora

Synthora requires Python 3.8+. You can install Synthora via pip:

!pip install synthora

from synthora.workflows import task
from synthora.workflows.base_task import BaseTask
from synthora.workflows.context.base import BaseContext
from synthora.workflows.scheduler import ThreadPoolScheduler
from synthora.workflows.scheduler.base import BaseScheduler
from synthora.workflows.scheduler.process_pool import ProcessPoolScheduler

Components of Workflow

A workflow in Synthora is composed of three main components: scheduler, task, and context.

• Task: The basic execution unit, usually a function.

• Scheduler: Invokes tasks either in parallel or in series according to specific strategies.

• Context: Allows tasks to read and store information during execution, and supports advanced features like loops and conditional statements.

Now that we understand the components of a workflow, let’s dive into how to initialize a task. There are two ways to do this:

1. Using decorators

2. Using classes.

@task
def add1(x: int, y: int) -> int:
    return x + y


@task(name="add2")
def add2(x: int, y: int) -> int:
    return x + y


def _add3(x: int, y: int) -> int:
    return x + y


add3 = BaseTask(_add3, name="add3")

Task Signature

Task signatures allow you to predefine task parameters, enabling users to implement more complex functionalities.

By using task signatures, you can set default values for the parameters of a task, making it easier to reuse tasks with predefined configurations. This is particularly useful when you need to run the same task multiple times with the same parameters or when you want to create more complex workflows by chaining tasks together.

add1 = add1.s(1)
# or
# add1 = add1.signature(1)

add1(2)  # will return 3

3

Task signatures also support immutable parameters, which ensure that the task ignores any external parameters passed to it during execution. This feature helps maintain consistency and predictability in task execution, especially in complex workflows.

try:
    add1(1, 2)  # will raise an error
except TypeError as e:
    print(e)

add1() takes 2 positional arguments but 3 were given

add2 = add2.si(1, 2)
# or
# add2 = add2.signature(1, 2, immutable=True)
# or
# add2 = add2.signature_immutable(1, 2)
add2(3, 3)  # will return 3, because the arguments are ignored

3

Serial and Parallel

In Synthora, tasks can be executed either serially or in parallel, providing flexibility in how workflows are structured.

Serial Tasks

Serial tasks are executed one after another, with each task’s output being passed as input to the next task. This allows for a linear flow of data through the tasks.

To add serial tasks, you can use the add_task and add_tasks methods of the scheduler:

flow = (
    BaseScheduler()
    .add_task(BaseTask(_add3))
    .add_tasks(BaseTask(_add3).s(1), BaseTask(_add3).s(2))
)

The flow can also use signatures to predefine task parameters, ensuring consistency and simplifying the workflow configuration.

flow = flow.s(1)
flow.run(1)

5

why the output is 5?

This is because the workflow will add the return value of the previous task as a parameter to the next task.

First Step: The flow gets the input: (1, 1), and passes it to the first task. The first task returns 1 + 1 = 2.

For the second task, we have pre-specified the input as 1, which, combined with the return value of the previous task (2), is passed as parameters. The second task returns 2 + 1 = 3.

Similarly, the third task returns 3 + 2 = 5.

Alternatively, you can use Python operators to declare serial workflows, which can simplify the creation process:

flow = BaseTask(_add3) >> BaseTask(_add3).s(1) >> BaseTask(_add3).s(2)

Additionally, the flow can be nested, allowing for more complex workflows.

flow = flow >> BaseTask(_add3).s(3)

Parallel Tasks

Parallel tasks are executed simultaneously, with each task receiving the same input parameters. This allows for concurrent processing and can significantly speed up the workflow when tasks are independent of each other.

To add parallel tasks, you can use the add_task_group method of the scheduler:

flow = ThreadPoolScheduler().add_task_group(
    [
        BaseTask(_add3).s(0),
        BaseTask(_add3).s(1),
        BaseTask(_add3).s(2),
        BaseTask(_add3).s(3),
    ]
)

flow.run(1)

[1, 2, 3, 4]

Unlike serial tasks, the input parameters for parallel tasks are passed to each task individually. For example:

task1 receives parameters 0, 1 task2 receives parameters 1, 1 task3 receives parameters 2, 1 task4 receives parameters 3, 1 Each task will process its own set of parameters independently and simultaneously.

Parallel tasks can also be declared using expressions:

flow = (
    BaseTask(_add3).s(0)
    | BaseTask(_add3).s(1)
    | BaseTask(_add3).s(2)
    | BaseTask(_add3).s(3)
)

Combining Serial and Parallel Tasks

Synthora allows you to combine serial and parallel tasks within the same workflow. This enables the creation of complex workflows that can handle various processing needs.

For example, you can create a workflow where two tasks are executed in parallel, and their results are then passed to a subsequent task:

flow = (BaseTask(_add3) | BaseTask(_add3)) >> BaseTask(_add3)
flow.run(1, 2)

6

In this example, BaseTask(_add3) | BaseTask(_add3) will execute two tasks in parallel, and the results will be passed to the next task in the sequence.

Both instances of BaseTask(_add3) will receive 1 and 2 as parameters. Each task will compute the sum of these parameters, resulting in 1 + 2 = 3. These two results (3 and 3) will then be passed as input to the subsequent task in the workflow.

The final task in the sequence will take these two 3s as its input parameters and compute their sum, resulting in 3 + 3 = 6. Thus, the final output of this combined serial and parallel workflow will be 6.

Advanced Workflow Usage — Context

To achieve advanced usage of the workflow, we introduce context to allow data management across tasks. To use context, please pass it as the first parameter of the function and annotate its type.

def add(ctx: BaseContext, x: int, y: int) -> int:
    with ctx:
        print(ctx.get_state(f"{x + y}"))
        if "ans" not in ctx:
            ctx["ans"] = [x + y]
        else:
            ctx["ans"] = ctx["ans"] + [x + y]
    return x + y


flow = ProcessPoolScheduler()
(
    flow
    | BaseTask(add, "2").s(1)
    | BaseTask(add, "3").s(2)
    | BaseTask(add, "4").s(3)
)


print(flow.run(1), flow.get_context()["ans"])

TaskState.RUNNING
TaskState.RUNNING
TaskState.RUNNING
[2, 3, 4] [2, 3, 4]

You can use any Scheduler with Context, and Synthora will automatically select the matching context based on the scheduler type.

Now, let’s use context to complete advanced features such as loops and if statements.

Loop

To implement a loop in a workflow, you can use the context to manage the loop state and control the flow of tasks. Here’s how to create a loop that runs a task multiple times until a condition is met:

1. Define a task that updates the context with the current state and checks the loop condition.

2. Use the context to set the cursor to the previous task if the loop condition is met.

def add(ctx: BaseContext) -> int:
    with ctx:
        a = ctx.get("a", 1)
        b = ctx.get("b", 1)
        cnt = ctx.get("cnt", 0)
        print(f"This is {cnt}th time")
        ctx["cnt"] = cnt + 1
        ctx["ans"] = a + b
        ctx["a"] = a + 1
        ctx["b"] = b + 1

        if a + b < 5:
            ctx.set_cursor(-1)
    return int(a + b)


flow = BaseTask(add) >> BaseTask(add).si()


print(flow.run(), flow.get_context()["ans"])

This is 0th time
This is 1th time
This is 2th time
This is 3th time
8 8

If

To implement conditional logic in a workflow, you can use the context to evaluate conditions and control the flow of tasks. Here’s how to create a conditional branch in a workflow:

1. Define a task that evaluates a condition and updates the context with the result.

2. Use the context to set the state of tasks based on the condition.

from typing import Any, List

from synthora.types.enums import TaskState


@task(name="skip")
def task1(a: int, b: int) -> int:
    return a + b


@task
def task2(a: int, b: int) -> int:
    print("hello", a, b)
    return a + b


@task
def task0() -> List[int]:
    print("task0")
    return [1, 2]


@task
def bratch(ctx: BaseContext, previous: Any) -> int:
    print("bratch", previous)
    with ctx:
        ctx.set_state("skip", TaskState.SKIPPED)

    return previous

bratch.flat_result = True
flow = task0 >> bratch >> (task1 | task2)
flow.run()

task0
bratch [1, 2]
hello 1 2

[3, 3]