LangGraph 流式三口径：messages 模式与 6 个真实报错

TL;DR：LangGraph 给你三个流式入口，但它们不是同一件事的三种写法。stream() 与 astream() 是「状态机视角」——按你指定的 stream_mode 把图运行时的数据吐出来；astream_events() 是「回调视角」——把 LangChain 回调事件（on_chain_start / on_chat_model_stream / on_chain_end …）原样转给你。我实际测试 langgraph 1.2.16 + langchain-core 1.6.12 之后，最反直觉的一条是：stream_mode="messages" 的 token 级输出不是图在流，是一个挂在回调链上的 StreamMessagesHandler 在流。这解释了三件事：为什么拼错模式名会静默返回 0 个 chunk、为什么子图里的 token 默认不冒泡、为什么 nostream 标签挡掉了 token 却挡不掉那条完整消息。

一、先把三个入口摆正

官方文档 Streaming 页开篇就写了一句很容易被跳过的话：

"For new applications, we recommend event streaming—the typed-projection API introduced in LangGraph v1.2."

也就是说，官方自己把 stream(mode=...) 这条路径定位成「需要直接访问图运行时事件或特定模式输出时」的接口，新应用推荐走 v1.2 引入的 event streaming。这不是说 stream-mode API 不能用了——它仍然是当下绝大多数项目在用的那一套——而是你选型前得知道有两条路。

三条入口的语义差别可以这样记：

astream() 不是「在 async 里跑的 stream」。它俩的 stream_mode 语义完全相同——我实测同一张图、同一种模式下，stream() 与 astream() 的 chunk 数都是 13，一个不差。真正的差别在「节点怎么跑」：同步 def 节点会被 LangGraph 丢进线程池，async def 节点才跑在事件循环里。这一条 Troubleshooting 里会给数字。

二、七个 stream_mode 的 chunk 形态（本机实测）

下面这张表是同一张图（START → chat → END）、同一个输入、只改 stream_mode 跑出来的。测试模型是一个自写的 BaseChatModel 子类，把 45 个字符的原文按每片 4 个字符切出来，这样 token 数可预期：ceil(45/4) = 12 片。

class FakeChatModel(BaseChatModel):
    text: str = "LangGraph streams tokens one chunk at a time."  # 45 字符
    per: int = 4
    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content=self.text))])
    def _stream(self, messages, stop=None, run_manager=None, **kwargs):
        for i in range(0, len(self.text), self.per):
            yield ChatGenerationChunk(message=AIMessageChunk(content=self.text[i:i+self.per]))

checkpoints 那一行值得单独说。官方文档 Checkpoints 小节写着 "Requires a checkpointer"，但代码不检查、不报错、不警告——你把 stream_mode="checkpoints" 传给一张没配 checkpointer 的图，得到的是 0 个 chunk 和一个正常结束的循环。补上 compile(checkpointer=MemorySaver()) 之后是 3 个 chunk，键为 config / metadata / next / parent_config / tasks / values。

三、version 参数：v1 的元组长度会随组合变

stream() 和 astream() 都有个 version 参数，默认 'v1'。官方迁移表把差异写得很清楚，但如果你只读过单模式的例子，第一次碰上组合用法会懵：

最后一行我实测确认过：父图包一层子图，stream_mode=["updates", "messages"], subgraphs=True 时，产出元组的长度集合是 {3}，形如 ("agent:078e6684-...",) + "messages" + (AIMessageChunk, {...})。而 subgraphs=False 时长度集合是 {2}。我实际测试时的 chunk 数从 2 跳到 15。

所以下面这种「看起来通用」的取值代码，在加了子图或第二个模式之后就会静默取错：

# 只在「单模式 + 无子图」下正确
for chunk in graph.stream(inputs, stream_mode="messages"):
    msg, meta = chunk        # 一旦开启子图，chunk 变成三元组，这里直接 ValueError

改成 v2 能一次性消掉这种分支：

for part in graph.stream(inputs, stream_mode=["updates", "messages"],
                         subgraphs=True, version="v2"):
    if part["type"] == "messages":
        msg, meta = part["data"]
        print(msg.content, end="", flush=True)

四、token 级输出的真相：是回调在流，不是图在流

这是本文的核心。

翻 langgraph/pregel/_messages.py，stream_mode="messages" 的实现体是一个回调处理器：

class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
    """A callback handler that implements stream_mode=messages."""

注册方式在 main.py 里，很直白：

if "messages" in stream_modes:
    run_manager.inheritable_handlers.append(
        messages_handler_cls(stream.put, subgraphs, parent_ns=...)
    )

它挂在 inheritable_handlers 上——会被继承到图内部所有 LangChain 调用。这解释了官方文档那句乍看有点奇怪的话：

> "Note that message events are emitted even when the LLM is run using .invoke rather than .stream"

我的实测：节点里明明写的是 model.invoke(...)，整段一次返回，但 stream_mode="messages" 依然拿到了 13 个 chunk——12 个 AIMessageChunk（'Lang'、'Grap'、'h st' … '.'）加 1 个 content='' 的收尾 chunk。12 片拼回去正好是那 45 个字符，一个不多一个不少，非空 chunk 占比 12/13 = 92.3%。那个空的收尾 chunk 是流结束的信号，不是脏数据——但它会让「按 chunk 数算进度条」这类写法产生一格偏差。

对应地，astream_events(version="v2") 在同一张图上产出 21 个事件，分布是 on_chain_start 2、on_chat_model_start 1、on_chat_model_stream 13、on_chat_model_end 1、on_chain_stream 2、on_chain_end 2。只看 on_chat_model_stream 的 13 个，拼接结果与原文完全一致。

同一个机制还解释了两件容易踩的事。

子图的 token 默认不冒泡。 官方原话是 "Without subgraphs=True, stream_mode='messages' on the parent graph will not emit token chunks from the inner agent's LLM calls"。实测对照：父图包子图，subgraphs=False 时父图侧 token chunk 数 0，subgraphs=True 时 13。官方还补了一句很实在的话——create_agent 返回的是已编译的图，所以当节点挂进父图就自动变成了子图，"this often shows up only after wrapping"。

nostream 标签挡的是 token，不是消息。 官方说带 nostream 的调用 "still run and produce output; their tokens are simply not emitted in messages mode"。我实测把模型设成 .with_config({"tags": ["nostream"]}) 后，chunk 数从 13 掉到 1，而那唯一的 1 个 chunk 是 AIMessage，content 是完整的 45 字符原文。也就是说：它挡掉了 12 个中间 token，但节点最终返回的那条完整消息照样进流。如果你的前端一边按 token 增量渲染、一边又消费最终消息，这里就是重复输出的来源。

顺手能用的四条判据

把上面几节压成四句话，方便照着选：

1. 只要 token：优先 astream_events(version="v2") 后过滤 on_chat_model_stream，或者 stream_mode="messages"；图里挂了子图就必须补 subgraphs=True，否则一个 token 都拿不到。

2. 只要状态：增量用 updates，快照用 values。别为了省事直接开 debug——它每个 chunk 都带着完整 payload，实测 2 个 chunk 的体量远超前两者。

3. 同时用多个模式：直接上 version="v2"，省掉按元组长度分叉的代码。

4. 排障先数 chunk 数：0 个 chunk 不代表图没跑完，八成是模式名拼错或者 checkpointer 没配——这两条都不报错。

五、Troubleshooting：6 个真实报错

以下 6 条全部本机复现，报错原文照抄。本文不含任何联盟推广链接。

报错一：stream_mode 拼错 → 静默 0 chunk

for c in graph.stream(inputs, stream_mode="tokens"):   # 本意是 "messages"
    ...
# 无异常、无警告，循环体一次都不进

实测 5 种错误写法——"tokens"、"message"（少个 s）、"Messages"（大小写）、"update"、"values "（尾空格）——全部返回 0 个 chunk 且不抛任何异常。因为未知模式只是进不了任何 if mode in stream_modes 分支。排错时先 print(len(list(...)))，别急着怀疑模型。

报错二：checkpoints 模式无 checkpointer → 静默 0 chunk

与报错一同源，但更隐蔽：官方文档写了 "Requires a checkpointer"，代码却不做校验。判据是「有没有给 compile() 传 checkpointer」。补上 MemorySaver() 后立刻从 0 变 3。

报错三：TypeError: tuple indices must be integers or slices, not str

for chunk in graph.stream(inputs, stream_mode="messages"):
    print(chunk["content"])     # ❌ chunk 是 (message_chunk, metadata) 元组

messages 模式的 chunk 是二元组，不是 dict。正确写法是 msg, meta = chunk。这条在开启 subgraphs=True 后会进一步恶化成解包数量不匹配，见第三节。

报错四：TypeError: 'async_generator' object is not iterable

for e in graph.astream_events(inputs, version="v2"):    # ❌ 用了 for 而不是 async for
    ...

astream_events 是异步生成器，只能在 async for 里消费。同步上下文里要拿事件，请用 stream(..., stream_mode=...)。

报错五：TypeError: 'async for' requires an object with __aiter__ method, got coroutine

async for e in graph.astream_events(inputs, version="v3"):   # ❌ v3 少了一次 await
    ...
# TypeError: 'async for' requires an object with __aiter__ method, got coroutine

version="v3" 返回的不是异步生成器，是协程。签名里写得明白：-> "AsyncIterator[StreamEvent] | Awaitable[Any]"。正确用法要先 await 一次：

stream = await graph.astream_events(inputs, version="v3")
async for e in stream:
    ...

实测 await 之后拿到的是 AsyncGraphRunStream，跑出 18 个事件；同时会收到 LangChainBetaWarning: The v3 streaming protocol on Pregel is experimental.——v3 目前是实验协议，生产环境别指望它稳定。

报错六：事件循环被 async def 里的 time.sleep() 卡死

这条没有异常，只有「界面卡住了」。我用 0.60 秒睡眠加 50 毫秒心跳任务做对照，读心跳次数：

（未阻塞的期望值约 12 次，实测两次 10 属于调度开销。）async def 只是让函数能进事件循环，不会把里面的阻塞调用变异步——time.sleep 在 async 节点里会直接冻住整条循环，流式输出也就跟着一起停。I/O 一律用 await，CPU 密集的活儿显式丢线程池。

六、本文的边界

• 无可用 API Key，全部数据来自本机自写的 BaseChatModel 子类，token 分片是人为按固定 4 字符切的，不是真实 tokenizer 的分片；真实模型的 chunk 粒度由厂商决定，本文结论讲的是管道行为，不是 tokenizer 行为。
• 未测 Postgres / Redis checkpointer 下的 checkpoints 与 tasks 模式，只用了 MemorySaver。
• 未测 async 节点加真实并发、Send 扇出，以及 astream_events 的 v3 协议细节（只验证了调用方式与事件总数）。
• 版本：langgraph 1.2.16、langgraph-checkpoint 4.3.0、langchain-core 1.6.12，Python 3.13 / Windows 11。官方最新为 1.2.17（GitHub Releases，2026-10-24 发布），本文结论未经该版本复验。

相关阅读

• LangGraph 持久化实测：checkpointer、interrupt/resume、time travel 与 8 个真实报错——本篇的下一层：状态怎么存。
• LangGraph 人审中断：interrupt 与 resume 怎么接得上——同系列，讲中断之后怎么醒。
• 从零手搓注意力机制：NumPy 逐行实现与 PyTorch 对拍——同样是「不靠外部凭据，本机跑出数字」的路子。

👉 立即参与小米 MiMo 开放平台：国内领先的 AI 大模型开放平台，高性价比推理服务

👉 立即参与阿里云 AI：汇集爆款 AI 产品，热门模型专属权益优惠券助力企业创新加速

📌 本文由 AI 辅助生成并经人工审核发布 | TechPassive — AI 驱动的内容测试站点，专注于效率工具与 SaaS 真实评测