Multi-Agent Coordination and Delegation
Multi-agent systems divide complex tasks among specialized agents, each with distinct expertise and tool access. Instead of one agent handling all tools, a manager agent routes requests to specialists: a research agent (web search, document retrieval), an analysis agent (data processing), and a writing agent (formatting output). This division improves reasoning accuracy, isolates failures, and scales to large workflows. This article teaches you to design, implement, and coordinate multi-agent systems in Python.
The key insight is that agents, like humans, work better in teams. A single agent trying to search the web, perform mathematical analysis, and write a report might confuse these tasks. Specialists, each with the right tools and focus, execute more reliably.
Manager-Worker Architecture
The most common multi-agent pattern is a hierarchical manager-worker structure:
- Manager agent receives the user's request, decides which workers to consult, and coordinates their results
- Worker agents are specialists with access to specific tools (research worker, analysis worker, writing worker)
- Aggregation happens at the manager level, combining worker outputs into a final response
import anthropic
from typing import Any
class ManagerAgent:
"""Routes tasks to specialized worker agents."""
def __init__(self):
self.client = anthropic.Anthropic()
self.model = "claude-3-5-sonnet-20241022"
# Define worker responsibilities
self.workers = {
"research": ResearchWorker(),
"analysis": AnalysisWorker(),
"writing": WritingWorker()
}
def route_request(self, user_request: str) -> str:
"""Manager decides which workers to involve."""
# Step 1: Analyze the request
analysis_prompt = f"""
User request: {user_request}
Analyze this request. Which of these workers would help?
- research: web search, data gathering, fact-checking
- analysis: math, statistics, data processing
- writing: formatting, summarization, report generation
Respond with a JSON object listing the workers to consult and what to ask each.
Example: {{"research": "Find current Bitcoin price", "analysis": "Compare to historical average"}}
"""
response = self.client.messages.create(
model=self.model,
max_tokens=1024,
messages=[{"role": "user", "content": analysis_prompt}]
)
analysis_text = response.content[0].text
print(f"Manager analysis: {analysis_text}")
# Parse the analysis (in production, use structured output)
# For now, assume JSON in the response text
import json
import re
json_match = re.search(r'{.*}', analysis_text, re.DOTALL)
if json_match:
work_plan = json.loads(json_match.group())
else:
work_plan = {}
# Step 2: Execute worker tasks
worker_results = {}
for worker_name, task in work_plan.items():
if worker_name in self.workers:
print(f"Delegating to {worker_name}: {task}")
worker_results[worker_name] = self.workers[worker_name].execute(task)
# Step 3: Synthesize results
synthesis_prompt = f"""
User request: {user_request}
Worker results:
{json.dumps(worker_results, indent=2)}
Synthesize these results into a comprehensive, coherent answer.
"""
synthesis_response = self.client.messages.create(
model=self.model,
max_tokens=2048,
messages=[{"role": "user", "content": synthesis_prompt}]
)
return synthesis_response.content[0].text
class ResearchWorker:
"""Handles information gathering."""
def __init__(self):
self.client = anthropic.Anthropic()
self.model = "claude-3-5-sonnet-20241022"
self.tools = [
{
"name": "search_web",
"description": "Search the internet for current information",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
},
{
"name": "fetch_article",
"description": "Retrieve full text of an article by URL",
"input_schema": {
"type": "object",
"properties": {
"url": {"type": "string"}
},
"required": ["url"]
}
}
]
def execute(self, task: str) -> str:
"""Execute a research task."""
messages = [{"role": "user", "content": task}]
for iteration in range(5):
response = self.client.messages.create(
model=self.model,
max_tokens=2048,
tools=self.tools,
messages=messages
)
if response.stop_reason == "end_turn":
return next(
(block.text for block in response.content if hasattr(block, 'text')),
"No result"
)
# Handle tool calls
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
# Simulate tool execution
if block.name == "search_web":
result = f"Search results for '{block.input.get('query')}': [Simulated results]"
elif block.name == "fetch_article":
result = f"Article from {block.input.get('url')}: [Simulated content]"
else:
result = "Unknown tool"
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "user", "content": tool_results})
return "Research incomplete (max iterations)"
class AnalysisWorker:
"""Handles data analysis and calculations."""
def execute(self, task: str) -> str:
"""Execute an analysis task."""
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2048,
messages=[
{"role": "user", "content": f"Analyze and calculate: {task}"}
]
)
return response.content[0].text
class WritingWorker:
"""Handles formatting and report generation."""
def execute(self, task: str) -> str:
"""Execute a writing task."""
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=2048,
messages=[
{"role": "user", "content": f"Format and write: {task}"}
]
)
return response.content[0].text
# Usage
if __name__ == "__main__":
manager = ManagerAgent()
result = manager.route_request(
"Research Bitcoin's price trend over the last year and write a brief market analysis."
)
print(f"\nFinal output:\n{result}")
Peer-to-Peer Agent Networks
In some domains, agents operate as peers without a strict hierarchy. For example, a debate system might have two agents arguing opposing positions:
def agents_debate(topic: str, num_rounds: int = 3) -> str:
"""Two agents debate a topic and reach consensus."""
client = anthropic.Anthropic()
agent_a_position = "pro"
agent_b_position = "con"
conversation = [
{
"role": "user",
"content": f"You argue the PRO position on: {topic}. Make your opening argument."
}
]
for round_num in range(num_rounds):
print(f"\n--- Round {round_num + 1} ---")
# Agent A argues
response_a = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=512,
messages=conversation
)
arg_a = response_a.content[0].text
print(f"Agent A (PRO): {arg_a}")
conversation.append({"role": "assistant", "content": arg_a})
conversation.append({
"role": "user",
"content": f"You argue the CON position. Respond to the above argument."
})
# Agent B argues
response_b = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=512,
messages=conversation
)
arg_b = response_b.content[0].text
print(f"Agent B (CON): {arg_b}")
conversation.append({"role": "assistant", "content": arg_b})
conversation.append({
"role": "user",
"content": f"Agent A responds to the above. Continue the debate."
})
# Consensus
conversation.append({
"role": "user",
"content": "Given the above debate, what are the strongest points from each side? Suggest a balanced conclusion."
})
response_consensus = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=conversation
)
return response_consensus.content[0].text
# Usage
consensus = agents_debate("Should AI agents be used for critical infrastructure decisions?")
print(f"\nConsensus: {consensus}")
Handling Disagreement and Failures
In multi-agent systems, agents may disagree or fail. Handle this gracefully:
def resolve_agent_disagreement(task: str, agents: list, method: str = "majority") -> str:
"""Get multiple agents' opinions and resolve disagreement."""
client = anthropic.Anthropic()
opinions = []
for agent_name in agents:
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=512,
messages=[
{"role": "user", "content": f"{agent_name} perspective: {task}"}
]
)
opinions.append({
"agent": agent_name,
"opinion": response.content[0].text
})
# Aggregate opinions
if method == "majority":
# Simplified: in practice, parse opinions for voting
consensus_prompt = f"""
Multiple agents provided opinions on: {task}
Opinions:
{json.dumps(opinions, indent=2)}
What is the consensus among these agents? Highlight any disagreements.
"""
else:
# method == "debate": have agents discuss opinions
consensus_prompt = f"""
Synthesize these diverse opinions into a balanced recommendation.
"""
consensus_response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=[{"role": "user", "content": consensus_prompt}]
)
return consensus_response.content[0].text
Key Takeaways
- Multi-agent systems assign specialized roles (research, analysis, writing) to improve accuracy and scale
- Manager-worker architectures use a coordinator to route tasks to specialists and aggregate results
- Peer-to-peer networks enable agents to collaborate without hierarchy (debates, consensus-building)
- Route requests based on agent specialization; agents should focus on what they do best
- Handle disagreement and failures by aggregating opinions or falling back to a primary agent
Frequently Asked Questions
How many agents should I have?
Start with 2-3 specialists. Each adds complexity and API calls. More than 5 agents often introduces coordination overhead that exceeds benefits. Use a single generalist agent for simple tasks.
Can agents call other agents?
Yes, but carefully. Nested agent calls multiply API costs and latency. Use a shallow hierarchy: manager calls workers, not workers calling other workers.
How do I ensure agents don't contradict each other?
Share context explicitly. Give all agents a summary of prior decisions. Use a shared "facts" document they reference. Agree on terminology upfront.
What if an agent fails?
Catch errors in the orchestration layer. If a research agent fails, skip its results or use a fallback search tool. Log failures and retry with a different model or approach.