Orchestrating Agents
Design and implement multi-agent systems with handoffs, routing, and coordination patterns.
Overview
Agent orchestration enables complex workflows by coordinating multiple specialized agents, each handling specific domains or tasks.
┌─────────────────────────────────────────────────────────────────────┐
│ AGENT ORCHESTRATION PATTERNS │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ SEQUENTIAL PARALLEL HIERARCHICAL │
│ ┌─────┐ ┌─────┐ ┌─────────┐ │
│ │ A │ │ A │ │ Manager │ │
│ └──┬──┘ └──┬──┘ └────┬────┘ │
│ │ ┌───┼───┐ ┌────┼────┐ │
│ ▼ ▼ ▼ ▼ ▼ ▼ ▼ │
│ ┌─────┐ ┌───┐┌───┐┌───┐ ┌───┐┌───┐┌───┐ │
│ │ B │ │ B ││ C ││ D │ │ A ││ B ││ C │ │
│ └──┬──┘ └───┘└───┘└───┘ └───┘└───┘└───┘ │
│ │ └───┬───┘ │
│ ▼ ▼ │
│ ┌─────┐ ┌───────────┐ │
│ │ C │ │ Combine │ │
│ └─────┘ └───────────┘ │
│ │
│ ROUTER-BASED SWARM (Handoffs) │
│ ┌─────────┐ ┌─────────────────────┐ │
│ │ Router │ │ Agent A │ │
│ └────┬────┘ │ ┌─────────────────┐ │ │
│ ┌───┼───┐ │ │transfer_to_B() │ │ │
│ ▼ ▼ ▼ │ │transfer_to_C() │ │ │
│ ┌───┐┌───┐┌───┐ │ └─────────────────┘ │ │
│ │ A ││ B ││ C │ └──────────┬──────────┘ │
│ └───┘└───┘└───┘ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ Agent B │ │
│ └─────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘The Routines Pattern
Routines define agent behavior with system prompts and available functions:
python
from dataclasses import dataclass
from typing import Callable, Any
@dataclass
class Agent:
name: str
instructions: str
functions: list[Callable]
model: str = "gpt-4o"
def transfer_to_sales():
"""Transfer conversation to sales agent."""
return sales_agent
def transfer_to_support():
"""Transfer conversation to support agent."""
return support_agent
def transfer_to_billing():
"""Transfer conversation to billing agent."""
return billing_agent
# Define specialized agents
triage_agent = Agent(
name="Triage Agent",
instructions="""You are the first point of contact for customers.
Determine what they need and transfer to the appropriate specialist:
- Sales questions → transfer_to_sales()
- Technical issues → transfer_to_support()
- Billing questions → transfer_to_billing()
Ask clarifying questions if needed before transferring.""",
functions=[transfer_to_sales, transfer_to_support, transfer_to_billing]
)
sales_agent = Agent(
name="Sales Agent",
instructions="""You are a sales specialist.
Help customers with:
- Product information
- Pricing questions
- Purchase decisions
- Upgrades and plans
If asked about technical issues, use transfer_to_support().""",
functions=[transfer_to_support, transfer_to_billing]
)
support_agent = Agent(
name="Support Agent",
instructions="""You are a technical support specialist.
Help customers with:
- Troubleshooting
- Bug reports
- Feature usage
- Account setup
If asked about billing, use transfer_to_billing().""",
functions=[transfer_to_billing, transfer_to_sales]
)
billing_agent = Agent(
name="Billing Agent",
instructions="""You are a billing specialist.
Help customers with:
- Invoice questions
- Payment issues
- Refunds
- Subscription management""",
functions=[transfer_to_sales, transfer_to_support]
)Swarm-Style Execution
Run agents with automatic handoffs:
python
from openai import OpenAI
import json
client = OpenAI()
def function_to_tool(func: Callable) -> dict:
"""Convert a Python function to OpenAI tool format."""
import inspect
sig = inspect.signature(func)
params = {}
required = []
for name, param in sig.parameters.items():
if param.annotation != inspect.Parameter.empty:
param_type = param.annotation.__name__ if hasattr(param.annotation, '__name__') else 'string'
else:
param_type = 'string'
params[name] = {"type": param_type}
if param.default == inspect.Parameter.empty:
required.append(name)
return {
"type": "function",
"function": {
"name": func.__name__,
"description": func.__doc__ or "",
"parameters": {
"type": "object",
"properties": params,
"required": required
}
}
}
def run_agent_loop(agent: Agent, messages: list, context_variables: dict = None):
"""Execute agent with tool calling loop."""
context_variables = context_variables or {}
current_agent = agent
while True:
# Convert functions to tools
tools = [function_to_tool(f) for f in current_agent.functions]
# Build messages with system prompt
full_messages = [
{"role": "system", "content": current_agent.instructions}
] + messages
# Call the model
response = client.chat.completions.create(
model=current_agent.model,
messages=full_messages,
tools=tools if tools else None
)
message = response.choices[0].message
messages.append({"role": "assistant", "content": message.content, "tool_calls": message.tool_calls})
# No tool calls - return response
if not message.tool_calls:
return {
"agent": current_agent,
"messages": messages,
"response": message.content
}
# Process tool calls
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments) if tool_call.function.arguments else {}
# Find and execute the function
func = next((f for f in current_agent.functions if f.__name__ == func_name), None)
if func:
result = func(**func_args)
# Check if result is an agent (handoff)
if isinstance(result, Agent):
current_agent = result
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": f"Transferred to {current_agent.name}"
})
else:
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result) if result else "Done"
})
# Usage
messages = [{"role": "user", "content": "I have a question about my invoice"}]
result = run_agent_loop(triage_agent, messages)
print(f"Final agent: {result['agent'].name}")
print(f"Response: {result['response']}")Router Pattern
Use a dedicated router agent to direct requests:
python
from pydantic import BaseModel
from enum import Enum
class Department(str, Enum):
SALES = "sales"
SUPPORT = "support"
BILLING = "billing"
GENERAL = "general"
class RoutingDecision(BaseModel):
department: Department
confidence: float
reasoning: str
ROUTER_PROMPT = """You are a request router. Analyze the user's message and determine which department should handle it.
Departments:
- SALES: Product inquiries, pricing, purchases, upgrades
- SUPPORT: Technical issues, troubleshooting, bugs, how-to questions
- BILLING: Invoices, payments, refunds, subscriptions
- GENERAL: Anything that doesn't fit the above categories
Provide your routing decision with confidence (0-1) and brief reasoning.
"""
async def route_request(user_message: str) -> RoutingDecision:
"""Route request to appropriate department."""
response = client.beta.chat.completions.parse(
model="gpt-4o-mini", # Fast model for routing
messages=[
{"role": "system", "content": ROUTER_PROMPT},
{"role": "user", "content": user_message}
],
response_format=RoutingDecision
)
return response.choices[0].message.parsed
async def handle_routed_request(user_message: str):
"""Handle request with routing."""
# Route the request
routing = await route_request(user_message)
# Select appropriate agent
agents = {
Department.SALES: sales_agent,
Department.SUPPORT: support_agent,
Department.BILLING: billing_agent,
Department.GENERAL: triage_agent
}
selected_agent = agents[routing.department]
# Run the selected agent
messages = [{"role": "user", "content": user_message}]
result = run_agent_loop(selected_agent, messages)
return {
"routing": routing,
"response": result["response"],
"final_agent": result["agent"].name
}Parallel Agent Execution
Run multiple agents simultaneously for independent tasks:
python
import asyncio
from concurrent.futures import ThreadPoolExecutor
async def run_agents_parallel(tasks: list[tuple[Agent, str]]) -> list[dict]:
"""Run multiple agents in parallel."""
async def run_single(agent: Agent, message: str):
messages = [{"role": "user", "content": message}]
return run_agent_loop(agent, messages)
results = await asyncio.gather(*[
run_single(agent, message)
for agent, message in tasks
])
return results
# Example: Research task with multiple perspectives
async def multi_perspective_analysis(topic: str):
"""Get analysis from multiple specialized agents."""
analyst_agent = Agent(
name="Analyst",
instructions="You are a data analyst. Provide quantitative insights.",
functions=[]
)
strategist_agent = Agent(
name="Strategist",
instructions="You are a business strategist. Provide strategic recommendations.",
functions=[]
)
critic_agent = Agent(
name="Critic",
instructions="You are a critical analyst. Identify risks and weaknesses.",
functions=[]
)
tasks = [
(analyst_agent, f"Analyze: {topic}"),
(strategist_agent, f"Provide strategy for: {topic}"),
(critic_agent, f"Critique and identify risks: {topic}")
]
results = await run_agents_parallel(tasks)
return {
"analyst": results[0]["response"],
"strategist": results[1]["response"],
"critic": results[2]["response"]
}Hierarchical Agent Structure
python
@dataclass
class ManagerAgent(Agent):
subordinates: list[Agent]
delegation_strategy: str = "automatic"
def create_manager_agent(name: str, subordinates: list[Agent]) -> ManagerAgent:
"""Create a manager that coordinates subordinate agents."""
# Create delegation functions
delegation_functions = []
for sub in subordinates:
def delegate_to(agent=sub):
return agent
delegate_to.__name__ = f"delegate_to_{sub.name.lower().replace(' ', '_')}"
delegate_to.__doc__ = f"Delegate task to {sub.name}"
delegation_functions.append(delegate_to)
instructions = f"""You are {name}, a manager coordinating specialized agents.
Your subordinates:
{chr(10).join(f'- {sub.name}: {sub.instructions[:100]}...' for sub in subordinates)}
Your role:
1. Understand the user's request
2. Break it down into sub-tasks if needed
3. Delegate to appropriate subordinates
4. Synthesize their responses into a coherent answer
Use delegation functions to assign work to subordinates.
"""
return ManagerAgent(
name=name,
instructions=instructions,
functions=delegation_functions,
subordinates=subordinates
)
# Example hierarchical structure
research_agent = Agent(
name="Research Agent",
instructions="You research and gather information on topics.",
functions=[]
)
writing_agent = Agent(
name="Writing Agent",
instructions="You write and edit content based on research.",
functions=[]
)
review_agent = Agent(
name="Review Agent",
instructions="You review content for accuracy and quality.",
functions=[]
)
content_manager = create_manager_agent(
"Content Manager",
[research_agent, writing_agent, review_agent]
)State Management
Track conversation state across agent handoffs:
python
from dataclasses import dataclass, field
from typing import Any
from datetime import datetime
@dataclass
class ConversationState:
session_id: str
current_agent: str
context: dict = field(default_factory=dict)
history: list = field(default_factory=list)
handoff_count: int = 0
created_at: datetime = field(default_factory=datetime.now)
def add_message(self, role: str, content: str, agent: str = None):
self.history.append({
"role": role,
"content": content,
"agent": agent or self.current_agent,
"timestamp": datetime.now().isoformat()
})
def record_handoff(self, from_agent: str, to_agent: str, reason: str):
self.handoff_count += 1
self.context["last_handoff"] = {
"from": from_agent,
"to": to_agent,
"reason": reason,
"count": self.handoff_count
}
self.current_agent = to_agent
def get_context_summary(self) -> str:
"""Generate context summary for agent continuity."""
summary = f"Session: {self.session_id}\n"
summary += f"Handoffs: {self.handoff_count}\n"
if self.context.get("customer_info"):
summary += f"Customer: {self.context['customer_info']}\n"
if self.context.get("issue_summary"):
summary += f"Issue: {self.context['issue_summary']}\n"
return summary
# Enhanced agent loop with state
def run_stateful_agent_loop(
agent: Agent,
user_message: str,
state: ConversationState
) -> dict:
"""Run agent loop with state tracking."""
state.add_message("user", user_message)
# Include context in system prompt
context_prompt = f"{agent.instructions}\n\nContext:\n{state.get_context_summary()}"
messages = [{"role": "system", "content": context_prompt}]
messages.extend([
{"role": m["role"], "content": m["content"]}
for m in state.history
])
# ... rest of agent loopEvaluation and Monitoring
python
from dataclasses import dataclass
from typing import Optional
@dataclass
class AgentMetrics:
agent_name: str
total_requests: int = 0
successful_completions: int = 0
handoffs_initiated: int = 0
handoffs_received: int = 0
avg_response_time: float = 0.0
error_count: int = 0
class OrchestrationMonitor:
def __init__(self):
self.metrics: dict[str, AgentMetrics] = {}
def record_request(self, agent_name: str, response_time: float, success: bool):
if agent_name not in self.metrics:
self.metrics[agent_name] = AgentMetrics(agent_name)
m = self.metrics[agent_name]
m.total_requests += 1
if success:
m.successful_completions += 1
else:
m.error_count += 1
# Update rolling average
m.avg_response_time = (
(m.avg_response_time * (m.total_requests - 1) + response_time)
/ m.total_requests
)
def record_handoff(self, from_agent: str, to_agent: str):
if from_agent in self.metrics:
self.metrics[from_agent].handoffs_initiated += 1
if to_agent in self.metrics:
self.metrics[to_agent].handoffs_received += 1
def get_report(self) -> dict:
return {
name: {
"success_rate": m.successful_completions / max(m.total_requests, 1),
"avg_response_time": m.avg_response_time,
"handoff_ratio": m.handoffs_initiated / max(m.total_requests, 1),
"error_rate": m.error_count / max(m.total_requests, 1)
}
for name, m in self.metrics.items()
}Best Practices
Agent Specialization
python
# BAD: One agent does everything
general_agent = Agent(
name="General",
instructions="Handle all customer requests including sales, support, billing...",
functions=[...] # Too many functions
)
# GOOD: Specialized agents with clear boundaries
sales_agent = Agent(
name="Sales",
instructions="Focus ONLY on sales. Transfer other requests.",
functions=[transfer_to_support, transfer_to_billing, create_quote, process_order]
)Clear Handoff Criteria
python
# Define explicit handoff conditions in instructions
SUPPORT_INSTRUCTIONS = """You are a technical support agent.
HANDOFF CRITERIA:
- Transfer to SALES if customer wants to upgrade or buy new products
- Transfer to BILLING if issue is about payments or invoices
- Stay in SUPPORT for: bugs, errors, how-to questions, configurations
When transferring, provide context: "Transferring to {department}. Summary: {issue_summary}"
"""