Agent Orchestration Patterns
Why This Matters
Orchestration is the backbone of multi-agent systems. Without proper orchestration, agents work in silos, duplicate efforts, or deadlock waiting for each other. Mastering these patterns lets you build systems where dozens of agents collaborate seamlessly—like an air traffic control tower managing hundreds of flights simultaneously.
Real-World Analogy: Think of a restaurant kitchen during dinner rush. The head chef (orchestrator) assigns stations (agents), sequences prep work (sequential), runs parallel cooking (grill + sauté + plating simultaneously), and handles dependencies (dessert waits for main course to finish). Without this coordination, you'd have chaos.
Orchestration Architecture Overview
Core Orchestration Implementation
import asyncio
import uuid
import time
import logging
from enum import Enum
from dataclasses import dataclass, field
from typing import Any, Callable, Coroutine, Optional
from collections import defaultdict
logger = logging.getLogger(__name__)
class AgentStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class AgentResult:
agent_id: str
status: AgentStatus
output: Any = None
error: Optional[str] = None
duration: float = 0.0
tokens_used: int = 0
metadata: dict = field(default_factory=dict)
@property
def is_success(self) -> bool:
return self.status == AgentStatus.COMPLETED
@dataclass
class AgentNode:
agent_id: str
agent_fn: Callable[..., Coroutine]
dependencies: list[str] = field(default_factory=list)
timeout: float = 300.0
retries: int = 3
retry_delay: float = 1.0
priority: int = 0
class OrchestrationEngine:
def __init__(self, max_concurrent: int = 10):
self.agents: dict[str, AgentNode] = {}
self.results: dict[str, AgentResult] = {}
self.max_concurrent = max_concurrent
self._semaphore = asyncio.Semaphore(max_concurrent)
self._execution_order: list[list[str]] = []
def register_agent(
self,
agent_id: str,
agent_fn: Callable[..., Coroutine],
dependencies: list[str] = None,
timeout: float = 300.0,
retries: int = 3,
):
self.agents[agent_id] = AgentNode(
agent_id=agent_id,
agent_fn=agent_fn,
dependencies=dependencies or [],
timeout=timeout,
retries=retries,
)
logger.info(f"Registered agent: {agent_id}")
def validate_dag(self) -> bool:
visited: set[str] = set()
rec_stack: set[str] = set()
def dfs(node_id: str) -> bool:
visited.add(node_id)
rec_stack.add(node_id)
for dep in self.agents[node_id].dependencies:
if dep not in self.agents:
raise ValueError(f"Agent '{node_id}' depends on unknown agent '{dep}'")
if dep not in visited:
if dfs(dep):
return True
elif dep in rec_stack:
raise ValueError(f"Circular dependency: {node_id} -> {dep}")
rec_stack.remove(node_id)
return False
for agent_id in self.agents:
if agent_id not in visited:
dfs(agent_id)
return True
def topological_sort(self) -> list[list[str]]:
in_degree = {aid: 0 for aid in self.agents}
for aid, node in self.agents.items():
for dep in node.dependencies:
in_degree[aid] += 1
levels: list[list[str]] = []
while in_degree:
level = [aid for aid, deg in in_degree.items() if deg == 0]
if not level:
raise ValueError("Circular dependency detected")
levels.append(level)
for aid in level:
del in_degree[aid]
for other_id, node in self.agents.items():
if other_id in in_degree and aid in node.dependencies:
in_degree[other_id] -= 1
self._execution_order = levels
return levels
async def _execute_agent(self, agent_id: str, context: dict) -> AgentResult:
node = self.agents[agent_id]
start_time = time.time()
async with self._semaphore:
for attempt in range(node.retries):
try:
agent_context = {
"agent_id": agent_id,
"dependencies_results": {
dep: self.results[dep]
for dep in node.dependencies
},
"attempt": attempt + 1,
**context,
}
output = await asyncio.wait_for(
node.agent_fn(agent_context),
timeout=node.timeout,
)
duration = time.time() - start_time
logger.info(f"Agent {agent_id} completed in {duration:.2f}s")
return AgentResult(
agent_id=agent_id,
status=AgentStatus.COMPLETED,
output=output,
duration=duration,
)
except asyncio.TimeoutError:
if attempt == node.retries - 1:
return AgentResult(
agent_id=agent_id,
status=AgentStatus.FAILED,
error=f"Timeout after {node.timeout}s",
duration=time.time() - start_time,
)
except Exception as e:
if attempt == node.retries - 1:
return AgentResult(
agent_id=agent_id,
status=AgentStatus.FAILED,
error=str(e),
duration=time.time() - start_time,
)
await asyncio.sleep(node.retry_delay * (2 ** attempt))
return AgentResult(agent_id=agent_id, status=AgentStatus.FAILED, error="Max retries exceeded")
async def execute(self, context: dict = None) -> dict[str, AgentResult]:
context = context or {}
self.validate_dag()
execution_levels = self.topological_sort()
logger.info(f"Execution plan: {len(execution_levels)} levels")
for level in execution_levels:
logger.info(f" Level: {level}")
for level in execution_levels:
tasks = [self._execute_agent(agent_id, context) for agent_id in level]
level_results = await asyncio.gather(*tasks)
for result in level_results:
self.results[result.agent_id] = result
if not result.is_success:
logger.warning(f"Agent {result.agent_id} failed: {result.error}")
return self.results
def get_summary(self) -> dict:
completed = sum(1 for r in self.results.values() if r.is_success)
failed = sum(1 for r in self.results.values() if r.status == AgentStatus.FAILED)
total_duration = sum(r.duration for r in self.results.values())
return {
"total_agents": len(self.agents),
"completed": completed,
"failed": failed,
"total_duration": round(total_duration, 2),
"execution_levels": len(self._execution_order),
}
Performance Considerations
| Metric | Sequential | Parallel | DAG |
|---|---|---|---|
| Latency | O(sum of all) | O(max of parallel) | O(critical path) |
| Throughput | Low | High | High |
| Cost | Low | Medium | Medium-High |
| Accuracy | High | High | High |
| Best For | Simple workflows | Independent tasks | Complex dependencies |
Security Considerations
- Input Validation: Validate agent inputs to prevent injection attacks when agents process external data
- Resource Limits: Use semaphores to prevent resource exhaustion from runaway parallel execution
- Timeout Enforcement: Always enforce timeouts to prevent denial-of-service from stuck agents
- Error Isolation: Isolate agent failures to prevent cascading crashes across the system
- Audit Logging: Log all agent executions for security monitoring and debugging
Mathematical Foundation
Amdahl's Law — Maximum speedup with parallelism:
Where is the parallelizable fraction and is the number of workers.
Critical Path Length:
Interview Questions
1. What is the difference between sequential and parallel orchestration patterns?
Answer: Sequential execution processes agents one after another, where each agent's output feeds into the next. This is simple but slow for independent tasks. Parallel execution runs multiple agents concurrently, significantly reducing latency for independent work. The key tradeoff is complexity: parallel execution requires careful handling of shared state, error propagation, and result aggregation. Sequential is best for pipelines where data transformation is required; parallel is best for independent analyses that can be merged later.
2. How does topological sort work in DAG-based orchestration?
Answer: Topological sort orders nodes in a directed acyclic graph so that all dependencies precede their dependents. The algorithm: 1) Compute in-degree for each node, 2) Enqueue nodes with in-degree 0, 3) Dequeue a node, reduce in-degree of its neighbors, 4) Enqueue neighbors when in-degree becomes 0, 5) Repeat until queue is empty. This produces a level-based execution plan where nodes at each level can run in parallel. The algorithm runs in O(V+E) time where V is vertices and E is edges.
3. What is Amdahl's Law and why does it matter for agent orchestration?
Answer: Amdahl's Law states that the maximum speedup from parallelization is limited by the sequential fraction of work: S = 1/(1-f) where f is the parallelizable fraction. If 20% of agent processing must be sequential (e.g., result aggregation), maximum speedup is 5x even with unlimited parallelism. This matters because it sets realistic performance expectations and guides optimization efforts toward reducing sequential bottlenecks rather than adding more parallel workers.
4. How do you handle circular dependencies in a DAG?
Answer: Circular dependencies create deadlocks where no agent can proceed because each waits for another. Detection methods: 1) DFS-based cycle detection using recursion stack, 2) Kahn's algorithm (topological sort fails if not all nodes processed), 3) Three-color marking (white/gray/black). Prevention: Enforce acyclic constraint at registration time, validate with test cases, use design patterns like observer instead of circular references. Resolution: Refactor to break the cycle, introduce a mediator agent, or use asynchronous callbacks.
5. What are the tradeoffs between fan-out/fan-in and pipeline patterns?
Answer: Fan-out/fan-in splits work across parallel agents then merges results—best for independent data processing (e.g., analyzing multiple documents). Pipeline chains agents sequentially where each transforms output—best for multi-stage processing (e.g., extract → transform → load). Tradeoffs: Fan-out has higher throughput but requires merge logic; pipeline has lower latency per item but higher total latency. Fan-out complexity increases with merge strategy; pipeline complexity increases with stage count.
6. How do you implement fault tolerance in parallel agent execution?
Answer: Implement retry with exponential backoff, circuit breakers, and graceful degradation. Each parallel task should have: 1) Configurable retry count with backoff (1s, 2s, 4s...), 2) Timeout per task to prevent hanging, 3) Fallback responses for non-critical tasks, 4) Error isolation so one failure doesn't cascade, 5) Dead letter queue for failed tasks requiring manual review. Use asyncio.gather with return_exceptions=True to collect partial results.
7. What is the critical path method and how does it optimize orchestration?
Answer: The critical path method identifies the longest sequence of dependent tasks that determines minimum execution time. Steps: 1) Calculate earliest start/finish for each task, 2) Calculate latest start/finish by working backward from deadline, 3) Tasks with zero slack are on the critical path. Optimization: Focus resources on critical path tasks, parallelize non-critical tasks, reduce duration of critical tasks. The critical path identifies which delays directly impact overall latency.
8. How would you design a dynamic orchestration system that adapts to workload?
Answer: Use adaptive strategies: 1) Monitor queue depth and latency metrics, 2) Auto-scale concurrent workers based on demand, 3) Dynamically adjust timeouts based on historical performance, 4) Route tasks to appropriate execution patterns (sequential for simple, parallel for complex), 5) Implement priority queues so critical tasks get resources first, 6) Use predictive scaling based on time-series patterns. Include feedback loops: measure actual vs. predicted performance, adjust heuristics, and learn optimal configurations over time.
Common Pitfalls
| Pitfall | Solution |
|---|---|
| Circular dependencies cause deadlock | Validate DAG at registration; use DFS cycle detection |
| Thundering herd on parallel execution | Use semaphores to limit concurrency; add jitter |
| Missing dependency results | Validate all dependencies exist before execution |
| Infinite retry loops | Set max retries with exponential backoff and timeout |
| Memory exhaustion from parallel tasks | Use bounded queues and backpressure |
| Dead code from failed dependencies | Propagate failures; skip dependent agents |
| Race conditions on shared state | Use locks or message passing instead of shared memory |
| Over-parallelization wastes resources | Use Amdahl's Law to determine optimal parallelism |
Summary with Key Takeaways
- Sequential orchestration is simple but slow; use for linear data transformation pipelines
- Parallel orchestration maximizes throughput for independent tasks; requires merge strategies
- DAG-based orchestration handles complex dependencies; requires topological sort and cycle detection
- Fan-out/fan-in pattern is ideal for map-reduce workloads with independent data processing
- Critical path analysis identifies bottlenecks and guides optimization efforts
- Fault tolerance requires retries, timeouts, and graceful degradation at each execution level
- Concurrency control via semaphores prevents resource exhaustion in parallel execution
- Amdahl's Law sets realistic expectations for parallel speedup based on sequential fraction
KnowledgeCheck
-
What algorithm is used to determine execution order in a DAG?
- a) Dijkstra's algorithm
- b) Topological sort
- c) Breadth-first search
- d) Binary search
-
According to Amdahl's Law, if 25% of work is sequential, what is the maximum speedup?
- a) 4x
- b) 2x
- c) 1.33x
- d) 10x
-
What pattern is best for processing multiple independent data items?
- a) Sequential pipeline
- b) Fan-out/fan-in
- c) Supervisor pattern
- d) Ring topology
-
What happens when circular dependencies are detected in a DAG?
- a) Execution continues with warnings
- b) Deadlock occurs; no tasks can proceed
- c) The system automatically breaks the cycle
- d) Dependencies are ignored
-
What limits parallel execution performance even with infinite workers?
- a) Network bandwidth
- b) Sequential fraction of work (Amdahl's Law)
- c) Memory availability
- d) CPU clock speed
-
How should retry logic handle repeated failures?
- a) Retry immediately without delay
- b) Use exponential backoff with maximum retry count
- c) Retry indefinitely
- d) Skip retry and fail immediately
Answers: 1-b, 2-a, 3-b, 4-b, 5-b, 6-b