langchain-ai/langgraph-human-in-the-loop

langgraph-human-in-the-loop は何をしますか？

このスキルは、LangGraph でヒューマンインザループパターン、承認のための一時停止、エラーハンドリングを実装する際に使用します。interrupt()、Command(resume=...)、承認/検証ワークフロー、および4層のエラー処理戦略をカバーします。

概要

LangGraph におけるヒューマンインザループ（HIL）は、エージェントの実行フローに人間の判断を介入させるための重要なパターンです。これは、高い信頼性が要求されるタスクや、エージェントが自律的に決定すべきでない状況で特に有用です。

中核となる概念

interrupt() 関数

interrupt() は、グラフの実行を一時停止し、人間の入力を待つための関数です。この関数は、以下のように使用します。

from langgraph.checkpoint import interrupt

# ユーザーへの指示を設定
def human_step(state):
    result = interrupt(
        {
            "question": "次の操作を承認しますか？",
            "context": state["current_context"]
        }
    )
    state["user_decision"] = result
    return state

Command(resume=...) の使用

実行が interrupt() で一時停止された後、Command(resume=...) を使用して実行を再開し、人間の入力をグラフに渡すことができます。

# 外部から実行を再開する例
from langgraph.types import Command

graph.update_state(config, Command(resume={"action": "approve"}))

承認ワークフロー

一般的な承認ワークフローは以下の手順で構成されます。

1. 実行の一時停止: interrupt() でグラフを停止し、承認待ち状態にする
2. 情報の提示: 人間が判断するために必要なコンテキストを提供する
3. 入力の受付: 外部から Command(resume=...) で応答を受け取る
4. 実行の再開: 受け取った応答に基づいてグラフの実行を続行する

検証ワークフロー

入力値やエージェントの出力を検証する必要がある場合、以下のパターンが利用できます。

from langgraph.checkpoint import interrupt

def validation_step(state):
    # 生成されたコンテンツを人間が検証
    validation_result = interrupt(
        {
            "generated_content": state["content"],
            "validation_guidelines": ["正確性", "トーン", "長さ"]
        }
    )
    
    if validation_result.get("approved"):
        state["content"] = validation_result.get("edit", state["content"])
        return state
    else:
        # 再生成や別のパスへ
        state["needs_revision"] = True
        return state

4層のエラー処理戦略

LangGraph アプリケーションでは、以下の4層のエラー処理戦略を推奨します。

1. 第1層: リトライ可能なエラー

   • 一時的な障害（ネットワークエラー、レート制限など）
   • retry_policy を使用して自動的に再試行

2. 第2層: ヒューマンインザループによる修正

   • 回復不能なエラーや予期しない入力
   • interrupt() で人間にエラーを通知し、修正を依頼

3. 第3層: 代替パスへのフォールバック

   • エラーが発生した場合、別の関数やモデルに切り替える
   • グラフ内で条件分岐を使用

4. 第4層: グレースフルデグラデーション

   • 全ての手段が失敗した場合、安全なデフォルト値または部分的な結果を返す
   • ユーザーにエラーを通知し、次のステップを提案

使用例

以下は、完全なヒューマンインザループパターンの例です。

from langgraph.graph import StateGraph, State
from langgraph.checkpoint import interrupt
from langgraph.types import Command
from typing import TypedDict

class MyState(TypedDict):
    input: str
    approved: bool
    result: str

def step1(state):
    state["result"] = process_initial(state["input"])
    return state

def approval_step(state):
    decision = interrupt({
        "result_to_approve": state["result"],
        "message": "この結果を承認しますか？"
    })
    
    if decision.get("approved"):
        state["approved"] = True
    else:
        state["approved"] = False
    return state

def step2_approved(state):
    state["result"] = finalize(state["result"])
    return state

def step2_rejected(state):
    state["result"] = "処理は拒否されました"
    return state

builder = StateGraph(MyState)
builder.add_node("step1", step1)
builder.add_node("approval", approval_step)
builder.add_node("approved", step2_approved)
builder.add_node("rejected", step2_rejected)

builder.add_edge("step1", "approval")
builder.add_conditional_edges(
    "approval",
    lambda state: "approved" if state["approved"] else "rejected"
)

builder.set_entry_point("step1")
graph = builder.compile()

ベストプラクティス

1. 一時停止の最小化: 必要な場合にのみ interrupt() を使用し、ユーザーエクスペリエンスを損なわないようにする
2. 明確な指示: interrupt() のメッセージは、人間が何をすべきか明確に理解できるように記述する
3. タイムアウト処理: 長期間応答がない場合のフォールバックを実装する
4. 状態の整合性: 一時停止と再開の際に状態が正しく保たれることを確認する
5. エラーハンドリング: 4層の戦略を適用し、あらゆる障害シナリオに対処する

参考資料

• LangGraph 公式ドキュメント: Human-in-the-loop
• LangGraph の Command リファレンス
• LangGraph エラーハンドリングガイド