エージェントは不確かで、タスクを正常に完了するために人間の入力を必要とする場合があります。あるいは思惑通りに動作していることを確認するために、実行前に人間の承認を必要とするかもしれません。LangGraph の永続化層は human-in-the-loop ワークフローをサポートし、ユーザ・フィードバックに基づいて一時停止や再開の実行を可能にします。
※ LangGraph 0.5 でドキュメント構成が大幅に変更されましたので、再翻訳しながら Google Colab ベースで動作するようにまとめ直します。モデルは可能な限り最新版を利用し、プロンプトは日本語を使用します。
LangGraph 0.5 on Colab : Get started : human-in-the-loop 制御の追加
作成 : クラスキャット・セールスインフォメーション
作成日時 : 07/01/2025
* 本記事は langchain-ai.github.io の以下のページを独自に翻訳した上で、補足説明を加えてまとめ直しています :
* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。
◆ お問合せ : 下記までお願いします。
- クラスキャット セールス・インフォメーション
- sales-info@classcat.com
- ClassCatJP
LangGraph 0.5 on Colab : Get started : human-in-the-loop 制御の追加
エージェントは不確か (unreliable) である可能性があり、タスクを正常に完了するために人間の入力を必要とする場合があります。同様に、一部のアクションについては、すべてが思惑通りに動作していることを確認するために、実行前に人間の承認を必要とするかもしれません。
LangGraph の永続化層 (persistence layer) は human-in-the-loop ワークフローをサポートし、ユーザ・フィードバックに基づいて一時停止や再開の実行を可能にします。この機能への主要なインターフェイスは interrupt 関数です。ノード内での interrupt の呼び出しは実行を一時停止します。Command を渡すことで、人間からの新しい入力とともに、実行が再開できます。interrupt は 幾つかの注意事項 はあるものの、Python の組み込み input() に人間工学的に類似しています。
0. 準備
%%capture --no-stderr
%pip install -U --quiet langgraph
%pip install -U --quiet langchain-tavily
import getpass
import os
def _set_env(key: str):
if key not in os.environ:
os.environ[key] = getpass.getpass(f"{key}:")
_set_env("TAVILY_API_KEY")
1. human_assistance ツールの追加
チャットボットにメモリを追加する チュートリアルからの既存のコードから始めて、チャットボットに human_assistance ツールを追加しましょう。このツールは interrupt を使用して人間から情報を受け取ります。
最初にチャットモデルを選択しましょう :
OpenAI
%pip install -U "langchain[openai]"
import os
from langchain.chat_models import init_chat_model
_set_env("OPENAI_API_KEY")
llm = init_chat_model("openai:gpt-4.1")
Anthropic
%pip install -U "langchain[anthropic]"
import os
from langchain.chat_models import init_chat_model
_set_env("ANTHROPIC_API_KEY")
llm = init_chat_model("anthropic:claude-4-sonnet-20250514")
追加の 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 = "私は AI エージェントの構築のために専門家のガイダンスが必要です。私のために支援をリクエストしていただけますか？"
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 =================================
私は AI エージェントの構築のために専門家のガイダンスが必要です。私のために支援をリクエストしていただけますか？
================================== Ai Message ==================================
Tool Calls:
human_assistance (call_6uAzNPxuv6Gpt3KA8uEEWNjE)
Call ID: call_6uAzNPxuv6Gpt3KA8uEEWNjE
Args:
query: AIエージェントの構築に関する専門家のガイダンスが必要です。支援をリクエストします。
チャットボットはツール呼び出しを生成しましたが、実行は中断されました。グラフ状態を調べれば、ツールノードで停止していることがわかります :
snapshot = graph.get_state(config)
snapshot.next
('tools',)
5. 実行の再開
実行を再開するには、ツールにより期待されるデータを含む Command オブジェクトを渡します。このデータの形式はニーズに応じてカスタマイズできます。この例では、キー "data" を含む dict を使用しています :
human_response = (
"私たち専門家がお手伝いします！エージェントを構築するには LangGraph をご検討ください。"
"単純な自律エージェントよりもはるかに信頼性が高く、拡張性も優れています。"
)
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 ==================================
Tool Calls:
human_assistance (call_6uAzNPxuv6Gpt3KA8uEEWNjE)
Call ID: call_6uAzNPxuv6Gpt3KA8uEEWNjE
Args:
query: AIエージェントの構築に関する専門家のガイダンスが必要です。支援をリクエストします。
================================= Tool Message =================================
Name: human_assistance
私たち専門家がお手伝いします！エージェントを構築するには LangGraph をご検討ください。単純な自律エージェントよりもはるかに信頼性が高く、拡張性も優れています。
================================== Ai Message ==================================
AIエージェント構築のための専門家からのアドバイスとして、「LangGraph」の活用が推奨されています。これは単純な自律エージェントよりも信頼性・拡張性に優れたフレームワークです。
もし具体的に何を実現したいのか、スキルレベル、用途、プラットフォーム選定など詳細がありましたら、さらに最適なアドバイスやステップをご案内できます。どのようなAIエージェントを構築したいか教えていただけますか？
入力は受信され、ツールメッセージとして処理されました。この呼び出しの 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)
以上