ストーリー

Human-in-the-Loopとは

なぜHITLが必要か

HITLの種類

1. 承認型（Approval）
   → エージェントがアクションを提案 → 人間が承認/却下 → 実行/中止

2. 修正型（Edit）
   → エージェントが出力を生成 → 人間が修正 → 修正版で続行

3. フィードバック型（Feedback）
   → エージェントが中間結果を提示 → 人間がフィードバック → 改善

4. エスカレーション型（Escalation）
   → エージェントが対応不能と判断 → 人間に引き継ぎ

LangGraphでの実装

中断と再開（Interrupt / Resume）

LangGraphの interrupt_before / interrupt_after を使って、特定のノードの前後でワークフローを中断できます。

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class OrderState(TypedDict):
    messages: Annotated[list, add_messages]
    order_id: str
    refund_amount: int
    human_approved: bool | None
    action_plan: str | None

def analyze_request(state: OrderState) -> OrderState:
    """問い合わせを分析してアクションプランを作成"""
    return {
        "action_plan": f"注文 {state['order_id']} に対して {state['refund_amount']}円の返金を実行",
        "messages": [AIMessage(content=f"返金プラン: {state['refund_amount']}円")]
    }

def human_review(state: OrderState) -> OrderState:
    """人間のレビュー - このノードの前でワークフローが中断する"""
    # interrupt_before で中断されるので、
    # このノードが実行される時には人間の判断が State に入っている
    return state

def process_refund(state: OrderState) -> OrderState:
    """返金処理を実行"""
    if state.get("human_approved"):
        return {"messages": [AIMessage(content="返金処理が完了しました")]}
    else:
        return {"messages": [AIMessage(content="返金は承認されませんでした")]}

# グラフ構築
workflow = StateGraph(OrderState)
workflow.add_node("analyze", analyze_request)
workflow.add_node("human_review", human_review)
workflow.add_node("process_refund", process_refund)

workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "human_review")
workflow.add_edge("human_review", "process_refund")
workflow.add_edge("process_refund", END)

# チェックポイント付きでコンパイル（中断/再開に必要）
memory = MemorySaver()
app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["human_review"]  # human_reviewノードの前で中断
)

中断と再開の実行フロー

# Phase 1: ワークフロー開始 → human_reviewの前で自動中断
config = {"configurable": {"thread_id": "refund-001"}}
result = app.invoke(
    {
        "messages": [HumanMessage(content="返金してほしい")],
        "order_id": "ORD-12345",
        "refund_amount": 15800,
        "human_approved": None,
        "action_plan": None
    },
    config
)
# → analyze ノードまで実行して中断

# Phase 2: 人間がレビューして承認
# （UI/Slackなどで承認を受け取る）
app.update_state(
    config,
    {"human_approved": True},
    as_node="human_review"
)

# Phase 3: ワークフロー再開
result = app.invoke(None, config)
# → human_review → process_refund → END
print(result["messages"][-1].content)
# "返金処理が完了しました"

承認フローの設計パターン

リスクレベルに応じた承認フロー

リスク判定:
    低リスク（情報照会のみ）→ 承認不要、自動実行
    中リスク（ステータス変更）→ エージェントが実行後に報告
    高リスク（金銭操作）→ 人間の事前承認が必須

def determine_risk_level(state: AgentState) -> str:
    """アクションのリスクレベルを判定"""
    action = state.get("proposed_action", "")

    HIGH_RISK_ACTIONS = ["refund", "cancel_order", "delete_account"]
    MEDIUM_RISK_ACTIONS = ["update_status", "change_address", "apply_coupon"]

    if action in HIGH_RISK_ACTIONS:
        return "require_approval"  # 人間の承認が必要
    elif action in MEDIUM_RISK_ACTIONS:
        return "execute_and_notify"  # 実行して報告
    else:
        return "auto_execute"  # 自動実行

タイムアウトの考慮

# 人間の応答にタイムアウトを設定
import asyncio

async def wait_for_human_approval(
    thread_id: str,
    timeout_seconds: int = 300  # 5分
) -> bool:
    start_time = asyncio.get_event_loop().time()

    while True:
        elapsed = asyncio.get_event_loop().time() - start_time
        if elapsed > timeout_seconds:
            # タイムアウト: 安全側に倒して未承認とする
            return False

        approval = await check_approval_status(thread_id)
        if approval is not None:
            return approval

        await asyncio.sleep(5)  # 5秒ごとにポーリング

まとめ

チェックリスト

• Human-in-the-Loopの必要性と4つの種類を理解した
• LangGraphの interrupt_before による中断・再開を理解した
• リスクレベルに応じた承認フローの設計方法を把握した
• タイムアウト処理の考慮点を理解した

次のステップへ

次は「状態の永続化」を学びます。チェックポイントによるワークフロー状態の保存と復元、メモリ管理について理解しましょう。

推定読了時間: 30分