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é.

hitlai agentisafety

Human-in-the-Loop patterns