Single-agent systems hit limits on complex tasks. Multi-agent systems -- where specialized agents collaborate -- can tackle problems no single agent handles well.
Why Multi-Agent?
A single general-purpose agent asked to "research competitors, write a comparison report, and create a slide deck" will do each task mediocrely. Specialized agents do each task well:
- Research Agent: web search, data gathering, fact-checking
- Writing Agent: structured prose, analysis, citations
- Design Agent: slide structure, visual hierarchy, formatting
The Orchestrator Pattern
import anthropic
from typing import Callable
client = anthropic.Anthropic()
class Agent:
def __init__(self, name: str, system_prompt: str, tools: list = None):
self.name = name
self.system_prompt = system_prompt
self.tools = tools or []
def run(self, task: str, context: str = "") -> str:
messages = [{"role": "user", "content": f"{context}\n\nTask: {task}".strip()}]
response = client.messages.create(
model="claude-opus-4-8",
max_tokens=4096,
system=self.system_prompt,
tools=self.tools,
messages=messages
)
return self._extract_text(response)
def _extract_text(self, response) -> str:
return "\n".join(
block.text for block in response.content
if hasattr(block, "text")
)
def run_orchestrator(task: str, agents: dict[str, Agent]) -> str:
orchestrator = Agent(
name="Orchestrator",
system_prompt=f"""You coordinate specialized agents to complete complex tasks.
Available agents: {', '.join(agents.keys())}
For each subtask, specify: AGENT: <name> | TASK: <description>
Combine their outputs into a final result."""
)
# Get the orchestrator's plan
plan = orchestrator.run(task)
# Parse and execute subtasks
results = {}
for line in plan.split('\n'):
if 'AGENT:' in line and 'TASK:' in line:
parts = line.split('|')
agent_name = parts[0].replace('AGENT:', '').strip()
subtask = parts[1].replace('TASK:', '').strip()
if agent_name in agents:
context = "\n".join(f"{k}: {v}" for k, v in results.items())
results[subtask] = agents[agent_name].run(subtask, context)
# Final synthesis
synthesis_prompt = f"""
Original task: {task}
Agent results:
{chr(10).join(f'{k}: {v}' for k, v in results.items())}
Synthesize these into a complete, coherent response.
"""
return orchestrator.run(synthesis_prompt)
Parallel Execution
For independent subtasks, run agents concurrently:
import asyncio
import anthropic
async_client = anthropic.AsyncAnthropic()
async def run_agent_async(agent: Agent, task: str) -> tuple[str, str]:
response = await async_client.messages.create(
model="claude-opus-4-8",
max_tokens=2048,
system=agent.system_prompt,
messages=[{"role": "user", "content": task}]
)
return agent.name, response.content[0].text
async def run_parallel(tasks: dict[str, tuple[Agent, str]]) -> dict[str, str]:
coroutines = [
run_agent_async(agent, task)
for agent, task in tasks.values()
]
results = await asyncio.gather(*coroutines)
return dict(results)
# Run research, writing, and data analysis simultaneously
results = asyncio.run(run_parallel({
"competitor_research": (research_agent, "Find top 5 competitors of Notion"),
"market_size": (data_agent, "Estimate the project management software market size"),
"user_trends": (research_agent, "What are users saying about Notion vs Obsidian?")
}))
Shared State / Memory
Agents need to share context without repeating work:
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class SharedMemory:
task_id: str
context: dict = field(default_factory=dict)
findings: list[dict] = field(default_factory=list)
completed_subtasks: list[str] = field(default_factory=list)
def add_finding(self, agent: str, content: str, confidence: float = 1.0):
self.findings.append({
"agent": agent,
"content": content,
"confidence": confidence,
"timestamp": datetime.utcnow().isoformat()
})
def get_context_summary(self) -> str:
if not self.findings:
return "No previous findings."
return "\n".join(
f"[{f['agent']}]: {f['content']}"
for f in self.findings[-10:] # Last 10 findings
)
memory = SharedMemory(task_id="research-001")
class MemoryAwareAgent(Agent):
def run_with_memory(self, task: str, memory: SharedMemory) -> str:
context = f"Previous findings:\n{memory.get_context_summary()}"
result = self.run(task, context)
memory.add_finding(self.name, result[:500]) # Store summary
return result
Agent Handoffs
Some tasks require one agent to pass control to another mid-task:
HANDOFF_TOOL = {
"name": "handoff_to_agent",
"description": "Transfer control to a specialized agent for a specific subtask",
"input_schema": {
"type": "object",
"properties": {
"agent": {"type": "string", "description": "Agent name to hand off to"},
"task": {"type": "string", "description": "Specific task for that agent"},
"context": {"type": "string", "description": "Context needed for the task"}
},
"required": ["agent", "task"]
}
}
def handle_handoff(agent_name: str, task: str, context: str, agents: dict) -> str:
if agent_name not in agents:
return f"Error: Unknown agent {agent_name}"
return agents[agent_name].run(task, context)
Cost and Latency Management
Multi-agent systems multiply API costs. Optimize:
# Use cheaper models for simple subtasks
AGENT_MODELS = {
"research": "claude-haiku-4-5-20251001", # Fast, cheap for search/retrieval
"writing": "claude-opus-4-8", # Best for prose
"coding": "claude-sonnet-5", # Good balance for code
"verification": "claude-haiku-4-5-20251001" # Quick fact checks
}
# Cache agent outputs for identical inputs
from functools import lru_cache
@lru_cache(maxsize=100)
def cached_agent_run(agent_name: str, task_hash: str) -> str:
return agents[agent_name].run(task_hash) # task_hash is a hash of the task
Multi-agent systems work best when: tasks are clearly divisible, agents can work in parallel, and each agent has a genuinely different specialty. For simple tasks, a single well-prompted agent is almost always better.