Multi-Agent Systems › Code Implementation
Michael BrenndoerferJanuary 3, 20269 min read
Language AI HandbookAll chapters Part 48 of 67Tool Use and Agents
Multi-Agent Systems › Code ImplementationOne section · Sections in this chapter8 / 14 ↓
Sections in this chapter14
Reading tools Definitions & inline links
Choose your expertise level to adjust how many terms are explained. Beginners see more tooltips, experts see fewer to maintain reading flow. Hover over underlined terms for instant definitions.
Article links
Make inline references clickable
Let's implement a simplified multi-agent coordination system that demonstrates the core mechanisms: an orchestrator that decomposes tasks, dispatches to specialized agents, and synthesizes results.
Setting Up the Agent Framework
In[3]:
Codefrom dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import List, Optional
class TaskStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class Message:
sender: str
recipient: str
message_type: str
content: str
conversation_id: str
timestamp: str = field(
default_factory=lambda: datetime.utcnow().isoformat()
)
@dataclass
class Task:
task_id: str
description: str
assigned_to: str
status: TaskStatus = TaskStatus.PENDING
result: Optional[str] = None
dependencies: List[str] = field(default_factory=list)
created_at: str = field(
default_factory=lambda: datetime.utcnow().isoformat()
)Out[4]:
ConsoleAgent framework classes defined: TaskStatus values: ['pending', 'running', 'completed', 'failed'] Message fields: sender, recipient, message_type, content, conversation_id, timestamp Task fields: task_id, description, assigned_to, status, result, dependencies
The Task dataclass carries all information about a work item: who should do it, its current status, and any prerequisite tasks that must complete first. The Message dataclass models explicit inter-agent communication with sender and recipient fields alongside type information. Together they form the minimal data model needed to implement dependency tracking and message routing.
Implementing the Shared Workspace
In[5]:
Codeclass SharedWorkspace:
"""Central coordination store for a multi-agent system."""
def __init__(self):
self.tasks: Dict[str, Task] = {}
self.messages: List[Message] = []
self.results: Dict[str, str] = {}
def add_task(self, task: Task) -> None:
self.tasks[task.task_id] = task
def update_task_status(
self, task_id: str, status: TaskStatus, result: Optional[str] = None
) -> None:
if task_id in self.tasks:
self.tasks[task_id].status = status
if result is not None:
self.tasks[task_id].result = result
self.results[task_id] = result
def get_ready_tasks(self) -> List[Task]:
"""Return tasks whose dependencies are all completed."""
ready = []
for task in self.tasks.values():
if task.status != TaskStatus.PENDING:
continue
all_deps_complete = all(
self.tasks.get(dep, Task("", "", "")).status
== TaskStatus.COMPLETED
for dep in task.dependencies
)
if all_deps_complete:
ready.append(task)
return ready
def post_message(self, message: Message) -> None:
self.messages.append(message)
def get_messages_for(self, recipient: str) -> List[Message]:
return [m for m in self.messages if m.recipient in (recipient, "all")]Implementing Specialized Agents
In[6]:
Codeimport json
from typing import Callable
class Agent:
"""A specialized agent that can execute tasks and communicate."""
def __init__(self, agent_id: str, role: str, execute_fn: Callable):
self.agent_id = agent_id
self.role = role
self._execute = execute_fn
def handle_task(self, task: Task, workspace: SharedWorkspace) -> str:
workspace.update_task_status(task.task_id, TaskStatus.RUNNING)
try:
result = self._execute(task, workspace)
workspace.update_task_status(
task.task_id, TaskStatus.COMPLETED, result
)
workspace.post_message(
Message(
sender=self.agent_id,
recipient="orchestrator",
message_type="task_completed",
content=json.dumps(
{"task_id": task.task_id, "summary": result[:100]}
),
conversation_id=task.task_id,
)
)
return result
except Exception as e:
workspace.update_task_status(task.task_id, TaskStatus.FAILED)
workspace.post_message(
Message(
sender=self.agent_id,
recipient="orchestrator",
message_type="task_failed",
content=json.dumps(
{"task_id": task.task_id, "error": str(e)}
),
conversation_id=task.task_id,
)
)
raiseNotice the try-except structure: whether the task succeeds or fails, the agent always sends a message to the orchestrator and always updates the task status. This guarantees that the orchestrator is never left waiting for a response that will never come due to an unhandled exception, which would cause the system to stall.
In[7]:
Code# Define mock execution functions simulating specialized agents
def researcher_execute(task: Task, workspace: SharedWorkspace) -> str:
"""Simulate a research agent finding relevant information."""
topic = task.description.replace("Research: ", "")
findings = {
"EV adoption rates": (
"Norway leads at 82% EV share. Germany at 18%, France at 15%, UK at 17%. "
"Market grew 35% YoY across EU in 2023."
),
"charging infrastructure": (
"EU has 630,000 public charging points. Target is 3.5M by 2030. "
"Netherlands has highest density; Eastern Europe lags significantly."
),
"policy incentives": (
"Germany offers 4,500 EUR subsidy on new EVs. France 5,000 EUR. "
"Norway eliminated VAT entirely. Many countries phase out ICE sales by 2035."
),
}
for key in findings:
if any(word in topic.lower() for word in key.split()):
return findings[key]
return f"Research complete for: {topic}. Key data points gathered."
def analyst_execute(task: Task, workspace: SharedWorkspace) -> str:
"""Synthesize results from completed upstream research tasks."""
dep_results = [workspace.results.get(dep, "") for dep in task.dependencies]
combined = " ".join(dep_results)
return (
"Key EV adoption factors identified: (1) Policy support via subsidies and ICE bans "
"drives adoption most strongly, as shown by Norway's extreme subsidy-adoption correlation. "
"(2) Charging infrastructure availability limits adoption in lower-density regions. "
"(3) Price parity with ICE vehicles is approaching in multiple segments by 2025."
)
def writer_execute(task: Task, workspace: SharedWorkspace) -> str:
"""Draft a structured report from analysis results."""
analysis = workspace.results.get(
task.dependencies[0], "No analysis available"
)
return (
f"## EV Adoption in Europe: Key Factors\n\n"
f"Based on current data: {analysis}\n\n"
f"**Recommendation**: Prioritize charging infrastructure investment alongside "
f"demand-side incentives for balanced adoption growth."
)Running the Orchestrator
In[8]:
Codefrom typing import Dict
class Orchestrator:
"""Manages task assignment, dependency tracking, and result synthesis."""
def __init__(self, workspace: SharedWorkspace):
self.workspace = workspace
self.agents: Dict[str, Agent] = {}
self.agent_id = "orchestrator"
def register_agent(self, agent: Agent) -> None:
self.agents[agent.agent_id] = agent
def decompose_and_run(self, goal: str, tasks: List[Task]) -> Dict[str, str]:
"""Add all tasks, then execute them in dependency order."""
for task in tasks:
self.workspace.add_task(task)
max_rounds = 20
for round_num in range(max_rounds):
ready = self.workspace.get_ready_tasks()
if not ready:
pending = [
t
for t in self.workspace.tasks.values()
if t.status == TaskStatus.PENDING
]
if not pending:
break
else:
break
for task in ready:
if task.assigned_to in self.agents:
self.agents[task.assigned_to].handle_task(
task, self.workspace
)
return {
tid: t.result
for tid, t in self.workspace.tasks.items()
if t.status == TaskStatus.COMPLETED
}In[9]:
Code# Assemble the multi-agent system
workspace = SharedWorkspace()
orchestrator = Orchestrator(workspace)
researcher_a = Agent("researcher_a", "EV researcher", researcher_execute)
researcher_b = Agent(
"researcher_b", "Infrastructure researcher", researcher_execute
)
researcher_c = Agent("researcher_c", "Policy researcher", researcher_execute)
analyst = Agent("analyst", "Data analyst", analyst_execute)
writer = Agent("writer", "Report writer", writer_execute)
for agent in [researcher_a, researcher_b, researcher_c, analyst, writer]:
orchestrator.register_agent(agent)
tasks = [
Task("t1", "Research: EV adoption rates", "researcher_a"),
Task("t2", "Research: charging infrastructure", "researcher_b"),
Task("t3", "Research: policy incentives", "researcher_c"),
Task(
"t4",
"Analyze: synthesize research findings",
"analyst",
dependencies=["t1", "t2", "t3"],
),
Task("t5", "Write: final report", "writer", dependencies=["t4"]),
]
results = orchestrator.decompose_and_run("EV adoption report", tasks)Out[10]:
Console=== Multi-Agent Pipeline Execution === Total tasks: 5 Completed tasks: 5 Messages exchanged: 5 [COMPLETED ] t1: Research: EV adoption rates [COMPLETED ] t2: Research: charging infrastructure [COMPLETED ] t3: Research: policy incentives [COMPLETED ] t4: Analyze: synthesize research findings [COMPLETED ] t5: Write: final report --- Final Report --- ## EV Adoption in Europe: Key Factors Based on current data: Key EV adoption factors identified: (1) Policy support via subsidies and ICE bans drives adoption most strongly, as shown by Norway's extreme subsidy-adoption correlation. (2) Charging infrastructure availability limits adoption in lower-density regions. (3) Price parity with ICE vehicles is approaching in multiple segments by 2025. **Recommendation**: Prioritize charging infrastructure investment alongside demand-side incentives for balanced adoption growth.
The orchestrator executes tasks in topological order determined by dependencies. The three research tasks complete independently (in parallel in a real async system), the analysis task waits for all three, and the writer produces the final report from the analysis. The message log captures every task completion and failure event for debugging and audit purposes.
Tracking Message Flow
In[11]:
Codedef analyze_message_flow(workspace: SharedWorkspace) -> Dict[str, int]:
"""Count messages by type and sender."""
type_counts: Dict[str, int] = {}
sender_counts: Dict[str, int] = {}
for msg in workspace.messages:
type_counts[msg.message_type] = type_counts.get(msg.message_type, 0) + 1
sender_counts[msg.sender] = sender_counts.get(msg.sender, 0) + 1
return {"by_type": type_counts, "by_sender": sender_counts}Out[12]:
ConsoleMessage flow analysis:
By type: {'task_completed': 5}
By sender: {'researcher_a': 1, 'researcher_b': 1, 'researcher_c': 1, 'analyst': 1, 'writer': 1}
Total coordination messages: 5Message flow analysis reveals the communication overhead of multi-agent coordination. Every task completion generates a message from the executing agent to the orchestrator, confirming that coordination cost scales linearly with the number of tasks in this design. In a real system with an LLM orchestrator, each message would also require the orchestrator to process the incoming message and decide whether to take action, adding additional LLM inference costs on top of the raw message count.
Simulating Failure and Recovery
In[13]:
Codedef unreliable_execute(task: Task, workspace: SharedWorkspace) -> str:
"""Simulate an agent that fails on first attempt."""
if not hasattr(unreliable_execute, "_attempts"):
unreliable_execute._attempts = {}
attempt_key = task.task_id
unreliable_execute._attempts[attempt_key] = (
unreliable_execute._attempts.get(attempt_key, 0) + 1
)
if unreliable_execute._attempts[attempt_key] < 2:
raise RuntimeError(
f"Transient failure on attempt {unreliable_execute._attempts[attempt_key]}"
)
return f"Succeeded on attempt {unreliable_execute._attempts[attempt_key]}"
class ResilientOrchestrator(Orchestrator):
"""Orchestrator with retry logic for failed tasks."""
def __init__(self, workspace: SharedWorkspace, max_retries: int = 3):
super().__init__(workspace)
self.max_retries = max_retries
self.retry_counts: Dict[str, int] = {}
def decompose_and_run(self, goal: str, tasks: List[Task]) -> Dict[str, str]:
for task in tasks:
self.workspace.add_task(task)
max_rounds = 30
for round_num in range(max_rounds):
ready = self.workspace.get_ready_tasks()
if not ready:
failed = [
t
for t in self.workspace.tasks.values()
if t.status == TaskStatus.FAILED
]
retryable = [
t
for t in failed
if self.retry_counts.get(t.task_id, 0) < self.max_retries
]
if retryable:
for task in retryable:
task.status = TaskStatus.PENDING
self.retry_counts[task.task_id] = (
self.retry_counts.get(task.task_id, 0) + 1
)
continue
break
for task in ready:
if task.assigned_to in self.agents:
try:
self.agents[task.assigned_to].handle_task(
task, self.workspace
)
except Exception:
pass
return {
tid: t.result
for tid, t in self.workspace.tasks.items()
if t.status == TaskStatus.COMPLETED
}In[14]:
Coderesilient_workspace = SharedWorkspace()
resilient_orch = ResilientOrchestrator(resilient_workspace, max_retries=3)
unreliable_agent = Agent(
"unreliable_researcher", "Unreliable researcher", unreliable_execute
)
resilient_orch.register_agent(unreliable_agent)
retry_tasks = [
Task("r1", "Research: find EV market data", "unreliable_researcher"),
]
retry_results = resilient_orch.decompose_and_run("Retry test", retry_tasks)Out[15]:
ConsoleTask status: completed
Retry count: 1
Result: Succeeded on attempt 2
Messages generated: 2
Message types: {'task_failed': 1, 'task_completed': 1}The resilient orchestrator detects failure, resets the task to pending, and dispatches it again. The unreliable agent fails on its first attempt but succeeds on the second. The message log records both the failure and the eventual success, making the recovery history fully auditable. In a production system you would also add exponential backoff between retries to avoid overwhelming a temporarily overloaded downstream service.
Key Parameters
The key parameters for multi-agent system design are:
- max_retries: Maximum retry attempts before a task is marked permanently failed. Higher values improve resilience against transient failures but can delay failure detection for persistent errors.
- max_rounds: Upper bound on orchestration loops. Prevents infinite loops in systems with circular dependencies or stuck agents.
- dependency graph depth: Deeper DAGs require more sequential execution rounds, limiting parallelism. Flatter graphs execute faster but may sacrifice output quality from synthesis.
- agent specialization depth: Highly specialized agents with narrow prompts and tools outperform generalist agents on in-domain tasks but fail on edge cases outside their specialization.