Multi-Agent Systems

Why multiple agents

A single LLM call has two constraints: context window (how much it can see at once) and attention (how well it can juggle many concerns simultaneously). A task like "research 10 competitors, analyse their pricing, and write a 50-page report" exceeds both.

Multi-agent systems solve this by dividing the task: one agent searches, another analyses, a third writes, a fourth edits. Each agent has a focused context and a single responsibility. An orchestrator coordinates them, passing results between agents and synthesising the final output.

Think of it like a consulting firm: the partner (orchestrator) defines the scope and assigns tasks; analysts, writers, and subject-matter experts (agents) do focused work; the partner assembles the final deliverable.

Orchestrator-agent pattern

import anthropic
from concurrent.futures import ThreadPoolExecutor

client = anthropic.Anthropic()

def run_agent(role: str, task: str, context: str = "") -> str:
    '''Run a specialist agent with a given role and task.'''
    system = f"You are a {role}. Be concise and return only what was asked."
    user_content = f"{context}\n\nTask: {task}" if context else f"Task: {task}"
    response = client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": user_content}]
    )
    return response.content[0].text

def orchestrate(goal: str) -> str:
    '''Orchestrator: plan tasks, run agents, synthesise results.'''

    # 1. Planner agent creates subtask assignments
    plan_response = run_agent(
        "task planner",
        f"Break this goal into 3 specialist subtasks. Return as a JSON list of {{'role': ..., 'task': ...}}\n\nGoal: {goal}"
    )
    import json, re
    match = re.search(r'\[.*\]', plan_response, re.DOTALL)
    subtasks = json.loads(match.group()) if match else []

    # 2. Run specialist agents (parallel where independent)
    results = {}
    with ThreadPoolExecutor(max_workers=len(subtasks)) as pool:
        futures = {pool.submit(run_agent, s["role"], s["task"]): s for s in subtasks}
        for future, subtask in futures.items():
            results[subtask["role"]] = future.result()

    # 3. Synthesiser agent combines everything
    context = "\n\n".join(f"## {role}\n{result}" for role, result in results.items())
    return run_agent("senior analyst", f"Synthesise these specialist reports into a final answer for: {goal}", context)

result = orchestrate("Summarise the key differences between FastAPI, Flask, and Django for a senior developer.")
print(result)

Parallel agent execution

import asyncio
from anthropic import AsyncAnthropic

async_client = AsyncAnthropic()

async def run_agent_async(role: str, task: str) -> tuple[str, str]:
    response = await async_client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=512,
        system=f"You are a {role}.",
        messages=[{"role": "user", "content": task}]
    )
    return role, response.content[0].text

async def parallel_research(topic: str) -> dict[str, str]:
    '''Run multiple research agents simultaneously.'''
    tasks = [
        run_agent_async("technical analyst",    f"Explain the technical architecture of {topic}"),
        run_agent_async("market analyst",       f"Describe the market position and competitors for {topic}"),
        run_agent_async("risk analyst",         f"Identify the main risks and limitations of {topic}"),
        run_agent_async("trend forecaster",     f"Predict where {topic} is heading in 2025"),
    ]
    results = await asyncio.gather(*tasks)
    return dict(results)

# Run all 4 agents concurrently — takes as long as the slowest, not sum of all
reports = asyncio.run(parallel_research("large language models"))
for role, report in reports.items():
    print(f"\n=== {role.upper()} ===\n{report[:200]}...")

Message-passing between agents

For sequential pipelines where Agent B needs Agent A's output, pass results as messages:

class AgentPipeline:
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.shared_state = {}

    def run_stage(self, agent_name: str, role: str, task: str, inputs: list[str] = None) -> str:
        # Build context from required inputs
        context_parts = []
        if inputs:
            for key in inputs:
                if key in self.shared_state:
                    context_parts.append(f"{key}: {self.shared_state[key]}")
        context = "\n".join(context_parts)

        result = run_agent(role, task, context)
        self.shared_state[agent_name] = result
        return result

pipeline = AgentPipeline()

# Stage 1: Research
pipeline.run_stage("research", "researcher",
    "Find the top 3 Python web frameworks by GitHub stars.")

# Stage 2: Analysis (uses research output)
pipeline.run_stage("analysis", "technical analyst",
    "Compare the frameworks from a scalability perspective.",
    inputs=["research"])

# Stage 3: Recommendation (uses both)
final = pipeline.run_stage("recommendation", "senior architect",
    "Write a one-paragraph recommendation for a startup.",
    inputs=["research", "analysis"])

print(final)

When to use multi-agent

Use multi-agent when:

• The task requires more context than fits in a single LLM window (long research, multi-document synthesis).
• Subtasks are genuinely independent and can run in parallel (research 5 topics simultaneously).
• Different subtasks need different specialisations (a creative writer + a fact-checker + a formatter).
• You want checks and balances — an "adversarial reviewer" agent that critiques the primary agent's output.

Don't use multi-agent when:

• A single agent can handle the task — complexity is a cost, not a benefit.
• The task is inherently sequential with no parallelism opportunity.
• Debugging ability matters — multi-agent systems are harder to trace and fix than single agents.

Failure modes

Context loss between agents. Agent B gets only what you explicitly pass from Agent A. If A's reasoning matters, pass the reasoning — not just the conclusion.

Contradictions between agents. Two specialist agents may give conflicting information. Your orchestrator needs explicit conflict-resolution logic, or the synthesiser will produce an inconsistent final answer.

Cascading failures. If Agent A produces a bad output and Agent B trusts it, Agent C gets contaminated input. Add validation after each agent: check that outputs meet minimum quality criteria before passing them downstream.

Gotchas

Cost scales linearly with agents. 5 agents each making 3 API calls = 15 API calls per orchestration. Add up the token budgets before deploying. Use cheaper models (Haiku) for specialist agents and reserve expensive models (Sonnet) for the orchestrator and synthesiser.

Parallel agents can hit rate limits. Launching 20 agents simultaneously with large prompts may exceed your RPM or TPM quota. Use a semaphore to cap concurrency:

sem = asyncio.Semaphore(5)   # max 5 concurrent agents
async def run_with_limit(role, task):
    async with sem:
        return await run_agent_async(role, task)

Agent roles must be orthogonal. If two agents have overlapping responsibilities ("summariser" and "report writer"), they'll duplicate work or contradict each other. Define clear, non-overlapping scopes for each agent role.