Human-in-the-Loop-Muster stellen einen Schluesselansatz zur Optimierung der Arbeit mit grossen Sprachmodellen und KI-Agenten dar. These patterns enable effective combination of human oversight with automated processes. Proper implementation of these patterns leads to more accurate results and higher reliability of AI systems.
Human-in-the-Loop Patterns for AI Agents¶
Human-in-the-Loop (HITL) patterns represent a key approach to safe deployment of AI agents in der Produktion systems. These patterns allow combining the power of automation with human judgment where it’s most needed. In this article, we’ll go through practical implementations of HITL patterns for various scenarios.
Types of Human-in-the-Loop Interactions¶
HITL patterns can be divided into several categories based on the level of human intervention:
- Approval-based - Agent executes action only after human approval
- Confidence-threshold - Human intervention only with low confidence
- Exception-handling - Human as fallback when agent fails
- Continuous oversight - Ongoing monitoring with ability to interrupt
Implementing Approval-based Pattern¶
The simplest HITL pattern requires explicit approval before executing an action. This approach is ideal for critical operations like data deletion or financial transactions.
class ApprovalBasedAgent:
def __init__(self, approval_service):
self.approval_service = approval_service
self.pending_actions = {}
async def execute_with_approval(self, action, context):
# Generate action proposal
proposal = await self.generate_action_proposal(action, context)
# Request approval
approval_id = await self.approval_service.request_approval({
'action_type': action.type,
'description': proposal.description,
'risk_level': proposal.risk_level,
'estimated_impact': proposal.impact,
'context': context
})
self.pending_actions[approval_id] = {
'action': action,
'proposal': proposal,
'timestamp': datetime.utcnow()
}
return approval_id
async def handle_approval_response(self, approval_id, approved, feedback=None):
if approval_id not in self.pending_actions:
raise ValueError(f"Unknown approval ID: {approval_id}")
action_data = self.pending_actions[approval_id]
if approved:
result = await self.execute_action(action_data['action'])
await self.log_execution(approval_id, result)
return result
else:
await self.log_rejection(approval_id, feedback)
return None
Confidence-threshold Pattern¶
A more sophisticated approach uses confidence scoring. The agent automatically executes actions with high confidence, while escalating uncertain cases to human operators.
class ConfidenceBasedAgent:
def __init__(self, confidence_threshold=0.8):
self.confidence_threshold = confidence_threshold
self.human_queue = HumanReviewQueue()
async def process_request(self, request):
# Get prediction with confidence score
prediction = await self.model.predict(request)
confidence = prediction.confidence
if confidence >= self.confidence_threshold:
# High confidence - execute automatically
result = await self.execute_action(prediction.action)
await self.log_automatic_execution(request, prediction, result)
return result
else:
# Low confidence - escalate to human
review_id = await self.human_queue.add_for_review({
'request': request,
'ai_suggestion': prediction,
'confidence': confidence,
'reason': 'Low confidence score'
})
return {
'status': 'pending_human_review',
'review_id': review_id,
'estimated_wait': await self.human_queue.get_estimated_wait()
}
async def handle_human_decision(self, review_id, decision, explanation=None):
review_item = await self.human_queue.get_item(review_id)
# Learn from human feedback
await self.update_model_with_feedback(
review_item['request'],
review_item['ai_suggestion'],
decision,
explanation
)
if decision.approved:
return await self.execute_action(decision.action)
else:
return {'status': 'rejected', 'reason': explanation}
Exception-handling Pattern¶
This pattern allows the agent to work autonomously, but automatically escalates problems to human operators during exceptional situations or errors.
class ExceptionHandlingAgent:
def __init__(self):
self.escalation_service = EscalationService()
self.retry_count = {}
self.max_retries = 3
async def execute_task(self, task):
task_id = task.id
try:
return await self.attempt_execution(task)
except AgentException as e:
return await self.handle_exception(task_id, e)
except Exception as e:
# Unexpected error - immediate escalation
return await self.escalate_immediately(task_id, e)
async def handle_exception(self, task_id, exception):
retry_count = self.retry_count.get(task_id, 0)
if retry_count < self.max_retries and exception.retryable:
# Attempt automatic recovery
self.retry_count[task_id] = retry_count + 1
await asyncio.sleep(2 ** retry_count) # Exponential backoff
return await self.execute_task(task)
else:
# Escalate to human
escalation_id = await self.escalation_service.create_escalation({
'task_id': task_id,
'error_type': type(exception).__name__,
'error_message': str(exception),
'retry_attempts': retry_count,
'context': await self.get_task_context(task_id)
})
return {
'status': 'escalated',
'escalation_id': escalation_id,
'message': 'Task requires human intervention'
}
Continuous Oversight Pattern¶
For long-term or critical operations, we implement continuous oversight with real-time interruption capability.
class ContinuousOversightAgent:
def __init__(self):
self.monitoring_service = MonitoringService()
self.active_sessions = {}
async def start_monitored_execution(self, task, supervisor_id):
session_id = str(uuid.uuid4())
# Create monitoring session
session = MonitoringSession(
session_id=session_id,
supervisor_id=supervisor_id,
task=task,
status='active'
)
self.active_sessions[session_id] = session
# Start task in separate task
execution_task = asyncio.create_task(
self.execute_with_monitoring(session_id, task)
)
# Monitoring loop
monitoring_task = asyncio.create_task(
self.monitor_execution(session_id)
)
return session_id
async def execute_with_monitoring(self, session_id, task):
session = self.active_sessions[session_id]
try:
for step in task.steps:
# Check if execution was interrupted
if session.status == 'interrupted':
await self.cleanup_interrupted_execution(session_id)
return {'status': 'interrupted_by_supervisor'}
# Execute step
step_result = await self.execute_step(step)
# Report progress
await self.monitoring_service.report_progress(
session_id, step.id, step_result
)
session.add_step_result(step.id, step_result)
return {'status': 'completed', 'results': session.results}
except Exception as e:
await self.monitoring_service.report_error(session_id, e)
raise
async def interrupt_execution(self, session_id, reason):
if session_id in self.active_sessions:
self.active_sessions[session_id].status = 'interrupted'
await self.monitoring_service.log_interruption(session_id, reason)
Best Practices for HITL Implementation¶
When implementing HITL patterns, it’s important to follow several key principles:
- Explicit escalation paths - Clearly define when and how escalation occurs
- Context preservation - Provide enough context for informed decisions
- Feedback loops - Implement mechanisms for learning from human decisions
- Timeout handling - Define what happens when human operator is unavailable
- Audit trails - Record all interactions for compliance and debugging
class HITLOrchestrator:
def __init__(self):
self.patterns = {
'approval': ApprovalBasedAgent(),
'confidence': ConfidenceBasedAgent(),
'exception': ExceptionHandlingAgent(),
'oversight': ContinuousOversightAgent()
}
self.audit_logger = AuditLogger()
async def route_request(self, request, pattern_type='auto'):
# Automatic pattern type selection
if pattern_type == 'auto':
pattern_type = await self.determine_pattern(request)
agent = self.patterns[pattern_type]
# Log processing start
await self.audit_logger.log_request_start(
request, pattern_type, agent.__class__.__name__
)
try:
result = await agent.process_request(request)
await self.audit_logger.log_success(request.id, result)
return result
except Exception as e:
await self.audit_logger.log_error(request.id, e)
raise
Zusammenfassung¶
Human-in-the-Loop patterns provide a robust framework for safe deployment of AI agents in der Produktion environments. Der Schluessel zum Erfolg liegt in proper selection of pattern type based on operation criticality, implementation of effective escalation mechanisms, and continuous improvement based on human feedback. Combining different HITL approaches enables creating a system that maximizes AI autonomy while maintaining human control where necessary.