流式传输

Alpha 版本提示： 本文档涵盖的是 v1-alpha 版本。内容尚不完整，且可能随时变更。如需查阅最新稳定版本，请参阅 v0 版本的 LangChain Python 文档或 LangChain JavaScript 文档。

LangChain 实现了一套流式传输系统，用于实时推送更新。流式传输对于提升基于大语言模型（LLM）构建的应用程序的响应速度至关重要。通过在完整响应尚未生成完毕时就逐步显示输出内容，流式传输显著改善了用户体验（UX），尤其在应对 LLM 的延迟时效果尤为明显。

从智能体进行流式传输

LangChain 的流式传输系统允许你将智能体运行过程中的实时反馈推送到你的应用程序中。使用 LangChain 流式传输可以实现的功能包括：

流式传输智能体进度 —— 在每个智能体步骤后获取状态更新。
流式传输 LLM 令牌 —— 在语言模型生成令牌时实时推送。
流式传输自定义更新 —— 发送用户自定义信号（例如 "已获取 10/100 条记录"）。
同时流式传输多种模式 —— 可选择 updates（智能体进度）、messages（LLM 令牌 + 元数据）或 custom（任意用户数据）。

智能体进度

要流式传输智能体进度，请使用 stream() 或 astream() 方法，并设置 stream_mode="updates"。此设置将在每个智能体步骤后发出一个事件。例如，如果你有一个仅调用一次工具的智能体，你应看到如下更新：

LLM 节点：包含工具调用请求的 AI 消息
工具节点：包含执行结果的工具消息
LLM 节点：最终的 AI 响应

流式传输智能体进度

from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """获取指定城市的天气。"""

    return f"{city}总是阳光明媚！"

agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[get_weather],
)
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "旧金山的天气怎么样？"}]},
    stream_mode="updates",
):
    for step, data in chunk.items():
        print(f"步骤: {step}")
        print(f"内容: {data['messages'][-1].content_blocks}")

输出

步骤: agent
内容: [{'type': 'tool_call', 'name': 'get_weather', 'args': {'city': 'San Francisco'}, 'id': 'call_OW2NYNsNSKhRZpjW0wm2Aszd'}]

步骤: tools
内容: [{'type': 'text', 'text': "旧金山总是阳光明媚！"}]

步骤: agent
内容: [{'type': 'text', 'text': '旧金山总是阳光明媚！'}]

LLM 令牌

要流式传输 LLM 生成的令牌，请使用 stream_mode="messages"。下面你可以看到智能体流式传输工具调用和最终响应的输出。

流式传输 LLM 令牌

from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """获取指定城市的天气。"""

    return f"{city}总是阳光明媚！"

agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[get_weather],
)
for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "旧金山的天气怎么样？"}]},
    stream_mode="messages",
):
    print(f"节点: {metadata['langgraph_node']}")
    print(f"内容: {token.content_blocks}")
    print("\n")

输出

节点: agent
内容: [{'type': 'tool_call_chunk', 'id': 'call_vbCyBcP8VuneUzyYlSBZZsVa', 'name': 'get_weather', 'args': '', 'index': 0}]

节点: agent
内容: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': '{"', 'index': 0}]

节点: agent
内容: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': 'city', 'index': 0}]

节点: agent
内容: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': '":"', 'index': 0}]

节点: agent
内容: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': 'San', 'index': 0}]

节点: agent
内容: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': ' Francisco', 'index': 0}]

节点: agent
内容: [{'type': 'tool_call_chunk', 'id': None, 'name': None, 'args': '"}', 'index': 0}]

节点: agent
内容: []

节点: tools
内容: [{'type': 'text', 'text': "旧金山总是阳光明媚！"}]

节点: agent
内容: []

节点: agent
内容: [{'type': 'text', 'text': '以下是'}]

节点: agent
内容: [{'type': 'text', 'text': '我'}]

节点: agent
内容: [{'type': 'text', 'text': '查到'}]

节点: agent
内容: [{'type': 'text', 'text': '的'}]

节点: agent
内容: [{'type': 'text', 'text': '：“'}]

节点: agent
内容: [{'type': 'text', 'text': "旧金山"}]

节点: agent
内容: [{'type': 'text', 'text': '总是'}]

节点: agent
内容: [{'type': 'text', 'text': '阳光'}]

节点: agent
内容: [{'type': 'text', 'text': '明媚'}]

节点: agent
内容: [{'type': 'text', 'text': '！”\n\n'}]

自定义更新

要在工具执行时流式传输更新，你可以使用 get_stream_writer。

流式传输自定义更新

from langchain.agents import create_agent
from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """获取指定城市的天气。"""
    writer = get_stream_writer()
    # 流式传输任意数据
    writer(f"正在查找城市 {city} 的数据")
    writer(f"已获取城市 {city} 的数据")
    return f"{city}总是阳光明媚！"

agent = create_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "旧金山的天气怎么样？"}]},
    stream_mode="custom"
):
    print(chunk)

输出

正在查找城市 San Francisco 的数据
已获取城市 San Francisco 的数据

如果你在工具内部添加了 get_stream_writer，则无法在 LangGraph 执行上下文之外调用该工具。

同时流式传输多种模式

你可以通过将流模式作为列表传递来指定多个流模式：stream_mode=["updates", "custom"]：

同时流式传输多种模式

from langchain.agents import create_agent
from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """获取指定城市的天气。"""
    writer = get_stream_writer()
    writer(f"正在查找城市 {city} 的数据")
    writer(f"已获取城市 {city} 的数据")
    return f"{city}总是阳光明媚！"

agent = create_agent(
    model="openai:gpt-5-nano",
    tools=[get_weather],
)

for stream_mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "旧金山的天气怎么样？"}]},
    stream_mode=["updates", "custom"]
):
    print(f"流模式: {stream_mode}")
    print(f"内容: {chunk}")
    print("\n")

输出

流模式: updates
内容: {'agent': {'messages': [AIMessage(content='', response_metadata={'token_usage': {'completion_tokens': 280, 'prompt_tokens': 132, 'total_tokens': 412, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 256, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-nano-2025-08-07', 'system_fingerprint': None, 'id': 'chatcmpl-C9tlgBzGEbedGYxZ0rTCz5F7OXpL7', 'service_tier': 'default', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--480c07cb-e405-4411-aa7f-0520fddeed66-0', tool_calls=[{'name': 'get_weather', 'args': {'city': 'San Francisco'}, 'id': 'call_KTNQIftMrl9vgNwEfAJMVu7r', 'type': 'tool_call'}], usage_metadata={'input_tokens': 132, 'output_tokens': 280, 'total_tokens': 412, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 256}})]}}

流模式: custom
内容: 正在查找城市 San Francisco 的数据

流模式: custom
内容: 已获取城市 San Francisco 的数据

流模式: updates
内容: {'tools': {'messages': [ToolMessage(content="旧金山总是阳光明媚！", name='get_weather', tool_call_id='call_KTNQIftMrl9vgNwEfAJMVu7r')]}}

流模式: updates
内容: {'agent': {'messages': [AIMessage(content='旧金山天气：旧金山总是阳光明媚！\n\n', response_metadata={'token_usage': {'completion_tokens': 764, 'prompt_tokens': 168, 'total_tokens': 932, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 704, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-nano-2025-08-07', 'system_fingerprint': None, 'id': 'chatcmpl-C9tljDFVki1e1haCyikBptAuXuHYG', 'service_tier': 'default', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--acbc740a-18fe-4a14-8619-da92a0d0ee90-0', usage_metadata={'input_tokens': 168, 'output_tokens': 764, 'total_tokens': 932, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 704}})]}}

禁用流式传输

在某些应用程序中，你可能需要为特定模型禁用单个令牌的流式传输。这在多智能体系统中非常有用，可以控制哪些智能体流式传输其输出。请参阅模型指南，了解如何禁用流式传输。

快速开始

核心组件

进阶使用

在生产环境使用

从智能体进行流式传输

智能体进度

LLM 令牌

自定义更新

同时流式传输多种模式

禁用流式传输

​从智能体进行流式传输

​智能体进度

​LLM 令牌

​自定义更新

​同时流式传输多种模式

​禁用流式传输

从智能体进行流式传输

智能体进度

LLM 令牌

自定义更新

同时流式传输多种模式

禁用流式传输