3장: Human-in-the-Loop 설계

from dataclasses import dataclass
from enum import Enum
 
class GateType(Enum):
    MANDATORY = "mandatory"      # 항상 승인 필요
    CONDITIONAL = "conditional"  # 조건 충족 시 승인 필요
    ADVISORY = "advisory"        # 알림만, 승인 불필요
 
@dataclass
class ApprovalGate:
    name: str
    gate_type: GateType
    condition: str | None  # 조건부 게이트의 활성화 조건
    timeout_hours: float
    escalation_on_timeout: str  # 타임아웃 시 에스컬레이션 대상
    approvers: list[str]  # 승인 가능한 역할/사용자 목록
 
# 고객 환불 워크플로우의 승인 게이트 예시
refund_gates = [
    ApprovalGate(
        name="small_refund",
        gate_type=GateType.CONDITIONAL,
        condition="amount > 50000",  # 5만원 초과 시
        timeout_hours=4,
        escalation_on_timeout="team_lead",
        approvers=["cs_agent", "cs_lead"],
    ),
    ApprovalGate(
        name="large_refund",
        gate_type=GateType.MANDATORY,
        condition="amount > 500000",  # 50만원 초과 시 항상
        timeout_hours=2,
        escalation_on_timeout="finance_manager",
        approvers=["finance_manager", "cs_director"],
    ),
]

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
 
class SupportState(TypedDict):
    inquiry: str
    proposed_action: str | None
    approval_status: str | None  # pending, approved, rejected
    confidence: float
 
def propose_action(state: SupportState) -> SupportState:
    """에이전트가 행동을 제안"""
    action = resolver_agent.propose(state["inquiry"])
    return {
        **state,
        "proposed_action": action.text,
        "confidence": action.confidence,
    }
 
def check_approval_needed(state: SupportState) -> str:
    """승인 필요 여부를 판단하는 라우터"""
    if state["confidence"] >= 0.95:
        return "auto_approve"
    if state["confidence"] >= 0.80:
        return "advisory_review"
    return "mandatory_approval"
 
def wait_for_approval(state: SupportState) -> SupportState:
    """사람의 승인을 대기 (interrupt 지점)"""
    # LangGraph의 interrupt 메커니즘이 여기서 워크플로우를 일시 정지
    # 외부 시스템(UI, Slack 등)에서 승인 입력을 받은 후 재개
    return state
 
graph = StateGraph(SupportState)
graph.add_node("propose", propose_action)
graph.add_node("wait_approval", wait_for_approval)
graph.add_node("execute", execute_action)
graph.add_node("auto_execute", execute_action)
 
graph.add_conditional_edges("propose", check_approval_needed, {
    "auto_approve": "auto_execute",
    "advisory_review": "auto_execute",  # 알림만 발송
    "mandatory_approval": "wait_approval",
})
graph.add_edge("wait_approval", "execute")
graph.add_edge("auto_execute", END)
graph.add_edge("execute", END)
 
# PostgreSQL 기반 체크포인트로 상태 영속화
checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
workflow = graph.compile(checkpointer=checkpointer)

@dataclass
class EscalationPolicy:
    triggers: list[EscalationTrigger]
    levels: list[EscalationLevel]
    notification_channels: list[str]
 
@dataclass
class EscalationLevel:
    level: int
    target: str  # 역할 또는 그룹
    sla_minutes: int  # 응답 SLA
    context_to_include: list[str]  # 포함할 컨텍스트
    auto_escalate_on_sla_breach: bool
 
support_escalation = EscalationPolicy(
    triggers=[
        EscalationTrigger(type="confidence_below", threshold=0.6),
        EscalationTrigger(type="retry_exceeded", max_retries=3),
        EscalationTrigger(type="timeout", minutes=30),
    ],
    levels=[
        EscalationLevel(
            level=1,
            target="senior_agent",
            sla_minutes=15,
            context_to_include=["inquiry", "agent_reasoning", "attempted_actions"],
            auto_escalate_on_sla_breach=True,
        ),
        EscalationLevel(
            level=2,
            target="team_lead",
            sla_minutes=30,
            context_to_include=["full_conversation", "system_logs"],
            auto_escalate_on_sla_breach=True,
        ),
        EscalationLevel(
            level=3,
            target="department_head",
            sla_minutes=60,
            context_to_include=["full_audit_trail"],
            auto_escalate_on_sla_breach=False,
        ),
    ],
    notification_channels=["slack", "email", "pagerduty"],
)

@dataclass
class ConfidenceSignals:
    llm_self_assessment: float  # LLM 자체 신뢰도 평가
    retrieval_relevance: float  # 참조 문서의 관련성
    historical_accuracy: float  # 유사 케이스의 과거 정확도
    input_clarity: float        # 입력 데이터의 명확성
 
def calculate_confidence(signals: ConfidenceSignals) -> float:
    """가중 평균으로 종합 신뢰도 산출"""
    weights = {
        "llm_self_assessment": 0.25,
        "retrieval_relevance": 0.30,
        "historical_accuracy": 0.30,
        "input_clarity": 0.15,
    }
 
    score = sum(
        getattr(signals, key) * weight
        for key, weight in weights.items()
    )
 
    return round(min(max(score, 0.0), 1.0), 3)

@dataclass
class AutonomyConfig:
    mode: str  # shadow, supervised, confidence_gated, autonomous
    auto_approve_threshold: float
    sample_review_rate: float  # 자율 모드에서의 샘플 리뷰 비율
    rollback_trigger: RollbackCondition
 
def evaluate_progression(metrics: PerformanceMetrics) -> AutonomyConfig:
    """성과 지표 기반 자율성 단계 평가"""
    if metrics.accuracy < 0.90:
        return AutonomyConfig(
            mode="supervised",
            auto_approve_threshold=1.0,  # 모든 결정에 리뷰
            sample_review_rate=1.0,
            rollback_trigger=RollbackCondition(error_rate=0.15),
        )
 
    if metrics.accuracy < 0.95:
        return AutonomyConfig(
            mode="confidence_gated",
            auto_approve_threshold=0.90,
            sample_review_rate=0.20,
            rollback_trigger=RollbackCondition(error_rate=0.10),
        )
 
    return AutonomyConfig(
        mode="autonomous",
        auto_approve_threshold=0.70,
        sample_review_rate=0.05,  # 5% 샘플 리뷰
        rollback_trigger=RollbackCondition(error_rate=0.05),
    )