Orchestrating multistep workflows

Most production AI systems are workflows: your code runs a fixed sequence of steps, some of them model calls. A handful of patterns cover most of them.

Chaining: the output of one step is the input to the next, with a check in between. Outline, check the outline, write the draft. Each step is simpler than doing everything at once, and the checks catch problems early.

Routing: a first step sorts the input and sends it down the right path. A refund request goes one way, a delivery question another, anything unclear to a person. Each path can have its own prompt, tools and model.

Parallel work: independent steps run at the same time and are combined afterwards. Summarise ten documents at once, then merge; or ask the same question three ways and take the majority answer.

from concurrent.futures import ThreadPoolExecutor

def summarise(doc):
    return doc.split(".")[0]   # stands in for a model call

docs = ["Rent rose 4%. Details follow.", "Vacancy fell. More below.", "Taxes are flat. See table."]
with ThreadPoolExecutor() as pool:
    summaries = list(pool.map(summarise, docs))
print(summaries)

Orchestrator and workers: one model call plans the subtasks, others carry them out, and the orchestrator combines the results. It's useful when you can't know the subtasks in advance, at the cost of more moving parts.

Check and revise: one step produces, another checks against clear criteria, and the first revises using the feedback, for a fixed number of rounds. It works when the criteria can actually be checked.

Whatever the pattern, the workflow's state lives in your code or a database: each step's input, output and status. Then a failed step can be retried on its own, a run can resume after a crash, and you can see exactly where things went wrong. Orchestration tools used for data pipelines (Airflow, Dagster, Prefect, Temporal and the like) are often used for this too.