LangChain ストリーミング出力 Streaming

ストリーミング出力により、AI の返答がタイピングのように一文字ずつ表示され、ユーザーエクスペリエンスが大幅に向上します。LangChain の Agent には、充実したストリーミング出力サポートが組み込まれています。

なぜストリーミング出力が必要か

invoke() を使用する場合、ユーザーは Agent がすべてのステップ（複数回のモデル呼び出し + ツール実行）を完了するまで結果を待つ必要があります。複雑なタスクでは、これに十数秒以上かかることもあります。

ストリーミング出力はこの問題を解決します。Token が生成されるたびに即座に返されるため、ユーザーは進行状況をリアルタイムで確認できます。

方式ユーザーエクスペリエンス適用シーン
invoke()待機 → 一度に完全な結果を確認スクリプト、API、バッチ処理
stream()すべての Token をリアルタイムで確認チャットUI、リアルタイム表示

stream_mode="messages"——Token 単位のストリーミング

これは最も細かい粒度のストリーミングモードで、各 chunk が 1 つの Token に対応します：

例

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash")
agent = create_agent(
    model=model,
    system_prompt="あなたはサンプル のアシスタントです。",
)

# stream_mode="messages" で Token ごとに返す
print("リアルタイムストリーミング出力：")
for msg_chunk, metadata in agent.stream(
    {"messages": [HumanMessage(content="サンプル を一言で紹介してください")]},
    stream_mode="messages",
):
    # msg_chunk は AIMessageChunk
    # 各 chunk にはごく一部の内容だけが含まれる
    if msg_chunk.content:
        print(msg_chunk.content, end="", flush=True)
print()  # 最後に改行

実行結果：

ライブストリーミング出力:
サンプルはプログラミング初心者向けの無料オンライン学習プラットフォームで、豊富な技術チュートリアルと実践的なサンプルを提供しています。

metadata について

metadata には、この chunk の出所情報が含まれています：

例

print("metadata 情報を確認：\n")
for msg_chunk, metadata in agent.stream(
    {"messages": [HumanMessage(content="こんにちは、自己紹介してください")]},
    stream_mode="messages",
):
    if msg_chunk.content and len(msg_chunk.content) > 5:
        print(f"内容: {msg_chunk.content}")
        print(f"発生元ノード: {metadata.get('langgraph_node')}")
        print(f"メッセージタイプ: {type(msg_chunk).__name__}")
        break  # 最初の意味のある chunk だけを見る

実行結果：

メタデータを確認:

内容: こんにちは！私はAIアシスタントです
ソースノード: model
メッセージタイプ: AIMessageChunk

stream_mode="updates"——Agent の実行過程を段階的に確認

このモードは「思考過程」を表示する必要がある UI を構築するときに非常に役立ちます：

例

from langchain.tools import tool


@tool
def search_course(keyword: str) -> str:
    """サンプル でコースを検索"""
    courses = {
        "python": "Python3 基礎チュートリアル（30章、20時間）",
        "html": "HTML 基礎チュートリアル（25章、15時間）",
    }
    return courses.get(keyword.lower(), "関連コースが見つかりません")


agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
    tools=[search_course],
    system_prompt="あなたはサンプル のコースアドバイザーです。",
)

# updates モードで各ステップを確認
print("=== Agent 実行過程 ==="\n")
for chunk in agent.stream(
    {"messages": [HumanMessage(content="Python コースを調べてください")]},
    stream_mode="updates",
):
    for node_name, update in chunk.items():
        print(f"[{node_name}]", end=" ")
        if "messages" in update:
            for msg in update["messages"]:
                if msg.type == "ai":
                    if hasattr(msg, 'tool_calls') and msg.tool_calls:
                        calls = [tc['name'] for tc in msg.tool_calls]
                        print(f"リクエスト呼び出し: {calls}")
                    elif msg.content:
                        print(f"返信: {msg.content[:80]}")
                elif msg.type == "tool":
                    print(f"ツール返却 [{msg.name}]: {msg.content}")

実行結果：

=== Agent 执行过程 ===

[model] 请求调用: ['search_course']
[tools] 工具返回 [search_course]: Python3 基础教程（30章，20小时）
[model] 返答: サンプルにはPython3基礎チュートリアル（全30章、学習時間約20時間）があり、Python初心者の入門に最適です。

stream_mode="custom"——カスタムイベントの送信

Middleware の runtime.stream_writer() を使用すると、ストリームにカスタムイベントを送信できます：

例

from langchain.agents.middleware import before_model, after_model


@before_model
def notify_before(state, runtime):
    """モデル呼び出しの前にカスタムイベントを送信"""
    runtime.stream_writer({
        "type": "status",
        "message": "考え中...",
    })
    return None


@after_model
def notify_after(state, runtime):
    """モデル呼び出しの後にカスタムイベントを送信"""
    last_msg = state["messages"][-1] if state.get("messages") else None
    has_tools = hasattr(last_msg, 'tool_calls') and last_msg.tool_calls

    if has_tools:
        tool_names = [tc['name'] for tc in last_msg.tool_calls]
        runtime.stream_writer({
            "type": "status",
            "message": f"ツールを呼び出しています: {', '.join(tool_names)}...",
        })
    else:
        runtime.stream_writer({
            "type": "status",
            "message": 回答が完了しました,
        })
    return None


agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
    tools=[search_course],
    middleware=[notify_before, notify_after],
    system_prompt="あなたはサンプル のコースアドバイザーです。",
)

# stream_mode=["updates", "custom"] を使用して両方のイベントを受け取る
print("=== ハイブリッドストリーミング出力 ==="\n")
for mode, chunk in agent.stream(
    {"messages": [HumanMessage(content="Python コースを調べてください")]},
    stream_mode=["updates", "custom"],
):
    if mode == "custom":
        print(f"[カスタムイベント] ステータス: {chunk['message']}")
    elif mode == "updates":
        for node_name, update in chunk.items():
            if "messages" in update:
                for msg in update["messages"]:
                    if msg.type == "ai" and msg.content:
                        print(f"[返信] {msg.content}")

実行結果：

=== 混合流式输出 ===

[カスタムイベント] ステータス: 思考中...
[カスタムイベント] ステータス: ツール呼び出し中: search_course...
[返答] サンプルにはPython3基礎チュートリアル（全30章、学習時間約20時間）があります。
[カスタムイベント] ステータス: 思考中...
[カスタムイベント] ステータス: 回答完了

stream_mode は組み合わせて使用できます。例: stream_mode=["updates", "custom", "messages"]。ただし、モードが多すぎるとストリーム内のイベント量が増えるため、必要に応じて選択することをお勧めします。

非同期ストリーミング出力

Web サービスでは、非同期ストリーミングを使用するとイベントループのブロックを回避できます：

例

import asyncio


async def stream_agent():
    """Agent を非同期ストリーミングで実行"""
    agent = create_agent(
        model=init_chat_model("deepseek:deepseek-v4-flash"),
        system_prompt="あなたはサンプル のアシスタントです。",
    )

    full_response = ""
    async for msg_chunk, metadata in agent.astream(
        {"messages": [HumanMessage(content="サンプル を一言で紹介してください")]},
        stream_mode="messages",
    ):
        if msg_chunk.content:
            full_response += msg_chunk.content
            print(msg_chunk.content, end="", flush=True)

    print(f"\n\n"完全な返信の長さ: {len(full_response)} 文字")


# 非同期関数を実行
asyncio.run(stream_agent())

FastAPI 統合例

例

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

app = FastAPI()
agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash"),
    system_prompt="あなたはサンプル のアシスタントです。",
)


@app.get("/chat")
async def chat(message: str):
    """チャットAPI。SSE ストリーミングレスポンスを返す"""
    async def generate():
        async for msg_chunk, metadata in agent.astream(
            {"messages": [HumanMessage(content=message)]},
            stream_mode="messages",
        ):
            if msg_chunk.content:
                # SSE 形式：data: xxx\n\n
                yield f"data: {msg_chunk.content}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
    )

# 起動：uvicorn main:app --reload

本番環境では、Agent インスタンスをグローバルシングルトンとして作成し、リクエストごとに再作成しないことをお勧めします。Agent の作成コストは非常に小さいですが（主にグラフのコンパイル）、インスタンスを再利用する方が効率的です。

stream_mode 早見表

モード粒度反復オブジェクト代表的な用途
messagesToken レベル(AIMessageChunk, metadata)タイピング効果、リアルタイムチャット
updatesノードレベル{node_name: state_update}思考過程の表示
valuesノードレベル（全量）完全な state状態スナップショット、デバッグ
customカスタム任意の dict進捗通知、状態プッシュ
debug詳細デバッグ情報開発段階での問題調査
その他の拡張