LangGraph : Get started : human-in-the-loop (人間介在型) 制御を追加する

エージェントは不確かで、タスクを正常に完了するために人間の入力を必要とする場合があります。あるいは思惑通りに動作していることを確認するために、実行前に人間の承認を必要とするかもしれません。LangGraph の永続化層は人間介在型 (human-in-the-loop) ワークフローをサポートし、ユーザ・フィードバックに基づいて一時停止や再開の実行を可能にします。

LangGraph : Get started : 基本 – human-in-the-loop (人間介在型) 制御を追加する

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

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

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

 

クラスキャット 人工知能 研究開発支援サービス ⭐️ リニューアルしました 😉

クラスキャット は人工知能に関する各種サービスを提供しています。お気軽にご相談ください :

  • 人工知能導入個別相談会(無償)実施中! [詳細]

  • 人工知能研究開発支援 [詳細]
    1. 自社特有情報を含むチャットボット構築支援
    2. 画像認識 (医療系含む) / 画像生成

  • PoC(概念実証)を失敗させないための支援 [詳細]

お問合せ : 下記までお願いします。

  • クラスキャット セールス・インフォメーション
  • sales-info@classcat.com
  • ClassCatJP

 

 

LangGraph : Get started : 基本 – human-in-the-loop (人間介在型) 制御を追加する

エージェントは不確か (unreliable) である可能性があり、タスクを正常に完了するために人間の入力を必要とする場合があります。同様に、一部のアクションについては、すべてが思惑通りに動作していることを確認するために、実行前に人間の承認を必要とするかもしれません。

LangGraph の永続化層 (persistence layer) は人間介在型 (human-in-the-loop) ワークフローをサポートし、ユーザ・フィードバックに基づいて一時停止や再開の実行を可能にします。この機能への主要なインターフェイスは割り込み (interrupt) 関数です。ノード内での interrupt の呼び出しは実行を一時停止します。Command を渡すことで、人間からの新しい入力とともに、実行が再開できます。interrupt は 幾つかの注意事項 はあるものの、Python の組み込み input() に人間工学的に類似しています。

 

1. human_assistance ツールの追加

チャットボットにメモリを追加する チュートリアルからの既存のコードから始めて、チャットボットに human_assistance ツールを追加しましょう。このツールは interrupt を使用して人間から情報を受け取ります。

最初にチャットモデルを選択しましょう :

OpenAI

pip install -U "langchain[openai]"
import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

llm = init_chat_model("openai:gpt-4.1")

Anthropic

pip install -U "langchain[anthropic]"
import os
from langchain.chat_models import init_chat_model

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")

追加の tool を使用してそれを StateGraph に組み込めるようになりました :

from typing import Annotated

from langchain_tavily import TavilySearch
from langchain_core.tools import tool
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

from langgraph.types import Command, interrupt

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

@tool
def human_assistance(query: str) -> str:
    """Request assistance from a human."""
    human_response = interrupt({"query": query})
    return human_response["data"]

tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    # Because we will be interrupting during tool execution,
    # we disable parallel tool calling to avoid repeating any
    # tool invocations when we resume.
    assert len(message.tool_calls) <= 1
    return {"messages": [message]}

graph_builder.add_node("chatbot", chatbot)

tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)

graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

Tip : For more information and examples of human-in-the-loop workflows, see Human-in-the-loop. This includes how to review and edit tool calls before they are executed.

 

2. グラフのコンパイル

前と同じように、チェックポインターを使用してグラフをコンパイルします :

memory = MemorySaver()

graph = graph_builder.compile(checkpointer=memory)

 

3. グラフの視覚化 (オプション)

Visualizing the graph, you get the same layout as before – just with the added tool!

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass

 

4. チャットボットにプロンプトを提供する

そして、新しい human_assistance ツールを起動する質問をチャットボットにプロンプトとして提供します :

user_input = "I need some expert guidance for building an AI agent. Could you request assistance for me?"
config = {"configurable": {"thread_id": "1"}}

events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()
================================ Human Message =================================

I need some expert guidance for building an AI agent. Could you request assistance for me?
================================== Ai Message ==================================

[{'text': "Certainly! I'd be happy to request expert assistance for you regarding building an AI agent. To do this, I'll use the human_assistance function to relay your request. Let me do that for you now.", 'type': 'text'}, {'id': 'toolu_01ABUqneqnuHNuo1vhfDFQCW', 'input': {'query': 'A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?'}, 'name': 'human_assistance', 'type': 'tool_use'}]
Tool Calls:
  human_assistance (toolu_01ABUqneqnuHNuo1vhfDFQCW)
 Call ID: toolu_01ABUqneqnuHNuo1vhfDFQCW
  Args:
    query: A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?

チャットボットはツール呼び出しを生成しましたが、実行は中断されました。グラフ状態を調べれば、ツールノードで停止していることがわかります :

snapshot = graph.get_state(config)
snapshot.next
('tools',)

 

実行の再開

実行を再開するには、ツールにより期待されるデータを含む Command オブジェクトを渡します。このデータの形式はニーズに応じてカスタマイズできます。この例では、キー "data" を含む dict を使用しています :

human_response = (
    "We, the experts are here to help! We'd recommend you check out LangGraph to build your agent."
    " It's much more reliable and extensible than simple autonomous agents."
)

human_command = Command(resume={"data": human_response})

events = graph.stream(human_command, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()
================================== Ai Message ==================================

[{'text': "Certainly! I'd be happy to request expert assistance for you regarding building an AI agent. To do this, I'll use the human_assistance function to relay your request. Let me do that for you now.", 'type': 'text'}, {'id': 'toolu_01ABUqneqnuHNuo1vhfDFQCW', 'input': {'query': 'A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?'}, 'name': 'human_assistance', 'type': 'tool_use'}]
Tool Calls:
  human_assistance (toolu_01ABUqneqnuHNuo1vhfDFQCW)
 Call ID: toolu_01ABUqneqnuHNuo1vhfDFQCW
  Args:
    query: A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?
================================= Tool Message =================================
Name: human_assistance

We, the experts are here to help! We'd recommend you check out LangGraph to build your agent. It's much more reliable and extensible than simple autonomous agents.
================================== Ai Message ==================================

Thank you for your patience. I've received some expert advice regarding your request for guidance on building an AI agent. Here's what the experts have suggested:

The experts recommend that you look into LangGraph for building your AI agent. They mention that LangGraph is a more reliable and extensible option compared to simple autonomous agents.

LangGraph is likely a framework or library designed specifically for creating AI agents with advanced capabilities. Here are a few points to consider based on this recommendation:

1. Reliability: The experts emphasize that LangGraph is more reliable than simpler autonomous agent approaches. This could mean it has better stability, error handling, or consistent performance.

2. Extensibility: LangGraph is described as more extensible, which suggests that it probably offers a flexible architecture that allows you to easily add new features or modify existing ones as your agent's requirements evolve.

3. Advanced capabilities: Given that it's recommended over "simple autonomous agents," LangGraph likely provides more sophisticated tools and techniques for building complex AI agents.
...
2. Look for tutorials or guides specifically focused on building AI agents with LangGraph.
3. Check if there are any community forums or discussion groups where you can ask questions and get support from other developers using LangGraph.

If you'd like more specific information about LangGraph or have any questions about this recommendation, please feel free to ask, and I can request further assistance from the experts.
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...

 
入力は受信され、ツールメッセージとして処理されました。この呼び出しの LangSmith トレース をレビューすると上記の呼び出しで行われた正確な動作を確認できます。最初のステップで状態がロードされて、チャットボットが中断したところから継続できることに注意してください。

 
Congratulations! interrupt を使用してチャットボットで human-in-the-loop 実行を追加し、必要に応じて人間の監視と介入を可能にしました。これは AI システムで作成できる UI の可能性を広げます。既に チェックポインター を追加しているので、基礎的な永続化層が動作している限りは、グラフをいつでも 無期限に 一時停止して、何もなかったのように再開できます。

以下のコードスニペットを確認してこのチュートリアルのグラフをレビューしてください :

OpenAI

pip install -U "langchain[openai]"
import os
from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-..."

llm = init_chat_model("openai:gpt-4.1")

Anthropic

pip install -U "langchain[anthropic]"
import os
from langchain.chat_models import init_chat_model

os.environ["ANTHROPIC_API_KEY"] = "sk-..."

llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
from typing import Annotated

from langchain_tavily import TavilySearch
from langchain_core.tools import tool
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import Command, interrupt

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

@tool
def human_assistance(query: str) -> str:
    """Request assistance from a human."""
    human_response = interrupt({"query": query})
    return human_response["data"]

tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    assert(len(message.tool_calls) <= 1)
    return {"messages": [message]}

graph_builder.add_node("chatbot", chatbot)

tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)

graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

 

以上