LangChain ヒューマンインザループ

本番環境では、メール送信・削除実行・支払い処理など、人による確認が必要な操作があります。

ヒューマンインザループ（HITL）は、エージェントを重要な場面で一時停止させ、人による承認を待ってから続行させます。

interrupt()——ツール内での実行一時停止

interrupt()この関数を使うと、ツールの実行途中で一時停止し、外部からの入力を待ってから続行できます：

例

from langgraph.types import interrupt

# ツール内で interrupt() を使って一時停止
def send_email(to: str, subject: str, body: str) -> str:
    """メールを送信する（人による承認が必要）"""
    # 実行を一時停止し、外部に承認リクエストを送信
    approval = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "このメールを送信しますか？"
    })

    # 外部から approval が渡されるのを待って続行
    if approval.get("approved"):
        return f"メールを {to} に送信しました"
    else:
        return f"メール送信は拒否されました：{approval.get('reason', 'ユーザーによるキャンセル')}"

interrupt() のワークフロー：

  1. ツールが interrupt() を呼び出す → エージェントの実行が一時停止
  2. 外部システムが中断情報を取得し、ユーザーに表示する
  3. ユーザーの決定後、Command(resume=...) で実行を再開する
  4. interrupt() がユーザーから渡された値を返し、ツールの実行を続行する

完全な例——承認フロー

例

from dotenv import load_dotenv
load_dotenv()

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@tool
def delete_course(course_name: str) -> str:
    """コースを削除する（承認が必要）。

    Args:
course_name: 削除するコース名
    """
    # 一時停止して承認を待つ
    approval = interrupt({
        "action": "delete_course",
        "course": course_name,
        "message": f"コース『{course_name}』の削除を確認しますか？この操作は取り消せません。"
    })

    if approval.get("confirmed"):
        return f"コース『{course_name}』は削除されました"
    else:
        return f"削除操作はキャンセルされました"


checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[delete_course],
    checkpointer=checkpointer,
    system_prompt="あなたはサンプル の管理者アシスタントです。",
)

config = {"configurable": {"thread_id": "admin-001"}}

# ステップ1: 削除リクエストを開始する（中断が発生します）
print("=== 実行開始 ===")
result = agent.invoke(
    {"messages": [HumanMessage(content="コース『古い Java チュートリアル』を削除してください")]},
    config=config,
)

# エージェントが一時停止しているか確認
state = agent.get_state(config)
print(f"状態: {state.next}")  # ('tools',) は tools ノードで一時停止していることを示す
print(f"中断情報: {state.tasks)

# ステップ2: 人による承認（ユーザーが「確認」をクリックしたことをシミュレート）
print("\n=== 人による承認 ===")
resume_value = {"confirmed": True, "operator": "管理者 张三"}
result = agent.invoke(
    Command(resume=resume_value),
    config=config,
)
print(f"最終応答: {result['messages'][-1].content}")

実行結果：

=== 开始执行 ===
ステータス: ('tools',)
中断信息: (Interrupt(value={'action': 'delete_course', ...}),)

=== 人工审批 ===
最終返答: 「古いJavaチュートリアル」コースを削除しました。

interrupt_before / interrupt_after パラメータ

ツール内で interrupt() を使うだけでなく、create_agent() でグローバルな中断ポイントを設定することもできます：

例

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[some_tool],
    checkpointer=checkpointer,

    # ツールノードの前に一時停止（ツールを呼び出すたびに承認が必要）
    interrupt_before=["tools"],

    # モデルノードの後に一時停止（モデルが応答するたびに確認できる）
    # interrupt_after=["model"],
)
パラメータ一時停止のタイミング適用シナリオ
interrupt_before=["tools"]ツールを実行するたびに前すべてのツール呼び出しに承認が必要
interrupt_before=["model"]モデルを呼び出すたびに前モデル処理前にメッセージを人為的にレビュー
interrupt_after=["model"]モデルが応答するたびに後モデル出力を確認してから続行するか決定
interrupt_after=["tools"]ツールを実行するたびに後ツールの結果を確認してから次のステップを決定

HITL の典型的なアーキテクチャ

実際の Web アプリケーションでは、HITL は通常次のように実装されます：

例

# バックエンド: ユーザーメッセージを受け取り、中断ポイントまで処理し、中断情報を返す
def handle_user_message(thread_id: str, message: str):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [HumanMessage(content=message)]},
        config=config,
    )
    state = agent.get_state(config)

    # 承認待ちかどうかを確認
    if state.tasks and state.tasks[0].interrupts:
        return {
            "status": "pending_approval",
            "interrupt": state.tasks[0].interrupts[0].value,
            "thread_id": thread_id,
        }

    return {
        "status": "completed",
        "reply": result["messages"][-1].content,
    }


# バックエンド: ユーザーの承認を処理
def handle_approval(thread_id: str, approved: bool, reason: str = ""):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        Command(resume={"confirmed": approved, "reason": reason}),
        config=config,
    )
    return {"status": "completed", "reply": result["messages"][-1].content}

HITL を使用するには Checkpointer が必要です。エージェントが interrupt() で一時停止するとき、状態を永続化しなければ、再開時に正しく実行を続行できないためです。

その他の拡張