LangChain Agent ワークフロー
前のセクションではcreate_agent()のパラメータについて学びました。
このセクションでは Agent の内部を詳しく見て、どのように動作するのか——モデルとツールがどのように連携するのか、Agent がいつ停止するのかを理解します。
Agent 実行ループ
Agent の核心は単純なループです：モデルを呼び出し → ツールが必要か確認 → ツールを実行 → 繰り返す。モデルがツール呼び出しを要求しなくなるまで繰り返し、Agent は停止して最終結果を返します。
以下では Agent の各ステップを追跡して、このプロセスを理解します。
例
from dotenv import load_dotenv
load_dotenv()
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 get_weather(city: str) -> str:
"""指定された都市の天気を取得します。
Args:
city: 都市名
"""
weather_data = {
"杭州": "晴れ、25°C",
"北京": "曇り、18°C",
}
return weather_data.get(city, f"{city} の天気データが見つかりません")
@tool
def get_time(city: str) -> str:
「指定された都市の現在時刻を取得します。
Args:
city: 都市名
"""
time_data = {
"杭州": "14:30",
"北京": "14:30",
"ニューヨーク": "02:30",
}
return time_data.get(city, f"{city} の時刻データが見つかりません")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather, get_time],
system_prompt="あなたは親切なアシスタントです。",
)
# stream_mode="updates" を使用すると各ステップを確認できます
print("=== Agent 実行プロセスの追跡 ===\n")
step = 0
for chunk in agent.stream(
{"messages": [HumanMessage(content="杭州の今の天気はどうですか？何時ですか？")]},
stream_mode="updates",
):
step += 1
print(f"--- ステップ {step} ---")
for node_name, update in chunk.items():
print(f"ノード: {node_name}")
if "messages" in update:
for msg in update["messages"]:
if hasattr(msg, 'tool_calls') and msg.tool_calls:
# AI メッセージにはツール呼び出しが含まれます
for tc in msg.tool_calls:
print(f" → ツール呼び出しを要求: {tc['name']}({tc['args']})")
elif msg.type == "tool":
print(f" → ツール結果 [{msg.name}]: {msg.content}")
elif msg.type == "ai" and msg.content:
print(f" → AI 応答: {msg.content[:100]}")
load_dotenv()
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 get_weather(city: str) -> str:
"""指定された都市の天気を取得します。
Args:
city: 都市名
"""
weather_data = {
"杭州": "晴れ、25°C",
"北京": "曇り、18°C",
}
return weather_data.get(city, f"{city} の天気データが見つかりません")
@tool
def get_time(city: str) -> str:
「指定された都市の現在時刻を取得します。
Args:
city: 都市名
"""
time_data = {
"杭州": "14:30",
"北京": "14:30",
"ニューヨーク": "02:30",
}
return time_data.get(city, f"{city} の時刻データが見つかりません")
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather, get_time],
system_prompt="あなたは親切なアシスタントです。",
)
# stream_mode="updates" を使用すると各ステップを確認できます
print("=== Agent 実行プロセスの追跡 ===\n")
step = 0
for chunk in agent.stream(
{"messages": [HumanMessage(content="杭州の今の天気はどうですか？何時ですか？")]},
stream_mode="updates",
):
step += 1
print(f"--- ステップ {step} ---")
for node_name, update in chunk.items():
print(f"ノード: {node_name}")
if "messages" in update:
for msg in update["messages"]:
if hasattr(msg, 'tool_calls') and msg.tool_calls:
# AI メッセージにはツール呼び出しが含まれます
for tc in msg.tool_calls:
print(f" → ツール呼び出しを要求: {tc['name']}({tc['args']})")
elif msg.type == "tool":
print(f" → ツール結果 [{msg.name}]: {msg.content}")
elif msg.type == "ai" and msg.content:
print(f" → AI 応答: {msg.content[:100]}")
実行結果：
=== Agent 执行过程追踪 ===
--- 步骤 1 ---
ノード: model
→ 请求调用工具: get_weather({'city': '杭州'})
→ 请求调用工具: get_time({'city': '杭州'})
--- 步骤 2 ---
ノード: tools
→ 工具结果 [get_weather]: 晴，25°C
→ 工具结果 [get_time]: 14:30
--- 步骤 3 ---
ノード: model
→ AI 回复: 杭州现在天气晴朗，气温25°C，当前时间是14:30。
この追跡から Agent が実行した3 つのステップ：
- ステップ 1（model ノード）：モデルが質問を受け取り、get_weather と get_time の 2 つのツールを呼び出す必要があると判断し、2 つの tool_call を返します
- ステップ 2（tools ノード）：2 つのツールを実行し、天気と時刻の結果を取得します
- ステップ 3（model ノード）：モデルがツール結果を受け取り、情報が十分であると判断し、最終応答を生成します
stream_mode の詳細
stream() は複数の stream_mode をサポートしており、それぞれ異なる粒度の情報を提供します：
| モード | 戻り値の内容 | 適用シーン |
|---|---|---|
| updates | 各ノード実行後の状態更新 | Agent の実行ステップを追跡し、中間結果を表示 |
| values | 各ノード実行後の完全な状態 | 各ステップで完全なメッセージ履歴を確認する必要がある |
| messages | Token 単位のメッセージストリーム | フロントエンドで AI のタイピング効果をストリーミング表示 |
| custom | カスタムイベント | Middleware が stream_writer を通じてカスタムイベントを送信 |
stream_mode="values"——完全な状態変化を確認する
例
print("=== stream_mode='values' ===\n")
for i, chunk in enumerate(agent.stream(
{"messages": [HumanMessage(content="杭州の天気はどうですか？")]},
stream_mode="values",
)):
messages = chunk.get("messages", [])
print(f"状態 {i}: {len(messages)} 件のメッセージ")
for msg in messages:
print(f" [{msg.type}] {str(msg.content)[:80]}")
if i >= 3: # 最初のいくつかの状態だけを見る
break
for i, chunk in enumerate(agent.stream(
{"messages": [HumanMessage(content="杭州の天気はどうですか？")]},
stream_mode="values",
)):
messages = chunk.get("messages", [])
print(f"状態 {i}: {len(messages)} 件のメッセージ")
for msg in messages:
print(f" [{msg.type}] {str(msg.content)[:80]}")
if i >= 3: # 最初のいくつかの状態だけを見る
break
実行結果：
=== stream_mode='values' === 状態0: 1 条消息 [human] 杭州天气怎么样？ 状態1: 2 条消息 [human] 杭州天气怎么样？ [ai] 状態2: 3 条消息 [human] 杭州天气怎么样？ [ai] [tool] 晴，25°C 状態3: 4 条消息 [human] 杭州天气怎么样？ [ai] [tool] 晴，25°C [ai] 杭州今天天气晴朗，气温25°C，适合出门活动。
stream_mode="messages"——Token 単位のストリーミング出力
例
print("=== stream_mode='messages'（ストリーミングタイピング効果）===\n")
for msg_chunk, metadata in agent.stream(
{"messages": [HumanMessage(content="サンプル を一言で紹介してください")]},
stream_mode="messages",
):
# msg_chunk は AIMessageChunk で、それぞれが 1 つの Token のみを含みます
if hasattr(msg_chunk, 'content') and msg_chunk.content:
print(msg_chunk.content, end="", flush=True)
print()
for msg_chunk, metadata in agent.stream(
{"messages": [HumanMessage(content="サンプル を一言で紹介してください")]},
stream_mode="messages",
):
# msg_chunk は AIMessageChunk で、それぞれが 1 つの Token のみを含みます
if hasattr(msg_chunk, 'content') and msg_chunk.content:
print(msg_chunk.content, end="", flush=True)
print()
Agent の終了条件
Agent はいつ停止するのでしょうか？主に以下のケースがあります：
| 終了条件 | 説明 | 例 |
|---|---|---|
| ツール呼び出しなし | モデルが返した AIMessage の tool_calls が空 | モデルがタスク完了と判断し、直接応答 |
| return_direct=True | ツールが直接返却としてマークされ、実行後すぐに終了 | クエリ型ツールで、結果がそのまま最終回答 |
| structured_response | モデルが構造化出力を生成 | response_format で指定された構造化出力が完了 |
| jump_to="end" | Middleware が状態制御を通じて能動的に終了 | 権限超過の問題を検出し、早期終了 |
例
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# ケース 1：モデルが直接応答（ツール不要）
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[], # ツールなし
)
result = agent.invoke({
"messages": [HumanMessage(content="サンプル を一言で紹介してください")]
})
# モデルが直接応答し、ループは 1 回だけ実行
print(f"ツールなしシナリオ、メッセージ数: {len(result['messages'])}")
print(f"応答: {result['messages'][-1].content[:80]}...")
# ケース 2：ツール呼び出しが必要（複数回のループ）
@tool
def search_course(keyword: str) -> str:
"""サンプル のコースを検索"""
return f"{keyword} 関連コースを 3 件見つけました"
agent_with_tools = create_agent(
model=model,
tools=[search_course],
)
result = agent_with_tools.invoke({
"messages": [HumanMessage(content="Python コースを検索")]
})
print(f"\nツールありシナリオ、メッセージ数: {len(result['messages'])}")
# 通常は：human, ai(tool_call), tool(result), ai(final) の順になります
for msg in result["messages"]:
print(f" [{msg.type}]", end="")
if hasattr(msg, 'tool_calls') and msg.tool_calls:
print(f" tool_calls: {[tc['name'] for tc in msg.tool_calls]}")
else:
print(f" {str(msg.content)[:50]}")
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
# ケース 1：モデルが直接応答（ツール不要）
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[], # ツールなし
)
result = agent.invoke({
"messages": [HumanMessage(content="サンプル を一言で紹介してください")]
})
# モデルが直接応答し、ループは 1 回だけ実行
print(f"ツールなしシナリオ、メッセージ数: {len(result['messages'])}")
print(f"応答: {result['messages'][-1].content[:80]}...")
# ケース 2：ツール呼び出しが必要（複数回のループ）
@tool
def search_course(keyword: str) -> str:
"""サンプル のコースを検索"""
return f"{keyword} 関連コースを 3 件見つけました"
agent_with_tools = create_agent(
model=model,
tools=[search_course],
)
result = agent_with_tools.invoke({
"messages": [HumanMessage(content="Python コースを検索")]
})
print(f"\nツールありシナリオ、メッセージ数: {len(result['messages'])}")
# 通常は：human, ai(tool_call), tool(result), ai(final) の順になります
for msg in result["messages"]:
print(f" [{msg.type}]", end="")
if hasattr(msg, 'tool_calls') and msg.tool_calls:
print(f" tool_calls: {[tc['name'] for tc in msg.tool_calls]}")
else:
print(f" {str(msg.content)[:50]}")
実行結果：
无工具场景，消息数: 2 返答: サンプルは初心者向けのプログラミング学習プラットフォームで、豊富なチュートリアルとサンプルを提供しています... 有工具场景，消息数: 4 [human] 搜索 Python 课程 [ai] tool_calls: ['search_course'] [tool] 找到 python 相关课程 3 门 [ai] サンプルで「Pythonコース」を検索し、関連コースを3件見つけました...
invoke と stream の比較
| メソッド | 戻り値のタイミング | 適用シーン | ユーザー体験 |
|---|---|---|---|
| invoke() | すべて完了後に一括で返却 | スクリプト、API インターフェース、バッチ処理 | 待機後に完全な結果を確認 |
| stream() | 中間状態を段階的に返却 | チャットインターフェース、プロセス表示が必要な場合 | 進行状況をリアルタイムで確認 |
| ainvoke() | 非同期ですべて完了後に返却 | Web サービス、非同期フレームワーク | イベントループをブロックしない |
| astream() | 非同期で段階的に返却 | WebSocket、SSE プッシュ | サーバー側からのリアルタイムプッシュ |
with_config でスレッド ID を渡す
checkpointer（後続の章で詳しく説明します）を使用している場合は、config で thread_id を渡して会話スレッドを管理する必要があります：
例
# config はランタイム設定を渡すために使用
# thread_id は異なる会話スレッドを区別するために使用
config = {"configurable": {"thread_id": "conversation-001"}}
# invoke 方式
result = agent.invoke(
{"messages": [HumanMessage(content="こんにちは")]},
config=config,
)
# stream 方式も config をサポート
for chunk in agent.stream(
{"messages": [HumanMessage(content="こんにちは")]},
config=config,
stream_mode="updates",
):
print(chunk)
# thread_id は異なる会話スレッドを区別するために使用
config = {"configurable": {"thread_id": "conversation-001"}}
# invoke 方式
result = agent.invoke(
{"messages": [HumanMessage(content="こんにちは")]},
config=config,
)
# stream 方式も config をサポート
for chunk in agent.stream(
{"messages": [HumanMessage(content="こんにちは")]},
config=config,
stream_mode="updates",
):
print(chunk)