Multi-Agent Systems

A single agent with twenty tools suffers from tool confusion: the model has too many choices and picks poorly. Context windows grow large and expensive as task history accumulates. A single failure stops the entire task. Multi-agent systems address all three: each agent has a narrow tool set, its context stays focused, and failures are contained to the responsible sub-agent.

The trade-off is coordination overhead. Every agent-to-agent handoff costs tokens and adds latency. Worker errors can propagate silently to the supervisor, which may incorporate incorrect data into its synthesis. Start with a single agent. Move to multi-agent architecture only when you have measured a specific failure that coordination solves.

Agents communicate via text. Unstructured natural language handoffs ("Go research the market for me") create ambiguity: the receiving agent does not know the expected output format, what context to use, or what to do if it cannot complete the task. Structured handoffs, expressed as JSON objects, eliminate that ambiguity.

A well-formed handoff includes four fields: the specific sub-task, the context the receiving agent needs (background information it would not otherwise have), the expected output format (so the supervisor can parse the result without natural language interpretation), and a fallback instruction (what to return if the task cannot be completed).

from dataclasses import dataclass
import json

@dataclass
class AgentHandoff:
    task_id: str
    assigned_to: str
    task_description: str
    context: dict
    expected_output_format: str
    fallback_instruction: str

def format_handoff(handoff: AgentHandoff) -> str:
    return json.dumps({
        "task_id": handoff.task_id,
        "assigned_to": handoff.assigned_to,
        "task": handoff.task_description,
        "context": handoff.context,
        "expected_output": handoff.expected_output_format,
        "fallback": handoff.fallback_instruction
    }, indent=2)

The supervisor pattern proceeds through three phases:

1. Planning: a capable model breaks the user request into parallel sub-tasks and assigns each to a specialist.
2. Execution: workers complete their assigned tasks independently.
3. Synthesis: the supervisor combines the worker outputs into a coherent result.

Use a capable model (such as Claude Opus) for the supervisor: it handles planning and synthesis, where reasoning quality matters most. Use cheaper, faster models (such as Claude Haiku) for workers doing routine extraction, classification, or research. This asymmetry can reduce costs by five to ten times without meaningful quality loss.

import anthropic, json
client = anthropic.Anthropic()

def run_specialist(agent_name: str, system_prompt: str, task: str) -> str:
    """Run a specialist agent and return its output as text."""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # cheaper model for workers
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    text_blocks = [b for b in response.content if hasattr(b, "text")]
    return text_blocks[0].text if text_blocks else "No output produced."

def run_supervisor(user_request: str) -> str:
    # Phase 1: plan
    plan_response = client.messages.create(
        model="claude-opus-4-6",  # capable model for planning
        max_tokens=1024,
        system="Break the request into research_task, finance_task, tech_task. Return JSON.",
        messages=[{"role": "user", "content": user_request}]
    )
    tasks = json.loads(plan_response.content[0].text)

    # Phase 2: run workers
    research = run_specialist("Research", RESEARCH_PROMPT, tasks["research_task"])
    finance = run_specialist("Finance", FINANCE_PROMPT, tasks["finance_task"])
    tech = run_specialist("Tech", TECH_PROMPT, tasks["tech_task"])

    # Phase 3: synthesise
    synthesis = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=3000,
        system="Synthesise specialist reports into a coherent recommendation.",
        messages=[{"role": "user", "content": f"Research: {research}
Finance: {finance}
Tech: {tech}"}]
    )
    return synthesis.content[0].text

Running workers sequentially when their tasks are independent wastes time. If the research agent, finance agent, and tech agent each take 10 seconds and their tasks do not depend on each other, sequential execution takes 30 seconds. Parallel execution takes 10 seconds plus a small coordination overhead.

Python's asyncio.gather() runs independent coroutines concurrently. Replace the sequential worker calls in the supervisor with async versions and gather them in a single call.

import asyncio

async def run_specialist_async(agent_name: str, system_prompt: str, task: str) -> tuple[str, str]:
    aclient = anthropic.AsyncAnthropic()
    response = await aclient.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=2048,
        system=system_prompt,
        messages=[{"role": "user", "content": task}]
    )
    text = response.content[0].text if response.content else ""
    return agent_name, text

async def run_parallel_workers(tasks: dict) -> dict:
    coroutines = [
        run_specialist_async("Research", RESEARCH_PROMPT, tasks["research_task"]),
        run_specialist_async("Finance", FINANCE_PROMPT, tasks["finance_task"]),
        run_specialist_async("Tech", TECH_PROMPT, tasks["tech_task"])
    ]
    results = await asyncio.gather(*coroutines)
    return {name: result for name, result in results}

When workers produce conflicting outputs (the research agent finds a large market; the finance agent finds it unprofitable), the supervisor must reconcile the disagreement rather than arbitrarily pick one. A dedicated judge agent pattern passes both claims and the relevant context to a separate model call, which evaluates the reasoning and provides a conclusion.

Worker outputs must include structured success or error indicators. A worker that returns the string "I completed the task" gives the supervisor nothing to parse. A worker that returns a JSON object with status, result, and optionally confidence allows the supervisor to handle partial results and errors programmatically.