Routing & Escalation: Smart Request Distribution¶
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