LangChain的流式输出详解

2025-11-03 AI大模型 LangChain, Python 评论

LangChain 实现了一套流式输出系统，能够实时展示应用程序的更新状态。

流式输出对于提升基于大语言模型（LLM）构建的应用程序的响应性至关重要。通过逐步显示输出内容，即使在完整响应尚未就绪时也能展示部分结果，流式输出显著改善了用户体验（UX），特别是在处理LLM的延迟问题时效果尤为明显。

概述

LangChain 的流式输出系统让你能够将智能体运行时的实时反馈展示到你的应用程序中。

LangChain 流式输出支持以下功能：

流式输出智能体进度 — 在智能体每一步执行后获取状态更新
流式输出LLM令牌 — 在语言模型生成令牌时实时流式输出
流式输出自定义更新 — 发射用户定义的信号（例如：”已获取 10/100 条记录”）
多模式流式输出 — 可选择 updates（智能体进度）、messages（LLM令牌 + 元数据）或 custom（任意用户数据）

智能体进度

要流式输出智能体的执行进度，使用 stream 或 astream 方法并设置 stream_mode="updates"。这将在每个智能体步骤执行后发射一个事件。

例如，如果一个智能体调用了一次工具，你应该看到以下更新：

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

# 流式输出智能体进度
from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city} 的天气总是晴朗！"

agent = create_agent(
    model="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}")

输出示例：

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

步骤: tools
内容: [{'type': 'text', 'text': '旧金山的天气总是晴朗！'}]

步骤: model
内容: [{'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="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")

输出示例（部分）：

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

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

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

节点: tools
内容: [{'type': 'text', 'text': '旧金山的天气总是晴朗！'}]

节点: model
内容: [{'type': 'text', 'text': '这是'}]

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

自定义更新

要在工具执行时流式输出自定义更新，可以使用 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="claude-sonnet-4-5-20250929",
    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="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
内容: {'model': {'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')]}}

禁用流式输出

在某些应用中，你可能需要禁用特定模型的单个令牌流式输出。

这在多智能体系统中非常有用，可以控制哪些智能体流式输出其输出。

请参阅模型指南了解如何禁用流式输出。

参数详解

stream_mode 参数

stream_mode 参数控制流式输出的类型和粒度：

"updates": 流式输出智能体执行的步骤级更新，适合需要了解执行流程的场景
"messages": 流式输出LLM生成的令牌级内容，适合需要实时显示生成文本的场景
"custom": 流式输出工具内部的自定义数据，适合需要展示详细进度信息的场景
多模式: 可以同时指定多种模式，如 ["updates", "custom"]，获取不同类型的流数据

异步支持

除了同步的 stream 方法，LangChain 还提供了异步版本的 astream 方法，适用于异步应用程序：

async for chunk in agent.astream(
    {"messages": [{"role": "user", "content": "问题内容"}]},
    stream_mode="updates"
):
    # 处理流式数据
    pass

使用建议

用户体验优化: 对于需要长时间运行的智能体任务，使用 "updates" 模式向用户展示进度
实时交互: 对于聊天应用，使用 "messages" 模式提供更流畅的对话体验
调试监控: 使用 "custom" 模式在开发过程中监控工具执行状态
性能考虑: 在不需要实时更新的场景下禁用流式输出以提升性能

通过合理利用 LangChain 的流式输出功能，你可以构建响应更快、用户体验更好的AI应用程序。

本文链接： https://www.guoweiwei.com/2025/11/LangChain的流式输出详解/

版权声明： 本博客所有文章除特别声明外，均采用 CC BY 4.0 CN协议许可协议。转载请注明出处！

郭伟伟连续创业者 & 技术合伙人

去过互联网大厂，也在国企混过日子，目前在创业公司，主要负责科技公司的产品、技术、运营等管理工作。