Routing & Escalation¶
Overview¶
Not all requests are equal. Some need simple agents, others need complex reasoning, and some need humans.
Routing directs requests to the right handler, maximizing efficiency and quality.
-
Request Classification¶
Routing by Complexity¶
class ComplexityRouter:
"""Route by task difficulty"""
def route(self, request):
"""Determine which agent handles this"""
complexity = self.analyze_complexity(request)
if complexity < 0.3:
# Simple request
return self.simple_agent # Fast, cheap
elif complexity < 0.7:
# Medium request
return self.standard_agent # Balanced
else:
# Complex request
return self.expert_agent # Slower, more capable
def analyze_complexity(self, request) -> float:
"""Score request complexity 0-1"""
factors = [
len(request.text) / 1000, # Length
self.requires_reasoning(request),
self.requires_multiple_tools(request),
self.has_ambiguity(request)
]
return sum(factors) / len(factors)
-
Routing by Risk¶
class RiskBasedRouter:
"""Route by decision risk"""
def route(self, request):
"""Choose handler based on risk"""
risk = self.assess_risk(request)
if risk < 0.2:
# Low risk: autonomous
return ('autonomous', self.simple_agent)
elif risk < 0.5:
# Medium risk: review required
return ('review', self.agent_with_review)
else:
# High risk: human decides
return ('escalate', self.human_queue)
def assess_risk(self, request) -> float:
"""Estimate risk of decision"""
factors = {
'financial_impact': self.estimate_cost_impact(request) / 1000,
'user_disruption': self.estimate_user_impact(request),
'compliance': self.check_compliance_risk(request),
'data_access': self.check_data_sensitivity(request)
}
# Weighted average
weights = {'financial_impact': 0.4, 'user_disruption': 0.3,
'compliance': 0.2, 'data_access': 0.1}
return sum(factors[k] * weights[k] for k in factors)
Load Balancing Strategies¶
Strategy 1: Round Robin¶
class RoundRobinLoadBalancer:
"""Distribute evenly across agents"""
def __init__(self, agents):
self.agents = agents
self.next_index = 0
def select_agent(self):
"""Pick next agent in rotation"""
agent = self.agents[self.next_index % len(self.agents)]
self.next_index += 1
return agent
Use When: All agents equally capable Pros: Simple, fair Cons: Ignores agent load
Strategy 2: Least-Loaded¶
class LeastLoadedBalancer:
"""Route to least busy agent"""
def select_agent(self, agents):
"""Pick agent with lowest current load"""
loads = {}
for agent in agents:
# Current requests + estimated queue time
load = agent.current_requests + agent.avg_queue_time
loads[agent.id] = load
# Select least loaded
return min(loads, key=loads.get)
Use When: Agents have variable latency Pros: Balances load better Cons: Adds monitoring overhead
-
Strategy 3: Performance-Based¶
class PerformanceWeightedBalancer:
"""Route based on agent performance"""
def __init__(self, agents):
self.agents = agents
self.performance_scores = {}
def select_agent(self):
"""Pick agent with best recent performance"""
# Each agent has success rate
weights = {}
for agent in self.agents:
performance = agent.recent_success_rate
current_load = agent.queue_size
# Score = performance / load
weights[agent.id] = performance / (1 + current_load)
# Weighted random selection
return weighted_random_choice(weights)
def update_performance(self, agent_id, success):
"""Update agent performance tracking"""
self.performance_scores[agent_id].record(success)
Use When: Agents have different capabilities Pros: Maximizes success Cons: Most complex
Escalation Policies¶
Risk-Based Escalation¶
class EscalationPolicy:
"""Define when to escalate to humans"""
def __init__(self):
self.policies = {
'financial': {
'threshold': 100, # $100+
'action': 'require_approval'
},
'data_access': {
'types': ['personal_data', 'health'],
'action': 'require_approval'
},
'safety_violation': {
'detected': True,
'action': 'escalate_immediately'
},
'confidence_low': {
'threshold': 0.7, # < 70% confidence
'action': 'escalate'
}
}
def check_escalation(self, request, agent_output):
"""Determine if request needs escalation"""
for policy_name, policy in self.policies.items():
if self.matches_policy(request, agent_output, policy):
return policy['action']
return 'approve' # No escalation needed
def escalate(self, request, agent_output, reason):
"""Send to human queue"""
escalation = {
'request': request,
'agent_recommendation': agent_output,
'reason': reason,
'priority': self.calculate_priority(reason),
'deadline': time.time() + 3600 # 1 hour SLA
}
self.human_queue.add(escalation)
return {'status': 'escalated', 'id': escalation.id}
Queue Management¶
Priority Queue with SLA¶
class PriorityQueueManager:
"""Manage request queue with SLAs"""
def __init__(self):
self.queues = {
'high_priority': [],
'normal': [],
'low_priority': []
}
self.sla_times = {
'high_priority': 300, # 5 minutes
'normal': 3600, # 1 hour
'low_priority': 86400 # 1 day
}
def enqueue(self, request, priority='normal'):
"""Add to appropriate queue"""
self.queues[priority].append({
'request': request,
'enqueued_at': time.time()
})
def dequeue_next(self):
"""Get next request respecting SLA"""
current_time = time.time()
# Check high priority first
for priority in ['high_priority', 'normal', 'low_priority']:
queue = self.queues[priority]
if queue:
item = queue[0]
# Check if SLA violated
age = current_time - item['enqueued_at']
if age > self.sla_times[priority]:
# SLA violated! Escalate
queue.pop(0)
self.escalate_for_sla_violation(item)
else:
queue.pop(0)
return item['request']
return None # No requests
-
3 Warnings¶
Warning 1: Unbounded Queues¶
# WRONG
while True:
request = receive_request()
queue.append(request) # No limit!
# Queue grows infinitely
# Memory exhausted → system crashes
# RIGHT
MAX_QUEUE_SIZE = 10000
while True:
request = receive_request()
if len(queue) >= MAX_QUEUE_SIZE:
reject_request(request) # Backpressure
else:
queue.append(request)
Warning 2: Ignoring Agent Capabilities¶
# WRONG
# Route all requests to best agent
best_agent = agents[0]
for request in all_requests:
route_to(best_agent, request)
# Best agent overwhelmed
# Other agents idle
# Cascading failures
# RIGHT
# Route to agent capability/load
for request in all_requests:
agent = select_by_capability_and_load(request)
route_to(agent, request)
Warning 3: No Escalation Path¶
# WRONG
# Route to human only on crash
if agent.crashed():
escalate_to_human()
# But agent quietly produces wrong answer!
# RIGHT
# Multiple escalation triggers
if agent.crashed():
escalate()
elif agent.confidence < 0.7:
escalate()
elif request.risk > threshold:
escalate()
-
Last Updated: August 9, 2026