Orchestrating 10+ Agents: Architecture Patterns That Scale
When you move beyond simple single-agent demos into production systems with ten, twenty, or fifty specialized agents working together, the coordination layer becomes the hardest problem to solve. This tutorial walks through the architectural patterns that actually work at scale—drawing from real-world deployments in customer support, code generation pipelines, and enterprise workflow automation.
What Is Multi-Agent Orchestration?
Multi-agent orchestration is the coordination layer that routes tasks, manages state, handles failures, and sequences work across a collection of specialized AI agents. An agent here means an autonomous unit—typically an LLM wrapped with tools, memory, and a specific system prompt—that performs one narrow job well. The orchestrator sits above these agents and decides who does what, in what order, and what happens when something goes wrong.
At 10+ agents, you can no longer rely on ad-hoc linear chains or a single flat router. You need intentional architecture. The patterns below are battle-tested approaches that prevent the system from collapsing under its own complexity.
Why Orchestration at Scale Matters
Without a proper orchestration layer, multi-agent systems suffer from predictable failure modes:
- Context overflow — passing every agent's output to every other agent blows past token limits quickly
- Circular loops — agents consult each other endlessly without converging on a decision
- Single points of failure — one slow or erroneous agent stalls the entire pipeline
- Observability gaps — when something goes wrong, you cannot trace which agent made which decision
- Cost explosions — redundant LLM calls multiply without governance
A well-architected orchestration layer addresses each of these by enforcing clear boundaries, structured communication, and graceful degradation.
Pattern 1: Centralized Orchestrator with Tool Registry
The most straightforward scalable pattern. A single orchestrator agent maintains a registry of available sub-agents (each exposed as a "tool") and decides which to invoke based on the current task. The orchestrator itself is an LLM with a system prompt that describes routing logic, while each sub-agent is a dedicated prompt + tool set.
Architecture Diagram (Conceptual)
┌─────────────────┐
│ Orchestrator │
│ (Router LLM) │
└───────┬─────────┘
│
┌───────────────┼───────────────┐
│ │ │
┌───────▼──────┐ ┌─────▼──────┐ ┌──────▼──────┐
│ BillingAgent │ │ TechAgent │ │ ShippingAgt │
│ (tool-1) │ │ (tool-2) │ │ (tool-3) │
└──────────────┘ └────────────┘ └─────────────┘
│
┌───────────────┼───────────────┐
┌───────▼──────┐ ┌─────▼──────┐ ┌──────▼──────┐
│ RefundAgent │ │ Escalation │ │ AnalyticsAgt │
│ (tool-4) │ │ Agent (5) │ │ (tool-6) │
└──────────────┘ └────────────┘ └─────────────┘
Implementation
import json
from typing import Dict, List, Optional
from dataclasses import dataclass, field
@dataclass
class AgentTool:
"""Each sub-agent is registered as a callable tool."""
name: str
description: str
handler: callable # async function that takes dict, returns dict
@dataclass
class OrchestratorState:
messages: List[Dict] = field(default_factory=list)
tool_calls: List[Dict] = field(default_factory=list)
final_answer: Optional[str] = None
class CentralizedOrchestrator:
"""
Single orchestrator that routes to 10+ sub-agents via tool calling.
Each sub-agent is a tool. The orchestrator decides which tool(s) to call,
in what sequence, based on user intent.
"""
def __init__(self, agents: List[AgentTool], llm_client):
self.agents = {agent.name: agent for agent in agents}
self.llm = llm_client
self.state = OrchestratorState()
def _build_tool_schema(self) -> List[Dict]:
"""Generate OpenAI-compatible tool definitions for all agents."""
return [
{
"type": "function",
"function": {
"name": agent.name,
"description": agent.description,
"parameters": {
"type": "object",
"properties": {
"reasoning": {
"type": "string",
"description": "Why this agent is needed"
},
"input_payload": {
"type": "object",
"description": "Arguments to pass to the agent"
}
},
"required": ["reasoning", "input_payload"]
}
}
}
for agent in self.agents.values()
]
async def run(self, user_query: str, max_steps: int = 10) -> str:
"""
Execute orchestration loop. Orchestrator may call multiple agents
sequentially, using results from earlier calls in later decisions.
"""
system_prompt = """You are the master orchestrator for a customer support system.
You have access to specialized agents as tools. Your job:
1. Analyze the user's request
2. Decide which agent(s) to invoke and in what order
3. Synthesize results into a coherent final response
4. If an agent returns an error or incomplete result, try an alternative agent
5. Do NOT exceed {max_steps} total tool calls. Prioritize resolution speed.
Available agents: {agent_descriptions}
""".format(
max_steps=max_steps,
agent_descriptions="\n".join(
f"- {name}: {agent.description}"
for name, agent in self.agents.items()
)
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_query}
]
step_count = 0
while step_count < max_steps:
response = await self.llm.chat(
messages=messages,
tools=self._build_tool_schema(),
tool_choice="auto"
)
if response.content and not response.tool_calls:
# Orchestrator decided to respond directly
return response.content
for tool_call in response.tool_calls:
agent_name = tool_call.function.name
if agent_name not in self.agents:
# Fallback: orchestrator tried an invalid agent
error_msg = f"Agent '{agent_name}' not found. Available: {list(self.agents.keys())}"
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps({"error": error_msg})
})
continue
args = json.loads(tool_call.function.arguments)
agent = self.agents[agent_name]
try:
result = await agent.handler(args.get("input_payload", {}))
result_str = json.dumps(result)
except Exception as e:
result_str = json.dumps({
"error": str(e),
"agent": agent_name,
"suggestion": "Try another agent or escalate"
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result_str
})
self.state.tool_calls.append({
"agent": agent_name,
"input": args,
"result": result_str
})
step_count += 1
return "Maximum orchestration steps reached. Please refine your request."
When to Use This Pattern
The centralized orchestrator works well for 10–30 agents when the routing logic is moderately complex but still understandable by a single LLM. It shines in customer support, where a single entry point classifies intent and dispatches to billing, technical, shipping, or refund specialists. The weakness: at 50+ agents, the tool schema becomes too large for reliable routing, and the orchestrator context window fills up with intermediate results.
Pattern 2: Hierarchical Orchestration with Domain Managers
When agent count exceeds ~30, split orchestration into layers. A top-level router classifies the domain (billing vs. technical vs. logistics), then hands off to a domain-specific orchestrator that manages its own sub-agents. This is recursive orchestration—each domain manager is itself a mini-orchestrator.
Implementation: Two-Level Hierarchy
class DomainManager:
"""
A mid-level orchestrator responsible for one business domain.
It owns 5-15 leaf agents and exposes a single 'run' interface
to the top-level orchestrator.
"""
def __init__(self, domain_name: str, leaf_agents: List[AgentTool], llm_client):
self.domain = domain_name
self.internal_orchestrator = CentralizedOrchestrator(leaf_agents, llm_client)
async def handle(self, domain_specific_query: str) -> Dict:
"""Entry point called by top-level orchestrator."""
result = await self.internal_orchestrator.run(
user_query=domain_specific_query,
max_steps=8
)
return {
"domain": self.domain,
"resolution": result,
"agents_used": [
tc["agent"] for tc in self.internal_orchestrator.state.tool_calls
]
}
class TopLevelOrchestrator:
"""
Routes to domain managers instead of leaf agents.
Each domain manager is registered as a tool.
"""
def __init__(self, domain_managers: Dict[str, DomainManager], llm_client):
self.domains = domain_managers
self.llm = llm_client
def _domain_tools(self) -> List[Dict]:
return [
{
"type": "function",
"function": {
"name": f"route_to_{domain}",
"description": f"Route the user to the {domain} department",
"parameters": {
"type": "object",
"properties": {
"summary_for_domain": {
"type": "string",
"description": "Summarize what the domain manager should resolve"
}
},
"required": ["summary_for_domain"]
}
}
}
for domain in self.domains.keys()
]
async def run(self, user_query: str) -> str:
messages = [
{"role": "system", "content": """You are the top-level router.
Classify the user's request into exactly one domain: billing, technical, logistics, or general.
Route to the appropriate domain manager with a clear summary.
If the request spans multiple domains, pick the most urgent one and mention others will be handled separately."""},
{"role": "user", "content": user_query}
]
response = await self.llm.chat(
messages=messages,
tools=self._domain_tools(),
tool_choice="auto"
)
if not response.tool_calls:
return response.content or "Unable to route your request."
# Execute the chosen domain route
for tool_call in response.tool_calls:
domain_key = tool_call.function.name.replace("route_to_", "")
if domain_key in self.domains:
args = json.loads(tool_call.function.arguments)
summary = args.get("summary_for_domain", user_query)
domain_result = await self.domains[domain_key].handle(summary)
return domain_result["resolution"]
return "Could not route to an appropriate department."
Scaling Beyond Two Levels
The hierarchy can go deeper. A logistics domain manager might further route to warehousing, last-mile-delivery, and returns sub-domains, each with their own leaf agents. The key rule: each orchestrator should manage no more than 15 direct children to keep routing reliable.
Pattern 3: Event-Driven Message Bus
For systems where agents must react to events asynchronously—monitoring pipelines, real-time alerting, streaming data processing—a message bus pattern decouples agents completely. Agents subscribe to event types, and a bus delivers messages. No single orchestrator exists; coordination emerges from subscription rules.
Implementation with an In-Memory Bus
import asyncio
from collections import defaultdict
from enum import Enum
from typing import Callable, Coroutine, Set
class EventType(Enum):
ORDER_PLACED = "order_placed"
PAYMENT_RECEIVED = "payment_received"
INVENTORY_LOW = "inventory_low"
FRAUD_DETECTED = "fraud_detected"
SHIPMENT_DELAYED = "shipment_delayed"
CUSTOMER_COMPLAINT = "customer_complaint"
@dataclass
class Event:
type: EventType
payload: Dict
correlation_id: str # ties related events together
timestamp: float
class MessageBus:
"""
Decentralized event bus. Agents subscribe to event types.
When an event is published, all subscribers are notified concurrently.
No agent knows about other agents—only the bus.
"""
def __init__(self):
self.subscribers: Dict[EventType, Set[Callable]] = defaultdict(set)
self.event_log: List[Event] = [] # append-only log for replay/debugging
def subscribe(self, event_type: EventType, handler: Callable[[Event], Coroutine]):
"""Register an async handler for a specific event type."""
self.subscribers[event_type].add(handler)
def unsubscribe(self, event_type: EventType, handler: Callable):
"""Remove a subscription."""
self.subscribers[event_type].discard(handler)
async def publish(self, event: Event):
"""Fire an event to all subscribers concurrently."""
self.event_log.append(event)
handlers = self.subscribers.get(event.type, set())
if not handlers:
return
# Run all handlers concurrently, collect results
tasks = [handler(event) for handler in handlers]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Log any exceptions but don't crash the bus
for handler, result in zip(handlers, results):
if isinstance(result, Exception):
print(f"Handler {handler.__name__} failed: {result}")
async def publish_and_wait(self, event: Event, timeout: float = 30.0):
"""Publish and wait for all handlers with a timeout."""
publish_task = asyncio.create_task(self.publish(event))
try:
await asyncio.wait_for(publish_task, timeout=timeout)
except asyncio.TimeoutError:
print(f"Event {event.type} processing timed out after {timeout}s")
# Example agent handlers
async def fraud_detection_agent(event: Event):
"""Agent that checks orders for fraud patterns."""
order = event.payload
risk_score = 0
if order.get("amount", 0) > 10000:
risk_score += 50
if order.get("shipping_address") != order.get("billing_address"):
risk_score += 30
if order.get("email_domain", "").endswith(".xyz"):
risk_score += 20
if risk_score > 60:
# Publish a new event — agents can trigger cascades
bus = get_bus() # global bus reference
await bus.publish(Event(
type=EventType.FRAUD_DETECTED,
payload={"order_id": order["id"], "risk_score": risk_score},
correlation_id=event.correlation_id,
timestamp=time.time()
))
async def inventory_update_agent(event: Event):
"""Updates inventory and checks for low stock."""
order = event.payload
# Deduct items from inventory database...
new_stock_level = await inventory_db.deduct(order["items"])
if new_stock_level < 10:
bus = get_bus()
await bus.publish(Event(
type=EventType.INVENTORY_LOW,
payload={"sku": order["items"][0]["sku"], "remaining": new_stock_level},
correlation_id=event.correlation_id,
timestamp=time.time()
))
async def notification_agent(event: Event):
"""Sends customer notifications for shipment delays."""
delay_info = event.payload
customer_id = delay_info["customer_id"]
await email_service.send(
to=customer_id,
template="shipment_delayed",
context=delay_info
)
# Wiring everything together
async def main():
bus = MessageBus()
# Subscribe agents to events they care about
bus.subscribe(EventType.ORDER_PLACED, fraud_detection_agent)
bus.subscribe(EventType.ORDER_PLACED, inventory_update_agent)
bus.subscribe(EventType.FRAUD_DETECTED, fraud_escalation_agent)
bus.subscribe(EventType.INVENTORY_LOW, restock_alert_agent)
bus.subscribe(EventType.SHIPMENT_DELAYED, notification_agent)
bus.subscribe(EventType.CUSTOMER_COMPLAINT, sentiment_analysis_agent)
# Simulate an incoming order
order_event = Event(
type=EventType.ORDER_PLACED,
payload={
"id": "ord-12345",
"amount": 15000,
"items": [{"sku": "widget-x", "quantity": 5}],
"shipping_address": "123 Main St",
"billing_address": "456 Fraud Ave",
"email_domain": "buyer.xyz"
},
correlation_id="corr-abc",
timestamp=time.time()
)
await bus.publish_and_wait(order_event)
# Inspect what happened
print(f"Events processed: {len(bus.event_log)}")
for evt in bus.event_log:
print(f" {evt.type.value} → corr_id={evt.correlation_id}")
When the Message Bus Excels
This pattern shines in systems with 20–100+ agents where workflows are reactive rather than request-response. It handles fan-out naturally (one order triggers fraud check, inventory update, and notification simultaneously). The trade-off: debugging is harder because there's no single execution trace. You must rely on the event log and correlation IDs for observability.
Pattern 4: Directed Acyclic Graph (DAG) Pipelines
When the sequence of agent work is known in advance—like a code generation pipeline (spec → design → implement → test → review)—a DAG-based orchestrator gives you predictable execution order with parallelism where possible. Each node is an agent; edges define dependencies.
DAG Pipeline Implementation
from graphlib import TopologicalSorter
import asyncio
@dataclass
class PipelineStep:
name: str
agent: callable # async function
depends_on: List[str] = field(default_factory=list) # step names this depends on
timeout_seconds: int = 120
retry_count: int = 2
class DAGOrchestrator:
"""
Executes a predefined DAG of agent steps.
Steps with no dependencies run in parallel.
Steps with dependencies wait for their upstream steps to complete.
"""
def __init__(self, steps: List[PipelineStep]):
self.steps = {step.name: step for step in steps}
self._validate_dag()
def _validate_dag(self):
"""Ensure no cycles and all dependencies exist."""
step_names = set(self.steps.keys())
for step in self.steps.values():
for dep in step.depends_on:
if dep not in step_names:
raise ValueError(f"Step '{step.name}' depends on unknown step '{dep}'")
# TopologicalSorter raises if cycles exist
graph = {name: step.depends_on for name, step in self.steps.items()}
TopologicalSorter(graph) # validates DAG property
async def run(self, initial_input: Dict) -> Dict:
"""
Execute the DAG. Returns results from every step keyed by step name.
Steps with zero dependencies receive initial_input.
Downstream steps receive outputs from their dependencies.
"""
results: Dict[str, Dict] = {} # step_name → output
in_progress: Dict[str, asyncio.Task] = {}
# Determine execution order
graph = {name: step.depends_on for name, step in self.steps.items()}
ts = TopologicalSorter(graph)
ts.prepare()
while ts.is_active():
# Get nodes whose dependencies are all resolved
ready = []
for node in ts.get_ready():
ready.append(node)
if not ready and not in_progress:
break
# Start ready nodes concurrently
for node_name in ready:
step = self.steps[node_name]
# Gather inputs from dependencies
if not step.depends_on:
step_input = initial_input
else:
step_input = {}
for dep in step.depends_on:
if dep in results:
step_input.update(results[dep])
task = asyncio.create_task(
self._execute_step(step, step_input)
)
in_progress[node_name] = task
# Wait for at least one task to complete
if in_progress:
done, _ = await asyncio.wait(
in_progress.values(),
return_when=asyncio.FIRST_COMPLETED
)
for completed_task in done:
# Find which node finished
for name, task in list(in_progress.items()):
if task == completed_task:
try:
result = await task
results[name] = result
ts.done(name)
except Exception as e:
results[name] = {"error": str(e), "step": name}
ts.done(name) # mark done even on failure to unblock downstream
finally:
del in_progress[name]
break
return results
async def _execute_step(self, step: PipelineStep, input_data: Dict) -> Dict:
"""Execute a single step with retry logic."""
last_error = None
for attempt in range(step.retry_count + 1):
try:
result = await asyncio.wait_for(
step.agent(input_data),
timeout=step.timeout_seconds
)
return {"output": result, "step": step.name, "attempt": attempt + 1}
except asyncio.TimeoutError:
last_error = f"Timeout after {step.timeout_seconds}s"
except Exception as e:
last_error = str(e)
if attempt < step.retry_count:
await asyncio.sleep(2 ** attempt) # exponential backoff
raise Exception(f"Step '{step.name}' failed after {step.retry_count + 1} attempts: {last_error}")
# Example: Code generation pipeline with 12 agents
async def spec_writer(input_data: Dict) -> str:
"""Agent: writes technical specification from requirements."""
...
async def architecture_designer(input_data: Dict) -> str:
"""Agent: designs system architecture from spec."""
...
async def database_schema_agent(input_data: Dict) -> str:
"""Agent: generates database schema from architecture."""
...
async def api_contract_agent(input_data: Dict) -> str:
"""Agent: defines API contracts."""
...
async def backend_implementer(input_data: Dict) -> str:
"""Agent: writes backend code."""
...
async def frontend_implementer(input_data: Dict) -> str:
"""Agent: writes frontend code."""
...
async def unit_test_writer(input_data: Dict) -> str:
"""Agent: generates unit tests."""
...
async def integration_test_writer(input_data: Dict) -> str:
"""Agent: generates integration tests."""
...
async def code_reviewer(input_data: Dict) -> str:
"""Agent: reviews generated code."""
...
async def security_scanner(input_data: Dict) -> str:
"""Agent: scans for vulnerabilities."""
...
async def documentation_agent(input_data: Dict) -> str:
"""Agent: generates documentation."""
...
async def deployment_config_agent(input_data: Dict) -> str:
"""Agent: generates deployment configs."""
...
# Wire the DAG
pipeline_steps = [
PipelineStep("spec", spec_writer, depends_on=[]),
PipelineStep("architecture", architecture_designer, depends_on=["spec"]),
PipelineStep("db_schema", database_schema_agent, depends_on=["architecture"]),
PipelineStep("api_contract", api_contract_agent, depends_on=["architecture"]),
# These two run in parallel after architecture is done
PipelineStep("backend", backend_implementer, depends_on=["db_schema", "api_contract"]),
PipelineStep("frontend", frontend_implementer, depends_on=["api_contract"]),
# Tests depend on their respective implementations
PipelineStep("unit_tests", unit_test_writer, depends_on=["backend", "frontend"]),
PipelineStep("integration_tests", integration_test_writer, depends_on=["backend"]),
# Review, security, docs, and deployment all run in parallel after tests
PipelineStep("code_review", code_reviewer, depends_on=["unit_tests"]),
PipelineStep("security_scan", security_scanner, depends_on=["unit_tests"]),
PipelineStep("docs", documentation_agent, depends_on=["unit_tests"]),
PipelineStep("deployment", deployment_config_agent, depends_on=["integration_tests", "security_scan"]),
]
dag = DAGOrchestrator(pipeline_steps)
results = await dag.run(initial_input={"project": "E-commerce checkout service"})
The DAG pattern gives you maximum predictability and is ideal for CI/CD-style agent pipelines. You know exactly which agents run and in what order. The downside: it cannot adapt dynamically to unexpected intermediate results. If the spec writer produces something surprising, the architecture designer still runs on schedule.
Pattern 5: Hybrid Router + DAG (The Production Sweet Spot)
In practice, large-scale systems combine patterns. A common production architecture uses a centralized router for initial intent classification, then fans out into domain-specific DAGs, with an event bus for cross-cutting concerns like logging, alerting, and auditing.
Hybrid Architecture Sketch
class HybridOrchestrator:
"""
Combines:
1. Intent classification (router pattern)
2. Domain-specific DAG execution
3. Event bus for observability and side-effects
"""
def __init__(self, router_llm, domain_dags: Dict[str, DAGOrchestrator], event_bus: MessageBus):
self.router = router_llm
self.dags = domain_dags
self.bus = event_bus
async def handle(self, user_request: str) -> Dict:
# Step 1: Classify intent
classification = await self._classify_intent(user_request)
domain = classification.get("domain", "general")
confidence = classification.get("confidence", 0.0)
# Step 2: Publish audit event
await self.bus.publish(Event(
type=EventType.REQUEST_RECEIVED,
payload={"request": user_request, "domain": domain, "confidence": confidence},
correlation_id=str(uuid.uuid4()),
timestamp=time.time()
))
# Step 3: Route to appropriate DAG
if domain in self.dags and confidence > 0.6:
dag = self.dags[domain]
dag_results = await dag.run(initial_input={
"user_request": user_request,
"classification": classification
})
# Step 4: Publish completion event
await self.bus.publish(Event(
type=EventType.REQUEST_COMPLETED,
payload={"domain": domain, "steps_completed": list(dag_results.keys())},
correlation_id=str(uuid.uuid4()),
timestamp=time.time()
))
return dag_results
# Fallback: low confidence → escalate to human or retry with broader context
return {"resolution": "escalated", "reason": f"Low confidence ({confidence}) for domain '{domain}'"}
async def _classify_intent(self, request: str) -> Dict:
"""Use router LLM to classify the request into a domain."""
response = await self.router.chat(
messages=[{
"role": "system",
"content": """Classify the request into one domain: billing, technical, logistics, sales.
Return JSON: {"domain": "...", "confidence": 0.0-1.0, "reasoning": "..."}"""
}, {
"role": "user",
"content": request
}]
)
return json.loads(response.content)
Best Practices for Orchestrating 10+ Agents
1. Enforce Strict Input/Output Contracts
Every agent should have a typed interface. Use Pydantic models or JSON Schema to validate inputs and outputs. When Agent A passes malformed data to Agent B, the orchestrator should catch it—not let the error cascade silently through five more agents.
from pydantic import BaseModel, ValidationError
class BillingAgentInput(BaseModel):
customer_id: str
transaction_id: str
amount: float
currency: str = "USD"
class BillingAgentOutput(BaseModel):
status: str # "processed", "refunded", "declined"
reference_id: str
fee: float
async def billing_agent_handler(input_payload: Dict) -> Dict:
try:
validated_input = BillingAgentInput.model_validate(input_payload)
except ValidationError as e:
return {"error": "Invalid input", "details": e.errors()}
# ... process billing ...
result = BillingAgentOutput(
status="processed",
reference_id="ref-abc",
fee=2.50
)
return result.model_dump()
2. Implement Circuit Breakers
If an agent fails 5 times in a row within a time window, the orchestrator should stop calling it and either use a fallback agent or escalate. This prevents cascading failures from bringing down the entire system.
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0):
self.threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.last_failure_time = 0.0
self.state = "closed" # closed, open, half-open
async def call(self, agent_fn, *args, **kwargs):
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "half-open"
else:
raise Exception("Circuit breaker is open — agent unavailable")
try:
result = await agent_fn(*args, **kwargs)
if self.state == "half-open":
self.state = "closed"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.threshold:
self.state = "open"
raise e
3. Log Every Orchestration Decision
At minimum, log: which agent was called, with what input, what it returned, how long it took, and whether it succeeded. This log becomes invaluable for debugging and cost attribution.
@dataclass
class OrchestrationLogEntry:
timestamp: float
agent_name: str
input_summary: str # truncated for storage
output_summary: str
duration_ms: float
success: bool
error: Optional[str] = None
tokens_used: Optional[int] = None
cost_estimate: Optional[float] = None
class ObservabilityMiddleware:
def __init__(self):
self.log: List[OrchestrationLogEntry] = []
async def wrap_agent(self, agent_name: str, handler: callable, input_data: Dict) -> Dict:
start = time.perf_counter()
try:
result = await handler(input_data)
duration = (time.perf_counter() - start) * 1000
self.log.append(OrchestrationLogEntry(
timestamp=time.time(),
agent_name=agent_name,
input_summary=str(input_data)[:200],
output_summary=str(result)[:200],
duration_ms=duration,
success=True
))
return result
except Exception as e:
duration = (time.perf_counter() - start) * 1000
self.log.append(OrchestrationLogEntry(
timestamp=time.time(),
agent_name=agent_name,
input_summary=str(input_data)[:200],
output_summary=None,
duration_ms=duration,
success=False,
error=str(e)
))
raise
def get_cost_by_agent(self) -> Dict[str, float]:
"""Aggregate costs per agent for billing/optimization."""
costs = defaultdict(float)
for entry in self.log:
if entry.cost_estimate:
costs[entry.agent_name] += entry.cost_estimate
return dict(costs)
4. Set Per-Agent Token Budgets
Each agent should have a maximum token limit for both input and output. The orchestrator should truncate or summarize context before passing it to an agent, rather than sending the entire conversation history.
def truncate_context_for_agent(full_context: str, max_tokens: int = 4000, tokenizer=None) -> str:
"""Truncate context to fit within an agent's token budget."""
if tokenizer is None:
# Rough estimate: 1 token ≈ 4 characters for English
estimated_tokens = len(full_context) // 4
else:
estimated_tokens = len(tokenizer.encode(full_context))
if estimated_tokens <= max_tokens:
return full_context
# Keep the beginning (system context) and end (recent messages)
# Drop the middle (older conversation turns)
chars_to_keep = max_tokens * 4
head_size = chars_to_keep // 3
tail_size = chars_to_keep -