Originally published on tamiz.pro.
Every team that builds a multi-agent system starts the same way: three agents chatting in a Jupyter notebook, orchestrator routes hard-coded in YAML, output rendered by print()
. The demo works. The stakeholders nod. And then someone tries to hit the endpoint with 50 concurrent requests and the whole stack collapses into recursive loops, unbounded token consumption, and a Postgres table that forgot its indexes.
Multi-agent architectures are not harder than monolithic ones because the code is more complex. They are harder because the system's failure modes are emergent — they live in the interactions between agents, not in any single component. A well-tested single agent can still produce a broken system when paired with another well-tested agent whose output format assumptions don't match the receiver's input contract.
This article is not about whether to use multi-agent patterns. It is about what actually breaks when you move from prototype to production, and the concrete engineering work required to stop treating production as an afterthought.
A monolithic service fails locally. A multi-agent system fails relationally.
In a single-agent pipeline, latency comes from one model call plus one reasoning path. In a multi-agent system, latency is the sum of every inter-agent message. Accuracy is the product of every agent's conditional probability distribution. If each agent has a 90% chance of producing valid output, a three-agent chain has a 72.9% chance of producing a fully valid result — and that is before you account for the combinatorial explosion of decision branches in the orchestrator.
The failure modes cluster into four categories:
Orchestrator collapse. The coordinator gets stuck in a routing loop, bouncing between agents without a termination condition. This is the most common cause of token overconsumption and the most subtle to detect in development.
Contract drift. Agent A outputs JSON with a field called explanation
. Agent B expects reasoning
. In a notebook, this causes one bad response. In production, it causes cascading silent failures where downstream agents receive malformed inputs and continue processing with garbage data.
State explosion. Every agent maintains its own context window, its own tool history, its own memory. Without explicit state management, these fragments diverge and the system produces contradictory outputs across what should be the same conversation turn.
Observability blindness. You cannot trace a user's request through three agents, five tool calls, and a feedback loop using standard APM. When production breaks, you will not know which agent produced which output or why the orchestrator chose a particular path.
Each of these requires a different class of engineering solution. Getting any one wrong does not just cause a bug — it makes the entire architecture unreliably fragile.
The orchestrator (sometimes called the supervisor, coordinator, or manager) is not a router. It is a state machine with termination logic.
A common prototype pattern looks like this:
async def orchestrate(user_input: str, context: dict) -> dict:
agents = [
Agent("researcher"),
Agent("analyst"),
Agent("writer"),
]
current = user_input
results = []
for agent in agents:
out = await agent.call(current, context)
results.append(out)
current = out # Pass everything to next agent
return {"output": results[-1], "history": results}
This works in a notebook. It fails in production because:
current = out
discards everything except the last agent's output.A production orchestrator needs five capabilities that are almost never present in prototypes:
Every orchestration loop must have explicit exit criteria. The three standard approaches:
Max iteration bound. Hard limit on loop passes. Simple but brittle — a legitimate complex task may need more iterations than your bound allows.
Confidence threshold. The orchestrator evaluates whether the current output meets a quality bar before proceeding. Requires a judgment agent or a scoring function.
Stability detection. Detect when the system's state has stopped changing significantly between iterations. The most sophisticated approach and the hardest to implement correctly.
Context should never be lost between agent calls. The minimal pattern is:
class ConversationState:
def __init__(self):
self.turns: list[dict] = []
self.context_summary: str = ""
self.tool_results: dict[str, Any] = {}
self.decision_log: list[dict] = []
def add_turn(self, agent: str, input_msg: str, output_msg: str):
self.turns.append({
"agent": agent,
"input": input_msg,
"output": output_msg,
"timestamp": time.time(),
})
self._update_summary()
def _update_summary(self):
if len(self.turns) > self.max_turns:
self.context_summary = summarize(self.turns[:self.max_turns])
else:
self.context_summary = ""
The key insight: context summarization is not optional in production. A 50-turn conversation with full message history will exceed every context window. You need a rolling summary strategy that preserves semantic content while bounding token usage.
Every routing decision the orchestrator makes should be logged with:
This is the foundation of post-mortem analysis. Without it, a production incident leaves you with no evidence of why the system took a particular path.
When an agent fails, the system should not crash. It should fall back to a simpler path:
async def orchestrate_with_fallback(
user_input: str,
state: ConversationState,
primary: Agent,
fallback: Agent
) -> dict:
try:
result = await primary.call(user_input, state)
return result
except (TimeoutError, AgentFailedError, OutputValidationFailed) as e:
logger.warning(f"Primary agent {primary.name} failed: {e}")
state.add_turn("fallback", user_input, "primary_failed")
return await fallback.call(user_input, state)
Production systems do not fail because agents break. They fail because the system treats agent failures as unrecoverable errors instead of routine operational events.
An agent contract defines: what the agent expects as input, what it guarantees as output, and what side effects it may produce. In prototypes, contracts are implicit. In production, they must be explicit, versioned, and validated.
Most teams define agent contracts as "agents output JSON." This is not a contract — it is a hope.
A real contract uses structured validation:
from pydantic import BaseModel, Field
from typing import Literal
class ResearchOutput(BaseModel):
query_type: Literal["factual", "opinion", "procedural"]
findings: list[dict[str, str]] = Field(
min_length=1,
max_length=50
)
confidence: float = Field(ge=0.0, le=1.0)
citations: list[str] = Field(min_length=0, max_length=10)
class Config:
json_schema_extra = {
"example": {
"query_type": "factual",
"findings": [
{"claim": "Rust achieves zero-cost abstractions",
"source": "https://doc.rust-lang.org/book"}
],
"confidence": 0.95,
"citations": ["rust-lang.org"]
}
}
Without schema validation at the contract boundary, you get silently corrupted data flowing between agents. The receiving agent parses invalid JSON, extracts wrong fields, and produces plausible-but-wrong output. This is far worse than an explicit failure — it is a correctness problem disguised as a working system.
Agent contracts change. The researcher agent adds a methodology
field in v2. The analyst agent in v1 does not expect it. If you deploy v2 of the researcher without updating the analyst, you have two choices:
The production approach is backward-compatible evolution: new fields are additive, and agents explicitly declare which contract version they support. The orchestrator routes based on version compatibility, not agent names.
Every agent boundary should have validation:
async def validated_agent_call(
agent: Agent,
input_schema: type[BaseModel],
output_schema: type[BaseModel],
raw_input: str
) -> dict:
validated_input = input_schema.model_validate_json(raw_input)
raw_output = await agent.call(validated_input)
try:
validated_output = output_schema.model_validate_json(raw_output)
except ValidationError as e:
logger.error(f"Agent output failed validation: {e}")
raise OutputValidationFailed(str(e))
return validated_output.model_dump()
This adds latency and complexity. It also prevents the most dangerous class of production bugs: agents producing output that looks correct but violates the contract in subtle ways.
Standard APM tools trace a single service through its calls. A multi-agent system has no single service — it has many agents making calls to each other, to LLM endpoints, and to external tools.
Each agent call must emit a trace with:
import traceloop
from traceloop.sdk import Traceloop
Traceloop.init(app_name="multi-agent-system")
@traceloop.traced("orchestrator.route")
async def orchestrate(user_input: str) -> dict:
with traceloop.SpanAttributes("orchestrator.input", user_input):
decision = await choose_agent(user_input)
with traceloop.SpanAttributes("orchestrator.decision", decision.agent):
result = await decision.agent.call(user_input)
return result
The critical attribute most teams miss: ** orchestrator.decision**. This logs
Each agent's token consumption should be tracked independently. This serves two purposes:
class TokenMeter:
def __init__(self):
self.agents: dict[str, TokenStats] = defaultdict(TokenStats)
def record(self, agent_name: str, prompt_tokens: int, completion_tokens: int):
self.agents[agent_name].prompt += prompt_tokens
self.agents[agent_name].completion += completion_tokens
self.agents[agent_name].total += prompt_tokens + completion_tokens
def get_cost_estimate(self) -> float:
return sum(
stats.prompt * PRICING[self.model].input +
stats.completion * PRICING[self.model].output
for stats in self.agents.values()
)
Every agent should expose: success rate, average latency, token usage trend, and output validation failure rate. When these metrics degrade, you should detect the degradation before users report issues.
The metric most commonly ignored: output validation failure rate. A healthy agent might have 99% success on input but only 60% success on output validation. This means 40% of its outputs violate the contract — and in a multi-agent chain, that propagates as silent corruption.
Testing a single agent is straightforward: feed input, check output. Testing a multi-agent system is harder because the output depends on the interaction of multiple agents, not just one.
Test each agent's contract independently:
async def test_researcher_contract():
agent = ResearcherAgent(model="gpt-4o")
test_cases = [
(
{"query_type": "factual", "topic": "Rust ownership"},
lambda out: isinstance(out.findings, list)
and len(out.findings) > 0
and 0.0 <= out.confidence <= 1.0
),
(
{"query_type": "procedural", "topic": "How to implement a linked list"},
lambda out: len(out.findings) > 0
and all("step" in f for f in out.findings)
),
]
for input_data, validator in test_cases:
result = await agent.call(**input_data)
assert validator(result), f"Contract violated for {input_data}"
Test that agents work correctly together:
async def test_researcher_to_analyst_chain():
researcher = ResearcherAgent()
analyst = AnalystAgent()
research_result = await researcher.call(
query_type="factual",
topic="Kubernetes scheduler algorithms"
)
analysis_result = await analyst.call(
research_findings=research_result.findings,
confidence=research_result.confidence
)
assert hasattr(analysis_result, "recommendation")
assert len(analysis_result.recommendation) > 0
This catches contract drift before it reaches production. Most contract violations are detected too late because teams test agents in isolation.
Simulate agent failures in the orchestrator:
async def test_orchestrator_resilience():
"""Verify the orchestrator degrades gracefully when agents fail."""
failing_agent = FailingAgent(name="researcher", fail_rate=1.0)
healthy_agent = ResearcherAgent()
orchestrator = Orchestrator(
primary=failing_agent,
fallback=healthy_agent
)
result = await orchestrator.orchestrate("Explain Rust lifetimes")
assert result.source == "researcher"
assert result.fallback_triggered == True
Production multi-agent systems should have 100% chaos test coverage for every agent path. An agent failing should never cause the system to fail.
Maintain a dataset of input-output pairs that represent correct behavior. Run them periodically to detect regressions:
REGRESSION_TESTS = [
{
"input": "What is the CAP theorem?",
"expected_fields": ["consistency", "availability", "partition_tolerance"],
"min_confidence": 0.8,
},
{
"input": "Implement a Redis cache in Python",
"expected_tools": ["code_generator", "testing_agent"],
"max_latency_ms": 30000,
},
]
async def run_regression_suite():
for test in REGRESSION_TESTS:
result = await orchestrator.orchestrate(test["input"])
for field in test.get("expected_fields", []):
assert field in result.output, f"Missing field: {field}"
if "min_confidence" in test:
assert result.confidence >= test["min_confidence"]
This is not optional for production. Without regression tests, every prompt update is a potential correctness risk.
Multi-agent systems have more state than any other application architecture. This state exists at three levels:
Conversation state: The current turn's inputs, outputs, and intermediate results.
Session state: Persistent information across turns (user preferences, accumulated knowledge).
System state: Agent configurations, routing rules, and tool registrations.
Each conversation should be modeled as a state machine with explicit transitions:
START → RECEIVING_INPUT → ROUTING_DECISION
→ AGENT_CALL → OUTPUT_VALIDATION
→ (VALID) → RETURN_OUTPUT → END
→ (INVALID) → REFETCH_OR_FALLBACK → OUTPUT_VALIDATION
Transitions that are not explicitly modeled become sources of bugs. What happens when validation fails? Does the system retry? Fall back? Notify the user? If none of these are encoded in the state machine, the behavior is undefined — and undefined behavior in production is a production incident.
Session state should survive process restarts and scale across replicas. The two standard approaches:
PostgreSQL with JSONB: Store sessions as JSONB documents with indexing on frequently queried fields. Simple, powerful, and avoids the complexity of dedicated session stores.
Redis with TTL: Better for high-throughput systems where sessions are short-lived. Faster reads but requires careful eviction policy design.
class SessionStore:
def __init__(self, db: AsyncpgPool):
self.db = db
async def get_session(self, session_id: str) -> ConversationState | None:
row = await self.db.fetchrow(
"SELECT state FROM sessions WHERE id = $1",
session_id
)
return ConversationState.from_dict(row["state"]) if row else None
async def save_session(self, session: ConversationState) -> None:
await self.db.execute(
"INSERT INTO sessions (id, state, updated_at)
VALUES ($1, $2, NOW())
ON CONFLICT (id) DO UPDATE SET state = $2, updated_at = NOW()",
session.id,
session.to_dict()
)
The most common production bug: agents receiving conversations so long that the context window fills with history and leaves no room for new input. Solutions:
The pattern most teams get wrong: context summarization without preservation of key facts. A summary that loses the user's original question is worse than no summary — it causes the agent to answer the wrong question.
Multi-agent systems can cost 10-100x more than single-agent systems for the same task. Each additional agent adds at least one LLM call, and orchestrator loops can multiply this further.
Set hard token budgets for each agent:
class AgentBudget:
def __init__(self, max_tokens: int, max_cost_usd: float):
self.max_tokens = max_tokens
self.max_cost = max_cost_usd
self.spent_tokens = 0
self.spent_cost = 0.0
def can_accept(self, estimated_tokens: int, estimated_cost: float) -> bool:
return (
self.spent_tokens + estimated_tokens <= self.max_tokens
and self.spent_cost + estimated_cost <= self.max_cost
)
def record(self, tokens: int, cost: float):
self.spent_tokens += tokens
self.spent_cost += cost
The orchestrator should prefer cheaper agents when quality is equivalent:
async def cost_aware_route(
request: str,
agents: list[Agent]
) -> Agent:
candidates = []
for agent in agents:
if not agent.budget.can_accept(
estimated_tokens=request.estimated_tokens(),
estimated_cost=request.estimated_cost()
):
continue
quality_score = await agent.estimate_quality(request)
cost_score = 1.0 / max(agent.pricing.per_token, 0.00001)
candidates.append((agent, quality_score * cost_score))
if not candidates:
raise BudgetExceededError("No agent within budget")
return max(candidates, key=lambda x: x[1])[0]
When multiple agents could process the same input independently, batch the requests:
async def batch_agent_calls(tasks: list[Task]) -> list[Result]:
"""Process independent agent calls in parallel."""
results = await asyncio.gather(
*[agent.call(task) for task in tasks],
return_exceptions=True
)
successful = [
r for r in results if not isinstance(r, Exception)
]
failures = [
(i, r) for i, r in enumerate(results)
if isinstance(r, Exception)
]
for i, err in failures:
logger.error(f"Batch item {i} failed: {err}")
return successful
Parallel execution does not reduce per-request cost, but it reduces total wall-clock time and can improve throughput enough to justify the architecture.
Multi-agent systems expand the attack surface in ways that single-agent systems do not:
Prompt injection between agents: Agent A's output becomes Agent B's input. If Agent A produces a prompt injection payload, Agent B will execute it. This is not theoretical — it is the same attack pattern as user-generated content injection, but now it happens internally.
Tool access escalation: Each agent may have different tool permissions. A misconfigured orchestrator might give a low-privilege agent access to high-privilege tools through an intermediate agent.
State poisoning: If session state is shared across users (a configuration error), one user's conversation could inject malicious context into another user's session.
Treat every agent's output as untrusted input to the next agent:
class SanitizedInput:
def __init__(self, raw: str):
self.raw = raw
self.sanitized = self._remove_injection_patterns(raw)
def _remove_injection_patterns(self, text: str) -> str:
patterns = [
r"(?i)ignore previous instructions",
r"(?i)system:\s*(?!\s*$)",
r"(?i)you are now a.*developer",
]
for pattern in patterns:
text = re.sub(pattern, "", text)
return text.strip()
More importantly: structure agent inputs. Never pass raw text between agents. Pass structured data (JSON, schema-validated objects). Injection attacks require semantic interpretation of text — structured data eliminates that attack vector.
Each agent should have only the tools it needs:
class AgentRegistry:
def __init__(self):
self.agents: dict[str, Agent] = {}
self.tool_policies: dict[str, set[str]] = {}
def register_agent(
self,
name: str,
agent: Agent,
allowed_tools: set[str]
):
self.agents[name] = agent
self.tool_policies[name] = allowed_tools
def get_tools(self, agent_name: str) -> set[str]:
return self.tool_policies.get(agent_name, set())
The orchestrator should enforce this policy. An agent requesting a tool it is not authorized to use should receive a permission error, not silent denial or accidental access.
Deploy new agent versions with contract validation at the boundary:
Orchestrator logic changes are riskier than agent changes because they affect all agent interactions. Use canary deployment:
Use feature flags to enable/disable specific agent paths without redeployment:
if feature_flags.is_enabled("use_advanced_researcher"):
agent = AdvancedResearcherAgent()
else:
agent = BasicResearcherAgent()
This allows A/B testing of agent configurations and rapid rollback without code deployment.
Q: How many agents is too many?
There is no hard limit, but the practical ceiling is around 5-7 agents per orchestration layer. Beyond that, the orchestrator's routing complexity grows superlinearly, and observability becomes extremely difficult. If you need more agents, use hierarchical orchestration: multiple coordinators, each managing a subset of agents.
Q: Should I use a single LLM for all agents or different models?
Use different models for different roles when the cost-benefit analysis justifies it. A cheap model for routing decisions, a expensive model for complex reasoning. But do not use different models without contract validation — different models have different output styles and failure modes.
Q: How do I handle agent-to-agent disagreements?
Implement a debate protocol: when agents disagree, route to a third-party arbitrator agent that evaluates both positions. This is expensive but necessary for high-stakes decisions. For routine tasks, simple majority voting or confidence-weighted selection is sufficient.
The single most important lesson: your prototype is not a system. A prototype proves that agents can talk to each other. A production system proves that they can talk to each other reliably, observably, cost-effectively, and securely under real-world conditions.
The gap between those two states is not closed by adding more agents or better prompts. It is closed by treating contracts as code, observability as a requirement, and failure as a predictable event rather than a surprise.
Multi-agent systems are one of the most powerful patterns in modern AI engineering. They are also one of the most unforgiving. The teams that ship are the ones that invest in the boring infrastructure — validation, tracing, state management, and testing — before they need it.
The teams that do not ship learn this lesson during a 2 AM production incident, and by then the cost of fixing it is ten times what it would have been upfront.