Multi-agent orchestration

Why this matters

A single agent with ten tools gets confused. It forgets which tool to use, mixes up contexts, and makes poor decisions as complexity grows. The solution is the same one we use in software engineering: decompose. Instead of one agent doing everything, you build specialized agents — a researcher, a coder, a reviewer — and orchestrate them with a supervisor.

Multi-agent patterns are how production AI systems handle complex workflows. Customer support bots route to specialist agents. Code generation pipelines have separate planning, writing, and review stages. Data analysis systems decompose into query builders, executors, and interpreters. If you are building agent systems at work, you will encounter multi-agent orchestration within months.

Core concepts

Supervisor pattern. One agent (the supervisor) receives the user request, decides which specialist to delegate to, reviews the result, and either returns it or delegates again. The supervisor does not do the work itself — it routes and quality-checks.

Handoffs. When one agent transfers control to another, it passes a structured context object — not the entire conversation history. A handoff includes: what the next agent needs to do, what context it needs, and what a good result looks like.

Message passing vs shared state. Agents can communicate by passing messages through the supervisor (hub-and-spoke) or by reading/writing to shared state (blackboard pattern). Hub-and-spoke is simpler to debug. Shared state is more flexible but creates race conditions and ordering issues.

Specialist agent design. Each sub-agent should have a narrow system prompt, a limited set of tools, and a clear output contract. A research agent has search tools and returns structured findings. A code agent has a sandbox and returns tested code. Narrow scope means better performance.

Working example

Here is a supervisor agent that delegates to specialist sub-agents for a research-and-summarize workflow.

import json
from dataclasses import dataclass, field
from openai import OpenAI

client = OpenAI()

@dataclass
class AgentResult:
    agent_name: str
    content: str
    metadata: dict = field(default_factory=dict)

def run_agent(name: str, system_prompt: str, user_message: str, model: str = "gpt-4o") -> AgentResult:
    """Run a single specialist agent and return its result."""
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_message},
        ],
        temperature=0.3,
    )
    return AgentResult(
        agent_name=name,
        content=response.choices[0].message.content,
        metadata={"model": model, "tokens": response.usage.total_tokens},
    )

# --- Specialist agents ---

def research_agent(topic: str) -> AgentResult:
    return run_agent(
        name="researcher",
        system_prompt=(
            "You are a research specialist. Given a topic, produce a structured "
            "research brief with key facts, statistics, and source references. "
            "Output valid JSON with keys: facts (list of strings), "
            "statistics (list of strings), sources (list of strings)."
        ),
        user_message=f"Research this topic thoroughly: {topic}",
    )

def writer_agent(research_brief: str, task: str) -> AgentResult:
    return run_agent(
        name="writer",
        system_prompt=(
            "You are a technical writer. Given research material and a writing "
            "task, produce clear, well-structured content. Use the research "
            "provided — do not invent facts."
        ),
        user_message=f"Writing task: {task}\n\nResearch material:\n{research_brief}",
    )

def reviewer_agent(draft: str, criteria: str) -> AgentResult:
    return run_agent(
        name="reviewer",
        system_prompt=(
            "You are an editorial reviewer. Evaluate the draft against the "
            "given criteria. Output JSON with keys: approved (bool), "
            "score (1-10), issues (list of strings), suggestions (list of strings)."
        ),
        user_message=f"Criteria: {criteria}\n\nDraft to review:\n{draft}",
    )

# --- Supervisor ---

def supervisor(task: str, max_revisions: int = 2) -> str:
    """Orchestrate research, writing, and review agents."""
    print(f"[Supervisor] Received task: {task}")

    # Step 1: Research
    print("[Supervisor] Delegating to researcher...")
    research = research_agent(task)
    print(f"[Supervisor] Research complete ({research.metadata['tokens']} tokens)")

    # Step 2: Write
    print("[Supervisor] Delegating to writer...")
    draft = writer_agent(research.content, task)
    print(f"[Supervisor] Draft complete ({draft.metadata['tokens']} tokens)")

    # Step 3: Review loop
    for revision in range(max_revisions):
        print(f"[Supervisor] Delegating to reviewer (revision {revision + 1})...")
        review = reviewer_agent(
            draft.content,
            criteria="Accuracy, clarity, completeness, and proper use of research.",
        )

        # Parse review result
        try:
            review_data = json.loads(review.content)
        except json.JSONDecodeError:
            print("[Supervisor] Reviewer returned non-JSON, accepting draft.")
            break

        if review_data.get("approved", False) or review_data.get("score", 0) >= 8:
            print(f"[Supervisor] Draft approved (score: {review_data.get('score')})")
            break

        # Revise based on feedback
        print(f"[Supervisor] Revision needed (score: {review_data.get('score')})")
        feedback = "\n".join(review_data.get("suggestions", []))
        draft = writer_agent(
            research.content,
            f"{task}\n\nRevise based on this feedback:\n{feedback}",
        )

    total_tokens = sum(r.metadata["tokens"] for r in [research, draft, review])
    print(f"[Supervisor] Complete. Total tokens: {total_tokens}")
    return draft.content

# Usage
result = supervisor("Explain the tradeoffs of using LLMs for code review in CI pipelines")
print(result)

The supervisor controls the flow, each specialist has a narrow job, and the review loop has a bounded number of revisions. Each agent result carries metadata for cost tracking.

Common mistakes

1. Passing full conversation history between agents. Each specialist should receive only the context it needs. Passing everything wastes tokens and confuses the specialist.
2. No output contracts. If the research agent returns free-form text one time and JSON another time, the writer agent cannot reliably consume it. Enforce output structure.
3. Unbounded review loops. The reviewer and writer can ping-pong forever. Always cap revision rounds.
4. Supervisor does specialist work. If the supervisor starts writing content or doing research, you have lost the decomposition benefit. Keep it as a router.
5. No cost tracking. Multi-agent flows multiply costs. Track tokens per agent per run from day one.

Try it yourself

Build a three-agent system for a task relevant to your work. For example: an agent that takes a bug report, delegates to a "reproducer" agent to write a test case, then delegates to a "fixer" agent to propose a patch, with a "reviewer" agent checking the result. Keep each agent's system prompt under 200 words and give each agent at most two tools.