Orchestrator-Workers

Intent

An orchestrator model decides at run time how to split the work into subtasks, hands each one to a worker, and combines the results. It is for tasks whose split cannot be predicted before looking at the input.

Problem

Paula leads the migration of a legacy system from an old framework to a new one. There are 400 files. Some only need their imports swapped; others have lifecycle logic whose semantics change; some depend on each other and must migrate together.

The first attempt was a single agent walking through the files. Halfway through, the context was full and the early decisions (how to handle the new event system) had been lost. The second attempt was a script that sent each file to an isolated call. Fast, but files that depended on each other came out inconsistent.

What works has both parts: someone who looks at the whole and decides how to split it, and workers who do each part with focus. The split changes for every module, so it cannot be fixed in code.

The question this pattern answers: how do you split a large task dynamically across several models and still get a coherent final result?

Solution

Separate three roles:

• Orchestrator: reads the task and its context, produces a list of subtasks with self-contained instructions and dependencies. It does not do the work.
• Workers: each receives one subtask, does it in its own context, and returns a result in an agreed format. Subtasks without dependencies run in parallel.
• Synthesizer: combines the results, resolves conflicts, and delivers the final product. It is often the orchestrator itself in a second call.

The difference from Sectioning is who decides the split: there, code; here, the model, looking at the input. If you can write the for loop that splits the work before seeing the input, you do not need an orchestrator.

Anthropic’s “Building effective agents” post (December 2024) describes this pattern with two examples: coding products that change several files per request, and search tasks that gather information from several sources. In June 2025, Anthropic published how it built Claude’s research system, with a lead agent that spawns subagents in parallel. The same write-up states the cost: the multi-agent system used about 15 times more tokens than a regular chat. The pattern buys time and coverage with money.

Worked example

Paula asks: “Migrate the payments module to the new framework”. The orchestrator reads the module tree and returns this plan:

{
  "conventions": [
    "Events use the new EventBus; never call emit() directly",
    "Domain errors inherit from PaymentError"
  ],
  "subtasks": [
    { "id": "models",  "instruction": "Migrate models/*.py to the new ORM", "depends_on": [], "done_when": "models tests pass" },
    { "id": "events",  "instruction": "Replace emit() with EventBus in events.py", "depends_on": [], "done_when": "events.py compiles" },
    { "id": "service", "instruction": "Migrate service.py using the new models and events", "depends_on": ["models", "events"], "done_when": "service tests pass" },
    { "id": "api",     "instruction": "Update the routes in api.py", "depends_on": ["service"], "done_when": "api tests pass" }
  ]
}

Code turns the dependencies into waves. Wave 1 runs models and events in parallel. Wave 2 runs service, which receives both results. Wave 3 runs api. That is four workers, three waves, and one synthesis at the end.

Notice two things. Each instruction is self-contained: the events worker does not need to read the rest of the module to know what to do. And the conventions field carries the decisions that apply to the whole module to every worker.

The shared-decision trap

Without conventions, each worker decides alone how to handle the event system. One uses EventBus, another keeps emit() behind an adapter, and both results compile. The synthesis gets two correct, incompatible solutions.

It is the same problem as Paula’s first attempt, only in parallel. The rule of thumb: every decision that more than one worker must make comes from the orchestrator, in the plan, before any worker starts.

Structure

flowchart TD
  IN(["1. Task"]):::io --> O(["2. Orchestrator plans"]):::llm
  O --> P[("3. Plan: subtasks and dependencies")]:::store
  P --> W1(["4. Worker"]):::llm
  P --> W2(["4. Worker"]):::llm
  P --> W3(["4. Worker"]):::llm
  W1 --> V["5. Validate results"]:::tool
  W2 --> V
  W3 --> V
  V --> S(["6. Synthesize"]):::llm
  S --> OUT(["7. Result"]):::io

1. The task arrives with the needed material or access to it.
2. The orchestrator analyzes it and decides the split.
3. The plan is a structured object: subtasks, each one’s instruction, dependencies, and done criterion.
4. Workers run in parallel when there is no dependency, each with its own context and budget.
5. Code validates each result (schema, tests, compilation) before moving on.
6. The synthesizer puts everything together and resolves inconsistencies.
7. The result goes out, with the plan and partial results recorded on the trace.

Pseudocode

schema Subtask { id: string, instruction: string, depends_on: list[string], done_when: string }
schema Plan { conventions: list[string], subtasks: list[Subtask] }
schema WorkResult { id: string, output: string, ok: bool, notes: string }

function orchestrate(task):
    # 2 and 3. the model decides the split
    plan = llm(ORCHESTRATOR_PROMPT, task) -> Plan
    validate_dag(plan)                                   # no cycles, no dangling ids
    trace.span("plan", plan)

    results = {}
    for wave in topological_waves(plan.subtasks):
        # 4. each wave runs in parallel
        parallel {
            for st in wave:
                ctx = plan.conventions + [st.instruction] + [results[d].output for d in st.depends_on]
                results[st.id] = llm(WORKER_PROMPT, ctx) -> WorkResult
        }
        # 5. validate before the next wave
        for st in wave:
            if not check(st.done_when, results[st.id]):
                results[st.id] = retry_once(st, results)
        budget.check()

    # 6. synthesize
    return llm(SYNTH_PROMPT, task, plan, results)        # 7

Real-world scenarios

• Deep research with citations. The orchestrator breaks the question into sub-questions that depend on what the first round finds. Workers research each one; the synthesizer writes the cited report.
• Coding agent. For changes across several modules, the orchestrator decides which files change and in what order. Each worker edits one module and runs its tests.
• Legacy code migration. The plan groups interdependent files, sets the order of the waves, and defines the done criterion (it compiles and passes the module’s tests). This is Paula’s case.

Applicability

• The task is too large for one context, and the split depends on the content.
• The parts are mostly independent, but not entirely.
• There is a per-part done criterion that code can check.
• Parallelism cuts total time meaningfully.
• The task’s value justifies spending several times the tokens of a single call.

When not to use

• The split is always the same: per file, per page, per section. Use Sectioning or Map-Reduce, with the split in code.
• The task fits in one agent with a clean context. An orchestrator adds cost and points of failure. Start with a ReAct Loop and split only when the context fills up.
• The parts depend on each other so much that none can run alone. Parallelism disappears, and you pay for the plan and the synthesis without saving time.
• The workers would be decorative roles passing text along. An “analyst”, a “reviewer”, and a “writer” that read the same context and rewrite the same answer split no work at all. See Decorative Multi-Agent.

Orchestrate or plan and execute

Plan-then-Execute also starts with a plan. The difference is who executes. There, a single agent follows the plan step by step, in the same context. Here, each subtask goes to a worker with its own context, and independent ones run at the same time.

Use Plan-then-Execute when the steps are sequential and need what came before. Use Orchestrator-Workers when the parts fit in separate contexts and the volume justifies parallelism.

How to implement

1. Write the plan schema: subtasks, self-contained instruction, dependencies, done criterion.
2. Write the orchestrator prompt with limits: maximum number of subtasks, size of each one.
3. Validate the plan in code: no cycles, no missing dependencies, within limits.
4. Write the worker prompt as a generic function that takes an instruction and its dependencies’ context.
5. Execute in waves, with parallelism inside each wave and a shared total budget.
6. Validate each result with a code check before releasing the next wave.
7. Write the synthesizer and give it the plan along with the results.
8. Build an eval with real tasks, measuring final quality, cost, and total time.

Variations

• Orchestrator with replanning. After each wave, the orchestrator revises the plan using the results. More adaptive, more expensive. Good for research, where the first round changes the questions of the second.
• Workers with tools. Each worker is an agent in a loop, not a single call. Needed for code and research; each worker is then a subagent, as in Subagent Delegation.
• Workers on a smaller model. The orchestrator and synthesizer use the large model; the workers use a cheaper one. It works when the plan’s instructions are specific enough.
• Supervisor. The orchestrator watches the workers during execution and can reassign tasks. More control, more orchestrator calls.

An illustrative cost calculation

Suppose the plan costs 6,000 tokens, each worker 20,000, and the synthesis 15,000. With 4 subtasks, the task spends about 101,000 tokens. If the orchestrator decides on 20 subtasks, the total goes past 420,000, and quality does not necessarily rise.

That is why the subtask limit lives in the prompt and also in code. The model picks the split; code decides how much it may cost.

Cost, latency, and failure modes

How to evaluate

• Final quality, measured by the task’s criterion (tests pass, report has valid sources).
• Plan quality: coverage (every part of the task appears) and overlap (repeated parts).
• Total cost and time, compared with the single-agent version.

{ "task": "Migrate the payments module to the new framework", "check": "compiles and passes the payments tests", "max_subtasks": 12 }

Evaluate the plan separately from the result. For 20 real tasks, write by hand the list of parts a good plan must cover. Then compare it with the generated plans: a missing part is a coverage failure, and two subtasks touching the same file is overlap. A bad plan spoils everything after it, so it is the first thing you fix.

In production

• A plan span with the full plan, and child worker spans with each subtask and result.
• Limits on subtasks and waves per task, with an alert when they hit.
• A dashboard of retries per subtask: a subtask type that always fails calls for a better worker prompt.
• A review queue for tasks whose synthesis had conflicts.

Pros and cons

Code

The tabs build the plan with structured output, run workers in parallel per wave, and synthesize. Python and TypeScript launch the waves with the language runtime. LangGraph launches each worker with Send, and state collects the returns.

import asyncio

import anthropic
from pydantic import BaseModel

client = anthropic.AsyncAnthropic()


class Subtask(BaseModel):
    id: str
    instruction: str
    depends_on: list[str]


class Plan(BaseModel):
    subtasks: list[Subtask]


async def plan(task: str) -> Plan:  # 2 and 3
    reply = await client.messages.parse(
        model="claude-opus-5-5", max_tokens=8000,
        system="Split the task into at most 12 self-contained subtasks. List dependencies by id.",
        messages=[{"role": "user", "content": task}], output_format=Plan,
    )
    return reply.parsed_output


async def work(st: Subtask, deps: dict[str, str]) -> str:  # 4
    context = "\n\n".join(f"Result of {d}:\n{deps[d]}" for d in st.depends_on)
    reply = await client.messages.create(
        model="claude-sonnet-5-5", max_tokens=16000,
        system="You carry out one subtask of a larger plan. Do only what was asked.",
        messages=[{"role": "user", "content": f"{st.instruction}\n\n{context}"}],
    )
    return reply.content[-1].text


async def orchestrate(task: str) -> str:
    p = await plan(task)
    pending, results = {s.id: s for s in p.subtasks}, {}
    while pending:
        wave = [s for s in pending.values() if all(d in results for d in s.depends_on)]
        if not wave:
            raise ValueError("plan has a cycle or a missing dependency")
        outputs = await asyncio.gather(*(work(s, results) for s in wave))  # parallel
        for s, out in zip(wave, outputs):
            results[s.id] = out  # 5. validate here (tests, schema) before moving on
            del pending[s.id]
    joined = "\n\n".join(f"[{k}]\n{v}" for k, v in results.items())
    final = await client.messages.create(  # 6
        model="claude-opus-5-5", max_tokens=16000,
        system="Combine the results into one coherent final product. Point out conflicts.",
        messages=[{"role": "user", "content": f"Task: {task}\n\n{joined}"}],
    )
    return final.content[-1].text

import Anthropic from "@anthropic-ai/sdk";
import { zodOutputFormat } from "@anthropic-ai/sdk/helpers/zod";
import { z } from "zod";

const client = new Anthropic();

const Plan = z.object({
  subtasks: z.array(z.object({ id: z.string(), instruction: z.string(), depends_on: z.array(z.string()) })),
});
type Subtask = z.infer<typeof Plan>["subtasks"][number];

const text = (m: Anthropic.Message) => m.content.find((b) => b.type === "text")?.text ?? "";

async function work(st: Subtask, results: Map<string, string>): Promise<string> {
  const context = st.depends_on.map((d) => `Result of ${d}:\n${results.get(d)}`).join("\n\n");
  const reply = await client.messages.create({
    model: "claude-sonnet-5-5", max_tokens: 16000,
    system: "You carry out one subtask of a larger plan. Do only what was asked.",
    messages: [{ role: "user", content: `${st.instruction}\n\n${context}` }],
  });
  return text(reply);
}

export async function orchestrate(task: string): Promise<string> {
  const { parsed_output: plan } = await client.messages.parse({
    model: "claude-opus-5-5", max_tokens: 8000,
    system: "Split the task into at most 12 self-contained subtasks. List dependencies by id.",
    messages: [{ role: "user", content: task }],
    output_config: { format: zodOutputFormat(Plan) },
  });
  if (!plan) throw new Error("orchestrator returned no plan");

  const results = new Map<string, string>();
  let pending = plan.subtasks;
  while (pending.length) {
    const wave = pending.filter((s) => s.depends_on.every((d) => results.has(d)));
    if (!wave.length) throw new Error("plan has a cycle or a missing dependency");
    const outs = await Promise.all(wave.map((s) => work(s, results))); // 4. parallel
    wave.forEach((s, i) => results.set(s.id, outs[i])); // 5. validate here
    pending = pending.filter((s) => !results.has(s.id));
  }

  const joined = [...results].map(([k, v]) => `[${k}]\n${v}`).join("\n\n");
  const final = await client.messages.create({
    model: "claude-opus-5-5", max_tokens: 16000,
    system: "Combine the results into one coherent final product. Point out conflicts.",
    messages: [{ role: "user", content: `Task: ${task}\n\n${joined}` }],
  });
  return text(final);
}

import operator
from typing import Annotated

from langchain.chat_models import init_chat_model
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send
from pydantic import BaseModel
from typing_extensions import TypedDict

class Task(BaseModel):
    id: str
    instruction: str

class Plan(BaseModel):
    tasks: list[Task]

class State(TypedDict):
    goal: str
    plan: list[dict]
    results: Annotated[list[str], operator.add]
    report: str

planner = init_chat_model("anthropic:claude-opus-5-5", temperature=0).with_structured_output(Plan)
worker_model = init_chat_model("anthropic:claude-haiku-4-5", temperature=0)
synth = init_chat_model("anthropic:claude-opus-5-5", temperature=0)

def plan(state: State) -> dict:
    # 2 and 3. the orchestrator only plans
    drafted = planner.invoke([
        ("system", "Split the task into independent subtasks. Do not execute."),
        ("human", state["goal"]),
    ])
    return {"plan": [t.model_dump() for t in drafted.tasks]}

def fan(state: State):
    # 4. one Send per worker; they run in parallel
    return [Send("work", {"goal": task["instruction"]}) for task in state["plan"]]

def work(state: State) -> dict:
    reply = worker_model.invoke([("human", state["goal"])])
    return {"results": [reply.text]}  # operator.add merges the lists

def synthesize(state: State) -> dict:
    # 5 and 6. validate in your code before this call when the criterion is a test
    report = synth.invoke([("human", "\n".join(state["results"]))])
    return {"report": report.text}

graph = StateGraph(State)
graph.add_node("plan", plan)
graph.add_node("work", work)
graph.add_node("synthesize", synthesize)
graph.add_edge(START, "plan")
graph.add_conditional_edges("plan", fan, ["work"])
graph.add_edge("work", "synthesize")
graph.add_edge("synthesize", END)
app = graph.compile()

Framework support

Check your understanding