Human-in-the-Loop設計

Human-in-the-Loopとは

HITL（Human-in-the-Loop）:
AIの意思決定プロセスに人間の判断を組み込む設計パターン

3つのレベル:
1. Human-on-the-Loop  → AIが実行、人間が監視
2. Human-in-the-Loop  → AIが提案、人間が承認してから実行
3. Human-over-the-Loop → 人間が方針決定、AIが補助

承認フローの設計

リスクレベル別の承認フロー

from enum import Enum
from typing import Optional
from dataclasses import dataclass
from datetime import datetime

class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    AUTO_APPROVED = "auto_approved"

@dataclass
class ApprovalRequest:
    request_id: str
    customer_id: str
    risk_level: str
    proposed_action: str
    rationale: str
    churn_probability: float
    approver: Optional[str] = None
    status: ApprovalStatus = ApprovalStatus.PENDING
    created_at: datetime = None
    decided_at: datetime = None
    feedback: Optional[str] = None

    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()

承認判定ロジック

def determine_approval_flow(risk_level: str, action_type: str) -> dict:
    """リスクレベルとアクション種類から承認フローを決定"""
    rules = {
        ("LOW", "report"): {
            "auto_approve": True,
            "approver_role": None,
        },
        ("LOW", "coupon"): {
            "auto_approve": True,
            "approver_role": None,
        },
        ("MEDIUM", "coupon"): {
            "auto_approve": False,
            "approver_role": "manager",
        },
        ("MEDIUM", "contract_change"): {
            "auto_approve": False,
            "approver_role": "senior_manager",
        },
        ("HIGH", "contract_change"): {
            "auto_approve": False,
            "approver_role": "senior_manager",
        },
        ("HIGH", "special_offer"): {
            "auto_approve": False,
            "approver_role": "director",
        },
    }

    key = (risk_level, action_type)
    return rules.get(key, {"auto_approve": False, "approver_role": "manager"})

フィードバックループの設計

@dataclass
class Feedback:
    feedback_id: str
    request_id: str
    customer_id: str
    action_taken: str
    outcome: str  # "retained", "churned", "unknown"
    feedback_text: Optional[str] = None
    accuracy_rating: Optional[int] = None  # 1-5
    usefulness_rating: Optional[int] = None  # 1-5
    timestamp: datetime = None

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now()

class FeedbackCollector:
    def __init__(self):
        self.feedbacks = []

    def collect(self, feedback: Feedback):
        self.feedbacks.append(feedback)

    def analyze(self):
        """フィードバックを分析して改善点を特定"""
        if not self.feedbacks:
            return {"message": "フィードバックなし"}

        import numpy as np
        accuracy_scores = [f.accuracy_rating for f in self.feedbacks if f.accuracy_rating]
        usefulness_scores = [f.usefulness_rating for f in self.feedbacks if f.usefulness_rating]
        outcomes = [f.outcome for f in self.feedbacks]

        return {
            "total_feedbacks": len(self.feedbacks),
            "avg_accuracy": np.mean(accuracy_scores) if accuracy_scores else None,
            "avg_usefulness": np.mean(usefulness_scores) if usefulness_scores else None,
            "retention_rate": outcomes.count("retained") / len(outcomes) if outcomes else None,
            "churn_rate": outcomes.count("churned") / len(outcomes) if outcomes else None,
        }

信頼度閾値の設計

class ConfidenceThreshold:
    """モデルの予測確率に応じた信頼度閾値の管理"""

    def __init__(self):
        self.thresholds = {
            "high_confidence": 0.8,   # 80%以上で高信頼
            "medium_confidence": 0.5, # 50-80%で中信頼
            "low_confidence": 0.0,    # 50%未満で低信頼
        }

    def classify_confidence(self, probability: float) -> str:
        if probability >= self.thresholds["high_confidence"]:
            return "high"
        elif probability >= self.thresholds["medium_confidence"]:
            return "medium"
        else:
            return "low"

    def get_recommendation(self, confidence: str) -> dict:
        recommendations = {
            "high": {
                "action": "自動実行可能",
                "human_review": False,
                "escalation": False,
            },
            "medium": {
                "action": "人間のレビューを推奨",
                "human_review": True,
                "escalation": False,
            },
            "low": {
                "action": "エスカレーション必須",
                "human_review": True,
                "escalation": True,
            },
        }
        return recommendations[confidence]

LangGraphへの統合

from langgraph.graph import StateGraph, END

# HITL対応のState拡張
class HITLState(ChurnAnalysisState):
    approval_required: Optional[bool]
    approval_status: Optional[str]
    confidence_level: Optional[str]
    feedback: Optional[dict]

# 承認チェックノード
def check_approval(state: HITLState) -> dict:
    """承認が必要かチェックする"""
    confidence = ConfidenceThreshold()
    prob = state.get("churn_probability", 0)
    conf_level = confidence.classify_confidence(prob)
    rec = confidence.get_recommendation(conf_level)

    return {
        "confidence_level": conf_level,
        "approval_required": rec["human_review"],
    }

# 承認待ちノード（LangGraphのinterrupt機能）
def await_approval(state: HITLState) -> dict:
    """人間の承認を待つ（LangGraphのinterrupt）"""
    # 実際にはLangGraphのinterrupt機能を使用
    # ここではシミュレーション
    return {
        "approval_status": "pending",
        "response": f"承認待ち: 信頼度={state['confidence_level']}, "
                    f"離反確率={state['churn_probability']:.1%}",
    }

HITLの全体フロー

予測結果
    ↓
[信頼度チェック]
    ├── 高信頼 → 自動実行 → 結果をフィードバック収集
    ├── 中信頼 → 人間レビュー → 承認/却下 → フィードバック収集
    └── 低信頼 → エスカレーション → 上位承認 → フィードバック収集
                                                    ↓
                                            [フィードバック分析]
                                                    ↓
                                            [閾値・ルールの更新]

まとめ

チェックリスト

• HITLの3つのレベルを説明できる
• リスクレベル別の承認フローを設計できる
• フィードバックループの仕組みを実装できる
• 信頼度閾値に基づく分岐を設計できる
• LangGraphにHITL機能を統合できる

次のステップへ

HITL設計ができた。次はコストとレイテンシの最適化に進み、エージェントの実運用を見据えた改善を行おう。

推定読了時間: 30分