Proven patterns for building multi-agent systems. Each pattern includes a description and runnable code snippet.
The most fundamental pattern: one agent assigns work to another and waits for a result.
When to use: Any time you need to break work into specialized subtasks.
from src.agent import Agent
from src.messaging import Message
from src.task_manager import TaskManager
class ManagerAgent(Agent):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.task_manager = TaskManager()
async def on_start(self):
task = self.task_manager.assign(
title="Review PR #42",
assignee="reviewer",
assigner=self.name,
)
await self.send_message(
to_agent="reviewer",
payload=task.model_dump(),
msg_type="task_assignment",
)
async def handle_message(self, message: Message):
if message.type == "task_result":
task_id = message.payload["task_id"]
print(f"Task {task_id} completed: {message.payload['result']}")The assigner verifies the assignee's work before marking it complete. This prevents agents from self-certifying their output.
When to use: Any time quality matters (code review, data analysis, deployment approval).
class VerifyingManager(Agent):
async def handle_message(self, message: Message):
if message.type == "task_result":
result = message.payload.get("result", {})
# Verify the work meets requirements
if self._verify(result):
print(f"Verified: {result}")
await self.send_message(
to_agent=message.from_agent,
payload={"status": "approved"},
msg_type="verification_result",
)
else:
# Send back for rework
await self.send_message(
to_agent=message.from_agent,
payload={
"status": "rejected",
"feedback": "Missing test coverage",
},
msg_type="verification_result",
)
def _verify(self, result: dict) -> bool:
return result.get("tests_passing", False)Send the same task or query to multiple agents and collect their responses. Useful for parallel processing or gathering diverse perspectives.
When to use: Parallel data processing, multi-reviewer code review, distributed search.
from src.orchestrator import Orchestrator
async def fan_out_example():
orch = Orchestrator(org_id="myorg")
orch.add_agent(worker_1)
orch.add_agent(worker_2)
orch.add_agent(worker_3)
await orch.start_all()
# Send the same work to all agents
await orch.broadcast(
payload={
"task": "Analyze dataset chunk",
"dataset_url": "s3://data/chunk-*.parquet",
},
msg_type="parallel_task",
)class CollectorAgent(Agent):
def __init__(self, expected_count: int, **kwargs):
super().__init__(**kwargs)
self.results = []
self.expected = expected_count
async def handle_message(self, message: Message):
if message.type == "partial_result":
self.results.append(message.payload)
print(f"Collected {len(self.results)}/{self.expected}")
if len(self.results) >= self.expected:
merged = self._merge_results(self.results)
print(f"All results collected: {merged}")
def _merge_results(self, results: list[dict]) -> dict:
# Combine partial results into a final answer
return {"total_records": sum(r.get("count", 0) for r in results)}Pass work through a chain of agents, each adding their contribution. Each agent processes the output of the previous one.
When to use: Multi-stage processing (e.g., build -> test -> deploy), content pipelines.
PIPELINE = ["builder", "tester", "deployer"]
class PipelineAgent(Agent):
"""Agent that processes work and forwards to the next stage."""
def __init__(self, next_agent: str | None = None, **kwargs):
super().__init__(**kwargs)
self.next_agent = next_agent
async def handle_message(self, message: Message):
if message.type == "pipeline_work":
# Process this stage
result = await self._process(message.payload)
if self.next_agent:
# Forward to next stage
await self.send_message(
to_agent=self.next_agent,
payload={
**message.payload,
f"{self.role}_result": result,
},
msg_type="pipeline_work",
)
else:
# Final stage — report back to originator
await self.send_message(
to_agent=message.payload.get("originator", message.from_agent),
payload={
**message.payload,
f"{self.role}_result": result,
"pipeline_complete": True,
},
msg_type="pipeline_result",
)
async def _process(self, data: dict) -> dict:
# Override in each stage
return {"status": "processed"}
# Wire up the pipeline
builder = PipelineAgent(name="builder", role="build", next_agent="tester", org_id="myorg")
tester = PipelineAgent(name="tester", role="test", next_agent="deployer", org_id="myorg")
deployer = PipelineAgent(name="deployer", role="deploy", next_agent=None, org_id="myorg")Track agent liveness by sending periodic heartbeats. The orchestrator detects agents that stop responding.
When to use: Any production deployment where you need to detect agent failures.
import asyncio
class HeartbeatAgent(Agent):
"""Agent that sends periodic heartbeats to the orchestrator."""
async def on_start(self):
asyncio.create_task(self._heartbeat_loop())
async def _heartbeat_loop(self):
while self._running:
await self.send_message(
to_agent="orchestrator",
payload={
"agent": self.name,
"status": "alive",
"uptime_seconds": self._uptime(),
},
msg_type="heartbeat",
)
await asyncio.sleep(10)
def _uptime(self) -> float:
# Track uptime since start
return 0.0 # implement with time.monotonic()The orchestrator side:
from src.orchestrator import Orchestrator
orch = Orchestrator(org_id="myorg", heartbeat_interval=10.0)
# Check health at any time
health = orch.get_agent_health()
# {'worker-1': True, 'worker-2': True, 'worker-3': False}
for name, healthy in health.items():
if not healthy:
print(f"ALERT: Agent '{name}' is not responding!")Persist agent state across restarts using the MemoryStore.
When to use: Long-running agents that need to remember context, configuration, or intermediate results.
from src.memory import MemoryStore
class StatefulAgent(Agent):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.memory = MemoryStore(f".memory/{self.name}")
async def on_start(self):
# Restore previous state
state = self.memory.load("state")
if state:
print(f"Restored state: processed {state['tasks_completed']} tasks")
else:
self.memory.save("state", {"tasks_completed": 0})
async def handle_message(self, message: Message):
# Process message...
state = self.memory.load("state")
state["tasks_completed"] += 1
self.memory.save("state", state)Real systems combine these patterns. For example, a CI/CD pipeline might use:
- Delegation — CEO assigns a deploy task
- Sequential Pipeline — build -> test -> deploy stages
- Verification — CEO verifies the deployment succeeded
- Heartbeat — monitor all agents throughout
- Memory — track deployment history
See examples/devops_agent.py for a concrete implementation.
- Architecture Guide — system design and component overview
- Getting Started — run your first agent
- agent.ceo — production-grade agent orchestration
- docs.agent.ceo — full documentation