Human-in-the-Loopエスカレーション

エージェントに人間の指示が必要な場合

エスカレーションのトリガーを定義する

from enum import Enum

class EscalationReason(Enum):
    LOW_CONFIDENCE = 'low_confidence'           # model uncertainty
    DESTRUCTIVE_ACTION = 'destructive_action'   # irreversible change
    AMBIGUOUS_TASK = 'ambiguous_task'           # unclear instructions
    RETRY_BUDGET_EXHAUSTED = 'retry_exhausted'  # too many failures
    POLICY_BOUNDARY = 'policy_boundary'         # approaching limit
    HUMAN_REQUESTED = 'human_requested'         # explicit request
    TIMEOUT = 'timeout'                         # took too long

ESCALATION_THRESHOLDS = {
    'min_confidence': 0.6,
    'max_retries': 5,
    'max_runtime_minutes': 30,
}

確信度の低さを検出する

from pydantic import BaseModel

class ConfidenceCheck(BaseModel):
    proposed_action: str
    confidence: float  # 0.0 to 1.0
    uncertainty_reason: str | None
    proceed: bool

async def check_confidence(context: str, proposed_action: str) -> ConfidenceCheck:
    return await judge_client.chat.completions.create(
        model='gpt-4o',
        response_model=ConfidenceCheck,
        messages=[{
            'role': 'user',
            'content': f'Context: {context}\n\nI am about to: {proposed_action}\n\nHow confident am I that this is correct? Be honest about uncertainty.'
        }]
    )

破壊的な操作を検出する

from dataclasses import dataclass
from typing import Callable

@dataclass
class Tool:
    name: str
    func: Callable
    destructive: bool = False
    description: str = ''

tools = [
    Tool('search_web',     search_web,      destructive=False),
    Tool('read_file',      read_file,       destructive=False),
    Tool('write_file',     write_file,      destructive=True,  description='Overwrites existing file'),
    Tool('send_email',     send_email,      destructive=True,  description='Sends real email to user'),
    Tool('delete_records', delete_records,  destructive=True,  description='Permanent DB deletion'),
]

def requires_approval(tool: Tool) -> bool:
    return tool.destructive

エージェントを一時停止して入力を待つ

import asyncio

async def escalate_and_wait(task_id: str, reason: EscalationReason, context: str,
                             question: str, timeout_hours: int = 24) -> str:
    # Save checkpoint
    save_checkpoint(load_checkpoint(task_id))
    # Create escalation record
    escalation_id = create_escalation(task_id, reason, context, question)
    # Notify reviewer
    notify_reviewer(escalation_id, question)
    # Wait for response (polling with timeout)
    deadline = asyncio.get_event_loop().time() + timeout_hours * 3600
    while asyncio.get_event_loop().time() < deadline:
        response = get_escalation_response(escalation_id)
        if response:
            return response.guidance
        await asyncio.sleep(60)  # check every minute
    raise TimeoutError(f'Escalation {escalation_id} not answered within {timeout_hours}h')

レビュアーインターフェースの構築

# FastAPI escalation endpoint
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class EscalationResponse(BaseModel):
    escalation_id: str
    decision: str  # 'approve', 'reject', 'guide'
    guidance: str | None = None
    reviewer_id: str

@app.post('/escalations/{escalation_id}/respond')
async def respond_to_escalation(esc_id: str, response: EscalationResponse):
    escalation = get_escalation(esc_id)
    if not escalation or escalation.status != 'pending':
        return {'error': 'Escalation not found or already resolved'}
    save_escalation_response(esc_id, response)
    return {'status': 'response_recorded', 'task_will_resume': True}

エージェントのコンテキストへの人間のガイダンスの注入

def inject_human_guidance(messages: list, decision: str, guidance: str | None) -> list:
    if decision == 'approve':
        messages.append({
            'role': 'user',
            'content': 'Supervisor: Your proposed action has been approved. Proceed.'
        })
    elif decision == 'reject':
        messages.append({
            'role': 'user',
            'content': f'Supervisor: Your proposed action was rejected. Instead: {guidance}'
        })
    elif decision == 'guide':
        messages.append({
            'role': 'user',
            'content': f'Supervisor: Additional guidance: {guidance}'
        })
    return messages

エスカレーション指標の追跡

def escalation_report(db_connection, days: int = 7) -> dict:
    # SQL query (pseudocode)
    rows = db_connection.execute('''
        SELECT
            reason,
            COUNT(*) as count,
            AVG(EXTRACT(EPOCH FROM (responded_at - created_at)) / 3600) as avg_response_hours,
            SUM(CASE WHEN decision = 'approve' THEN 1 ELSE 0 END) as approvals,
            SUM(CASE WHEN decision = 'reject' THEN 1 ELSE 0 END) as rejections
        FROM escalations
        WHERE created_at > NOW() - INTERVAL '%s days'
        GROUP BY reason
        ORDER BY count DESC
    ''' % days).fetchall()
    return [dict(r) for r in rows]

自律性の段階的な拡大

# Autonomy expansion strategy:
# Week 1: escalate for ALL destructive actions
# Week 2: auto-approve file writes to /tmp (low-risk), escalate others
# Week 4: auto-approve all file writes, escalate only email/DB changes
# Week 8: auto-approve emails under 10 recipients, escalate mass emails

# Track approval rates per trigger type:
# Tool: write_file    -> 98% approve -> safe to automate
# Tool: send_email    -> 89% approve -> near-automate with content check
# Tool: delete_records -> 43% approve -> always escalate

緊急オーバーライドとタスクのキャンセル

async def run_agent_with_cancel(task_id: str, messages: list) -> str:
    for step in range(MAX_ITERATIONS):
        # Check cancel flag at start of every step
        if redis_client.get(f'agent:cancel:{task_id}'):
            save_final_status(task_id, 'cancelled')
            return 'Task cancelled by operator.'
        response = await get_next_action(messages)
        if response.is_final:
            return response.answer
        result = await execute_tool(response.tool, response.args)
        messages.append({'role': 'user', 'content': result})
        save_checkpoint_after_step(task_id, step, messages)
    return 'Max iterations reached'

エスカレーションしきい値の調整

# Threshold tuning guide:
# Escalation rate vs quality trade-off:
#
# confidence_threshold=0.8  -> 35% escalation rate (too many)
# confidence_threshold=0.6  -> 12% escalation rate (target)
# confidence_threshold=0.4  ->  4% escalation rate (too few)
#
# Weekly review of escalation decisions:
# - Approval rate > 90%: lower threshold (too conservative)
# - Approval rate < 70%: raise threshold (not catching real issues)
# - Target: 75-85% approval rate

クイックチェック

レッスンのまとめ