LangGraph : Prebuilt エージェント : ストリーミング

by

ストリーミングは応答性の高いアプリケーションを構築するためのキーです。ストリーミングしたいデータには幾つか種類があります : エージェント進捗、LLM トークン、そしてカスタムアップデートです。

LangGraph : Prebuilt エージェント : ストリーミング

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

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

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

 

 

LangGraph : Get started : Prebuilt エージェント : ストリーミング

ストリーミングは応答性の高いアプリケーションを構築するためのキーです。ストリーミングしたいデータには幾つか種類があります :

  1. エージェント進捗 – エージェントグラフの各ノードが実行された後の更新情報の取得。

  2. LLM トークン – 言語モデルにより生成中のトークンをストリーム。

  3. カスタムアップデート – 実行中にツールからのカスタムデータを出力します (e.g., “Fetched 10/100 records”)

一度に 複数の種類のデータ をストリームできます。


Waiting is for pigeons.

 

エージェント進捗

エージェント進捗をストリーミングするには、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.

 

追加リソース

 

以上