Multi-Agent System Design
Agent Architecture Patterns
4 min read
AI agents are systems that use LLMs to reason, plan, and take actions. Understanding different architecture patterns is essential for designing scalable agent systems.
Single Agent Pattern
The simplest pattern—one agent handles the entire task.
class SingleAgent:
def __init__(self, llm, tools: list):
self.llm = llm
self.tools = {tool.name: tool for tool in tools}
async def run(self, task: str) -> str:
messages = [{"role": "user", "content": task}]
while True:
response = await self.llm.complete(
messages=messages,
tools=list(self.tools.values())
)
if response.tool_calls:
# Execute tools
for call in response.tool_calls:
result = await self.tools[call.name].execute(call.args)
messages.append({
"role": "tool",
"content": result,
"tool_call_id": call.id
})
else:
# Final response
return response.content
Best for: Simple tasks, < 5 tools, clear single-path solutions.
Multi-Agent Patterns
1. Coordinator Pattern
A central coordinator delegates tasks to specialized agents.
┌─────────────────────────────────────────────────────────────┐
│ Coordinator Agent │
│ │
│ "Break down task, assign to specialists, synthesize" │
│ │
├─────────────┬─────────────┬─────────────┬──────────────────┤
│ │ │ │ │
▼ ▼ ▼ ▼ │
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│Research │ │ Code │ │ Review │ │ Test │ │
│ Agent │ │ Agent │ │ Agent │ │ Agent │ │
└─────────┘ └─────────┘ └─────────┘ └─────────┘ │
│
└──────────────── Results back to Coordinator ────────────────┘
class CoordinatorAgent:
def __init__(self, llm, specialist_agents: dict):
self.llm = llm
self.specialists = specialist_agents
async def run(self, task: str) -> str:
# Step 1: Plan the task
plan = await self._create_plan(task)
# Step 2: Execute each step with appropriate specialist
results = []
for step in plan.steps:
agent = self.specialists[step.agent_type]
result = await agent.run(step.instruction)
results.append({
"step": step.name,
"result": result
})
# Step 3: Synthesize results
return await self._synthesize(task, results)
async def _create_plan(self, task: str) -> Plan:
prompt = f"""Break down this task into steps.
For each step, specify which specialist should handle it.
Available specialists: {list(self.specialists.keys())}
Task: {task}
Return a JSON plan."""
response = await self.llm.complete(prompt)
return Plan.parse(response)
2. Pipeline Pattern
Agents process sequentially, each building on the previous output.
class PipelineAgent:
def __init__(self, stages: list):
self.stages = stages # List of (agent, transform_fn)
async def run(self, initial_input: str) -> str:
current_output = initial_input
for agent, transform in self.stages:
# Transform input for this stage
stage_input = transform(current_output)
# Run agent
current_output = await agent.run(stage_input)
return current_output
# Example: Document processing pipeline
pipeline = PipelineAgent([
(extraction_agent, lambda x: f"Extract key information: {x}"),
(analysis_agent, lambda x: f"Analyze this information: {x}"),
(summary_agent, lambda x: f"Summarize the analysis: {x}")
])
3. Debate/Critic Pattern
Multiple agents discuss and refine answers.
class DebateSystem:
def __init__(self, proposer, critic, judge):
self.proposer = proposer
self.critic = critic
self.judge = judge
async def run(self, question: str, max_rounds: int = 3) -> str:
# Initial proposal
proposal = await self.proposer.run(question)
for round in range(max_rounds):
# Critic reviews
critique = await self.critic.run(
f"Question: {question}\nProposal: {proposal}\n"
"Identify weaknesses and suggest improvements."
)
# Check if critique is satisfied
if "no significant issues" in critique.lower():
break
# Proposer revises
proposal = await self.proposer.run(
f"Question: {question}\n"
f"Previous answer: {proposal}\n"
f"Critique: {critique}\n"
"Revise your answer addressing the critique."
)
# Judge makes final decision
return await self.judge.run(
f"Question: {question}\n"
f"Final proposal: {proposal}\n"
"Provide the final, polished answer."
)
Choosing the Right Pattern
| Pattern | Use Case | Complexity | Latency |
|---|---|---|---|
| Single Agent | Simple tasks, few tools | Low | Low |
| Coordinator | Complex tasks, multiple domains | Medium | Medium |
| Pipeline | Sequential processing | Low | High |
| Debate | High-stakes decisions | High | Very High |
Interview Tip
When designing agent systems in interviews, always consider:
- Task decomposition - Can it be broken into subtasks?
- Tool selection - What capabilities does each agent need?
- Failure handling - What if an agent fails mid-task?
- Latency budget - Multi-agent adds latency
Next, we'll explore tool registry design for dynamic agent capabilities. :::