Human-in-the-Loop

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