Human-in-the-Loop patterns
Human-in-the-Loop vzory představují klíčový přístup k optimalizaci práce s velkými jazykovými modely a AI agenty. Tyto vzory umožňují efektivní kombinaci lidského dohledu s automatizovanými procesy. Správná implementace těchto vzorů vede k přesnějším výsledkům a vyšší spolehlivosti AI systémů.
Human-in-the-Loop patterns pro AI agenty
Human-in-the-Loop (HITL) patterns představují klíčový přístup k bezpečnému nasazení AI agentů v produkčních systémech. Tyto vzory umožňují kombinovat sílu automatizace s lidským úsudkem tam, kde je to nejvíce potřeba. V tomto článku si projdeme praktické implementace HITL patterns pro různé scénáře.
Typy Human-in-the-Loop interakcí
HITL patterns můžeme rozdělit do několika kategorií podle míry lidské intervence:
- Approval-based - Agent vykonává akci až po lidském schválení
- Confidence-threshold - Lidská intervence pouze při nízké confidence
- Exception-handling - Human jako fallback při selhání agenta
- Continuous oversight - Průběžné monitorování a možnost přerušení
Implementace Approval-based pattern
Nejjednodušší HITL pattern vyžaduje explicitní schválení před vykonáním akce. Tento přístup je ideální pro kritické operace jako mazání dat nebo finanční transakce.
class ApprovalBasedAgent:
def __init__(self, approval_service):
self.approval_service = approval_service
self.pending_actions = {}
async def execute_with_approval(self, action, context):
# Generuj návrh akce
proposal = await self.generate_action_proposal(action, context)
# Požádej o schválení
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
Sofistikovanější přístup využívá confidence scoring. Agent automaticky vykonává akce s vysokou confidence, zatímco nejisté případy eskaluje k lidskému operátorovi.
class ConfidenceBasedAgent:
def __init__(self, confidence_threshold=0.8):
self.confidence_threshold = confidence_threshold
self.human_queue = HumanReviewQueue()
async def process_request(self, request):
# Získej prediction s confidence score
prediction = await self.model.predict(request)
confidence = prediction.confidence
if confidence >= self.confidence_threshold:
# Vysoká confidence - vykonej automaticky
result = await self.execute_action(prediction.action)
await self.log_automatic_execution(request, prediction, result)
return result
else:
# Nízká confidence - eskaluj k člověku
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)
# Učení z lidského 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
Tento pattern umožňuje agentovi pracovat autonomně, ale při výjimečných situacích nebo chybách automaticky eskaluje problém k lidskému operátorovi.
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:
# Neočekávaná chyba - okamžitá eskalace
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:
# Pokus o automatické opravení
self.retry_count[task_id] = retry_count + 1
await asyncio.sleep(2 ** retry_count) # Exponential backoff
return await self.execute_task(task)
else:
# Eskalace k člověku
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
Pro dlouhodobé nebo kritické operace implementujeme kontinuální oversight s možností přerušení v reálném čase.
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())
# Vytvoř monitoring session
session = MonitoringSession(
session_id=session_id,
supervisor_id=supervisor_id,
task=task,
status='active'
)
self.active_sessions[session_id] = session
# Spusť task v separátním tasku
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:
# Kontrola, zda nebyla execution přerušena
if session.status == 'interrupted':
await self.cleanup_interrupted_execution(session_id)
return {'status': 'interrupted_by_supervisor'}
# Vykonej krok
step_result = await self.execute_step(step)
# Reportuj 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 pro HITL implementaci
Při implementaci HITL patterns je důležité dodržovat několik klíčových principů:
- Explicitní escalation paths - Jasně definuj, kdy a jak probíhá eskalace
- Context preservation - Poskytni dostatek kontextu pro informované rozhodnutí
- Feedback loops - Implementuj mechanismy pro učení z lidských rozhodnutí
- Timeout handling - Definuj, co se stane při nedostupnosti lidského operátora
- Audit trails - Zaznamenej všechny interakce pro compliance a 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'):
# Automatická volba pattern typu
if pattern_type == 'auto':
pattern_type = await self.determine_pattern(request)
agent = self.patterns[pattern_type]
# Zaloguj začátek zpracování
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
Shrnutí
Human-in-the-Loop patterns poskytují robustní framework pro bezpečné nasazení AI agentů v produkčních prostředích. Klíčem k úspěchu je správná volba pattern typu podle kritičnosti operace, implementace efektivních escalation mechanismů a kontinuální zlepšování na základě lidského feedbacku. Kombinace různých HITL approaches umožňuje vytvořit systém, který maximalizuje autonomii AI při zachování lidské kontroly tam, kde je to nezbytné.