LangGraph : Prebuilt エージェント : Human-in-the-loop

by

エージェントでツール呼び出しをレビュー、編集、承認するには、LangGraph の組み込み Human-In-the-Loop (HIL) 機能、特に interrupt() プリミティブを使用できます。LangGraph は人間の入力を受け取るまで、実行を 無期限に一時停止することを可能にします。

LangGraph : Prebuilt エージェント : Human-in-the-loop

作成 : クラスキャット・セールスインフォメーション
作成日時 : 06/14/2025

* 本記事は langchain-ai.github.io の以下のページを独自に翻訳した上で、補足説明を加えてまとめ直しています :

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

LangGraph : Get started : Prebuilt エージェント : Human-in-the-loop

エージェントでツール呼び出しをレビュー、編集、そして承認するには、LangGraph の組み込み Human-In-the-Loop (HIL) 機能、特に interrupt() プリミティブを使用できます。

LangGraph は、人間の入力を受け取るまで、実行を 無期限に – 数分、数時間、あるいは数日間でも – 一時停止することを可能にします。

これは、エージェント状態が データベースにチェックポイントされる ために可能です、これによりシステムが実行コンテキストを永続化して、後で中断したところからワークフローを再開して継続することを可能にします。

human-in-the-loop コンセプトを深く理解するには、concept ガイド をご覧ください。


人間はエージェントの出力を (続行する前に) レビューして編集できます。これは、リクエストされたツール呼び出しがセンシティブであったり人間の監視を必要とするかもしれないアプリケーションでは特に重要です。

 

ツール呼び出しのレビュー

ツールに人間による承認ステップを追加するには :

  1. ツールで interrupt() を使用して実行を一時停止する。

  2. 人間の入力に基づいて Command(resume=…) で再開して続行する。

API リファレンス: InMemorySaver | interrupt | create_react_agent

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
from langgraph.prebuilt import create_react_agent

# An example of a sensitive tool that requires human review / approval
def book_hotel(hotel_name: str):
    """Book a hotel"""
    response = interrupt(  
        f"Trying to call `book_hotel` with args {{'hotel_name': {hotel_name}}}. "
        "Please approve or suggest edits."
    )
    if response["type"] == "accept":
        pass
    elif response["type"] == "edit":
        hotel_name = response["args"]["hotel_name"]
    else:
        raise ValueError(f"Unknown response type: {response['type']}")
    return f"Successfully booked a stay at {hotel_name}."

checkpointer = InMemorySaver() 

agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_hotel],
    checkpointer=checkpointer, 
)

stream() メソッドを使用してエージェントを実行し、config オブジェクトを渡してスレッド ID を指定します。これは、エージェントが次回の呼び出しで同じ会話を再開させることを可能にします。

config = {
   "configurable": {
      "thread_id": "1"
   }
}

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
    config
):
    print(chunk)
    print("\n")

人間の入力に基づいて、エージェントを Command(resume=…) で再開して続行させます。

from langgraph.types import Command

for chunk in agent.stream(
    Command(resume={"type": "accept"}),  
    # Command(resume={"type": "edit", "args": {"hotel_name": "McKittrick Hotel"}}),
    config
):
    print(chunk)
    print("\n")

 

Agent Inbox での使用

任意のツールに interrupt を追加するためにラッパーを作成できます。

下の例は Agent Inbox UI と Agent Chat UI と互換なリファレンス実装を提供しています。

Wrapper that adds human-in-the-loop to any tool

from typing import Callable
from langchain_core.tools import BaseTool, tool as create_tool
from langchain_core.runnables import RunnableConfig
from langgraph.types import interrupt 
from langgraph.prebuilt.interrupt import HumanInterruptConfig, HumanInterrupt

def add_human_in_the_loop(
    tool: Callable | BaseTool,
    *,
    interrupt_config: HumanInterruptConfig = None,
) -> BaseTool:
    """Wrap a tool to support human-in-the-loop review.""" 
    if not isinstance(tool, BaseTool):
        tool = create_tool(tool)

    if interrupt_config is None:
        interrupt_config = {
            "allow_accept": True,
            "allow_edit": True,
            "allow_respond": True,
        }

    @create_tool(  
        tool.name,
        description=tool.description,
        args_schema=tool.args_schema
    )
    def call_tool_with_interrupt(config: RunnableConfig, **tool_input):
        request: HumanInterrupt = {
            "action_request": {
                "action": tool.name,
                "args": tool_input
            },
            "config": interrupt_config,
            "description": "Please review the tool call"
        }
        response = interrupt([request])[0]  
        # approve the tool call
        if response["type"] == "accept":
            tool_response = tool.invoke(tool_input, config)
        # update tool call args
        elif response["type"] == "edit":
            tool_input = response["args"]["args"]
            tool_response = tool.invoke(tool_input, config)
        # respond to the LLM with user feedback
        elif response["type"] == "response":
            user_feedback = response["args"]
            tool_response = user_feedback
        else:
            raise ValueError(f"Unsupported interrupt response type: {response['type']}")

        return tool_response

    return call_tool_with_interrupt

任意のツールに interrupt() を追加するために add_human_in_the_loop ラッパーを使用できます、この場合、ツール内にそれを追加する必要はありません :

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

checkpointer = InMemorySaver()

def book_hotel(hotel_name: str):
   """Book a hotel"""
   return f"Successfully booked a stay at {hotel_name}."


agent = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[
        add_human_in_the_loop(book_hotel), 
    ],
    checkpointer=checkpointer,
)

config = {"configurable": {"thread_id": "1"}}

# Run the agent
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
    config
):
    print(chunk)
    print("\n")

人間の入力に基づいて Command(resume=…) でエージェントを再開して続行します。

from langgraph.types import Command 

for chunk in agent.stream(
    Command(resume=[{"type": "accept"}]),
    # Command(resume=[{"type": "edit", "args": {"args": {"hotel_name": "McKittrick Hotel"}}}]),
    config
):
    print(chunk)
    print("\n")

 

追加リソース

 

以上