ストリーミングは応答性の高いアプリケーションを構築するためのキーです。ストリーミングしたいデータには幾つか種類があります : エージェント進捗、LLM トークン、そしてカスタムアップデートです。
LangGraph : Prebuilt エージェント : ストリーミング
作成 : クラスキャット・セールスインフォメーション
作成日時 : 06/09/2025
* 本記事は langchain-ai.github.io の以下のページを独自に翻訳した上で、補足説明を加えてまとめ直しています :
* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。
◆ お問合せ : 下記までお願いします。
- クラスキャット セールス・インフォメーション
- sales-info@classcat.com
- ClassCatJP
LangGraph : Get started : Prebuilt エージェント : ストリーミング
ストリーミングは応答性の高いアプリケーションを構築するためのキーです。ストリーミングしたいデータには幾つか種類があります :
- エージェント進捗 – エージェントグラフの各ノードが実行された後の更新情報の取得。
- LLM トークン – 言語モデルにより生成中のトークンをストリーム。
- カスタムアップデート – 実行中にツールからのカスタムデータを出力します (e.g., “Fetched 10/100 records”)
一度に 複数の種類のデータ をストリームできます。
エージェント進捗
エージェント進捗をストリーミングするには、stream_mode=”updates” とともに stream() or astream() メソッドを使用します。これはすべてのエージェントステップ後にイベントを出力します。
例えば、ツールを 1 回呼び出すエージェントがある場合、以下のアップデートを見るはずです :
- LLM ノード : ツール呼び出しリクエストを含む AI メッセージ
- ツールノード : 実行結果を含むツールメッセージ
- LLM ノード : 最終的な AI レスポンス
同期
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_weather],
)
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="updates"
):
print(chunk)
print("\n")
非同期
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_weather],
)
async for chunk in agent.astream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="updates"
):
print(chunk)
print("\n")
LLM トークン
LLM により生成中のトークンをストリームするには、stream_mode=”messages” を使用します :
同期
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_weather],
)
for token, metadata in agent.stream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="messages"
):
print("Token", token)
print("Metadata", metadata)
print("\n")
非同期
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_weather],
)
async for token, metadata in agent.astream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="messages"
):
print("Token", token)
print("Metadata", metadata)
print("\n")
ツール更新
ツールの実行中に更新をストリームするには、get_stream_writer を使用できます :
同期
from langgraph.config import get_stream_writer
def get_weather(city: str) -> str:
"""Get weather for a given city."""
writer = get_stream_writer()
# stream any arbitrary data
writer(f"Looking up data for city: {city}")
return f"It's always sunny in {city}!"
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_weather],
)
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="custom"
):
print(chunk)
print("\n")
非同期
from langgraph.config import get_stream_writer
def get_weather(city: str) -> str:
"""Get weather for a given city."""
writer = get_stream_writer()
# stream any arbitrary data
writer(f"Looking up data for city: {city}")
return f"It's always sunny in {city}!"
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_weather],
)
async for chunk in agent.astream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="custom"
):
print(chunk)
print("\n")
複数のストリーミングモード
ストリームモードをリスト: stream_mode=[“updates”, “messages”, “custom”] として渡すことで、複数のストリーミングモードを指定できます :
同期
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_weather],
)
for stream_mode, chunk in agent.stream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode=["updates", "messages", "custom"]
):
print(chunk)
print("\n")
非同期
agent = create_react_agent(
model="anthropic:claude-3-7-sonnet-latest",
tools=[get_weather],
)
async for stream_mode, chunk in agent.astream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode=["updates", "messages", "custom"]
):
print(chunk)
print("\n")
ストリーミングの無効化
一部のアプリケーションでは、指定のモデルに対して個々のトークンのストリーミングを無効にする必要があるかもしれません。これは、マルチエージェント システムにおいてどのエージェントが出力をストリーミングするかを制御するのに役立ちます。
See the Models guide to learn how to disable streaming.
追加リソース
以上