第 14 章 Streaming：七种 stream_mode 详解

流式输出（Streaming）是 LLM 应用区别于传统 RPC 服务的关键体验维度。大型语言模型单次推理耗时在数百毫秒到数十秒区间，若等待完整生成结果再一次性返回，首字延迟（Time To First Token，TTFT）会显著拖慢交互节奏，用户在等待期间看到空白界面，容易误判为系统卡死而重复提交；长任务场景（长文生成、多步 Agent、Map-Reduce 检索、多工具调度）执行过程中用户无法感知进度、无法提前打断、无法预览中间产物，严重影响可用性与可解释性。LangGraph 的流式机制在 Pregel 超步边界与 LLM 调用层之间埋入标准化事件钩子，将每个超步的状态变化、每个 token 的生成、每个节点的自定义进度事件统一组织为可订阅的异步流，供前端逐段渲染、后端实时落盘或监控系统消费。本章基于 LangGraph 1.2.x 正式 API，系统梳理七种 stream_mode 的行为、参数、最小示例与生产陷阱。

14.1 graph.stream / graph.astream 签名与 v2 输出格式

CompiledStateGraph 提供同步 stream() 与异步 astream() 两个流式入口，二者语义完全一致，仅迭代器协议不同：同步返回普通 Iterator，异步返回 AsyncIterator（需 async for 迭代）。LangGraph 1.2.x 中二者签名如下：

python

def stream(
    self,
    input: dict | Any,
    config: RunnableConfig | None = None,
    *,
    stream_mode: StreamMode | list[StreamMode] = "values",
    output_keys: str | list[str] | None = None,
    interrupt_before: All | list[str] | None = None,
    interrupt_after: All | list[str] | None = None,
    debug: bool | None = None,
    subgraphs: bool = False,
) -> Iterator[dict | tuple[str, dict]]: ...

参数	类型	说明
input	dict / Any	初始 State 输入，结构必须对齐图的 input_schema，未声明字段会被忽略
config	RunnableConfig / None	运行时配置；启用 checkpointer 时必须含 `configurable.thread_id`
stream_mode	str / list[str]	流模式标识，支持七种取值；传列表时可同时启用多种模式
output_keys	str / list[str] / None	仅推送指定 State 字段，用于裁剪 payload 体积
interrupt_before / after	"all" / list[str] / None	本次调用的动态断点，会覆盖编译期同名参数
debug	bool / None	覆盖编译期 debug 开关，None 表示不覆盖
subgraphs	bool	是否透传子图内部事件，默认 False（子图作为黑盒）

自 LangGraph 0.2 起引入 version="v2" 输出格式，1.2.x 默认为 v2。v2 将所有流事件统一为三元组结构：

text

{"type": <stream_mode_name>, "ns": "<namespace::path>", "data": <mode_specific_payload>}

type 字段标明事件来源模式；ns（namespace）字段为命名空间路径，父图与子图名以 :: 串联，根图层为空字符串；data 字段为各模式专属 payload 对象。当 stream_mode 传入单字符串时，迭代项直接是 data 载荷（保持向后兼容）；当传入列表（多模式组合）时，迭代项为 (mode_name, payload) 两元组，便于调用方按 mode 分流处理。v1 格式下不同模式返回结构差异较大（values 返回裸 State dict、messages 返回 (message_metadata, chunk) 元组），迁移到 v2 时前端 JSON 解析逻辑需要改为统一从 data 字段取载荷、以 type 字段做分支判断，并注意单模式与多模式返回值形态的差异（单模式直接 yield payload，多模式 yield tuple）。

最小可运行示例（v2 values 模式骨架）：

python

# langgraph==1.2.x
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    msg: str

def node_a(s: State) -> dict:
    return {"msg": s["msg"] + " ->a"}

def node_b(s: State) -> dict:
    return {"msg": s["msg"] + " ->b"}

builder = StateGraph(State)
builder.add_node("a", node_a).add_node("b", node_b)
builder.add_edge(START, "a").add_edge("a", "b").add_edge("b", END)
graph = builder.compile()

for ev in graph.stream({"msg": "start"}, stream_mode="values"):
    print(ev)
# 输出：{'msg': 'start'}
#      {'msg': 'start ->a'}
#      {'msg': 'start ->a ->b'}

需要说明的是，流式迭代器只有在被迭代（for 循环/异步 for）时才会驱动图执行。如果只调用 graph.stream(...) 而不迭代，图不会运行。流结束的标志是迭代器正常终止（StopIteration/StopAsyncIteration），此时图已执行到 END；若图运行中抛出未捕获异常，异常会从迭代器中 throw 出来。

14.2 七种 stream_mode 逐一详解

1) "values" 模式。每个超步（Superstep）结束、Channel 合并完成后，推送一次完整 State 快照，payload 为 State 字典，结构与 invoke() 返回值完全一致。该模式适合做全局状态看板、调试回放、后端全量持久化、离线审计等"需要完整上下文"的场景。其代价是 State 体积较大时（如 messages 列表累积数千条、或承载二进制字段、原始文档），每次推送都要序列化整份 State，容易出现单条消息上 MB 级 JSON，在网络传输、反序列化与前端 diff 渲染侧形成卡顿，严重时阻塞事件循环导致其他流事件延迟。生产中应配合 output_keys 参数裁剪字段，或在 State Schema 中避免把巨量原始数据放入顶层字段——原始长文档/二进制应放在外部 Store（BaseStore）或对象存储中，State 只存引用。

python

for ev in graph.stream({"msg": "hi"}, stream_mode="values"):
    print("SNAPSHOT:", ev)

2) "updates" 模式。仅推送当前超步内被调度节点产生的 State 增量，payload 格式为 {node_name: {field: value}}。同一超步内多个节点（Fan-out）并发执行时，各节点的增量被聚合到同一个字典中按 node_name 索引。该模式带宽占用小、语义清晰，是前端增量合并渲染的默认首选：前端维护一份本地 State 镜像，每次按 node_name 与 field 合并（按字段 reducer 规则）即可得到最新视图，无需每次全量替换。stream() 方法未显式指定 stream_mode 时默认即为 "updates"。

python

for ev in graph.stream({"msg": "hi"}, stream_mode="updates"):
    print("DELTA:", ev)
# 输出：{'a': {'msg': 'hi ->a'}}
#      {'b': {'msg': 'hi ->a ->b'}}

3) "messages" 模式。从图内所有 ChatModel 调用中抽取token 级增量，payload 为 AIMessageChunk 对象。每个 chunk 上携带元数据属性：langgraph_node（产生该 chunk 的节点名）、langgraph_step（超步序号）、langgraph_checkpoint_ns（checkpoint 命名空间，子图场景下标明来源）、tags（RunnableConfig 中传入的标签集合）。前端据此实现打字机效果，并通过 tags 或 langgraph_node 字段过滤掉工具节点、反思节点、规划节点等非目标输出，避免内部推理 token、工具调用参数泄露到用户界面。拼接 AIMessageChunk 时应使用 Message 的 __add__ 运算符（即 acc = acc + chunk），可自动合并 content、tool_calls、response_metadata、usage_metadata 等字段；若只用字符串 += chunk.content 会丢失 tool_calls 与 token 用量统计。

python

for meta, chunk in graph.stream({"messages": [("user", "你好")]}, stream_mode="messages"):
    node_name = getattr(chunk, "langgraph_node", None)
    if node_name == "agent" and isinstance(chunk.content, str):
        print(chunk.content, end="", flush=True)

4) "custom" 模式。节点函数通过 langgraph.config.get_stream_writer() 获取写入器，主动向流推送任意自定义数据（该数据不进入 State、不影响执行流）。payload 结构完全由业务代码定义，典型用途包括长任务进度百分比（{"percent":60}）、阶段性产物草稿（{"draft":"..."}）、中间日志、检索召回片段预览、子任务完成通知、工具调用前/后的状态提示等。写入器是同步/异步双态可调用对象：在同步 def 节点中以 w(data) 同步调用，在 async def 节点中以 await w(data) 异步调用，混用会触发 RuntimeError: Attempted to call async stream writer from sync context 或事件被静默丢弃。写入操作是非阻塞的——数据进入异步队列后立即返回，不会因消费者未读取而阻塞节点执行，但队列有上限，消费者不及时读取会导致背压生效。

python

from langgraph.config import get_stream_writer

def long_task(s: State):
    w = get_stream_writer()
    for i in range(3):
        w({"stage": i, "text": f"进度 {i}/3"})
        # ... 实际处理（LLM 调用、工具调用等）
    return {"msg": "done"}

5) "checkpoints" 模式。每当 checkpointer 写入新快照（含三种来源：超步正常结束、interrupt 中断触发、update_state 外部更新），推送一条 checkpoint 事件。payload 包含：config（本快照的 RunnableConfig，含 thread_id 与 checkpoint_id）、parent_config（父快照 config，根快照为 None）、metadata（source 枚举值、step 超步序号、writes 节点写入、sends 待派发 Send 列表等）、values（该快照的完整 State 值）、next（下一步将调度的节点名元组）。该模式用于实现时间旅行调试、外部审计落盘、增量备份、状态同步到外部存储等场景。未配置 checkpointer 时该模式不产出任何事件。

6) "tasks" 模式。推送每个 Pregel 任务的生命周期事件。任务是 Pregel 对"一次节点调用"的内部表示，每条事件含 type（"spawn" 任务开始、"finish" 任务成功结束、"error" 任务异常）与对应 payload：spawn 含 task_id、path（命名空间元组）、input（节点入参）；finish 含 result（节点返回值）；error 含 error 异常对象。path 字段以元组形式标明任务所在子图与节点，如 ("parent", "research", "retrieve")，便于定位"是哪个子图的哪个节点出了错"。该模式用于构建执行计划可视化面板、分布式链路追踪、任务级耗时统计。

7) "debug" 模式。最细粒度调试事件，涵盖节点入参快照、出参快照、执行耗时、Channel 写入详情、条件路由 path 函数返回值、Barrier 等待情况等内部调度事件，单轮典型运行可产生数十到上百条事件。debug 模式面向本地开发阶段排查问题（例如"为什么某个节点没被触发""为什么路由走了错误分支"），生产环境必须禁用：输出量大、可能包含敏感明文数据（用户输入、密钥、token），且序列化会显著拖慢执行。

14.3 多模式组合、事件流、子图透传与前端对接

多模式组合通过传入列表启用：stream_mode=["values", "messages", "custom"]。此时迭代器中不同模式的事件按时间序交错到达，迭代项为 (mode_name, payload) 两元组，调用方按 mode 分发到不同处理器即可。多模式共用同一异步迭代器与同一事件队列，不会互相阻塞；某一模式消费者慢（如写入数据库）会通过背压影响所有模式的推送速率，因此消费侧应做异步解耦（把事件放入独立的 asyncio.Queue 或 Kafka 再慢处理）。

python

async for mode, data in graph.astream(
    {"messages": [("user", "你好，自我介绍一下")]},
    stream_mode=["messages", "custom"],
):
    if mode == "messages":
        if getattr(data, "langgraph_node", None) == "agent":
            print(data.content, end="", flush=True)
    elif mode == "custom":
        print("\n[progress]", data)

graph.astream_events()（v3 事件流）基于 LangChain LCEL（LangChain Expression Language）的标准事件系统，将图内所有 Runnable（LLM、Tool、Chain、Prompt、Retriever）的 on_chain_start/stream/end、on_chat_model_start/stream/end、on_tool_start/end、on_retriever_start/end 等事件以统一 schema 推送，事件对象包含 event（事件类型）、name（Runnable 名称）、run_id、tags、metadata、data（含 input/output/chunk 等）字段。与 stream_mode 的关键差异：stream_mode 面向"LangGraph 超步与 State"视角，产出是 Pregel 调度的产物，包含 custom/checkpoints/tasks 等图级事件；astream_events 面向"LangChain Runnable 树"视角，产出与图结构无关，所有 Runnable 都会产出事件，适合对接 LangSmith、OpenTelemetry、LangFuse 等通用可观测体系。前端打字机与 Agent 进度场景优先用 stream_mode（messages+custom，结构简单、语义清晰），全链路追踪与跨系统可观测场景优先用 astream_events。

子图流式透传通过 subgraphs=True 开启。开启后子图内部的各模式事件会透传到父图迭代器，事件的 ns 字段（tuple 路径）标明来源，如 "research::retrieve" 表示名为 research 的子图中的 retrieve 节点。前端可按 ns.split("::") 渲染层级进度条（"研究子图 → 检索节点"）。未开启 subgraphs 时，子图作为整体只在其入口（节点被调度）与出口（节点返回）产出一次父图层事件，内部 token、custom 事件、tasks 事件都被屏蔽。

禁用节点流式输出。对敏感节点（密钥计算、用户隐私数据处理、内部审批逻辑）需要阻止其 token 或中间数据出现在流中。方式一：节点调用 LLM 时在 RunnableConfig 中传入 tags=["langsmith:hidden"]（阻止 LangSmith 记录）和自定义标签（如 tags=["internal"]），前端在 messages 模式事件上按标签过滤；方式二：对节点使用 @traceable(exclude_inputs=True, exclude_outputs=True) 标注，阻止其输入输出进入 LangSmith 与 events 流；方式三：在图外的 SSE 网关层对 ns、langgraph_node 做黑名单过滤。必须注意：messages 模式默认推送图内所有 ChatModel 输出，包括 ToolNode 周边的 LLM 调用、反思节点、规划节点，若不显式过滤，工具参数 JSON、内部推理草稿可能泄露到前端。

前端对接实践。浏览器端主要通过两种协议消费 LangGraph 流：(1) SSE（Server-Sent Events）：后端以 text/event-stream Content-Type 逐事件输出 data: {...}\n\n，前端用 EventSource 或 fetch+ReadableStream 接收；(2) fetch ReadableStream：直接读取 HTTP 响应流，手动按行解析，适合需要自定义 header（鉴权）的 POST 场景。v2 流的 JSON 解析必须采用"逐行/逐块增量解析"策略：从流中读取字节块，按 \n 切分，对每个完整行 JSON.parse，不能等整个响应结束再一次性 parse。messages 模式的 AIMessageChunk 在 SSE 中被序列化为 JSON（含 content、tool_calls、response_metadata 等字段），前端需要按 LangChain 序列化协议反序列化（或简化为只取 content 字段做打字机渲染，tool_calls 走单独通道处理）。AIMessageChunk 拼接时不要用字符串 +=，应累积后在流结束时通过 AIMessageChunk 加法合并为完整 AIMessage 再写入 State。

完整示例：打字机效果+进度提示 Agent。

python

# langgraph==1.2.x
import asyncio
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.config import get_stream_writer
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage

class S(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def agent(s: S):
    w = get_stream_writer()
    w({"phase": "thinking", "detail": "正在整理思路..."})
    # 模拟一个小停顿
    import time; time.sleep(0.3)
    w({"phase": "writing", "detail": "开始生成回答..."})
    r = llm.invoke(s["messages"])
    w({"phase": "done", "detail": "回答生成完毕"})
    return {"messages": [r]}

g = StateGraph(S).add_node("agent", agent)
g.add_edge(START, "agent").add_edge("agent", END)
graph = g.compile()

async def main():
    async for mode, data in graph.astream(
        {"messages": [("user", "用三句话介绍你自己")]},
        stream_mode=["messages", "custom"],
    ):
        if mode == "custom":
            print("\n[progress]", data)
        elif mode == "messages":
            if getattr(data, "langgraph_node", None) == "agent":
                c = data.content
                if isinstance(c, str):
                    print(c, end="", flush=True)

asyncio.run(main())

本章小结

LangGraph 的流式体系围绕统一的 v2 三元组结构 {type, ns, data} 组织，七种 stream_mode 分别覆盖：完整状态快照（values，每超步推送全量 State）、节点增量（updates，每节点推送 State delta，默认模式）、LLM token 流（messages，AIMessageChunk 打字机效果，带节点元数据）、节点自定义事件（custom，通过 get_stream_writer 主动推送进度/草稿/日志）、检查点事件（checkpoints，快照落盘即推送，用于审计与时间旅行）、任务生命周期（tasks，任务 spawn/finish/error，用于执行计划可视化）、细粒度调试（debug，仅开发期使用）。多模式通过传列表在同一迭代器中交错推送；astream_events 提供基于 LCEL 的通用 Runnable 事件树，适用于可观测体系；subgraphs=True 透传子图事件，ns 字段标明来源路径；敏感节点通过 tags/traceable 标注禁用流式输出。前端推荐 SSE + fetch ReadableStream，按事件 type 分发到消息区、进度区、调试区；AIMessageChunk 通过 + 运算符合并而非字符串拼接。

常见错误

State 过大导致 values 模式卡死前端。现象：前端 EventSource 长时间无新事件、浏览器标签冻结、单条 SSE 消息数 MB。原因：messages 未裁剪、State 中放入长文本/二进制原文，values 模式每超步序列化整份 State。修复：前端默认使用 updates 模式；必须用 values 时配合 output_keys 只暴露必要字段；长列表字段用 reducer 增量追加，前端按增量渲染；原始大字段放 Store 或对象存储，State 只存引用。
messages 模式未过滤 tags 导致工具节点消息泄露。现象：界面上突然出现 ToolMessage 文本、SQL 查询体、工具调用参数 JSON 等内部信息。原因：messages 模式默认推送图内所有 ChatModel 输出，包括绑定在 ToolNode 周边的 LLM 调用、反思/规划节点。修复：前端渲染前判断 chunk.langgraph_node 白名单，或在 LLM 调用时传入 tags=["visible"]，前端只渲染带 visible 标签的 chunk；内部节点统一打 tags=["internal"]。
custom 模式 writer 在同步节点 await（或异步节点同步调用）报错。现象：RuntimeError: Attempted to call async stream writer from sync context，或事件不出现。原因：节点函数是同步 def，却对 writer 使用了 await；或反之在 async 节点中把 writer 当同步可调用。修复：同步节点 w(data)；异步节点 await w(data)；获取 writer 后立即根据节点类型选择调用方式。
SSE 经过 Nginx/CDN 被缓冲。现象：后端已 flush，前端迟迟收不到数据，直到响应结束一次性全部到达。原因：Nginx 默认开启 proxy_buffering，CDN 默认开启响应缓冲，SSE 被整段缓存。修复：响应头加入 X-Accel-Buffering: no、Content-Type: text/event-stream、Cache-Control: no-cache、Connection: keep-alive；Nginx 配置 proxy_buffering off; proxy_read_timeout 300s;；CDN 关闭对应路径缓存。
AIMessageChunk 用字符串 += 拼接丢失 tool_calls 和 token 统计。现象：流式渲染正常，但结束后存到 State 的 message 丢失工具调用信息与 usage_metadata。原因：str(chunk.content) 只取文本，丢弃了 tool_calls、response_metadata、usage_metadata。修复：使用 acc = acc + chunk（Message 加法）累积，最后取 acc.content 渲染，acc 整体写入 State。

生产建议

默认流模式使用 ["messages", "custom"] 服务前端对话界面；后端审计落盘单独订阅 "checkpoints" 模式，避免同一流承担多职责；调试面板单独订阅 "tasks" 模式；
前端实现统一事件总线层：按 type/ns/langgraph_node 分发到消息区、进度区、调试区，不直接在 UI 组件中解析裸 JSON；
长任务 Agent 必须用 custom 模式推送结构化进度（阶段名 stage、百分比 percent、预计剩余 token、当前子任务），提升可解释性与用户信任感；
Nginx/CDN 层显式关闭 SSE 缓冲并设置合理空闲超时（300s），配合客户端自动重连（Last-Event-ID 机制）与 thread_id 断点续传；
debug 模式仅在开发/灰度环境开启，生产环境所有内部节点默认 tags 追加 internal，敏感节点追加 hidden，在 SSE 网关层做黑名单过滤；
AIMessageChunk 拼接统一封装工具函数（前端/后端各一份），避免业务代码散落手工 content 拼接导致 tool_calls 字段丢失、token 统计不准；
后端消费流时用独立 asyncio.Task 异步迭代，避免阻塞主请求线程；高并发场景使用 Redis Pub/Sub 或消息队列中转流事件，WebSocket 网关水平扩展；
stream_mode 列表中避免一次性开启全部七种模式，按场景按需组合，减少序列化开销与带宽占用。

C 语言

语言点

深度

新标准

面试整理

第 14 章 Streaming：七种 stream_mode 详解 ​

14.1 graph.stream / graph.astream 签名与 v2 输出格式 ​

14.2 七种 stream_mode 逐一详解 ​

14.3 多模式组合、事件流、子图透传与前端对接 ​

本章小结 ​

常见错误 ​

生产建议 ​

第 14 章 Streaming：七种 stream_mode 详解

14.1 graph.stream / graph.astream 签名与 v2 输出格式

14.2 七种 stream_mode 逐一详解

14.3 多模式组合、事件流、子图透传与前端对接

本章小结

常见错误

生产建议