The Problem
The supervisor pattern requires a central planner to make all routing decisions. For complex analytical workflows, that's exactly what you want. But for real-time customer-facing flows, the supervisor's extra planning call adds latency that users feel. The swarm pattern eliminates the central planner: each agent decides for itself whether it's the right agent, and if not, it routes to the one that is.
The key insight: routing decisions are often local — the current agent can tell from the user's latest message whether it's still in scope. A billing agent doesn't need a supervisor to know it shouldn't handle "my app keeps crashing."
Swarm architecture
No central supervisor. Each agent has a set of transfer_to_{name} tools that it can call to hand off the conversation to another agent. The runtime manages the active agent and processes transfers.
The mechanism:
- Runtime starts with the triage agent (or whatever agent is most general)
- Agent processes the user message
- If in scope: respond normally
- If out of scope: call
transfer_to_{appropriate_agent}(handoff_note="...") - Runtime switches the active agent, passes full context
- New agent continues from where the previous agent left off
Each agent is a specialist: it knows exactly what it handles, and it knows when to pass the baton.
In a swarm, an agent calls transfer_to_billing_agent(). What must be included in the transfer payload for the handoff to be seamless?
The receiving agent starts cold — it knows nothing about the conversation history or what was already resolved. Without full context, it will ask the user to repeat themselves (frustrating UX) or make incorrect assumptions. The handoff note tells it why it's being transferred — so it can pick up exactly where the previous agent left off.
Designing handoffs
A poor handoff: "Here's the conversation. Good luck." A good handoff has three parts:
1. Conversation history — every message exchanged so far, so the receiving agent has full context without asking the user to repeat themselves.
2. Handoff note — a structured summary written by the transferring agent:
- Why is the transfer happening?
- What was resolved in this conversation?
- What is the open question the receiving agent needs to answer?
3. Context objects — structured data that the receiving agent needs:
- Customer ID, account status, order ID
- Any data already looked up (don't re-query what you already have)
- Flags like
is_premium_customer: true
# Good handoff structure
Handoff(
conversation_history=messages,
handoff_note=(
"Customer asking about shipping delay for order #ORD-8891. "
"Verified billing — no issues there. Issue is with logistics. "
"Customer is frustrated — priority resolution needed."
),
context={
"customer_id": "C-12345",
"order_id": "ORD-8891",
"is_premium": True,
"resolved_issues": ["billing_confirmed_ok"],
},
transfer_history=["billing_agent → shipping_agent"],
)
Two agents in a swarm keep handing off to each other: Agent A transfers to Agent B, Agent B transfers back to Agent A. How do you prevent this?
Circular handoffs are the #1 failure mode in swarm systems. A transfer history tracker with a maximum transfer count (typically 5–10 total transfers per conversation) prevents loops. When the limit is hit, escalate to human — you've found a case not covered by any agent's capabilities.
Swarm vs. supervisor: trade-off table
| Dimension | Swarm | Supervisor |
|---|
| Latency | Lower — no supervisor call | Higher — extra planning call |
| Routing control | Decentralized (each agent decides) | Centralized (supervisor decides) |
| Auditability | Harder — decisions distributed | Easy — single decision log |
| Loop risk | High — need loop prevention | Low — supervisor controls flow |
| Best for | Low-latency customer-facing flows | Complex analytical multi-step workflows |
Use swarm for: customer support, real-time assistants, domain routing with well-defined agent scopes.
Use supervisor for: research pipelines, report generation, analytical workflows with many sequential dependencies.
Build
Build it
Before the full implementation, here's a compact simulation of keyword-based swarm routing with anti-loop protection.
# Minimal swarm routing simulation (no dependencies)
AGENT_KEYWORDS = {
"billing_agent": ["charge", "invoice", "payment", "refund"],
"technical_agent": ["bug", "crash", "error", "broken"],
"shipping_agent": ["ship", "delivery", "track", "order"],
}
def route(message, current_agent, transfer_history, max_transfers=5):
msg = message.lower()
for agent, keywords in AGENT_KEYWORDS.items():
if agent != current_agent and any(k in msg for k in keywords):
if len(transfer_history) >= max_transfers:
return current_agent, "escalate_to_human"
transfer_history.append(f"{current_agent} -> {agent}")
return agent, "transferred"
return current_agent, "handled"
messages = [
"Was I charged twice on my last invoice?",
"Also, the app keeps crashing on login.",
"And where is my order? It shipped 3 days ago.",
]
active_agent = "billing_agent"
transfer_history = []
for msg in messages:
active_agent, action = route(msg, active_agent, transfer_history)
print(f"[{action}] active_agent={active_agent} msg={msg[:40]!r}")
print(f"\nTotal transfers: {len(transfer_history)} -> {transfer_history}")
The implementation below creates a 3-agent customer support swarm (billing, technical, shipping) with anti-loop protection and full context transfer.
"""
lesson_04_03_swarm.py
Swarm pattern: agents hand off to each other based on capability, with
full context transfer and anti-loop protection.
"""
import os
from dataclasses import dataclass, field
from typing import Optional
import anthropic
ANTHROPIC_KEY = os.getenv("ANTHROPIC_API_KEY")
client = anthropic.Anthropic(api_key=ANTHROPIC_KEY) if ANTHROPIC_KEY else None
MAX_TRANSFERS = 8 # Anti-loop protection
@dataclass
class Handoff:
"""Context passed during an agent-to-agent transfer."""
to_agent: str
conversation_history: list[dict]
handoff_note: str
context: dict = field(default_factory=dict)
transfer_history: list[str] = field(default_factory=list)
@dataclass
class AgentResponse:
"""What an agent returns after processing a message."""
message: str
handoff: Optional[Handoff] = None
class SwarmAgent:
"""
Base class for swarm agents.
Each agent has a domain and a set of transfer tools for routing.
"""
def __init__(self, name: str, domain: str, system_prompt: str):
self.name = name
self.domain = domain
self.system_prompt = system_prompt
self._peer_agents: dict[str, "SwarmAgent"] = {}
def register_peers(self, agents: dict[str, "SwarmAgent"]):
"""Register other agents this agent can transfer to."""
self._peer_agents = {k: v for k, v in agents.items() if k != self.name}
def process(
self,
user_message: str,
conversation_history: list[dict],
context: dict,
transfer_history: list[str],
) -> AgentResponse:
"""
Process a user message. Return a response or a handoff.
"""
# Build a description of available transfer targets
transfer_options = "\n".join(
f"- transfer_to_{name}: {agent.domain}"
for name, agent in self._peer_agents.items()
)
prompt = (
f"You are the {self.name}. Your domain: {self.domain}\n\n"
f"If this message is in your domain, answer it directly.\n"
f"If it's outside your domain, respond with:\n"
f"TRANSFER: <agent_name>\\nNOTE: <brief handoff note explaining why>\n\n"
f"Available agents:\n{transfer_options}\n\n"
f"Context: {context}\n\n"
f"Customer message: {user_message}"
)
if not client:
# Mock: simple keyword routing for demo without API key
response_text = self._mock_routing(user_message)
else:
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=300,
system=self.system_prompt,
messages=conversation_history + [{"role": "user", "content": prompt}],
)
response_text = response.content[0].text
# Parse TRANSFER directive
if response_text.strip().startswith("TRANSFER:"):
lines = response_text.strip().split("\n")
target_name = lines[0].replace("TRANSFER:", "").strip()
note = lines[1].replace("NOTE:", "").strip() if len(lines) > 1 else "Transferring."
if target_name not in self._peer_agents:
return AgentResponse(
message=f"I'm not the right agent for that, but I can't route to {target_name}. "
f"Please contact support directly."
)
return AgentResponse(
message=f"Let me connect you with our {target_name.replace('_', ' ')}...",
handoff=Handoff(
to_agent=target_name,
conversation_history=conversation_history,
handoff_note=note,
context=context,
transfer_history=transfer_history + [f"{self.name} → {target_name}"],
),
)
return AgentResponse(message=response_text)
def _mock_routing(self, message: str) -> str:
msg_lower = message.lower()
# Billing agent keywords
if self.name == "billing_agent":
if any(w in msg_lower for w in ["charge", "invoice", "payment", "bill", "refund"]):
return "I can help with that billing question. [Mock billing answer]"
elif any(w in msg_lower for w in ["bug", "crash", "error", "broken"]):
return "TRANSFER: technical_agent\nNOTE: Customer reporting a technical issue — outside billing scope."
elif any(w in msg_lower for w in ["ship", "delivery", "track", "arrive"]):
return "TRANSFER: shipping_agent\nNOTE: Customer asking about shipping — not a billing issue."
# Technical agent keywords
elif self.name == "technical_agent":
if any(w in msg_lower for w in ["bug", "crash", "error", "broken", "issue"]):
return "I can help debug that. [Mock technical answer]"
elif any(w in msg_lower for w in ["ship", "delivery", "track"]):
return "TRANSFER: shipping_agent\nNOTE: Shipping question — beyond technical support scope."
# Shipping agent keywords
elif self.name == "shipping_agent":
if any(w in msg_lower for w in ["ship", "delivery", "track", "arrive", "order"]):
return "I can check your shipment status. [Mock shipping answer]"
return f"I'm the {self.name}. [Mock default answer for: {message[:40]}]"
class SwarmRuntime:
"""
Manages the active agent and processes inter-agent transfers.
Enforces transfer limits to prevent infinite loops.
"""
def __init__(self, agents: dict[str, SwarmAgent], entry_agent: str):
self.agents = agents
self.entry_agent = entry_agent
self.max_transfers = MAX_TRANSFERS
# Register peers for all agents
for agent in agents.values():
agent.register_peers(agents)
def run_conversation(self, messages: list[str]) -> list[dict]:
"""
Process a list of user messages through the swarm.
Returns a list of turn records showing routing decisions.
"""
active_agent_name = self.entry_agent
conversation_history: list[dict] = []
context: dict = {}
transfer_history: list[str] = []
trace: list[dict] = []
for user_message in messages:
print(f"\n[user] {user_message}")
conversation_history.append({"role": "user", "content": user_message})
# Anti-loop check
if len(transfer_history) >= self.max_transfers:
print(f" [swarm] Transfer limit reached ({self.max_transfers}). Escalating to human.")
trace.append({
"message": user_message,
"active_agent": active_agent_name,
"response": "I'm escalating this to a human agent for specialized help.",
"transferred_to": None,
"escalated": True,
})
break
agent = self.agents[active_agent_name]
response = agent.process(user_message, conversation_history, context, transfer_history)
turn = {
"message": user_message,
"active_agent": active_agent_name,
"response": response.message,
"transferred_to": None,
"escalated": False,
}
if response.handoff:
h = response.handoff
transfer_history = h.transfer_history
context.update(h.context)
active_agent_name = h.to_agent
print(f" [swarm] HANDOFF: {turn['active_agent']} → {active_agent_name}")
print(f" [swarm] Note: {h.handoff_note}")
print(f" [swarm] Transfers so far: {transfer_history}")
turn["transferred_to"] = active_agent_name
turn["handoff_note"] = h.handoff_note
# Get response from the new agent immediately
new_agent = self.agents[active_agent_name]
follow_up = new_agent.process(
user_message, h.conversation_history, context, transfer_history
)
conversation_history.append({"role": "assistant", "content": follow_up.message})
print(f" [{active_agent_name}] {follow_up.message}")
else:
conversation_history.append({"role": "assistant", "content": response.message})
print(f" [{active_agent_name}] {response.message}")
trace.append(turn)
return trace
def build_support_swarm() -> SwarmRuntime:
"""Build a 3-agent customer support swarm."""
billing = SwarmAgent(
name="billing_agent",
domain="billing questions, payment issues, invoices, refunds, subscription charges",
system_prompt=(
"You are a billing specialist. You handle payment-related questions. "
"If the question is not billing-related, transfer to the appropriate agent."
),
)
technical = SwarmAgent(
name="technical_agent",
domain="technical bugs, app crashes, API errors, integration issues",
system_prompt=(
"You are a technical support engineer. You handle software bugs and technical issues. "
"If the question is not technical, transfer to the appropriate agent."
),
)
shipping = SwarmAgent(
name="shipping_agent",
domain="order shipping, delivery tracking, estimated arrival, logistics",
system_prompt=(
"You are a shipping specialist. You handle order tracking and delivery questions. "
"If the question is not shipping-related, transfer to the appropriate agent."
),
)
agents = {
"billing_agent": billing,
"technical_agent": technical,
"shipping_agent": shipping,
}
return SwarmRuntime(agents=agents, entry_agent="billing_agent")
if __name__ == "__main__":
swarm = build_support_swarm()
# Conversation that requires 2 handoffs:
# billing → technical → shipping
conversation = [
"Hi, I have a question about my last invoice — was I charged correctly?",
"Thanks! Actually, I'm also getting an error when I try to log in to the dashboard.",
"Got it. One more thing — where is my order? I ordered 3 days ago.",
]
print("=" * 60)
print("CUSTOMER SUPPORT SWARM — 3-Agent Demo")
print("=" * 60)
trace = swarm.run_conversation(conversation)
print("\n" + "=" * 60)
print("CONVERSATION TRACE")
print("=" * 60)
for i, turn in enumerate(trace, 1):
print(f"\nTurn {i}:")
print(f" User: {turn['message'][:60]}")
print(f" Agent: {turn['active_agent']}")
if turn.get("transferred_to"):
print(f" → Transferred: {turn['transferred_to']}")
print(f" → Note: {turn.get('handoff_note', '')[:80]}")
elif turn.get("escalated"):
print(f" → ESCALATED TO HUMAN")