LangGraph流式输出实战：如何用stream_mode优化大模型交互体验（附通义千问案例）
LangGraph流式输出实战：如何用stream_mode优化大模型交互体验（附通义千问案例）
如果你正在构建基于大模型的智能体应用，大概率遇到过这样的场景：用户问了一个问题，界面却陷入漫长的等待，光标闪烁，没有任何反馈。十几秒甚至几十秒后，一个完整的答案才突然出现。这种“黑盒”式的交互体验，在如今追求即时反馈的时代，已经很难满足用户期待。
这正是流式输出要解决的核心痛点。它让智能体的思考过程变得可见，将漫长的等待拆解为一个个可感知的进度更新。而LangGraph作为当前最流行的智能体编排框架之一，其stream_mode参数正是实现这一体验的关键。但面对updates、messages、values、custom、debug这五种模式，很多开发者会感到困惑：它们到底有什么区别？我的场景该用哪个？组合起来又该怎么用？
这篇文章，我将从一个实践者的角度，带你彻底搞懂LangGraph的流式输出。我不会只复述官方文档，而是结合真实的开发经验，通过一个完整的通义千问API案例，拆解每种模式的应用场景、代码实现和背后的设计逻辑。你会发现，用好stream_mode，不仅能提升用户体验，更是你调试复杂智能体工作流的利器。
1. 流式输出：从“黑盒”到“白盒”的智能体交互革命
在传统的大模型调用中，无论是简单的model.invoke()，还是早期的智能体实现，我们采用的往往是“请求-等待-完整响应”的模式。智能体内部可能经历了多轮思考、工具调用、信息整合，但对用户而言，这一切都发生在不可见的后台。这种模式带来几个显著问题：
- 用户体验差：长时间的空白等待会导致用户焦虑，怀疑请求是否失败。
- 调试困难：当智能体输出不符合预期时，开发者很难定位问题出在哪个环节（是工具调用出错？还是模型理解有误？）。
- 资源感知弱：用户无法感知到智能体正在进行的“有价值”的工作（例如，正在查询数据库、正在调用某个API），降低了过程的可信度。
流式输出改变了这一切。它的核心思想是将智能体执行过程中的中间状态和进度，实时地、增量地推送给客户端。这就像把智能体的“大脑”活动直播出来。LangGraph通过stream()和astream()方法，以及关键的stream_mode参数，为我们提供了多种维度的“直播视角”。
为了直观感受，我们先看一个最简单的对比。假设我们有一个查询天气并生成报告的智能体。
传统阻塞式调用：
# 用户发出请求后，开始漫长等待...
final_result = agent.invoke({"messages": [{"role": "user", "content": "上海天气怎么样？"}]})
# 几十秒后，控制台一次性打印出最终报告
print(final_result['messages'][-1].content)
# 输出：经过查询，上海当前天气晴朗，气温25度。基于此天气，建议进行户外活动...
流式输出调用（以updates模式为例）：
# 用户发出请求后，立即开始接收反馈
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "上海天气怎么样？"}]},
stream_mode="updates"
):
print(f"步骤更新: {chunk}")
# 输出会是这样的：
# 步骤更新: {'model': {'messages': [AIMessage(content='', tool_calls=[...])]}} # 模型思考，决定调用工具
# 步骤更新: {'tools': {'messages': [ToolMessage(content='上海：晴，25°C')]}} # 工具返回天气数据
# 步骤更新: {'model': {'messages': [AIMessage(content='上海当前天气晴朗，气温25度。')]}} # 模型生成最终回答
流式输出将一次性的等待，变成了一个连续的、有状态的对话过程。接下来，我们深入LangGraph提供的五种“直播视角”。
2. 核心五模式深度解析：找到你的“监控仪表盘”
stream_mode的五种选项，可以理解为五种不同精细度和维度的监控仪表盘。选择哪一种，取决于你想观察什么。
2.1 updates：聚焦状态变化的“增量日志”
这是最常用、也最直观的模式。它只关心状态（State）中哪些字段被修改了，并返回这些修改。
- 输出内容：一个字典，键是发生更新的节点名，值是该节点对状态所做的更新（也是一个字典）。
- 核心用途：
- 前端进度展示：理想用于构建进度条或步骤指示器。例如，显示“模型思考中 -> 调用查询工具 -> 生成回答”。
- 轻量级调试：快速查看工作流是否按预期步骤执行。
- 增量更新客户端状态：前端可以只根据变化的部分更新UI，效率更高。
实战代码片段：
from langgraph.prebuilt import create_react_agent
from langchain_community.chat_models.tongyi import ChatTongyi
import asyncio
# 初始化通义千问模型
llm = ChatTongyi(
model="qwen-max", # 或 qwen-turbo, qwen-plus
temperature=0,
)
# 定义一个模拟工具
def get_weather(city: str) -> str:
"""查询指定城市的天气"""
# 模拟一个耗时操作
import time
time.sleep(0.5)
return f"{city}的天气是晴朗，25摄氏度。"
# 创建智能体
agent = create_react_agent(model=llm, tools=[get_weather])
print("=== updates 模式流式输出 ===")
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "杭州的天气如何？"}]},
stream_mode="updates"
):
print(f"更新块: {chunk}")
print("-" * 40)
在这个例子中，updates模式会输出三个块，分别对应模型决定调用工具、工具执行返回结果、模型生成最终回答这三个步骤的状态更新。
2.2 messages：逐词流淌的“LLM思维流”
这个模式专门用于流式传输大模型本身生成的令牌（token）。即使你在节点内部使用的是llm.invoke()（非流式调用），只要设置了stream_mode="messages"，LangGraph也能将模型输出拆分成令牌流出来。
- 输出内容：一个元组
(message_chunk, metadata)。message_chunk是LLM输出的消息块（如AIMessageChunk），metadata包含了丰富的元数据，如哪个节点调用的LLM (langgraph_node)、步骤ID、模型信息等。 - 核心用途：
- 实现类ChatGPT的打字机效果：这是前端展示模型“实时思考”最关键的模式。
- 精细化监控与过滤：结合
metadata，可以只订阅特定节点或特定标签的模型输出。
实战代码片段：实现打字机效果
print("=== messages 模式流式输出 (逐词效果) ===")
for msg_chunk, metadata in agent.stream(
{"messages": [{"role": "user", "content": "用一句话介绍西湖"}]},
stream_mode="messages"
):
# 只处理模型节点产生的文本内容
if msg_chunk.content and metadata.get("langgraph_node") == "model":
print(msg_chunk.content, end="", flush=True) # end="" 确保不换行，flush=True立即刷新缓冲区
print() # 最后换行
提示：
metadata中的langgraph_node字段极其有用，它可以帮你区分当前流出的令牌是来自“模型思考”节点，还是来自其他可能调用LLM的节点。
2.3 values：每一步的“全局状态快照”
与updates的增量视角相反，values模式在每个节点执行后，都返回状态的完整快照。
- 输出内容：执行到当前步骤时，状态（State）对象的完整值。
- 核心用途：
- 状态回溯与调试：当逻辑复杂时，查看每一步的完整状态有助于理解数据是如何被一步步加工和传递的。
- 需要全量数据的下游处理：某些后处理逻辑可能需要基于某个中间步骤的完整状态进行计算。
- 注意：对于大型状态，
values模式会产生较多数据量，因为每次都会重复传输未变化的字段。
对比表格：updates vs values
| 特性 | stream_mode="updates" | stream_mode="values" |
|---|---|---|
| 数据量 | 小（仅增量） | 大（全量快照） |
| 内容 | {节点名: {更新的字段: 新值}} | {字段1: 值1, 字段2: 值2, ...} (完整状态) |
| 适用场景 | 前端进度更新、轻量调试 | 深度调试、需要完整状态的分析 |
| 类比 | Git 的 diff | Git 的完整文件 |
2.4 custom：自由度最高的“自定义广播”
这是最强大的模式，允许你从图节点或工具的内部，向外发送任意自定义数据。
- 实现方式：在节点或工具函数中，通过
get_stream_writer()获取一个“写入器”，然后调用它来发送数据。 - 核心用途：
- 工具执行进度报告：例如，一个数据库查询工具可以流式报告“已扫描1000/10000条记录”。
- 集成非LangChain原生流式LLM：即使你用的LLM客户端不兼容LangChain接口，也能通过
custom模式将其流式输出接入LangGraph的流。 - 发送任意业务数据：如发送中间计算结果、调试信息、或给前端的结构化指令。
实战代码片段：在工具中发送进度更新
from langgraph.config import get_stream_writer
def complex_data_query(query: str) -> str:
"""一个模拟的复杂数据查询工具"""
writer = get_stream_writer()
writer({"type": "progress", "message": "开始解析查询语句...", "percentage": 0})
# 模拟处理
import time; time.sleep(0.3)
writer({"type": "progress", "message": "连接数据库中...", "percentage": 30})
time.sleep(0.5)
writer({"type": "progress", "message": "获取到500条记录，正在分析...", "percentage": 70})
time.sleep(0.4)
writer({"type": "progress", "message": "分析完成，生成摘要。", "percentage": 100})
return "分析结果：趋势向好。"
# 创建使用此工具的智能体
agent_with_progress = create_react_agent(model=llm, tools=[complex_data_query])
print("=== custom 模式流式输出 (工具进度) ===")
for chunk in agent_with_progress.stream(
{"messages": [{"role": "user", "content": "分析一下Q2的销售数据"}]},
stream_mode="custom"
):
print(f"自定义进度: {chunk}")
前端可以监听这些custom事件，并实时更新一个进度条，极大提升用户对耗时操作的感知。
2.5 debug：开发者的“全链路透视镜”
这是信息量最大的模式，旨在输出图执行过程中尽可能多的信息，包括每个节点的输入、输出、内部事件等。它主要用于开发和调试阶段，帮助你深入理解LangGraph内部的工作机制。
- 注意：
debug模式的输出格式可能随着LangGraph版本更新而变化，且数据量非常庞大，不建议在生产环境使用。
3. 组合拳实战：通义千问智能体与多模式流式集成
理解了单个模式，我们来看如何在实际项目中组合使用它们。一个成熟的AI应用前端，往往需要同时接收多种流式数据：既要LLM的逐词输出（messages），又要知道智能体进行到了哪一步（updates），可能还需要工具的自定义进度（custom）。
LangGraph允许你通过传递一个列表给stream_mode来同时启用多种模式。
综合案例：构建一个带进度反馈的查询智能体
假设我们要构建一个智能体，它可以根据用户问题决定是调用“天气查询”还是“数据查询”工具，并且我们希望在前端实现：
- 显示智能体的执行步骤。
- 显示耗时的数据查询工具的进度。
- 以打字机效果流式显示模型的最终回答。
import asyncio
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_community.chat_models.tongyi import ChatTongyi
from langgraph.config import get_stream_writer
# 定义状态
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
final_answer: str
# 初始化通义千问
llm = ChatTongyi(model="qwen-turbo", temperature=0.1)
# 工具1：快速天气查询
def get_weather_tool(city: str) -> str:
return f"{city}天气：晴，26℃。"
# 工具2：带进度反馈的复杂数据查询
def query_sales_data_tool(period: str) -> str:
writer = get_stream_writer()
steps = ["连接数据仓库", "提取Q2数据", "计算环比", "生成报告"]
for i, step in enumerate(steps):
writer({
"type": "tool_progress",
"tool": "query_sales_data",
"step": step,
"progress": (i + 1) / len(steps)
})
asyncio.sleep(0.2) # 模拟耗时
return f"{period}销售数据报告：营收同比增长15%。"
# 构建一个简单的图
def should_use_tool(state: AgentState):
"""判断是否需要调用工具，这里简化为总是调用一个工具"""
# 在实际应用中，这里会是LLM判断的逻辑
# 为了示例，我们直接返回一个工具调用
from langchain_core.messages import AIMessage, ToolCall
tool_call = ToolCall(name="query_sales_data", args={"period": "Q2"}, id="1")
return {"messages": [AIMessage(content="", tool_calls=[tool_call])]}
def call_tool(state: AgentState):
"""执行工具调用"""
last_msg = state["messages"][-1]
tool_call = last_msg.tool_calls[0]
if tool_call["name"] == "query_sales_data":
result = query_sales_data_tool(**tool_call["args"])
else:
result = "工具未找到。"
from langchain_core.messages import ToolMessage
return {"messages": [ToolMessage(content=result, tool_call_id=tool_call["id"])]}
def generate_final_answer(state: AgentState):
"""基于工具结果生成最终答案"""
# 这里应该调用LLM，但为简化示例，我们直接拼接
tool_result = state["messages"][-1].content
answer = f"根据分析，{tool_result} 这表明业务发展健康。"
return {"final_answer": answer, "messages": [AIMessage(content=answer)]}
# 组装图
builder = StateGraph(AgentState)
builder.add_node("decide", should_use_tool)
builder.add_node("execute_tool", call_tool)
builder.add_node("answer", generate_final_answer)
builder.add_edge(START, "decide")
builder.add_edge("decide", "execute_tool")
builder.add_edge("execute_tool", "answer")
builder.add_edge("answer", END)
graph = builder.compile()
# **核心：同时使用三种流式模式**
print("启动综合流式输出...")
inputs = {"messages": [{"role": "user", "content": "帮我分析一下Q2的销售情况"}]}
for stream_mode, chunk in graph.stream(
inputs,
stream_mode=["updates", "messages", "custom"] # 组合三种模式
):
if stream_mode == "updates":
print(f"[步骤更新] {chunk}")
elif stream_mode == "messages":
msg, meta = chunk # 注意：组合模式下messages也是元组形式
if msg.content:
print(f"[模型输出] {msg.content}", end="", flush=True)
elif stream_mode == "custom":
print(f"[工具进度] {chunk}")
print("\n--- 执行结束 ---")
在这个例子中，一次graph.stream()调用会同时产生三种类型的数据流。你的客户端（或后端处理逻辑）需要根据stream_mode（即返回的元组的第一个元素）来区分和处理它们，将其分别用于更新步骤UI、刷新进度条和渲染文字流。
4. 高级技巧与避坑指南
掌握了基础用法，一些高级技巧和常见陷阱能让你用得更顺手。
4.1 精准过滤：只获取你关心的流
当图很复杂时，流式数据可能很多。你可以利用metadata进行过滤。
- 按节点过滤LLM令牌：只接收来自特定节点的
messages。for msg, meta in graph.stream(inputs, stream_mode="messages"): if meta.get("langgraph_node") == "research_agent": # 只关心研究节点 process_token(msg) - 按标签过滤：在初始化LLM时添加
tags，然后按标签过滤。from langchain.chat_models import init_chat_model summary_llm = init_chat_model("openai:gpt-4o-mini", tags=["summary"]) # ... 在图中使用这个llm ... for msg, meta in graph.stream(inputs, stream_mode="messages"): if meta.get("tags") == ["summary"]: process_summary_token(msg)
4.2 异步环境（Python < 3.11）的特殊处理
在Python 3.11以下版本中，异步上下文传播需要手动处理，否则get_stream_writer()和流式元数据可能无法正常工作。
- 关键点1：在异步节点函数中，必须显式接收
config参数，并将其传递给LLM的异步调用（如ainvoke）。async def async_node(state: State, config): # 注意这里的config response = await llm.ainvoke([...], config=config) # 必须传递config return {...} - 关键点2：在异步节点中使用
custom模式时，不能使用get_stream_writer()，而需要将writer作为函数参数。from langgraph.types import StreamWriter async def async_node_with_custom(state: State, writer: StreamWriter): # 直接接收writer writer({"progress": "start"}) # ... do work writer({"progress": "end"}) return {...}
4.3 性能与兼容性考量
- 禁用流式传输：在多智能体系统中，你可能只想让主智能体流式输出，而让后台工作的子智能体静默执行。可以在初始化模型时设置
disable_streaming=True。from langchain_openai import ChatOpenAI silent_llm = ChatOpenAI(model="gpt-4o", disable_streaming=True) - 子图流式传输：默认情况下，流式输出只包含顶层图的节点。如果你希望子图的执行细节也流式输出，需要在调用时设置
subgraphs=True。输出会附带命名空间信息，以区分数据来自哪个子图。
流式输出不仅仅是锦上添花的功能，它正在成为构建专业级、高用户体验AI应用的标配。通过深入理解和灵活运用LangGraph的stream_mode，你可以将智能体从神秘的“黑箱”转变为透明、可信、交互流畅的合作伙伴。从简单的updates进度提示，到复杂的多模式custom数据集成，这套工具链为你提供了充分的控制力。下次当你觉得智能体响应“太慢”时，不妨先别优化模型或代码，试试打开流式输出，或许用户体验的瓶颈就此突破。
更多推荐
23
0- 0
已为社区贡献6条内容
所有评论(0)