DEV Community

AI Builders
AI Builders

Posted on

Multi-Agent Orchestration in 2026: Supervisor, Handoffs & Swarm Patterns in Python (From My New Book)

Multi-Agent Orchestration in 2026: Supervisor, Handoffs & Swarm Patterns in Python

One agent. One context window. One tool set. That is the default setup for most AI applications, and it works — until the task grows past what a single context window can hold.

I hit that wall the same way most people do: my "one smart agent" started with two tools, then five, then ten. Every added tool definition ate context. Every past message stayed in the window. The agent got slower, more expensive, and started making silly routing mistakes. The fix was not a bigger model — it was splitting the work across specialized agents that each keep a small, focused context.

This is the orchestration chapter from my new book, condensed into three patterns you can actually ship today, plus one complete runnable crew in Python.

The ceiling every single agent hits

Three problems show up in the same order, every time:

  1. Context bloat. Tool definitions, system prompts, and conversation history all share one window. Past ~8-10 tool definitions, the model starts ignoring the ones it used least — which is usually the one you need most.
  2. One failure mode. A single wrong tool call derails the whole pipeline. There is no second opinion, no reviewer, no recovery path.
  3. Serial work. One agent does everything in sequence, so independent sub-tasks (research three sources, review two drafts) never run in parallel.

None of these are model-quality problems. They are architecture problems.

The decision rule

Do not start with multi-agent. Start with one agent and a couple of tools. Add orchestration only when you hit one of these:

  • You have two or more clearly separable specialists (research, drafting, review, formatting).
  • You have independent work that should run in parallel (translate five docs, screen ten leads).
  • Your context is overflowing and most of it is tool definitions or intermediate results.

If none of those apply, a single agent with tools is the cheaper, simpler, more reliable answer. Multi-agent is a scaling tool, not a feature.

Pattern 1: Supervisor routing

The supervisor pattern is the simplest and most common. A lightweight agent looks at the task, picks one worker from a registry, and the worker runs. The supervisor is not a boss — it is a classifier with a function list.

import json
import os
from openai import OpenAI

client = OpenAI()  # reads OPENAI_API_KEY; set OPENAI_BASE_URL for other providers
MODEL = os.getenv("CREW_MODEL", "gpt-4o-mini")


def llm_json(system: str, user: str, temperature: float = 0.2) -> dict:
    resp = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        temperature=temperature,
    )
    text = resp.choices[0].message.content
    start, end = text.find("{"), text.rfind("}")
    return json.loads(text[start : end + 1])


WORKERS = {
    "research": "Find and summarize sources for a topic",
    "draft": "Write a draft section from research notes",
    "review": "Review a draft against a quality checklist",
}


def route(task: str) -> str:
    system = (
        "You are a router. Pick the single worker best suited for the task.\n"
        f"Workers:\n{json.dumps(WORKERS, indent=2)}\n"
        'Reply with JSON only: {"worker": "<name>", "reason": "<one line>"}'
    )
    return llm_json(system, task)["worker"]
Enter fullscreen mode Exit fullscreen mode

Two practical notes from running this in production:

  • Use a cheap, fast model for the router. Routing is classification, not reasoning. Your expensive model should be doing the actual work, not deciding which worker to call.
  • Validate the router's output. If the returned worker name is not in the registry, fall back to a default worker instead of crashing. A wrong route is recoverable; a crash is not.

Pattern 2: Handoffs

Not all work is tree-shaped. Sometimes a worker needs to pass the baton mid-task: a writer finishes a section and hands it to a reviewer, who hands it back with notes. OpenAI's Swarm experiment popularized this pattern, and it lives on in the Agents SDK — but the pattern itself is framework-independent. A handoff is just a control signal in the agent's output that tells the runtime to switch the active agent.

AGENTS = {
    "writer": {"system": "You write technical sections. Hand off to 'reviewer' when done."},
    "reviewer": {"system": "You review sections for accuracy and clarity. Return a verdict."},
}


def parse_handoff(reply: str):
    marker = "HANDOFF:"
    if marker not in reply:
        return None
    payload = reply.split(marker, 1)[1].strip()
    return json.loads(payload)


def run_agent(agent_name: str, task: str, context: str = "", max_handoffs: int = 5) -> str:
    name = agent_name
    messages = [{"role": "user", "content": task}]
    for _ in range(max_handoffs):
        agent = AGENTS[name]
        reply = client.chat.completions.create(
            model=MODEL,
            messages=[
                {"role": "system", "content": agent["system"] + "\nShared context:\n" + context},
                *messages,
            ],
        ).choices[0].message.content

        handoff = parse_handoff(reply)
        if handoff:
            name = handoff["to"]
            messages.append({"role": "assistant", "content": reply})
            messages.append(
                {"role": "user", "content": f"Take over. Task: {handoff.get('brief', task)}"}
            )
            continue
        return reply
    raise RuntimeError("Handoff limit reached")
Enter fullscreen mode Exit fullscreen mode

The writer finishes, emits HANDOFF: {"to": "reviewer", "brief": "..."}, and the runtime switches agents. Two rules keep this from becoming a mess:

  • Always cap handoffs. An unbounded handoff loop is a cost explosion with extra steps. Five is a sane default.
  • Hand off a summary, not the transcript. Pass the result and a short brief. If the next agent needs the full history, you have a context problem, not an orchestration problem.

Pattern 3: Fan-out / fan-in

When sub-tasks are independent, run them in parallel and merge the results. This is where multi-agent stops being about context and starts being about wall-clock time.

from concurrent.futures import ThreadPoolExecutor


def fan_out(task: str, chunks: list[str], worker_fn) -> list[str]:
    with ThreadPoolExecutor(max_workers=min(4, len(chunks))) as pool:
        return list(pool.map(lambda c: worker_fn(task, c), chunks))
Enter fullscreen mode Exit fullscreen mode

Concurrency is where rate limits bite. Cap the pool, and wrap each worker call in a retry with exponential backoff — ten parallel calls that all hit a 429 at once is the classic failure.

The full crew: research → draft → review, with a verify loop

Here is the complete, runnable version. It combines all three patterns in one small file: a supervisor-style registry, parallel research, and a review loop that keeps revising until the editor passes the draft — or the revision budget runs out.

import json
import os
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI

client = OpenAI()
MODEL = os.getenv("CREW_MODEL", "gpt-4o-mini")


def llm_json(system: str, user: str, temperature: float = 0.2) -> dict:
    resp = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        temperature=temperature,
    )
    text = resp.choices[0].message.content
    start, end = text.find("{"), text.rfind("}")
    return json.loads(text[start : end + 1])


def research_worker(topic: str, angle: str) -> str:
    return llm_json(
        "You are a research analyst. Return {\"summary\": \"...\"}",
        f"Summarize the key facts about '{topic}' from the angle: {angle}",
    )["summary"]


def draft_worker(topic: str, research: str) -> str:
    return llm_json(
        "You are a technical writer. Return {\"draft\": \"...\"}",
        f"Write a 200-word section on '{topic}' using this research:\n{research}",
    )["draft"]


def review_worker(draft: str) -> dict:
    return llm_json(
        "You are an editor. Return {\"verdict\": \"pass\" or \"revise\", \"issues\": \"...\"}",
        f"Check this draft for accuracy, clarity and jargon. Draft:\n{draft}",
        temperature=0.0,
    )


def run_crew(topic: str, max_revisions: int = 2) -> str:
    angles = ["practical implementation", "common pitfalls", "cost and performance"]
    with ThreadPoolExecutor(max_workers=3) as pool:
        research = "\n".join(pool.map(lambda a: research_worker(topic, a), angles))

    draft = draft_worker(topic, research)
    for _ in range(max_revisions):
        verdict = review_worker(draft)
        if verdict["verdict"] == "pass":
            return draft
        draft = draft_worker(topic, research + "\n\nEditor feedback:\n" + verdict["issues"])
    return draft


if __name__ == "__main__":
    print(run_crew("why multi-agent systems fail in production"))
Enter fullscreen mode Exit fullscreen mode

To run it:

pip install openai
export OPENAI_API_KEY=sk-...
python crew.py "why multi-agent systems fail in production"
Enter fullscreen mode Exit fullscreen mode

The script works against any OpenAI-compatible endpoint — set OPENAI_BASE_URL and CREW_MODEL to point at the provider you already use.

Production rules I learned the hard way

  1. Wrap every agent call in a retry with timeout. Agents fail, providers rate-limit, networks hiccup. tenacity with exponential backoff and a total deadline turns flaky calls into boring ones.
  2. Set a spend budget per run. Count tokens per hop and stop the crew if the budget is blown. An agent loop that runs away is the most expensive bug you will ship this year.
  3. Trace every hop. Log agent name, input size, latency, and token count to a JSONL file. When a crew produces a bad result, the trace is the only way to find which agent caused it.
  4. Keep the router context tiny. The supervisor should see worker names and descriptions, not the full conversation. Its job is routing, and routing is easier with less noise.
  5. Test worker selection, not just final output. Add golden tasks where the correct worker is known in advance, and assert the router picks it. This is the orchestration equivalent of unit tests.

When NOT to build multi-agent

If one agent with two tools solves the task, use it. Multi-agent adds latency (extra round-trips), cost (more tokens per task), and failure surfaces (more agents, more things to break). And if your agents need to share mutable state — a shared database row they all write to — you are building a distributed system, not a crew. Reconsider before you go down that path.

The sweet spot: two to five specialized agents, a cheap router, bounded handoffs, and a hard budget. That combination ships, scales, and — most importantly — fails loudly instead of failing silently.


This is the orchestration chapter from my new book, The AI Agents Playbook. The full playbook has 15 more production patterns, ready-to-use prompts, and n8n workflows for agents that actually ship: AI Agents Playbook — use code LAUNCH11 for 11% off.

More from the book series: AI Agent Memory in 2026: Architectures That Actually Scale and LLM Evals in 2026: How to Test AI Agents Before They Break in Production.

Top comments (0)