Human-in-the-Loop: Hybrid Autonomy at Scale¶
Overview¶
Fully autonomous agents are risky. Fully human teams don't scale. Hybrid systems with intelligent escalation are the future.
The Hybrid Autonomy Spectrum¶
Level 1: Autonomous (No Review)¶
class FullyAutonomousAgent:
"""Agent decides and executes alone"""
def handle_request(self, request):
# Low risk, high volume
decision = self.analyze(request)
if decision.risk < 0.1:
# Execute directly
result = self.execute(decision)
self.log_audit(request, decision, result)
return result
else:
# Too risky, escalate
return self.escalate(request)
When: Routine, low-risk, high-volume tasks
Examples: Status check, FAQ answer, account lookup
Level 2: Autonomous with Review¶
class ReviewableAutonomousAgent:
"""Agent decides, human can review"""
def handle_request(self, request):
# Medium risk
decision = self.analyze(request)
result = self.execute(decision)
# Always create reviewable record
review_record = {
'request': request,
'agent_decision': decision,
'result': result,
'timestamp': time.time(),
'can_rollback': True
}
# Add to review queue (async)
self.review_queue.add(review_record)
# Return result to user
return result
def human_review_callback(self, review_id, human_decision):
"""Human can overturn decision"""
record = self.review_queue.get(review_id)
if human_decision == 'rollback':
# Undo the action
self.undo(record['result'])
self.log_rollback(record)
elif human_decision == 'approve':
self.log_approval(record)
When: Medium-risk decisions that can be rolled back
Examples: Refund < $100, temporary ban, discount
Level 3: Approval Required¶
class ApprovalRequiredAgent:
"""Agent recommends, human approves before execution"""
def handle_request(self, request):
# Medium-high risk
recommendation = self.analyze(request)
confidence = recommendation.confidence
# Create approval request
approval_request = {
'request': request,
'recommendation': recommendation,
'confidence': confidence,
'deadline': time.time() + 3600 # 1 hour SLA
}
# Wait for approval
approval = self.approval_queue.add_and_wait(approval_request)
if approval.approved:
# Execute with approval
result = self.execute(recommendation)
self.log_execution_with_approval(request, approval, result)
return result
else:
# Rejected by human
return {
'status': 'rejected',
'reason': approval.reason,
'escalation_info': approval.comments
}
When: High-risk decisions with moderate volume
Examples: Account deletion, large refund, policy override
Level 4: Human Decision with Agent Input¶
class HumanDecisionAgent:
"""Human decides, agent provides analysis"""
def handle_request(self, request):
# Very high risk, low volume
# Agent provides analysis only
analysis = self.deep_analyze(request)
recommendations = [
self.recommendation_1(analysis),
self.recommendation_2(analysis),
self.recommendation_3(analysis)
]
# Create human decision request
decision_request = {
'request': request,
'analysis': analysis,
'agent_recommendations': recommendations,
'deadline': time.time() + 7200, # 2 hour SLA
'priority': 'urgent',
'experts_needed': ['compliance', 'security']
}
# Route to expert humans
decision = self.expert_queue.add_and_wait(decision_request)
if decision.approved:
result = self.execute_human_decision(decision)
self.log_human_decision(request, decision, result)
return result
When: Critical decisions
Examples: Account takeover investigation, regulatory decision, data breach
Scaling Human Oversight¶
Load-Based Escalation¶
class LoadBasedEscalation:
"""Only escalate when humans have capacity"""
def __init__(self):
self.human_queue = PriorityQueue()
self.max_human_load = 20 # Max pending reviews
def should_escalate(self, request, confidence):
"""Decide if this needs human review"""
# Always escalate critical
if request.risk > 0.8:
return True
# Escalate low confidence
if confidence < 0.6:
# But only if humans have capacity
if len(self.human_queue) < self.max_human_load:
return True
return False
Template-Based Escalation¶
class TemplateBasedApprovals:
"""Pre-approve common patterns"""
def __init__(self):
self.templates = {
'refund_under_50': {
'auto_approve': True,
'auto_execute': True
},
'ban_repeat_offender': {
'auto_approve': True,
'require_review': False
},
'data_access_personal': {
'auto_approve': False,
'require_approval': True
}
}
def match_template(self, request, decision):
"""Check if decision matches pre-approved template"""
for template_name, template in self.templates.items():
if self.matches_template(decision, template):
if template['auto_approve']:
return 'auto_approved'
elif template['require_approval']:
return 'requires_approval'
return 'manual_review'
Feedback Loop Integration¶
Learning from Human Decisions¶
class HumanFeedbackLearning:
"""Agent learns from human corrections"""
def record_feedback(self, request, agent_decision, human_decision):
"""Record when humans override agent"""
if agent_decision != human_decision:
feedback = {
'request': request,
'agent_was_wrong': True,
'correct_answer': human_decision,
'categories': self.extract_categories(request)
}
self.feedback_store.add(feedback)
self.update_performance_metrics(feedback)
def retrain_from_feedback(self):
"""Periodically retrain on feedback"""
# Get recent corrections
corrections = self.feedback_store.get_recent(days=7)
if len(corrections) > 100:
# Enough data to retrain
self.agent.finetune(corrections)
self.evaluate_performance()
3 Warnings ⚠️¶
Warning 1: Bottleneck at Humans¶
# ❌ WRONG
# Every decision goes to human
if request.risk > 0:
escalate_to_human()
# Humans overloaded
# SLAs missed
# Requests pile up
# ✅ RIGHT
# Clear escalation criteria
if request.risk > 0.7:
escalate_to_expert()
elif agent.confidence < 0.6:
escalate_to_human()
else:
execute_autonomously()
# Humans handle only truly critical
Warning 2: Ignoring Human Feedback¶
# ❌ WRONG
# Collect human decisions
human_queue.add_and_wait(request)
# But never use feedback to improve
# Agent keeps making same mistakes
# ✅ RIGHT
# Use feedback for improvement
human_decision = human_queue.add_and_wait(request)
if human_overrode_agent():
record_training_example()
update_agent_weights()
Warning 3: No Escalation Path Fallback¶
# ❌ WRONG
if agent.unsure():
escalate_to_human()
# But human queue is full!
# Request lost
# ✅ RIGHT
if agent.unsure():
if human_queue.has_capacity():
escalate_to_human()
else:
# Fallback: return to user
return {
'status': 'unable_to_process',
'reason': 'Expert review unavailable',
'try_again': time.time() + 300
}
Last Updated: August 9, 2026