From Prototype to Production: Hard-Won Lessons Building Multi-Agent Systems That Actually Ship

Originally published on tamiz.pro.

The Gap Between Demo and Day One

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.

1. How Multi-Agent Systems Fail Differently

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.

2. The Orchestrator Is the Hardest Component You Will Build

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:

1. There is no depth limit — if an agent returns an instruction to re-query the researcher, the loop continues indefinitely until tokens run out or the system times out.
2. There is no context merging strategy — current = out discards everything except the last agent's output.
3. There is no error handling — if the analyst fails, the writer never runs, and the user sees a timeout instead of a graceful degradation.
4. There is no cost control — every agent call is unbounded in token usage.

A production orchestrator needs five capabilities that are almost never present in prototypes:

Termination conditions

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 preservation

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):
        # Rolling summary to prevent context overflow
        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.

Decision logging

Every routing decision the orchestrator makes should be logged with:

• The input that triggered the decision
• The available agents and their capabilities
• The chosen agent and the reasoning
• The cost and latency of the call

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.

Graceful degradation

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.

3. Agent Contracts Are Where Most Prototypes Die

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.

The JSON schema trap

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.

Contract versioning

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:

1. The analyst ignores the new field (safe but loses information)
2. The analyst fails on unexpected input (correct but disruptive)

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.

Input/output validation as first-class concern

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:
    # Validate input against contract
    validated_input = input_schema.model_validate_json(raw_input)

    # Call agent
    raw_output = await agent.call(validated_input)

    # Validate output against contract
    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.

4. Observability: You Cannot Fix What You Cannot Trace

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.

Distributed tracing for agents

Each agent call must emit a trace with:

• Trace ID: correlates all spans across the entire request
• Span IDs: one per agent call, one per tool call, one per orchestrator decision
• Attributes: agent name, model used, token counts, latency, decision rationale

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 why the orchestrator chose a particular agent for a particular turn. Without this, a production incident becomes a guessing game about routing logic.

Token cost tracking per agent

Each agent's token consumption should be tracked independently. This serves two purposes:

1. Cost attribution: When the monthly bill spikes, you need to know which agent drove the cost, not just that "the system" did.
2. Performance debugging: If Agent A consistently uses 3x more tokens than expected, something is wrong with its prompts or its output filtering.

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:
        # Model-specific pricing
        return sum(
            stats.prompt * PRICING[self.model].input +
            stats.completion * PRICING[self.model].output
            for stats in self.agents.values()
        )

Agent-level health metrics

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.

5. Testing Multi-Agent Systems: The Hard Part

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.

Unit testing agent contracts

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}"

Integration testing agent chains

Test that agents work correctly together:

async def test_researcher_to_analyst_chain():
    researcher = ResearcherAgent()
    analyst = AnalystAgent()

    # Researcher produces output
    research_result = await researcher.call(
        query_type="factual",
        topic="Kubernetes scheduler algorithms"
    )

    # Analyst must be able to consume researcher's output
    # without validation errors
    analysis_result = await analyst.call(
        research_findings=research_result.findings,
        confidence=research_result.confidence
    )

    # Check that the chain produced a coherent result
    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.

Chaos testing orchestration

Simulate agent failures in the orchestrator:

async def test_orchestrator_resilience():
    """Verify the orchestrator degrades gracefully when agents fail."""

    # Inject failure into the primary agent
    failing_agent = FailingAgent(name="researcher", fail_rate=1.0)
    healthy_agent = ResearcherAgent()

    orchestrator = Orchestrator(
        primary=failing_agent,
        fallback=healthy_agent
    )

    # Should not raise — should fall back
    result = await orchestrator.orchestrate("Explain Rust lifetimes")

    # Result should come from the fallback agent
    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.

Regression testing with golden outputs

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.

6. State Management: The Silent Complexity

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.

Conversation state machines

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 persistence

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()
        )

Context window management

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:

1. Rolling summaries: Replace old turns with summaries after N turns.
2. Selective history: Only include turns relevant to the current query.
3. Context budgeting: Reserve a fixed token budget for history and enforce it strictly.

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.

7. Cost Control: The Business Constraint

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.

Per-agent budgets

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

Cost-aware routing

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]

Batch processing for independent tasks

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
    )

    # Filter out failures but log them
    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.

8. Security: The Multi-Agent Attack Surface

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.

Prompt injection defense

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:
        # Remove common injection patterns
        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.

Least-privilege tool access

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.

9. Deployment Patterns That Work

Blue-green with contract validation

Deploy new agent versions with contract validation at the boundary:

1. Deploy new agent alongside old (blue-green)
2. Route 5% of traffic to new agent
3. Validate all new agent outputs against existing contract
4. If validation pass rate > 99.5%, increase traffic
5. If validation fails, automatically roll back

Canary for orchestrator changes

Orchestrator logic changes are riskier than agent changes because they affect all agent interactions. Use canary deployment:

1. Deploy new orchestrator to a subset of users
2. Monitor error rates, latency, and cost
3. Compare against control group
4. Expand gradually if metrics are healthy

Feature flags for agent paths

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.

Frequently Asked Questions

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.

What I Wish I Knew Before Shipping

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.