Master proven orchestration patterns for coordinating multiple AI agents in enterprise environments -from sequential pipelines to dynamic routing and hierarchical supervision.
TL;DR
Jump to Sequential pipelines · Jump to Parallel execution · Jump to Dynamic routing · Jump to Hierarchical supervision · Jump to Implementation
Building a single AI agent is straightforward. Building multiple agents that work together reliably is where most enterprise implementations stumble. Without deliberate orchestration patterns, you get agents that duplicate work, miss handoffs, or worse -produce conflicting outputs that confuse users.
I've reviewed agent architectures from 30+ enterprise teams over the past year. The successful ones share a common trait: they use explicit orchestration patterns rather than hoping agents "figure it out." This guide breaks down the four patterns that handle virtually every enterprise workflow, with code examples and failure modes to avoid.
Key insight: Agent orchestration isn't about making agents smarter -it's about making their collaboration predictable.
Here's a mistake I see repeatedly: teams invest heavily in prompt tuning and model selection whilst ignoring how agents coordinate. The result? Ten agents that each work brilliantly in isolation but produce rubbish when combined.
Consider this real scenario from a fintech company: they built separate agents for fraud detection, transaction classification, and customer communication. Each agent achieved 92%+ accuracy in testing. But when deployed together, customers received contradictory messages -one agent flagged a transaction as fraud whilst another sent a "payment successful" notification.
The issue wasn't agent quality. It was orchestration. They had no defined handoff protocol, no shared state management, and no conflict resolution logic.
Without orchestration:
With orchestration:
According to Gartner's 2024 AI Orchestration Survey, enterprises with formal orchestration patterns report 68% fewer production incidents and 3.2× faster time-to-resolution compared to ad-hoc implementations (Gartner, 2024).
The sequential pattern is your workhorse -agents execute in a defined order, each consuming the previous agent's output. Think assembly line: Agent A completes its task, hands off to Agent B, which hands off to Agent C.
Ideal for:
Not suitable for:
Stripe Engineering documented their payment reconciliation system that processes 450M+ transactions monthly using a five-agent sequential pipeline (Stripe Engineering Blog, 2024):
Results: Reduced median reconciliation time from 4.2 hours to 58 minutes. Error rate dropped from 2.8% to 0.3%.
from typing import Dict, Any
from openai import OpenAI
client = OpenAI()
class SequentialPipeline:
"""Orchestrates agents in sequential order."""
def __init__(self, agents: list):
self.agents = agents
self.execution_log = []
def execute(self, initial_input: Dict[str, Any]) -> Dict[str, Any]:
"""Run agents sequentially, passing output to next agent."""
current_data = initial_input
for i, agent in enumerate(self.agents):
try:
# Log execution start
self.execution_log.append({
"agent": agent.name,
"stage": i + 1,
"input": current_data,
"status": "started"
})
# Execute agent
result = agent.run(current_data)
# Validate output schema
if not agent.validate_output(result):
raise ValueError(f"{agent.name} produced invalid output")
# Update data for next agent
current_data = result
# Log success
self.execution_log[-1].update({
"status": "completed",
"output": result
})
except Exception as e:
# Log failure and halt pipeline
self.execution_log[-1].update({
"status": "failed",
"error": str(e)
})
raise
return current_data
# Example usage: Document processing pipeline
class ExtractionAgent:
name = "extraction"
def run(self, data):
# Extract text from PDF using OCR
extracted_text = ocr_service.process(data["document_url"])
return {"text": extracted_text, "metadata": data.get("metadata", {})}
def validate_output(self, result):
return "text" in result and len(result["text"]) > 0
class AnalysisAgent:
name = "analysis"
def run(self, data):
# Analyse document structure and extract entities
analysis = client.chat.completions.create(
model="gpt-4-turbo",
messages=[{
"role": "user",
"content": f"Analyse this document and extract key entities:\n\n{data['text']}"
}]
)
return {**data, "entities": parse_entities(analysis)}
def validate_output(self, result):
return "entities" in result
# Build and execute pipeline
pipeline = SequentialPipeline([
ExtractionAgent(),
AnalysisAgent(),
ClassificationAgent(),
StorageAgent()
])
result = pipeline.execute({"document_url": "https://..."})
1. Cascading errors: One agent's failure kills the entire pipeline.
Fix: Implement checkpointing -save intermediate results so you can resume from the failure point rather than restarting from scratch.
2. Bottlenecks: Slow agent blocks all downstream agents.
Fix: Add timeout limits and fallback logic. If Agent B takes >30 seconds, skip it and flag for manual review.
3. Data bloat: Each agent adds fields, creating massive payloads.
Fix: Define strict output schemas. Each agent returns only what downstream agents need.
| Pipeline stage | Input size | Output size | Cumulative overhead |
|---|---|---|---|
| Stage 1 (extraction) | 2 KB | 8 KB | +300% |
| Stage 2 (analysis) | 8 KB | 15 KB | +650% |
| Stage 3 (classification) | 15 KB | 18 KB | +800% |
| Stage 4 (storage) | 18 KB | 3 KB | +50% (cleaned) |
Notice how unchecked growth compounds. Set explicit size limits at each stage.
Parallel execution runs multiple agents simultaneously on the same input, then aggregates results. Like brainstorming: everyone generates ideas independently, then you combine the best ones.
Ideal for:
Not suitable for:
Shopify's Sidekick system uses parallel execution to route 2.3M customer queries monthly across 47 support categories (Shopify Engineering, 2024). Three agents run simultaneously:
All three complete within 400-600ms. The orchestrator aggregates results to determine routing: urgent billing issues go to specialist team, simple queries get auto-responses with help articles.
Results: Reduced median response time from 3.2 hours to 8 minutes for tier-1 queries. Customer satisfaction score increased from 3.8 to 4.6 (out of 5).
import asyncio
from typing import List, Dict, Any
class ParallelOrchestrator:
"""Executes agents in parallel and aggregates results."""
def __init__(self, agents: List[Any], aggregator: callable):
self.agents = agents
self.aggregator = aggregator
async def execute_agent(self, agent, input_data: Dict) -> Dict:
"""Execute single agent asynchronously."""
try:
result = await agent.run_async(input_data)
return {"agent": agent.name, "result": result, "status": "success"}
except Exception as e:
return {"agent": agent.name, "error": str(e), "status": "failed"}
async def execute_all(self, input_data: Dict) -> Dict:
"""Run all agents in parallel."""
tasks = [self.execute_agent(agent, input_data) for agent in self.agents]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out failures
successful_results = [r for r in results if r.get("status") == "success"]
# Aggregate results
if len(successful_results) < len(self.agents) * 0.5:
raise Exception("Too many agent failures")
return self.aggregator(successful_results)
# Example: Multi-source research synthesis
class WebSearchAgent:
name = "web_search"
async def run_async(self, data):
# Search web for relevant information
results = await web_search(data["query"])
return {"sources": results, "confidence": 0.85}
class DatabaseAgent:
name = "database"
async def run_async(self, data):
# Query internal knowledge base
results = await db_search(data["query"])
return {"sources": results, "confidence": 0.92}
class APIAgent:
name = "api"
async def run_async(self, data):
# Fetch from third-party APIs
results = await api_fetch(data["query"])
return {"sources": results, "confidence": 0.78}
def aggregate_research(results: List[Dict]) -> Dict:
"""Combine results from multiple agents."""
all_sources = []
total_confidence = 0
for r in results:
all_sources.extend(r["result"]["sources"])
total_confidence += r["result"]["confidence"]
# Deduplicate and rank by confidence
unique_sources = deduplicate(all_sources)
avg_confidence = total_confidence / len(results)
return {
"sources": unique_sources[:10], # Top 10
"aggregated_confidence": avg_confidence
}
# Execute parallel research
orchestrator = ParallelOrchestrator(
agents=[WebSearchAgent(), DatabaseAgent(), APIAgent()],
aggregator=aggregate_research
)
result = await orchestrator.execute_all({"query": "AI agent best practices"})
Choosing how to combine parallel agent outputs is critical:
1. Voting/consensus: Agents independently classify; majority wins.
2. Weighted ensemble: Combine outputs using confidence scores.
3. Best-of-N selection: Pick the single best output.
4. Merging: Combine complementary information.
Dynamic routing uses a controller agent to decide which specialist agent(s) should handle a task. Like a hospital triage nurse: assess the patient, route to appropriate specialist.
Ideal for:
Not suitable for:
Notion routes user queries across 8 specialist agents based on intent (Notion Engineering, 2024):
A lightweight router agent (GPT-3.5 Turbo, <100ms latency) analyses each query and selects the appropriate specialist.
Results: Increased task success rate from 78% to 91% by routing to domain-optimised agents. Reduced average cost per query by 34% (simple queries use cheaper agents).
class DynamicRouter:
"""Routes requests to appropriate specialist agents."""
def __init__(self, router_agent, specialists: Dict[str, Any]):
self.router = router_agent
self.specialists = specialists
def route(self, request: Dict) -> Dict:
"""Determine which agent(s) should handle request."""
# Router analyses request and selects agents
routing_decision = self.router.decide(request)
selected_agents = [
self.specialists[name]
for name in routing_decision["agents"]
]
# Execute selected agents
results = []
for agent in selected_agents:
result = agent.run(request)
results.append({"agent": agent.name, "output": result})
return {
"routing": routing_decision,
"results": results
}
class RouterAgent:
"""Lightweight agent that decides routing."""
def decide(self, request: Dict) -> Dict:
prompt = f"""
Analyse this request and select appropriate specialist agents.
Request: {request["query"]}
Context: {request.get("context", "None")}
Available specialists:
- document_agent: Summarisation, Q&A about documents
- data_agent: Table creation, data analysis
- writing_agent: Content generation
- search_agent: Information retrieval
Return JSON:
{{
"agents": ["agent_name1", "agent_name2"],
"reasoning": "Why these agents were selected",
"confidence": 0.0-1.0
}}
"""
response = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return parse_routing_decision(response)
# Use router
router = DynamicRouter(
router_agent=RouterAgent(),
specialists={
"document_agent": DocumentAgent(),
"data_agent": DataAgent(),
"writing_agent": WritingAgent(),
"search_agent": SearchAgent()
}
)
result = router.route({
"query": "Create a table comparing our Q4 revenue across regions",
"context": "Financial analysis"
})
Intent-based routing: Classify user intent → map to specialist.
Complexity-based routing: Simple queries → cheap agents; complex → expensive agents.
Load-based routing: Distribute across agent instances to prevent overload.
Hierarchical supervision mimics organisational structure: a supervisor agent coordinates multiple worker agents, makes high-level decisions, and handles escalations.
Ideal for:
Not suitable for:
Glean (enterprise search) uses a three-tier hierarchy for document ingestion (Glean Engineering, 2024):
Tier 1 - Supervisor: Orchestrates workflow, monitors quality, handles errors Tier 2 - Coordinators: Manage document type (PDF, DOCX, HTML, code files) Tier 3 - Workers: Execute specific tasks (OCR, entity extraction, embedding generation)
The supervisor monitors worker output quality. If entity extraction confidence drops below 80%, the supervisor triggers a quality review subprocess and potentially re-routes to a more capable (expensive) model.
Results: Reduced document processing errors by 64%. Improved handling of edge cases (scanned documents, non-English text) without explicit rules.
class SupervisorAgent:
"""Coordinates worker agents and manages quality."""
def __init__(self, workers: List[Any], quality_threshold: float = 0.85):
self.workers = workers
self.quality_threshold = quality_threshold
def supervise(self, task: Dict) -> Dict:
"""Coordinate workers and ensure quality."""
# Create execution plan
plan = self.plan_execution(task)
results = []
for step in plan["steps"]:
# Assign to worker
worker = self.select_worker(step)
# Execute with monitoring
result = self.execute_with_monitoring(worker, step)
# Quality check
if result["quality_score"] < self.quality_threshold:
# Retry with different worker or escalate
result = self.handle_quality_issue(worker, step, result)
results.append(result)
return self.aggregate_results(results)
def plan_execution(self, task: Dict) -> Dict:
"""Supervisor creates high-level plan."""
prompt = f"""
Create an execution plan for this task:
{task["description"]}
Break into steps, identify which worker handles each step.
Available workers: {[w.name for w in self.workers]}
Return structured plan.
"""
# Supervisor uses reasoning to create plan
plan = supervisor_llm.generate(prompt)
return plan
def handle_quality_issue(self, worker, step, poor_result):
"""Supervisor decides how to handle low-quality output."""
# Try different worker
alternative_workers = [w for w in self.workers if w != worker]
for alt_worker in alternative_workers:
retry_result = self.execute_with_monitoring(alt_worker, step)
if retry_result["quality_score"] >= self.quality_threshold:
return retry_result
# If all workers fail, escalate to human
return self.escalate_to_human(step, poor_result)
Proactive supervision: Supervisor creates detailed plan before workers start.
Reactive supervision: Workers execute independently; supervisor intervenes only on failures.
Hybrid supervision: Light planning upfront + monitoring during execution.
Don't force a pattern -let your workflow's structure dictate the choice.
Decision tree:
Does your workflow have strict order dependencies?
├─ Yes → Sequential pipeline
└─ No
├─ Can tasks run independently on same input?
│ └─ Yes → Parallel execution
└─ No
├─ Do different inputs need different processing?
│ └─ Yes → Dynamic routing
└─ No
├─ Is the workflow complex with quality requirements?
│ └─ Yes → Hierarchical supervision
└─ No → You might not need orchestration
60% of workflows fit sequential pipelines. Don't over-engineer with hierarchical supervision if sequential works.
Complexity ladder (start at bottom, move up only if needed):
You can't debug what you can't see. Instrument every orchestration decision:
import logging
import time
class ObservableOrchestrator:
"""Orchestrator with built-in logging and metrics."""
def execute(self, workflow):
start_time = time.time()
logger.info(f"Starting workflow: {workflow.id}")
logger.info(f"Pattern: {workflow.pattern}")
logger.info(f"Input: {workflow.input}")
try:
result = workflow.run()
duration = time.time() - start_time
logger.info(f"Workflow {workflow.id} completed in {duration:.2f}s")
logger.info(f"Result: {result}")
# Track metrics
metrics.record("workflow.duration", duration, {"pattern": workflow.pattern})
metrics.increment("workflow.success", {"pattern": workflow.pattern})
return result
except Exception as e:
duration = time.time() - start_time
logger.error(f"Workflow {workflow.id} failed after {duration:.2f}s: {e}")
metrics.increment("workflow.failure", {"pattern": workflow.pattern})
raise
Track these metrics:
Happy paths are easy. Your orchestration lives or dies on how it handles failures.
Failure injection tests:
def test_agent_timeout():
"""Verify pipeline handles slow agents."""
pipeline = SequentialPipeline([
FastAgent(),
SlowAgent(delay=60), # Inject 60s delay
FastAgent()
])
with pytest.raises(TimeoutError):
pipeline.execute(test_data, timeout=10)
# Verify partial results were saved
assert pipeline.checkpoint_exists()
def test_agent_error_propagation():
"""Verify errors don't cascade silently."""
pipeline = SequentialPipeline([
WorkingAgent(),
FailingAgent(), # Always raises error
WorkingAgent()
])
with pytest.raises(AgentError) as e:
pipeline.execute(test_data)
# Verify error includes context
assert "FailingAgent" in str(e)
assert "stage 2" in str(e)
Test these scenarios:
Adding orchestration logic for every edge case creates brittle systems. The Salesforce Einstein team reported that 40% of their orchestration code handled scenarios that occurred <0.1% of the time (Salesforce Engineering, 2024).
Fix: Handle the 95% case cleanly. Let the 5% fail gracefully with human escalation.
Parallel execution and hierarchical supervision increase LLM API costs. One enterprise team we advised was spending $18K/month on orchestration overhead -supervisor agents monitoring worker agents.
Fix: Cost model your patterns. Use cheaper models (GPT-3.5, Claude Haiku) for orchestration decisions.
| Pattern | Relative cost | When to optimise |
|---|---|---|
| Sequential | 1.0× (baseline) | N/A (already efficient) |
| Parallel | 1.5-3× | >1000 requests/day |
| Dynamic routing | 1.1-1.3× | >5000 requests/day |
| Hierarchical | 2-4× | >500 requests/day |
Hardcoding agent interfaces makes patterns inflexible. If you upgrade one agent, you break the entire pipeline.
Fix: Define strict contracts (input/output schemas). Use protocol buffers or JSON schemas to enforce interfaces.
from pydantic import BaseModel
class AgentInput(BaseModel):
query: str
context: dict
metadata: dict
class AgentOutput(BaseModel):
result: dict
confidence: float
metadata: dict
class Agent:
"""All agents implement this interface."""
def run(self, input: AgentInput) -> AgentOutput:
raise NotImplementedError
Week 1: Audit your current agent workflows
Week 2: Implement one pattern
Week 3: Deploy and monitor
Month 2+: Scale and add patterns
Agent orchestration transforms unpredictable multi-agent chaos into reliable, production-grade systems. Start with sequential pipelines for 80% of workflows, add parallel execution where speed matters, introduce dynamic routing as your agent pool grows, and layer in hierarchical supervision only when complexity demands it. The enterprises getting ROI from AI agents aren't building smarter agents -they're orchestrating them better.
Q: Can you combine orchestration patterns? A: Absolutely. Advanced workflows often nest patterns -for example, a sequential pipeline where each stage uses parallel execution internally. The Notion AI system uses dynamic routing to select a specialist agent, which then coordinates a sequential pipeline of sub-agents.
Q: How do you handle agent versioning in orchestration? A: Use semantic versioning for agents and explicit compatibility declarations. If Agent A requires Agent B v2.x, the orchestrator verifies compatibility before execution. Alternatively, run multiple versions simultaneously and route based on compatibility requirements.
Q: What's the performance overhead of orchestration? A: Minimal for sequential and parallel (typically <50ms). Dynamic routing adds 100-300ms for routing decisions. Hierarchical supervision adds 200-500ms for planning. For most workflows, orchestration overhead is <10% of total execution time.
Q: Should I build custom orchestration or use a framework? A: Frameworks (LangGraph, CrewAI, OpenAI Agents SDK) handle 80% of orchestration needs and reduce development time by 60-70%. Build custom orchestration only if you have unique requirements (e.g., strict latency constraints, proprietary agent protocols, or complex regulatory requirements).
Further reading:
External references: