Production Multi-Agent Systems: Architecture Patterns for Enterprise AI

A technical deep-dive into designing, orchestrating, and deploying multi-agent systems that actually work in production.

Why Multi-Agent Systems?

Single-agent architectures hit a ceiling fast. You end up with a monolithic prompt doing everything — retrieval, reasoning, tool execution, validation, output formatting. The context window becomes a junk drawer. The prompt becomes unmanageable. Testing becomes impossible.

Multi-agent systems solve this by decomposing complex workflows into specialized agents with clear boundaries. Each agent owns a domain. Each agent has a focused context. The system becomes composable, testable, and auditable.

But most multi-agent tutorials stop at “hello world” demos. This post covers what actually matters in production: orchestration patterns, state management, communication protocols, failure modes, and cost control.

---

Core Architecture Patterns

1. Orchestrator Pattern

The simplest and most common pattern. One central agent coordinates all others.

┌─────────────────────┐
│    Orchestrator      │
│  (Router + Planner)  │
└─────────┬───────────┘
          │
    ┌─────┼─────┐
    ▼     ▼     ▼
┌──────┐┌──────┐┌──────┐
│ Agent││ Agent││ Agent│
│  A   ││  B   ││  C   │
└──────┘└──────┘└──────┘

When to use: Linear workflows, clear task decomposition, single entry point.

Implementation:

class Orchestrator:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
        self.planner = PlannerAgent()
    
    async def execute(self, task: str) -> Result:
        plan = await self.planner.decompose(task)
        
        results = []
        for step in plan.steps:
            agent = self.agents[step.agent_name]
            context = self._build_context(results, step)
            result = await agent.run(step.instruction, context)
            results.append(result)
        
        return self._synthesize(results)

Tradeoffs:

• ✅ Simple to implement and reason about

• ✅ Easy to add new agents

• ❌ Orchestrator becomes a bottleneck

• ❌ Single point of failure

• ❌ Latency stacks sequentially

---

2. Supervisor Pattern

Orchestrator with decision authority. The supervisor agent doesn’t just route — it evaluates, decides, and can loop or abort.

┌──────────────────────┐
│     Supervisor       │
│  Route → Evaluate →  │
│  Decide → Loop/Abort │
└──────────┬───────────┘
           │
     ┌─────┼─────┐
     ▼     ▼     ▼
   ┌───┐ ┌───┐ ┌───┐
   │ A │ │ B │ │ C │
   └───┘ └───┘ └───┘
     │     │     │
     └─────┼─────┘
           ▼
     ┌──────────┐
     │Supervisor│ ← feedback loop
     └──────────┘

When to use: Workflows requiring quality gates, retry logic, or dynamic routing.

Key difference from Orchestrator: The supervisor receives agent outputs and decides what happens next. It can reject, retry, or redirect.

class Supervisor:
    async def execute(self, task: str) -> Result:
        context = {”task”: task, “history”: []}
        
        while not self._is_complete(context):
            next_agent = self._route(context)
            result = await next_agent.run(
                self._build_prompt(context)
            )
            
            evaluation = await self._evaluate(result, context)
            context[”history”].append({
                “agent”: next_agent.name,
                “result”: result,
                “evaluation”: evaluation
            })
            
            if evaluation.should_abort:
                return self._handle_failure(context)
        
        return self._finalize(context)

Production considerations:

• Set a max iteration count to prevent infinite loops

• Track token usage per evaluation cycle

• Log every routing decision for debugging

---

3. Hierarchical Pattern

Agents organized in a tree. Each level has a supervisor that manages the level below.

           ┌───────────┐
           │  Root     │
           │ Supervisor│
           └─────┬─────┘
          ┌──────┼──────┐
          ▼      ▼      ▼
      ┌──────┐┌──────┐┌──────┐
      │Team A││Team B││Team C│
      │Supv. ││Supv. ││Supv. │
      └──┬───┘└──┬───┘└──┬───┘
         │       │       │
       ┌─┴─┐   ┌─┴─┐   ┌─┴─┐
       ▼   ▼   ▼   ▼   ▼   ▼
      A1  A2  B1  B2  C1  C2

When to use: Complex enterprise workflows with team-like decomposition. Think: research team, compliance review pipeline, content production line.

Implementation pattern:

class TeamSupervisor:
    def __init__(self, team_name: str, agents: list[Agent]):
        self.team_name = team_name
        self.agents = {a.name: a for a in agents}
    
    async def execute(self, task: str, available_agents: dict) -> Result:
        # This team only sees its own agents
        relevant_agents = {
            k: v for k, v in available_agents.items()
            if k in self.agents
        }
        
        # Delegate to appropriate team member
        agent_name = self._select_agent(task)
        return await self.agents[agent_name].run(task)

When the hierarchy gets deep (>3 levels), you need:

• Async communication between levels

• Timeout propagation upward

• Result aggregation at each supervisor level

---

4. Mesh / Peer-to-Peer Pattern

No central coordinator. Agents communicate directly with each other through a shared message bus or state store.

┌──────┐     ┌──────┐
│AgentA│◄───►│AgentB│
└──┬───┘     └───┬──┘
   │   ┌─────┐   │
   └──►│Bus/ │◄──┘
       │State│
   ┌──►│Store│◄──┐
   │   └─────┘   │
┌──┴───┐     ┌───┴──┐
│AgentC│◄───►│AgentD│
└──────┘     └──────┘

When to use: Parallel workflows, event-driven systems, agents with equal authority.

Implementation with shared state:

class AgentMesh:
    def __init__(self, agents: list[Agent], state_store: StateStore):
        self.agents = {a.name: a for a in agents}
        self.state = state_store
        self.message_queue = asyncio.Queue()
    
    async def run_parallel(self, tasks: dict[str, str]):
        “”“Run multiple agents in parallel with shared state.”“”
        
        async def run_agent(name: str, task: str):
            context = await self.state.get_context(name)
            result = await self.agents[name].run(task, context)
            await self.state.update(name, result)
            await self.message_queue.put({
                “from”: name,
                “result”: result
            })
        
        # Fire all agents concurrently
        await asyncio.gather(*[
            run_agent(name, task) 
            for name, task in tasks.items()
        ])

Critical production concerns:

• Race conditions on shared state — use locks or event sourcing

• Message ordering guarantees

• Dead letter queues for failed messages

---

5. Pipeline Pattern

Agents arranged in a linear sequence. Each agent processes the output of the previous one.

┌──────┐   ┌──────┐   ┌──────┐   ┌──────┐
│Input │──►│Agent │──►│Agent │──►│Output│
│      │   │  A   │   │  B   │   │      │
└──────┘   └──────┘   └──────┘   └──────┘
                │                  ▲
                ▼                  │
           ┌──────┐                │
           │Agent │                │
           │  C   │────────────────┘
           └──────┘

When to use: Document processing, ETL-like workflows, content transformation chains.

class Pipeline:
    def __init__(self, stages: list[tuple[str, Agent]]):
        self.stages = stages
    
    async def execute(self, input_data: Any) -> Any:
        current = input_data
        
        for stage_name, agent in self.stages:
            result = await agent.run(
                f”Process stage: {stage_name}”,
                {”input”: current}
            )
            current = result.output
        
        return current

Performance optimization: Pipeline stages can overlap for streaming — start processing stage N+1 before stage N completes if data is chunkable.

---

Communication Protocols

Structured Message Passing

Agents need to communicate with structured contracts, not free-form text.

@dataclass
class AgentMessage:
    sender: str
    recipient: str
    message_type: MessageType  # REQUEST, RESPONSE, EVENT, ERROR
    payload: dict
    metadata: MessageMetadata
    timestamp: datetime

@dataclass
class MessageMetadata:
    correlation_id: str
    parent_id: Optional[str]
    priority: int  # 0=low, 5=critical
    ttl_seconds: int

Message Types

Type Purpose Example REQUEST Ask another agent to do work “Analyze this document” RESPONSE Return results Document analysis output EVENT Broadcast state change “New document available” ERROR Report failure “Retrieval timeout” HEARTBEAT Health check “I’m alive”

State Management Strategies

Option 1: Shared State Store (Redis/PostgreSQL)

class SharedState:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    async def get(self, key: str, agent_id: str) -> dict:
        raw = await self.redis.get(f”state:{key}”)
        return json.loads(raw) if raw else {}
    
    async def update(self, key: str, data: dict, agent_id: str):
        current = await self.get(key, agent_id)
        current.update(data)
        current[”_last_updated_by”] = agent_id
        current[”_timestamp”] = datetime.utcnow().isoformat()
        await self.redis.set(f”state:{key}”, json.dumps(current))

Option 2: Event Sourcing

class EventSourcedState:
    def __init__(self, event_store):
        self.events = event_store
    
    async def append(self, event: StateEvent):
        await self.events.append(event)
    
    async def get_current_state(self, entity_id: str) -> dict:
        events = await self.events.get_all(entity_id)
        state = {}
        for event in events:
            state = self._apply_event(state, event)
        return state

Tradeoffs:

Approach Pros Cons Shared Store Simple, fast reads Concurrency issues, tight coupling Event Sourcing Full audit trail, replayable Storage overhead, complexity Message Passing Decentralized, scalable No single source of truth

---

Production Concerns

1. Observability

Multi-agent systems are hard to debug. You need:

class ObservableAgent:
    def __init__(self, name: str, tracer: Tracer):
        self.name = name
        self.tracer = tracer
    
    async def run(self, task: str, context: dict) -> Result:
        with self.tracer.start_span(
            f”agent.{self.name}.run”,
            attributes={
                “agent.name”: self.name,
                “task.length”: len(task),
                “context.keys”: list(context.keys())
            }
        ) as span:
            try:
                result = await self._execute(task, context)
                span.set_attribute(”result.length”, len(str(result)))
                return result
            except Exception as e:
                span.record_exception(e)
                span.set_status(StatusCode.ERROR, str(e))
                raise

What to trace per agent:

• Input/output (with PII redaction)

• Token usage (input + output)

• Latency per call

• Tool invocations

• Routing decisions

2. Cost Control

Multi-agent systems multiply token costs. A single user query might trigger 5–10 agent calls.

class CostController:
    def __init__(self, budget_per_query: float = 0.50):
        self.budget = budget_per_query
        self.spent = 0.0
    
    async def check_budget(self, estimated_tokens: int, model: str) -> bool:
        cost = self._estimate_cost(estimated_tokens, model)
        if self.spent + cost > self.budget:
            logger.warning(
                f”Budget exceeded: {self.spent:.4f} + {cost:.4f} > {self.budget}”
            )
            return False
        return True
    
    def record_usage(self, input_tokens: int, output_tokens: int, model: str):
        cost = self._calculate_cost(input_tokens, output_tokens, model)
        self.spent += cost

Cost optimization strategies:

• Use cheaper models for routing/trivial tasks (Haiku for routing, Sonnet for reasoning)

• Cache agent outputs for repeated queries

• Set per-agent token budgets

• Implement circuit breakers for runaway costs

3. Failure Handling

class ResilientAgent:
    def __init__(self, agent: Agent, max_retries: int = 3):
        self.agent = agent
        self.max_retries = max_retries
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60
        )
    
    async def run(self, task: str, context: dict) -> Result:
        if not self.circuit_breaker.allow_request():
            raise CircuitOpenError(”Agent circuit breaker is open”)
        
        for attempt in range(self.max_retries):
            try:
                result = await self.agent.run(task, context)
                self.circuit_breaker.record_success()
                return result
            except RateLimitError:
                wait_time = (2 ** attempt) * 1.0  # Exponential backoff
                await asyncio.sleep(wait_time)
            except Exception as e:
                self.circuit_breaker.record_failure()
                if attempt == self.max_retries - 1:
                    raise

4. Evaluation

You can’t improve what you don’t measure.

@dataclass
class AgentEvaluation:
    task_completion: float  # 0-1, did the agent complete its task?
    output_quality: float   # 0-1, quality of the output
    latency_ms: float
    token_cost: float
    error_rate: float
    
class MultiAgentEvaluator:
    def evaluate_system(
        self, 
        test_cases: list[TestCase],
        system: AgentSystem
    ) -> SystemEvaluation:
        results = []
        for case in test_cases:
            output = system.execute(case.input)
            results.append(self._score(case, output))
        
        return SystemEvaluation(
            mean_task_completion=np.mean([r.task_completion for r in results]),
            p95_latency=np.percentile([r.latency_ms for r in results], 95),
            total_cost=sum(r.token_cost for r in results),
            error_rate=np.mean([r.error_rate for r in results])
        )

---

Decision Framework

When to Use Multi-Agent

Factor Single Agent Multi-Agent Task complexity Simple, linear Complex, multi-domain Context requirements Fits in one window Exceeds one window Auditability Optional Required Team structure Solo Specialized roles Error handling Simple retries Complex recovery Cost sensitivity Low High (need optimization)

Pattern Selection Guide

Is the workflow linear?
├── Yes → Pipeline Pattern
└── No → Is there a central decision maker?
    ├── Yes → Supervisor Pattern
    └── No → Do agents need to share state frequently?
        ├── Yes → Mesh Pattern
        └── No → Is the hierarchy deep (>3 levels)?
            ├── Yes → Hierarchical Pattern
            └── No → Orchestrator Pattern

---

Real-World Example: Enterprise Document Processing

Here’s a production multi-agent system for processing financial documents:

# System Architecture:
# Intake Agent → Classification Agent → Extraction Agent → 
# Validation Agent → Compliance Agent → Output Agent

class DocumentProcessingSystem:
    def __init__(self):
        self.agents = {
            “intake”: IntakeAgent(),        # File parsing, OCR
            “classify”: ClassificationAgent(),  # Document type
            “extract”: ExtractionAgent(),    # Data extraction
            “validate”: ValidationAgent(),   # Schema validation
            “compliance”: ComplianceAgent(), # Regulatory checks
            “output”: OutputAgent(),         # Format & deliver
        }
        self.supervisor = SupervisorAgent(
            routing_rules=RoutingRules.load(”document_routing.yaml”)
        )
    
    async def process(self, document: Document) -> ProcessedDocument:
        # Supervisor determines which agents to invoke
        plan = await self.supervisor.create_plan(document)
        
        results = {}
        for step in plan.ordered_steps():
            agent = self.agents[step.agent_name]
            context = {
                “document”: document,
                “previous_results”: results,
                “requirements”: step.requirements
            }
            
            result = await agent.process(
                instruction=step.instruction,
                context=context
            )
            
            # Supervisor validates before proceeding
            validation = await self.supervisor.validate_step(
                step, result
            )
            
            if not validation.passed:
                result = await self._handle_failure(step, result, validation)
            
            results[step.agent_name] = result
        
        return await self.agents[”output”].format(results)

---

Common Pitfalls

1. Over-Decomposition

Don’t create an agent for every tiny task. Each agent adds:

• Context-switching overhead

• Communication latency

• Cost multiplication

• Debugging complexity

Rule of thumb: If an agent does less than 3 meaningful operations, it should probably be a function, not an agent.

2. Circular Dependencies

Agent A calls Agent B calls Agent C calls Agent A. Deadlock or infinite loop.

Solution: Implement call depth tracking.

class DepthLimitedAgent:
    MAX_DEPTH = 5
    
    async def run(self, task: str, context: dict, depth: int = 0):
        if depth >= self.MAX_DEPTH:
            raise RecursionDepthError(
                f”Max agent depth {self.MAX_DEPTH} exceeded”
            )
        
        context[”_call_depth”] = depth + 1
        return await self._execute(task, context)

3. Unstructured Communication

Agents passing free-form text between each other. Information gets lost, hallucinated, or mangled.

Solution: Define strict input/output schemas per agent.

class AgentSchema:
    input_schema = {
        “type”: “object”,
        “properties”: {
            “document_text”: {”type”: “string”},
            “document_type”: {”type”: “string”, “enum”: [”invoice”, “contract”, “report”]},
            “extraction_fields”: {”type”: “array”, “items”: {”type”: “string”}}
        },
        “required”: [”document_text”, “document_type”]
    }
    
    output_schema = {
        “type”: “object”,
        “properties”: {
            “extracted_data”: {”type”: “object”},
            “confidence”: {”type”: “number”, “minimum”: 0, “maximum”: 1},
            “flags”: {”type”: “array”}
        },
        “required”: [”extracted_data”, “confidence”]
    }

4. Ignoring Failure Modes

Every agent can fail. Every communication can fail. Plan for it.

Failure matrix:

Failure Impact Mitigation Agent timeout Workflow stalls Timeout + fallback Agent hallucination Corrupted output Validation agent State corruption Data loss Event sourcing Cost overrun Budget breach Per-query limits Rate limiting Delays Exponential backoff Agent disagreement Conflicts Tiebreaker agent

---

Conclusion

Multi-agent systems are not always the answer. Start with a single agent. Decompose when complexity demands it. Choose the pattern that matches your workflow, not the one that looks most impressive on a diagram.

The patterns in this post — Orchestrator, Supervisor, Hierarchical, Mesh, Pipeline — cover 90% of production use cases. Start simple. Add observability from day one. Control costs aggressively. And always have a fallback.

The gap between a demo and production is measured in failure handling, observability, and cost control. Build those first.

---

About the Author

Seyhun Akyurek
AI Solution Architect & Delivery Lead · UAE & KSA
seyhunakyurek.com

Book a free 20-minute automation audit — I’ll map what’s automatable in your delivery. No pitch, no obligation.

Published: August 2026
#AI #MultiAgentSystems #Architecture #EnterpriseAI #ProductionML