LangGraph 设计与实现

第13章 流式输出与调试

作者 杨艺韬 · 14,703 字 · 发布于 · 更新于

13.1 引言

在构建 LLM 应用时，流式输出（streaming）不仅是用户体验的刚需，更是生产系统的可观测性基石。用户期望看到 AI 回复的"打字机效果"，而开发者需要实时监控图执行的每一步——哪个节点正在运行、状态如何变化、中间结果是什么。LangGraph 通过七种 StreamMode 和精心设计的 StreamPart 类型体系，提供了从粗粒度到细粒度的完整流式输出方案。

LangGraph 在 1.1.0 引入了 v2 Stream API（PR #6961 "type safe stream/invoke w/ proper output type coercion"），本专栏基准的 1.1.6 沿用这套接口。它将所有流式数据统一为带类型标签的 StreamPart 字典，解决了 v1 中多种 stream_mode 混合使用时类型模糊的问题。v2 API 使得每个流式事件都是自描述的：你可以通过 part["type"] 直接判断事件类型，无需依赖上下文推断。

本章将从 StreamMode 的枚举定义出发，逐一分析七种模式的语义和实现，深入 StreamProtocol 和 StreamMessagesHandler 的源码，揭示流式输出在 Pregel 循环中的注入点和数据流转路径。

本章要点

  1. 七种 StreamMode 的语义区别——values/updates/custom/messages/checkpoints/tasks/debug
  2. v2 Stream API 的 StreamPart 类型体系——带类型标签的联合类型
  3. StreamWriter 的注入机制——如何在节点中发射自定义流式数据
  4. StreamMessagesHandler 的回调实现——LLM token 级别的流式捕获
  5. StreamProtocol 在 Pregel 循环中的集成方式

13.2 StreamMode 枚举

13.2.1 七种模式定义

StreamMode 定义在 libs/langgraph/langgraph/types.py 中，是一个字面量联合类型：

StreamMode = Literal[
    "values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"
]

每种模式对应一种输出粒度和视角：

StreamMode 输出内容 触发时机 典型消费者
values 完整状态快照 每个超步结束后 前端展示完整状态
updates 节点名 + 该节点的输出 每个节点执行后 调试、日志
custom 用户自定义数据 节点内调用 StreamWriter 进度条、中间结果
messages LLM 消息 token LLM 流式生成时 打字机效果
checkpoints 检查点快照 检查点创建时（需配置 checkpointer，否则不发） 持久化监控
tasks 任务开始/结束事件 任务生命周期变化 执行监控
debug checkpoints + tasks 同上两者 开发调试

13.2.2 模式组合

stream 方法支持同时订阅多种模式：

# v1 API：多模式返回元组
for mode, data in graph.stream(input, stream_mode=["values", "messages"]):
    if mode == "values":
        print(f"State: {data}")
    elif mode == "messages":
        print(f"Token: {data}")

# v2 API：统一的 StreamPart（stream_mode 仍要显式列出，
# 不传时 StateGraph 编译出的图只发 "updates"）
for part in graph.stream(input, stream_mode=["values", "messages"], version="v2"):
    if part["type"] == "values":
        print(f"State: {part['data']}")
    elif part["type"] == "messages":
        msg, meta = part["data"]
        print(f"Token: {msg.content}")

v1 下多模式必须传列表：_output 用 isinstance(stream_mode, list) 判断是否给数据加上 mode 标签（libs/langgraph/langgraph/pregel/main.py:3624-3627），传元组 ("updates", "custom") 时两种事件照样都会产出，但不带 mode，无从区分（实跑确认）。

13.3 StreamPart 类型体系

13.3.1 v2 API 的核心创新

v2 Stream API 将每个流式事件包装为一个带 type 字段的 TypedDict，使得类型判别成为可能：

# libs/langgraph/langgraph/types.py:329
StreamPart = TypeAliasType(
    "StreamPart",
    ValuesStreamPart[OutputT]
    | UpdatesStreamPart
    | MessagesStreamPart
    | CustomStreamPart
    | CheckpointStreamPart[StateT]
    | TasksStreamPart
    | DebugStreamPart[StateT],
    type_params=(StateT, OutputT),
)

每个具体的 StreamPart 类型都包含三个公共字段：

classDiagram
    class StreamPart {
        <<TypedDict>>
        +type: str
        +ns: tuple[str, ...]
        +data: Any
    }
    class ValuesStreamPart {
        +type: "values"
        +ns: tuple[str, ...]
        +data: OutputT
        +interrupts: tuple[Interrupt, ...]
    }
    class UpdatesStreamPart {
        +type: "updates"
        +ns: tuple[str, ...]
        +data: dict[str, Any]
    }
    class MessagesStreamPart {
        +type: "messages"
        +ns: tuple[str, ...]
        +data: tuple[AnyMessage, dict]
    }
    class CustomStreamPart {
        +type: "custom"
        +ns: tuple[str, ...]
        +data: Any
    }
    class TasksStreamPart {
        +type: "tasks"
        +ns: tuple[str, ...]
        +data: TaskPayload | TaskResultPayload
    }
    class CheckpointStreamPart {
        +type: "checkpoints"
        +ns: tuple[str, ...]
        +data: CheckpointPayload
    }
    class DebugStreamPart {
        +type: "debug"
        +ns: tuple[str, ...]
        +data: DebugPayload
    }
    StreamPart <|-- ValuesStreamPart
    StreamPart <|-- UpdatesStreamPart
    StreamPart <|-- MessagesStreamPart
    StreamPart <|-- CustomStreamPart
    StreamPart <|-- TasksStreamPart
    StreamPart <|-- CheckpointStreamPart
    StreamPart <|-- DebugStreamPart

13.3.2 ns 字段：命名空间追踪

每个 StreamPart 都包含 ns（namespace）字段，它是一个字符串元组，标识事件来自图的哪个层级：

  • ()：来自顶层图
  • ("outer:<task_id>",)：来自父图节点 outer 调用的子图——每段是"调用子图的那个节点名:任务 ID"，不是子图的名字
  • ("outer:<task_id>", "inner:<task_id>")：来自嵌套两层的子图

这使得在启用 subgraphs=True 时，消费者可以精确识别每个事件的来源（实跑一个父图节点 outer 内嵌子图，子图事件的 ns 是 ('outer:0d34c449-3d8c-99b9-69ec-60ba5356d33c',)）。v1 下同样的信息放在元组最前面：单模式时产出 (ns, data)，多模式时产出 (ns, mode, data)（libs/langgraph/langgraph/pregel/main.py:3624-3629）。不开 subgraphs 时，子图内部的 updates / values 等事件不会汇入父图的流。

13.3.3 ValuesStreamPart

class ValuesStreamPart(TypedDict, Generic[OutputT]):
    type: Literal["values"]
    ns: tuple[str, ...]
    data: OutputT
    interrupts: tuple[Interrupt, ...]

values 模式在每个超步结束后发射完整的状态快照。data 的类型与图的输出类型一致。interrupts 字段记录了该步中发生的中断——这是 v2 API 特有的增强，v1 中中断信息需要通过 __interrupt__ 键间接获取。

13.3.4 UpdatesStreamPart

class UpdatesStreamPart(TypedDict):
    type: Literal["updates"]
    ns: tuple[str, ...]
    data: dict[str, Any]

updates 模式在每个节点执行后发射该节点的输出。data 是一个字典，键是节点名，值是节点返回的更新。如果多个节点在同一超步并行执行，每个节点的更新会单独发射。

13.3.5 MessagesStreamPart

class MessagesStreamPart(TypedDict):
    type: Literal["messages"]
    ns: tuple[str, ...]
    data: tuple[AnyMessage, dict[str, Any]]

messages 模式捕获 LLM 的 token 级流式输出。data 是一个二元组：消息对象（通常是 AIMessageChunk）和元数据字典。元数据包含 langgraph_step、langgraph_node、langgraph_triggers 等上下文信息。

13.3.6 CustomStreamPart

class CustomStreamPart(TypedDict):
    type: Literal["custom"]
    ns: tuple[str, ...]
    data: Any

custom 模式承载用户通过 StreamWriter 发射的任意数据。这是最灵活的模式——你可以用它发送进度百分比、中间计算结果、或任何自定义结构。

13.4 StreamWriter 注入机制

13.4.1 定义与签名

StreamWriter = Callable[[Any], None]

StreamWriter 的类型定义极为简洁——它就是一个接收单个参数、无返回值的可调用对象。要拿到它，节点函数可以声明一个名为 writer 的参数，注解写 StreamWriter 或干脆不写注解（注解成别的类型则不注入，见下一节）：

def my_node(state: State, writer: StreamWriter) -> dict:
    writer({"progress": 0.5})  # 发射自定义流式数据
    # ... 执行业务逻辑 ...
    writer({"progress": 1.0})
    return {"result": "done"}

在 LangGraph 1.1.6 中，StreamWriter 也通过 Runtime 对象提供：

def my_node(state: State, runtime: Runtime) -> dict:
    runtime.stream_writer({"progress": 0.5})
    return {"result": "done"}

不想改函数签名时，还可以在节点或函数式 API 的 task 里调用 langgraph.config.get_stream_writer()，它返回的就是当前 Runtime 的 stream_writer（libs/langgraph/langgraph/config.py:195-196；docstring 注明 Python 3.11 以下的异步运行里不可用，因为依赖 contextvar 传播）。这条限制其实不只针对 get_stream_writer()：astream 订阅 custom 时注入的 writer 闭包自己也要调用 get_config() 算 ns（libs/langgraph/langgraph/pregel/main.py:2999-3012），在 pin 环境的 Python 3.10 上实跑，异步运行里即便用 writer 参数写 custom 数据，也会抛 RuntimeError: Called get_config outside of a runnable context。

13.4.2 注入流程

注入分两步。第一步在包装节点函数时：RunnableCallable.__init__（libs/langgraph/langgraph/_internal/_runnable.py）用 inspect.signature 扫一遍参数，逐条比对常量表 KWARGS_CONFIG_KEYS——

# 节选（libs/langgraph/langgraph/_internal/_runnable.py:132 起；省略了 store 的 Optional 变体一行等）
KWARGS_CONFIG_KEYS = (
    ("config",   (RunnableConfig, "RunnableConfig", ...),               "N/A",           ...),
    ("writer",   (StreamWriter, "StreamWriter", inspect.Parameter.empty), "stream_writer", lambda _: None),
    ("store",    (BaseStore, "BaseStore", ...),                         "store",         ...),
    ("previous", (ANY_TYPE,),                                           "previous",      ...),
    ("runtime",  (ANY_TYPE,),                                           "N/A",           ...),
)

参数名必须叫 writer，注解要落在第二列里——StreamWriter、字符串 "StreamWriter"，或者 inspect.Parameter.empty（即不写注解）（runtime 一行的 ANY_TYPE 表示不校验注解），且参数种类要是 POSITIONAL_OR_KEYWORD 或 KEYWORD_ONLY；匹配上了才记入 self.func_accepts。第二步在每次调用时：从 config[CONF][CONFIG_KEY_RUNTIME] 取出 Runtime，按表里第三列的属性名（writer → Runtime.stream_writer）取值注入。这也解释了为什么 runtime.stream_writer(...) 和 writer(...) 拿到的是同一个对象。

这个对象本身是 Pregel.stream() 订阅了 custom 时现场定义的闭包（libs/langgraph/langgraph/pregel/main.py:2628-2641）：它从 get_config() 读出当前任务的 checkpoint_ns、截掉最后一段作为 ns，然后直接把 (ns, "custom", data) 放进输出队列，不经过 StreamProtocol：

sequenceDiagram
    participant Node as 节点函数
    participant SW as stream_writer 闭包
    participant Queue as 输出队列
    participant Client as 消费者
    Node->>SW: writer({"progress": 0.5})
    SW->>Queue: stream.put((ns, "custom", data))
    Queue-->>Client: 消费者读取

13.4.3 无订阅时的空操作

当调用方没有订阅 custom 模式时，StreamWriter 会变成一个空操作（no-op）：stream() 在这种情况下注入的是一个函数体只有 pass 的局部 stream_writer（libs/langgraph/langgraph/pregel/main.py:2644-2647）；Runtime.stream_writer 字段的默认值也是空函数（libs/langgraph/langgraph/runtime.py:77）：

# libs/langgraph/langgraph/runtime.py:77
def _no_op_stream_writer(_: Any) -> None: ...

这个设计确保了节点代码不需要检查 stream_mode 是否包含 "custom"——写入操作总是安全的，只是在没有订阅者时数据会被静默丢弃。子图是个容易踩的例外：子图作为节点运行时自己的 stream_mode 默认是 "values"，只有父图以 subgraphs=True 运行、把流放进 config[CONF][CONFIG_KEY_STREAM] 时，子图才会沿用父图的 stream_writer（libs/langgraph/langgraph/pregel/main.py:2642-2643）；否则子图用的是它自己那个 pass 空函数——它不是 _no_op_stream_writer 本身，Runtime.merge 会优先采用它而不是父图的 writer（libs/langgraph/langgraph/runtime.py:196-198）。实跑：父图 stream_mode=["updates", "custom"] 而不开 subgraphs 时，子图节点里写的 custom 数据一条也收不到。

13.5 StreamProtocol 与 StreamChunk

13.5.1 StreamProtocol

StreamProtocol 是流式输出的传输层抽象，定义在 libs/langgraph/langgraph/pregel/protocol.py 中：

StreamChunk = tuple[tuple[str, ...], str, Any]
# (命名空间元组, 模式字符串, 数据)

class StreamProtocol:
    __slots__ = ("modes", "__call__")

    modes: set[StreamMode]
    __call__: Callable[[Self, StreamChunk], None]

    def __init__(
        self,
        __call__: Callable[[StreamChunk], None],
        modes: set[StreamMode],
    ) -> None:
        self.__call__ = cast(Callable[[Self, StreamChunk], None], __call__)
        self.modes = modes

StreamProtocol 只有两个属性：

  • modes：订阅的模式集合，用于过滤
  • __call__：接收 StreamChunk 的回调函数

13.5.2 __slots__ 优化：StreamProtocol 的内存布局

看 StreamProtocol.__slots__ = ("modes", "__call__") 这一行——给一个类加 slots：

class StreamProtocol:
    __slots__ = ("modes", "__call__")
    modes: set[StreamMode]
    __call__: Callable[[Self, StreamChunk], None]

Python __slots__ 的作用——禁止实例动态新增属性，并用固定槽位代替 __dict__ 存储字段。带来两个好处：实例不再携带一个字典（省内存），属性访问走槽位描述符而不是字典查找（省时间）。StreamProtocol 在一次带子图的 streaming 中会创建多个实例（每层图一个，再加上 DuplexStream 的包装），而 stream.modes 的成员检查处在每个事件都要经过的热路径上，因此这里的取舍是明确的。

为什么把 __call__ 作为 slot？——注意源码里 __call__ 是实例属性而不是类方法：self.__call__ = cast(..., __call__)。这样每个 StreamProtocol 实例都能携带自己的回调闭包（DuplexStream 返回的就是这样一个实例），把"往哪里发"这件事变成运行时可绑定的数据，而不是靠子类化。它之所以能直接 stream(chunk) 调用，恰恰是因为 Python 仍在类型上查找 __call__：__slots__ 在类上为 __call__ 生成了一个槽位描述符，调用实例时解释器在类上找到这个描述符，再由它取出该实例槽位里存的回调函数来执行。

13.5.3 DuplexStream

在子图场景中（父图以 subgraphs=True 运行时），子图的流式输出需要同时发送到子图自己的流和父图的流式管道。DuplexStream 实现了这种多路复用：

def DuplexStream(*streams: StreamProtocol) -> StreamProtocol:
    def __call__(value: StreamChunk) -> None:
        for stream in streams:
            if value[1] in stream.modes:  # value[1] 是模式字符串
                stream(value)

    return StreamProtocol(__call__, {mode for s in streams for mode in s.modes})
graph LR
    Node[节点输出] --> Duplex[DuplexStream]
    Duplex -->|"mode in modes_A"| StreamA[子图 Stream]
    Duplex -->|"mode in modes_B"| StreamB[父图 Stream]

13.5.4 DuplexStream 的闭包 + 过滤合并

回到上一节的 DuplexStream 源码（libs/langgraph/langgraph/pregel/_loop.py:133），两个关键设计：

① 内部 __call__ 闭包捕获 streams——每次调用 DuplexStream 都创建新闭包、streams 被 capture 进去。调用 duplex(chunk) 时按 modes 过滤分发——只有订阅了这个 mode 的 stream 才收到。

② modes 合并——用 set comprehension {mode for s in streams for mode in s.modes} 构造一个超集——返回的 StreamProtocol 订阅的 modes 是所有 streams 订阅的 mode 的并集。上层看到这个超集、往 duplex 投递任何 mode 的 chunk——duplex 内部再按子 stream 过滤。

这就是事件系统的"fan-out + filter"模式——一个入口、多个出口、每个出口独立过滤。DuplexStream 用不到十行代码就实现了这套分发的核心，不需要专门的 message broker。

子图场景下的完整数据流：

  1. 子图节点 emit chunk
  2. 子图的 StreamProtocol 接收
  3. 父图以 subgraphs=True 运行时（libs/langgraph/langgraph/pregel/main.py:2702-2703 把父图的流放进 config[CONF][CONFIG_KEY_STREAM]），子图的 PregelLoop 会把自己的流包成 DuplexStream(own_stream, parent_stream)（libs/langgraph/langgraph/pregel/_loop.py:255-256）
  4. 同时分发到两个 stream——子图自己的流（子图的 invoke 靠它收集返回值）收到 + 父图的流也收到（如果订阅了该 mode）

这个机制让子图的 streaming 既不影响子图自身的运行、又能透过到父图——values / updates / tasks / checkpoints / debug 这些经 PregelLoop._emit 发出的事件，嵌套 streaming 就靠这个。custom 和 messages 不走这条路：前者用的是父图传下来的 stream_writer（§13.4.3），后者由挂在父图回调管理器上的 StreamMessagesHandler 直接捕获（§13.6）。

13.5.5 流式数据在 PregelLoop 中的注入

PregelLoop 在超步的几个关键位置发射流式事件，都走同一个 _emit 入口（行号取自 pregel/_loop.py）：

class PregelLoop:
    def tick(self) -> bool:
        ...
        self.tasks = prepare_next_tasks(...)
        # 新检查点：checkpoints 事件（:495）
        self._emit("checkpoints", map_debug_checkpoint, ...)
        ...
        # 本超步要跑的任务：tasks 开始事件（:531）
        self._emit("tasks", map_debug_tasks, self.tasks.values())
        return True

    def after_tick(self) -> None:
        # 任务跑完，写入落到 channel
        self.updated_channels = apply_writes(...)
        # 输出通道有变化才发 values 事件（:557）
        if not self.updated_channels.isdisjoint(...):
            self._emit("values", map_output_values, self.output_keys, writes, self.channels)
        ...

    def output_writes(self, task_id, writes, *, cached=False) -> None:
        # 单个任务完成时：updates 事件 + tasks 结束事件
        ...

注意两点：values 是在 after_tick 里、且只有输出通道确实被更新过才发（另有两处例外：任务产生中断时，output_writes（libs/langgraph/langgraph/pregel/_loop.py:994-1002）会补发一条带 __interrupt__ 的 values；顶层图因中断退出时，_suppress_interrupt（libs/langgraph/langgraph/pregel/_loop.py:870）会先应用已有写入、再发最后一条（同文件 889-914 行））；而 updates 与任务结束事件走的是 output_writes——它按单个任务触发，所以同一超步里并行的多个节点会各自发一条。

13.5.6 StreamChunk = tuple[tuple[str, ...], str, Any] 的三元组设计

StreamChunk 是 LangGraph 内部流式事件的wire format：

StreamChunk = tuple[tuple[str, ...], str, Any]
#              (命名空间元组,    模式字符串, 数据)

为什么是 tuple 而不是 dict 或 class？——三点考虑：

① 轻量——每个 token、每个节点更新都会产生一个 chunk，tuple 是 Python 里开销最小的复合结构之一：没有 __dict__、不可变，创建和解包都有专门的字节码。

② 解构简单——ns, mode, data = chunk 一行解构，热路径上的 value[1] in stream.modes（见 DuplexStream）直接按位置取，不必经过键查找。

③ 类型别名能表达清楚——StreamChunk = tuple[tuple[str, ...], str, Any] 让类型检查器知道结构，简洁之外仍有静态检查。

代价——字段顺序即 API。想加第 4 个字段就得改动所有解构点。LangGraph 用版本化的对外接口（v1 / v2）来隔离这种演进——v2 的 StreamPart 是 TypedDict，加字段不破坏既有解构。

内部用 tuple、外部用 TypedDict 是 LangGraph 的分层设计——内部追求性能、对外保证 API 灵活。转换发生在 pregel/main.py 的 _output：version="v2" 时把三元组拼成 {"type", "ns", "data"} 字典；v1 下用户拿到的仍是按 stream_mode / subgraphs 裁剪过的元组或裸数据（§13.10.7）。

13.5.7 PregelLoop._emit 的函数式事件发射

PregelLoop 里 emit 事件的模式是高阶函数 + mapper：

# pregel/_loop.py
def _emit(self, mode, values, *args, **kwargs):
    if self.stream is None:
        return
    # 订阅了 "debug" 的消费者，同样能收到 checkpoints / tasks 事件（外面套一层壳）
    debug_remap = mode in ("checkpoints", "tasks") and "debug" in self.stream.modes
    if mode not in self.stream.modes and not debug_remap:
        return
    for v in values(*args, **kwargs):
        if mode in self.stream.modes:
            self.stream((self.checkpoint_ns, mode, v))
        if debug_remap:
            self.stream((self.checkpoint_ns, "debug", {...}))  # 包上 step/timestamp/type

三层解耦：

① 订阅守卫——mode not in self.stream.modes and not debug_remap 就直接返回，没订阅的模式连 mapper 都不会调用——省掉整段读通道、组装数据的开销。debug_remap 这个额外条件正是 §13.2.1 表格里 "debug = checkpoints + tasks" 的实现：debug 订阅者会以包了一层 {step, timestamp, type, payload} 的形式收到这两类事件。

② mapper 是函数——map_debug_checkpoint / map_debug_tasks 等——每个 mode 有自己的专用 mapper。mapper 只负责 "把内部状态转成 stream chunk"、不关心订阅逻辑、不关心发送。

③ self.stream((ns, mode, chunk)) 的 uniform 发送——任何 mapper 产生的 chunk 走同一个 stream channel。

这种 "订阅守卫 + 懒计算 mapper + 统一发送" 的模式让 PregelLoop 用很少的代码发出五种事件（values / updates / checkpoints / tasks / debug；custom 和 messages 不经 _emit）。添加新 stream mode 的主要工作是：

  1. 在 StreamMode 字面量里登记新值
  2. 定义一个新 mapper
  3. 在 PregelLoop 的合适位置调一次 _emit("new_mode", new_mapper, ...)

整个事件系统是 "组合型设计"——不是每个 event type 写一份发射代码、而是统一框架 + 多种 mapper。这种模式在 LangChain Core 的 callback、React scheduler 的 markTaskXxx 都有体现——"事件类型是数据、不是代码结构"。

13.6 StreamMessagesHandler：LLM Token 捕获

13.6.1 回调机制

messages 模式的实现依赖于 LangChain 的回调系统。StreamMessagesHandler 同时继承了 BaseCallbackHandler 和 _StreamingCallbackHandler，在 LLM 调用的各个生命周期阶段捕获消息：

# libs/langgraph/langgraph/pregel/_messages.py:42
class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
    run_inline = True  # 异步回调路径上内联执行，避免顺序/加锁问题

    def __init__(self, stream, subgraphs, *, parent_ns=None):
        self.stream = stream
        self.subgraphs = subgraphs
        self.metadata: dict[UUID, Meta] = {}
        self.seen: set[int | str] = set()
        self.parent_ns = parent_ns

13.6.2 Token 流的捕获链路

sequenceDiagram
    participant LLM as ChatModel
    participant Handler as StreamMessagesHandler
    participant Stream as StreamProtocol
    participant Client as 消费者
    LLM->>Handler: on_chat_model_start(metadata)
    Note over Handler: 记录 run_id -> (ns, metadata)
    loop 每个 token
        LLM->>Handler: on_llm_new_token(chunk)
        Handler->>Handler: 检查 metadata[run_id]
        Handler->>Stream: _emit(meta, chunk.message)
        Stream->>Client: ("messages", (AIMessageChunk, metadata))
    end
    LLM->>Handler: on_llm_end(response)
    Note over Handler: 发射最终完整消息（去重）
    Handler->>Handler: 清理 metadata[run_id]

关键实现细节：

  1. 元数据追踪：on_chat_model_start 时记录 run_id 到 (ns, metadata) 的映射
  2. token 发射：on_llm_new_token 时根据 run_id 找到元数据，构造 (message, metadata) 元组发射
  3. 去重：使用 seen 集合跟踪已发射的消息 ID，避免在 on_llm_end 时重复发射
  4. 子图过滤：如果 subgraphs=False，跳过命名空间深度 > 0 的事件

13.6.3 节点输出中的消息捕获

StreamMessagesHandler 不仅捕获 LLM 的 token，还捕获节点返回值中的消息对象：

# StreamMessagesHandler.on_chain_end，libs/langgraph/langgraph/pregel/_messages.py:217
# （签名在源码里是多行带类型）
def on_chain_end(self, response, *, run_id, **kwargs):
    if meta := self.metadata.pop(run_id, None):
        if isinstance(response, Command):
            self._find_and_emit_messages(meta, response.update)
        elif isinstance(response, Sequence) and any(
            isinstance(value, Command) for value in response
        ):
            for value in response:
                if isinstance(value, Command):
                    self._find_and_emit_messages(meta, value.update)
                else:
                    self._find_and_emit_messages(meta, value)
        else:
            self._find_and_emit_messages(meta, response)

_find_and_emit_messages 方法扫描返回值（不递归，只看顶层字段和字段里的一层序列），从字典、BaseModel、dataclass 等结构中提取 BaseMessage 对象。这使得 messages 模式能够捕获所有经过节点的消息，而不仅是 LLM 直接输出的 token。

13.6.4 三个真实实现细节：ordering、ID 兜底、4 路 dispatch

上面的流程对了，但有三处libs/langgraph/langgraph/pregel/_messages.py 里的工程细节容易漏看——没它们，token 流式不会这么可靠：

1. run_inline = True 为什么必要（line 50-51，源码原注释 "We want this callback to run in the main thread to avoid order/locking issues."）：

这个属性只在 langchain-core 的异步回调路径上起作用：_ahandle_event_for_handler 遇到同步 handler 时，若 run_inline 为假，就用 run_in_executor 把回调丢进线程池，并与其他 handler 经 asyncio.gather 并发执行；run_inline = True 的 handler 则在事件循环线程里直接调用，而且 ahandle_event 会先按顺序跑完所有 inline handler，再并发其余的（libs/core/langchain_core/callbacks/manager.py:375-386、436-452）。StreamMessagesHandler 的 metadata、seen 都是不加锁的普通 dict / set，内联执行让所有回调在同一线程里依次落地，这就是源码注释说的 "order/locking issues"。同步路径（stream）上回调本来就在触发它的线程里同步执行，不受这个属性影响。

2. 消息 ID 的 UUID 兜底（libs/langgraph/langgraph/pregel/_messages.py:90-97，兜底在 94-95 行）：

def _emit(self, meta, message, *, dedupe=False):
    if dedupe and message.id in self.seen:
        return
    else:
        if message.id is None:
            message.id = str(uuid4())   # ← 没 ID 就补一个
        self.seen.add(message.id)
        self.stream((meta[0], "messages", (message, meta[1])))

seen: set[int | str] 去重依赖 message.id——但 LangChain 的 BaseMessage 里 id 字段是 Optional[str]。对 chat model 流出的 token，langchain-core 已经兜过一次：流式生成时没有 id 的 chunk 会被补上由 LC_ID_PREFIX（libs/core/langchain_core/language_models/chat_models.py:588-594，值为 "lc_run-"）和 run_id 拼成的 id，所以各 token 块与 on_llm_end 里的完整消息共用同一个 id，完整消息会被 dedupe 挡掉（实跑 GenericFakeChatModel 只流出三个 AIMessageChunk，没有第四条完整消息）。

_emit 的兜底针对的是其余没有 id 的消息，典型是节点自己构造后返回的 AIMessage(...)：第一次发射时当场生成 UUID 填进去、再记 seen。代价是轻微的 side-effect（修改了用户传入的 message 对象），但正因为改的是同一个对象，这条消息之后再出现时（例如被 add_messages 带着这个 id 存进状态，又出现在后续节点的输入里）就能被识别出来；实跑中流出的 id 与最终状态里的 id 一致。反过来，若不补 id 就直接 seen.add(message.id)，seen 里会混进 None，此后所有 id 为 None 的新消息都会被当成重复吞掉。

3. _find_and_emit_messages 的 4 路 dispatch（line 99-113）：

# libs/langgraph/langgraph/pregel/_messages.py:99（签名在源码里带类型）
def _find_and_emit_messages(self, meta, response):
    if isinstance(response, BaseMessage):                # ① 直接是消息
        self._emit(meta, response, dedupe=True)
    elif isinstance(response, Sequence):                 # ② 消息序列
        for value in response:
            if isinstance(value, BaseMessage):
                self._emit(meta, value, dedupe=True)
    else:
        for value in _state_values(response):            # ③ state 字典值
            if isinstance(value, BaseMessage):
                self._emit(meta, value, dedupe=True)
            elif isinstance(value, Sequence):            # ④ state 里嵌套序列
                for item in value:
                    if isinstance(item, BaseMessage):
                        self._emit(meta, item, dedupe=True)

这 15 行代码覆盖 LangGraph 节点返回值的四种典型形态：

  • ① 返回值本身就是一条消息——函数式 API 的 entrypoint 可以这样返回（实跑能流出）；StateGraph 节点这样写不报错，但这条消息不会写进状态（实跑 invoke 后 messages 仍为空），只会被 messages 模式流出
  • ② 返回值是消息列表——同样适用于 entrypoint 这类直接返回值的场景（实跑两条都能流出）；StateGraph 节点返回列表会抛 InvalidUpdateError: Expected dict, got [...]（实跑）
  • ③ 节点 return {"messages": [...], "other_state": ...}——经典 state dict 形态，_state_values 把 dict 展平成 values
  • ④ state dict 里某个 key 的 value 是 [msg, msg, ...]——messages 键就是这形态、4 路里专门处理嵌套序列

return {"messages": [msg]} 和函数式 API 里的 return msg 都能被流式机制感知。这种只看两层的"朴素结构发现"让用户不必为流式输出专门安排消息的位置——但放得更深（比如嵌在字段的字典里）就发现不了。

但这种"发现式"处理也有代价——对 BaseMessage 以外的自定义消息类型无效。用户如果定义了 class MyMessage(BaseModel) 当消息用、不继承 BaseMessage——isinstance(..., BaseMessage) 永远 False、流式系统看不到这种消息。这个边界需要使用者自己清楚：想走流式就必须用 LangChain 的 BaseMessage 或其子类。

13.6.5 on_chat_model_start 的 ns 提取和 TAG_NOSTREAM 过滤

libs/langgraph/langgraph/pregel/_messages.py:123-143 的 on_chat_model_start 有几段关键逻辑：

    def on_chat_model_start(
        self,
        serialized: dict[str, Any],
        messages: list[list[BaseMessage]],
        *,
        run_id: UUID,
        parent_run_id: UUID | None = None,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> Any:
        if metadata and (not tags or (TAG_NOSTREAM not in tags)):
            ns = tuple(cast(str, metadata["langgraph_checkpoint_ns"]).split(NS_SEP))[
                :-1
            ]
            if not self.subgraphs and len(ns) > 0 and ns != self.parent_ns:
                return
            if tags:
                if filtered_tags := [t for t in tags if not t.startswith("seq:step")]:
                    metadata["tags"] = filtered_tags
            self.metadata[run_id] = (ns, metadata)

三层过滤：

① TAG_NOSTREAM 检查——用户可以给 Runnable 打上 LangGraph 定义的 TAG_NOSTREAM tag（值为 "nostream"，libs/langgraph/langgraph/constants.py:23）、表示 "这个 LLM 调用我不想流式化"。典型场景：主 agent 流式、但内部的 tool-calling sub-model 不流式。handler 看到这个 tag 就跳过——metadata 不记录、后续 on_llm_new_token 也不会 emit。

② len(ns) > 0 and ns != self.parent_ns——如果用户 stream(subgraphs=False) 且事件来自子图、跳过。但 parent_ns 例外——显式从子图启动 streaming 时保留（source docstring 里的 example 场景）。这种"默认不穿透子图、但显式启动的子图透过"的语义需要这个精细判断。

③ seq:step tag 过滤——LangChain 内部给 Runnable sequence 的每一步打 seq:step:N tag、对用户毫无意义的内部信息。handler 过滤掉这些、只保留业务相关的 tag——对外暴露的 metadata 更干净。

ns 的 [:-1] 处理也值得看——langgraph_checkpoint_ns 的最后一段是当前 frame、前面的段是父图 ns。streaming event 的 ns 应该是父 ns、不包括自己——所以 [:-1] 截掉最后一段。也就是说，事件的 ns 标的是"这个事件产生于哪个（子）图"，而不是"哪个任务"。

13.6.6 on_chain_start 的预加载 seen 去重

看 _messages.py:206-215：

self.metadata[run_id] = (ns, metadata)
for value in _state_values(inputs):
    if isinstance(value, BaseMessage):
        if value.id is not None:
            self.seen.add(value.id)
    elif isinstance(value, Sequence) and not isinstance(value, str):
        for item in value:
            if isinstance(item, BaseMessage):
                if item.id is not None:
                    self.seen.add(item.id)

on_chain_start 时把 input state 里所有已有消息的 ID 都塞进 seen——为什么？

因为用户节点可能返回 "原始 state 加了新消息"——比如 return {"messages": state["messages"] + [new_msg]}。如果不先标记老消息、_find_and_emit_messages 会重新 emit 所有老消息 + 新消息——客户端看到重复。

预加载 seen 让老消息在 on_chain_end 的遍历里被 dedup 跳过——只有真正新的消息被发出。这就是 LangGraph 能识别 "增量消息"的底层机制——依赖 BaseMessage.id 作为识别标识 + seen 集合记忆。

这里的 isinstance(value, Sequence) and not isinstance(value, str) 检查排除 str——因为 Python 里 str 是 Sequence[str]、不加排除会试图对字符串的每个 char 做 isinstance(BaseMessage)——无意义但不出错、就是浪费 CPU。显式排除 str 是防御性编程。

13.6.7 tap_output_iter 和 tap_output_aiter 的空实现

libs/langgraph/langgraph/pregel/_messages.py:115-121 有两个空方法：

# libs/langgraph/langgraph/pregel/_messages.py:115-121
def tap_output_aiter(
    self, run_id: UUID, output: AsyncIterator[T]
) -> AsyncIterator[T]:
    return output

def tap_output_iter(self, run_id: UUID, output: Iterator[T]) -> Iterator[T]:
    return output

它们来自 _StreamingCallbackHandler——handler 实现者可以用来"拦截并改写 stream 输出"（在迭代器外面套一层）。StreamMessagesHandler 只观察、不改写，所以原样返回 output。

为什么要实现但什么都不做？——_StreamingCallbackHandler 在 langchain-core 里是一个 @typing.runtime_checkable 的 Protocol（libs/core/langchain_core/tracers/_streaming.py），不是 ABC：它不会阻止实例化，但 langchain-core 内部会用 isinstance(handler, _StreamingCallbackHandler) 做判断。最关键的一处在 BaseChatModel._should_stream（libs/core/langchain_core/language_models/chat_models.py:527-529）：模型没被显式关闭流式、调用方也没传 stream 参数时，只要回调里有这类 handler，连 invoke() 也会改走流式生成——这正是节点里写的是 llm.invoke(...)，messages 模式照样能逐 token 流出的原因（实跑 GenericFakeChatModel 即如此）。另一处是 Runnable 的 transform 路径（libs/core/langchain_core/runnables/base.py:2311-2318），找到这类 handler 就调用它的 tap_output_iter 包一层迭代器。协议的 runtime_checkable 只检查方法是否存在，所以必须把这两个方法写出来，isinstance 才为真；写成恒等返回，就等于"我登记了，但不改动任何东西"。

顺带一提，langgraph/pregel/_messages.py:22-25 对这个 import 做了 try/except ImportError 兜底，失败时把 _StreamingCallbackHandler 退化成 object——它是 langchain-core 的私有符号，这样即使所装的 langchain-core 里没有它，模块也能正常导入。

13.6.8 on_chain_error 的 metadata 清理——防内存泄漏

_messages.py 里的 on_chain_error（line 242+）：

def on_chain_error(self, error, *, run_id, parent_run_id=None, **kwargs):
    self.metadata.pop(run_id, None)

看似简单的一行、实际是防止残留状态的关键。

self.metadata: dict[UUID, Meta] 在 on_chat_model_start / on_chain_start 时塞入、在对应的 on_end / on_error 里 pop 出。如果只有 on_end 清理、on_error 不清理——节点或 LLM 报错时 metadata 不清、run_id 会一直留在 dict 里。好在 handler 是每次 stream() 调用时新建的（libs/langgraph/langgraph/pregel/main.py:2617-2625），残留最多活到本次运行结束，但在一次长时间运行的图里仍会越积越多。

对称地定义 on_error + pop 保证无论正常结束还是异常结束、metadata 都被清理。这种 "try/finally 式的清理" 是有状态 callback handler 的常见写法——在 on_start 里登记的，都要在 on_end 和 on_error 里成对清掉。

同一个文件里的 on_llm_error（line 175-183）也是同样的一行 self.metadata.pop(run_id, None)——成对出现的清理是这类回调 handler 的基本工程习惯，资源泄漏往往就漏在"只写了成功路径"上。

13.7 各 StreamMode 的输出映射函数

13.7.1 mapper 函数的命名规律

这些 mapper 函数分住两个文件，命名模式一致：

# libs/langgraph/langgraph/pregel/_io.py:100
map_output_values(...)      # line 100 → values mode
map_output_updates(...)     # line 118 → updates mode

# libs/langgraph/langgraph/pregel/debug.py
map_debug_tasks(...)        # line  37 → tasks（兼 debug）
map_debug_task_results(...) # line  83 → tasks（兼 debug）
map_debug_checkpoint(...)   # line 121 → checkpoints（兼 debug）

命名模式：

  • map_ 前缀——表明这是转换函数（functional programming vocabulary）
  • output_ / debug_ 中缀——事件来源分类
  • 后面描述具体事件类型

为什么名字里还留着 debug_ 前缀？——因为 debug mode 就是 checkpoints + tasks 的聚合：同一个 mapper 产出的 chunk 会同时发给 checkpoints（或 tasks）订阅者和 debug 订阅者。

但发射端只调用一次——分发由 _emit 内部的 debug_remap 完成（见 §13.5.7）：

self._emit("checkpoints", map_debug_checkpoint, ...)
# _emit 内部：若订阅了 "debug"，同一批 chunk 再包一层 {step, timestamp, type, payload} 发一遍

一个 mapper 服务多个 mode——这是 debug mode 得以复用而不重复实现的技术基础。

13.7.2 map_output_values

# libs/langgraph/langgraph/pregel/_io.py:100
def map_output_values(
    output_channels: str | Sequence[str],
    pending_writes: Literal[True] | Sequence[tuple[str, Any]],
    channels: Mapping[str, BaseChannel],
) -> Iterator[dict[str, Any] | Any]:
    if isinstance(output_channels, str):
        if pending_writes is True or any(
            chan == output_channels for chan, _ in pending_writes
        ):
            yield read_channel(channels, output_channels)
    else:
        if pending_writes is True or {
            c for c, _ in pending_writes if c in output_channels
        }:
            yield read_channels(channels, output_channels)

values 模式读取所有输出 Channel 的当前值，构成完整的状态快照。只有当相关 Channel 在本步被写入时才发射。

13.7.3 map_output_updates

# libs/langgraph/langgraph/pregel/_io.py:118
def map_output_updates(
    output_channels: str | Sequence[str],
    tasks: list[tuple[PregelExecutableTask, Sequence[tuple[str, Any]]]],
    cached: bool = False,
) -> Iterator[dict[str, Any | dict[str, Any]]]:
    output_tasks = [
        (t, ww) for t, ww in tasks
        if (not t.config or TAG_HIDDEN not in t.config.get("tags", EMPTY_SEQ))
        and ww[0][0] != ERROR
        and ww[0][0] != INTERRUPT
    ]
    ...

updates 模式将每个任务的写入按节点名分组。隐藏节点（标记了 TAG_HIDDEN）和错误/中断写入被过滤掉。输出格式为 {node_name: update_value}。

13.7.4 Debug 输出

debug 模式组合了 checkpoints 和 tasks 的输出，包裹在带时间戳和步数的外层结构中：

# libs/langgraph/langgraph/types.py:220
class _DebugTaskPayload(TypedDict):
    step: int
    timestamp: str        # ISO 8601 格式
    type: Literal["task"]
    payload: TaskPayload

class _DebugTaskResultPayload(TypedDict):
    step: int
    timestamp: str
    type: Literal["task_result"]
    payload: TaskResultPayload

class _DebugCheckpointPayload(TypedDict, Generic[StateT]):
    step: int
    timestamp: str
    type: Literal["checkpoint"]
    payload: CheckpointPayload[StateT]
graph TB
    subgraph debug 模式输出
        D[DebugPayload] --> DC[type: checkpoint]
        D --> DT[type: task]
        D --> DTR[type: task_result]
        DC --> CP[CheckpointPayload<br/>config, metadata, values, next, tasks]
        DT --> TP[TaskPayload<br/>id, name, input, triggers]
        DTR --> TRP[TaskResultPayload<br/>id, name, error, interrupts, result]
    end

13.7.5 _state_values 的三类型 dispatch

_state_values（line 31-39）是 handler 里的一个小 utility、但设计精致：

# libs/langgraph/langgraph/pregel/_messages.py:31
def _state_values(obj: Any) -> Sequence[Any]:
    """Extract top-level field values from a state object (dict, BaseModel, or dataclass)."""
    if isinstance(obj, dict):
        return list(obj.values())
    elif isinstance(obj, BaseModel):
        return [getattr(obj, k) for k in type(obj).model_fields]
    elif is_dataclass(obj) and not isinstance(obj, type):
        return [getattr(obj, f.name) for f in fields(obj)]
    return ()

三种 state schema 的统一处理：

  • dict——直接 list(values())、Python 原生容器
  • BaseModel（pydantic）——用 type(obj).model_fields 得字段名、然后 getattr 取值
  • dataclass——用 fields(obj) 得 Field 对象、取 .name 再 getattr

is_dataclass(obj) and not isinstance(obj, type)——dataclass 类本身也 is_dataclass() == True、但我们要的是 instance、不是 class。加 not isinstance(obj, type) 排除 class 对象。

为什么不用 fields(obj) 直接取 value？——因为 fields(obj) 返回的是 Field 对象、不是当前值。要通过 getattr(obj, f.name) 才能取当前值——和 BaseModel 路径同构。

三种 state 形态的支持是 LangGraph 对用户 state schema 自由度的体现——用户可以用 TypedDict（同构 dict）、BaseModel、dataclass 任一种——streaming 都能无感知地提取字段值。

兜底返回 ()——对不认识的 obj 返回空序列、不 panic。这让 handler 能安全处理各种用户数据——未知类型就跳过、不报错。

13.7.6 TAG_HIDDEN 过滤 vs TAG_NOSTREAM——两种过滤的语义差异

注意 map_output_updates 里有一个过滤条件：

output_tasks = [
    (t, ww) for t, ww in tasks
    if (not t.config or TAG_HIDDEN not in t.config.get("tags", EMPTY_SEQ))
    and ww[0][0] != ERROR
    and ww[0][0] != INTERRUPT
]

TAG_HIDDEN 和 §13.6.5 讲的 TAG_NOSTREAM 是两种不同的 tag：

  • TAG_HIDDEN ——节点要对流式输出隐藏。典型是 框架内部节点：StateGraph 为 __start__ 建的输入节点就带 TAG_HIDDEN（libs/langgraph/langgraph/graph/state.py:1300-1302）——用户不关心这些。updates / tasks（及 debug 里的任务事件）会过滤掉 TAG_HIDDEN 节点，messages 模式的 on_chain_start 也不登记它（libs/langgraph/langgraph/pregel/_messages.py:199），所以它返回的消息不会被流出；但它对通道的写入照样反映在 values 里。
  • TAG_NOSTREAM ——仅对 messages 模式里的 LLM token 生效（检查的是 chat model 这次运行的 tags，例如 llm.with_config(tags=[TAG_NOSTREAM])）。这是 "这个 LLM 我不想流式化" 的语义——values / updates 等仍然会看到这个节点的输出；而且只要节点把 LLM 的回复放进返回值，on_chain_end 仍会把这条完整消息作为一个 messages 事件发出（实跑：token 不再逐个流出，但节点结束时收到一条完整的 AIMessage）。

两种 tag 的设计反映了 "可见性" 的多维度——同一个节点可能需要：

  • 不出现在 state 更新日志（TAG_HIDDEN）
  • 但 token 还是要流（TAG_NOSTREAM 不打）
  • 或反过来：显示进度但不流 token（TAG_HIDDEN 不打、TAG_NOSTREAM 打）

四种组合覆盖不同的 observability 需求。这就是 LangGraph 对流式细粒度控制的体现——用户可以按需调每个维度。

13.8 v2 Stream API 的使用

13.8.1 基本用法

# v2 API：version="v2" 参数
# 注意 stream_mode 不传时取 self.stream_mode：裸 Pregel 是 "values"（main.py:596），
# 而 StateGraph.compile() 造出来的 CompiledStateGraph 被设成 "updates"（graph/state.py:1150）。
# 想同时收到下面几类事件必须显式列出来
async for part in graph.astream(
    input, stream_mode=["values", "updates", "messages", "custom"], version="v2"
):
    match part["type"]:
        case "values":
            state = part["data"]
            interrupts = part["interrupts"]
        case "updates":
            for node, update in part["data"].items():
                print(f"{node}: {update}")
        case "messages":
            message, metadata = part["data"]
            print(message.content, end="", flush=True)
        case "custom":
            handle_custom(part["data"])

13.8.2 version="v2" 返回的 async 迭代器语义

v2 API 的 async for part in graph.astream(..., version="v2") 里、part 是严格顺序的——LangGraph 保证：

① 同一条 stream 上的事件按发生顺序到达——不会 "后发生的 update 先到"。这是 StreamProtocol 的单线程消费者模式保证的——事件写入 queue 后消费者按 FIFO 读。

② 不同 mode 的事件交错——messages 和 values 可以同时订阅、交错到达——但同一种 mode 内部保证顺序。

③ 消费节奏影响推进节奏，但缓冲区本身是无界的——astream 是一个异步生成器：它在 yield 处挂起，只有消费者取走下一个 part、生成器恢复执行，图才会继续往前推。所以下面这种写法确实会把消费速度反压到图的推进上：

async for part in graph.astream(input, version="v2"):
    await save_to_database(part)  # 慢操作
    # ↑ 生成器在此挂起，图的下一轮 tick 也随之等待

但要看清它的边界。LangGraph 内部的缓冲区是 libs/langgraph/langgraph/_internal/_queue.py 里的 AsyncQueue（异步路径）和 SyncQueue（同步路径），两者的 docstring 都写明是 unbounded FIFO queue——AsyncQueue() 不传 maxsize，等于无界。也就是说：并不存在"队列写满就阻塞生产者"这一层机制。真正的节流来自生成器的 yield 点，而在两个 yield 点之间（比如一次 runner tick 里若干个节点并发产出 token），事件仍会先无限制地堆进队列。对于单步就能产生海量 custom 事件的节点，这一点值得留意。

还有一个相关细节：订阅了 messages / custom、开了 subgraphs=True 或图设了 stream_eager 时，stream() 会创建一个等待队列有新数据的 waiter（libs/langgraph/langgraph/pregel/main.py:2705-2726；异步版在 3080 起），PregelRunner.tick 借它在节点还没跑完时就把已产出的事件交还给消费者——这是 token 能边生成边送达的前提。代价是 runner 的单任务快速路径条件（单任务、无 timeout、无 waiter，libs/langgraph/langgraph/pregel/_runner.py:164）不再满足，即使一步只有一个节点也会提交到线程池执行（第 7 章 §7.13 有实跑）。这时消费者在 yield 处挂起，线程池里正在跑的节点并不会停，它产出的事件照样进队列；等待的是下一轮 tick。

13.8.3 v1 vs v2 对比

graph LR
    subgraph "v1 API"
        V1S["stream(mode='values')"] --> V1O["Iterator[dict]"]
        V1M["stream(mode=['values','messages'])"] --> V1T["Iterator[tuple[str, Any]]"]
    end
    subgraph "v2 API"
        V2["stream(version='v2')"] --> V2O["Iterator[StreamPart]"]
        V2O --> V2D["part.type 判别"]
    end

v2 的主要优势：

  1. 类型安全：每个 StreamPart 都有明确的 type 字段，可用于类型窄化
  2. 统一接口：无论订阅多少种模式，输出格式一致
  3. 中断信息内联：ValuesStreamPart 直接包含 interrupts 字段
  4. 泛型支持：StreamPart[StateT, OutputT] 携带完整的类型信息

13.8.4 GraphOutput：invoke 的 v2 返回值

v2 API 还改进了 invoke 的返回类型：

@dataclass(frozen=True)
class GraphOutput(Generic[OutputT]):
    value: OutputT
    interrupts: tuple[Interrupt, ...] = ()

不再返回裸字典，而是一个带类型的容器：

result = graph.invoke(input, version="v2")
print(result.value)       # 输出值
print(result.interrupts)  # 中断信息

13.9 TaskPayload 与 TaskResultPayload

13.9.1 任务生命周期事件

tasks 模式提供了细粒度的任务执行监控：

class TaskPayload(TypedDict):
    """任务开始事件"""
    id: str              # 唯一任务 ID
    name: str            # 节点名
    input: Any           # 任务输入
    triggers: list[str]  # 触发原因

class TaskResultPayload(TypedDict):
    """任务结束事件"""
    id: str              # 唯一任务 ID
    name: str            # 节点名
    error: str | None    # 错误信息
    interrupts: list[dict]  # 中断列表
    result: dict[str, Any]  # 输出结果
stateDiagram-v2
    [*] --> TaskStart: tasks 事件 (TaskPayload)
    TaskStart --> TaskEnd: tasks 事件 (TaskResultPayload)
    TaskEnd --> [*]
    state TaskStart {
        id: 任务ID
        name: 节点名
        input: 输入数据
        triggers: 触发器列表
    }
    state TaskEnd {
        id: 任务ID
        name: 节点名
        result: 输出数据
        error: 错误信息
        interrupts: 中断列表
    }

13.9.2 triggers: list[str] 字段的溯源价值

TaskPayload 的 triggers 字段是一个常被忽略的字段：

class TaskPayload(TypedDict):
    id: str
    name: str
    input: Any
    triggers: list[str]  # ← 触发原因列表

docstring 说它是"导致这个任务执行的触发器"，但实现要看清：PULL 任务的 triggers 取的是 tuple(sorted(proc.triggers))（libs/langgraph/langgraph/pregel/_algo.py:591）——该节点订阅的全部触发通道，而不是本步实际被写入的那几个；PUSH 任务（Send 产生的）则固定是 ("__pregel_push",)。运行时它是元组而非列表。实跑一个 START → a, b、[a, b] → c、c 再 Send 到 d 的图，tasks 模式里看到的是：

  • a：('branch:to:a',)
  • b：('branch:to:b',)
  • c：('branch:to:c', 'join:a+b:c')——branch:to:c 这一路本次并没有被写，仍然列在里面
  • d：('__pregel_push',)

溯源场景：debug 时一个任务突然被触发、不知道为什么——triggers 能告诉你它可能因哪些通道被触发（branch:to:X 对应普通边 / 条件边 / Command(goto=...)，join:... 对应多源汇合边，__pregel_push 对应 Send），缩小排查范围；但要确定本步究竟是哪条通道被写，还得结合上游任务的 result 或检查点里的 channel_versions。

Pregel 的超步模型里、任务是 reactive 的（channel 写入触发下游节点）——但这种 reactive 关系对用户是半黑盒。triggers 把节点的订阅关系摆了出来，是打开这个黑盒的一条线索，而不是完整的因果链。

13.9.3 实际应用

for part in graph.stream(input, stream_mode="tasks", version="v2"):
    payload = part["data"]
    if "input" in payload:
        # 任务开始
        print(f"Starting task {payload['name']} (id={payload['id']})")
    else:
        # 任务结束
        if payload.get("error"):
            print(f"Task {payload['name']} failed: {payload['error']}")
        else:
            print(f"Task {payload['name']} completed: {payload['result']}")

13.9.4 TaskPayload vs TaskResultPayload 的 input vs result 分开

注意 Task 的开始和结束用两个独立的 TypedDict——TaskPayload 和 TaskResultPayload——不是一个 Payload 带 status 字段。

class TaskPayload(TypedDict):
    id: str
    name: str
    input: Any
    triggers: list[str]

class TaskResultPayload(TypedDict):
    id: str
    name: str
    error: str | None
    interrupts: list[dict]
    result: dict[str, Any]

为什么不合成 {id, name, status: "start"/"end", data: ...} 一个 dict？——因为：

① 字段异构——start 有 input + triggers、end 有 error + interrupts + result——合成一个 dict 会让两种状态都要检查所有字段、且很多字段 Optional。

② TypedDict 类型精度——两个独立的 TypedDict 让 type checker 能根据 start vs end 精确推导可用字段。合成一个则退化成 "all Optional"、类型检查无用。

③ 消费者清晰——用户代码 if "input" in payload 就能判别开始/结束——字段缺失天然表达语义。不需要额外 status 字段。

共用字段 id + name——因为 start 和 end 要配对——用户可以存 pending_tasks[id] = info_from_start、遇到 end 时 pending_tasks.pop(id)——形成任务生命周期的完整记录。

这就是"用不同类型表达不同状态、避免 status enum + Optional"的 Python 模式——和 Rust 的 enum variants 异曲同工。

13.10 设计决策

13.10.1 为什么有七种模式而不是一种？

七种模式的枚举就是 StreamMode 这个 Literal（langgraph-1.1.6/libs/langgraph/langgraph/types.py:118，成员依次是 values / updates / checkpoints / tasks / debug / messages / custom）。设计多种流式模式的核心原因是消费者的多样性：

  • 前端 UI 只关心 LLM 文本流（messages）
  • 管理面板需要完整状态（values）
  • 日志系统需要增量更新（updates）
  • 监控系统需要任务级事件（tasks）
  • 开发者调试需要全部信息（debug）

一种模式无法同时满足所有需求，而如果把所有信息都塞进一种模式，消费者需要过滤大量不相关的数据。

13.10.2 回调 vs 返回值

messages 模式使用回调（StreamMessagesHandler，langgraph-1.1.6/libs/langgraph/langgraph/pregel/_messages.py:42）而非返回值来捕获 LLM token。这是因为 LLM 的流式输出发生在节点执行过程中，而不是执行完成后。回调机制允许在不修改节点代码的前提下拦截中间数据。

13.10.3 StreamWriter 作为 Callable 而非类

StreamWriter 被定义为 Callable[[Any], None]（langgraph-1.1.6/libs/langgraph/langgraph/types.py:134）而非一个类，这是刻意的简约设计。节点代码只需要调用 writer(data) 即可，不需要了解任何框架内部细节。这种设计也使得测试更容易——你可以用任意 lambda 替代：

# 测试时
collected = []
my_node(state, writer=collected.append)
assert collected == [{"progress": 0.5}, {"progress": 1.0}]

13.10.4 run_inline = True 的必要性

StreamMessagesHandler 设置了 run_inline = True（langgraph-1.1.6/libs/langgraph/langgraph/pregel/_messages.py:50），这意味着在异步回调路径上，它在事件循环线程里同步执行，而不是被派到线程池（§13.6.4）。这个选择是为了避免两个问题：

  1. 顺序保证：token 必须按生成顺序到达消费者
  2. 锁竞争：handler 内部的 metadata / seen 不加锁，被多个线程并发改写会出竞争

代价是 LLM 生成的速度会略微受到回调处理时间的影响，但对于 token 级别的事件，这个开销可以忽略。

13.10.5 为什么不给 StreamMode 加 errors 或 warnings？

有一个合理的疑问——为什么没有专门的 errors 或 warnings stream mode？

原因是错误和警告在现有 mode 里已经被表达：

  • 节点错误——在 TaskResultPayload.error 里，通过 tasks mode 流出来（langgraph-1.1.6/libs/langgraph/langgraph/types.py:160）
  • 任务中断——在 TaskResultPayload.interrupts 里，同样通过 tasks mode（langgraph-1.1.6/libs/langgraph/langgraph/types.py:162）
  • 框架级异常——通过 Python 原生的 exception 机制、不走 stream

单独加 errors mode 会造成信息碎片化——用户要同时订阅 tasks + errors 才能完整监控——违反"最小订阅"原则。

从现有设计看，mode 大体按语义分工、尽量避免冗余——唯一的有意重叠是 debug，它就是 checkpoints + tasks 套了一层壳（§13.5.7）。

类似地、没有 state_changes mode——因为 updates 已经覆盖（每个节点的 update 是 state change）、没有 logs mode——因为 debug 覆盖了。

用户要更细粒度的错误监控？——订阅 tasks 模式、自己在消费端分类。LangGraph 把复杂度推给消费者的灵活性——框架不猜用户要什么、提供原始信号让用户自组织。

13.10.6 TypeAliasType vs Union——PEP 695 的应用

StreamPart 的定义（langgraph-1.1.6/libs/langgraph/langgraph/types.py:329）用了 TypeAliasType：

StreamPart = TypeAliasType(
    "StreamPart",
    ValuesStreamPart[OutputT]
    | UpdatesStreamPart
    | MessagesStreamPart
    | CustomStreamPart
    | CheckpointStreamPart[StateT]
    | TasksStreamPart
    | DebugStreamPart[StateT],
    type_params=(StateT, OutputT),
)

TypeAliasType 来自 Python 3.12 (PEP 695) 或 typing_extensions 的 backport——让 type alias 成为一等公民、而不是普通变量赋值。

老写法：

StreamPart = Union[ValuesStreamPart, UpdatesStreamPart, ...]
# 或
StreamPart: TypeAlias = Union[...]

这种 "变量赋值式 type alias" 其实也能带 TypeVar 成为泛型别名，但有些局限：

  • 类型参数的顺序由 TypeVar 在表达式里首次出现的顺序隐式决定——对 StreamPart 来说，OutputT 先于 StateT 出现，写成 StreamPart[S, O] 就会对错位
  • 不是独立的 "type" 在 type system 里——只是一个引用

TypeAliasType：

  • 一等公民——运行时是独立的别名对象，有自己的名字
  • 显式声明 type_params——type_params=(StateT, OutputT) 把参数顺序钉死，与 union 里的出现顺序无关。这个顺序在 1.1.2 改过一次：#7134（"fix: stream part generic order"，commit 682814e9）把 (OutputT, StateT) 换成了 (StateT, OutputT)，提交说明直言这"technically breaking"

langgraph 这里是从 typing_extensions 导入的 TypeAliasType（libs/langgraph/langgraph/types.py:22），所以不要求 Python 3.12。

13.10.7 version="v2" 作为 stream 方法参数

v1/v2 的切换不是不同的方法——而是 stream(version=...) 参数（langgraph-1.1.6/libs/langgraph/langgraph/pregel/main.py:2499，astream 的对应重载在同文件 :2837），默认值是 "v1"。调用方的两种写法（用法示例，不是 main.py 的源码）：

# v1（默认）
for chunk in graph.stream(input, stream_mode=["values"]):
    ...

# v2
for part in graph.stream(input, version="v2"):
    ...

为什么同一个方法、不同返回格式？——因为stream 的核心逻辑（任务调度、checkpoint、channel 更新）完全一样——只是输出包装方式不同。用 version 参数切换、避免 stream_v1 / stream_v2 两个函数的代码重复。

内部实现：分叉只发生在最末端的 _output 函数（libs/langgraph/langgraph/pregel/main.py:3572），stream() / astream() 把 version 一路传给它：

# libs/langgraph/langgraph/pregel/main.py:3607（节选）
if mode in stream_mode:
    if version == "v2":
        if mode == "values":
            # pop __interrupt__ into typed field, coerce data
            ints: tuple[Interrupt, ...] = ()
            if isinstance(payload, dict):
                ints = payload.pop(INTERRUPT, ())
                if output_mapper:
                    payload = output_mapper(payload)
            yield {"type": mode, "ns": ns, "data": payload, "interrupts": ints}
        elif mode in ("checkpoints", "debug"):
            ...  # 对 payload 里的 values 做类型转换后同样产出三字段字典
        else:
            yield {"type": mode, "ns": ns, "data": payload}
    elif stream_subgraphs and isinstance(stream_mode, list):
        yield (ns, mode, payload)
    elif isinstance(stream_mode, list):
        yield (mode, payload)
    elif stream_subgraphs:
        yield (ns, payload)
    else:
        yield payload

任务调度、checkpoint、channel 更新全部共用、仅在输出层分叉——这是 API 版本演进的健康方式。用户代码里只改一个参数就能切换——迁移成本最小。

默认 version="v1"——签名里写死的就是 version: Literal["v1", "v2"] = "v1"（langgraph-1.1.6/libs/langgraph/langgraph/pregel/main.py:2499），保持向后兼容。默认值今后是否切到 v2，1.1.6 的源码里看不出来。

13.11 小结

本章系统分析了 LangGraph 的流式输出架构。七种 StreamMode 覆盖了从完整状态快照到 LLM token 的各个粒度层级。v2 Stream API 通过 StreamPart 类型体系实现了类型安全的流式消费，每个事件都是自描述的。StreamWriter 以极简的函数签名为节点提供了自定义流式输出的能力，而 StreamMessagesHandler 通过 LangChain 回调系统在不侵入业务代码的前提下捕获 LLM 的 token 流。

整个流式系统的设计理念可以归结为：分层解耦，按需订阅。生产者（节点、LLM、框架内部）不关心消费者订阅了什么；消费者通过 stream_mode 声明自己感兴趣的事件类型；StreamProtocol 作为中间层负责路由和过滤。这种设计使得新增流式模式的改动集中在少数几处：登记 StreamMode、定义新的 StreamPart 类型和对应的映射函数，再在 PregelLoop 里加一次 _emit。

下一章我们将探讨 LangGraph 的 Runtime 与 Context 机制，了解如何将运行时依赖（如用户身份、数据库连接）安全地注入到图的执行过程中。

13.12 跨章节呼应

与《React 19 内核探秘》专栏第 4 章调度器的呼应——LangGraph 的 streaming 走回调（StreamMessagesHandler）+ 事件归并（_find_and_emit_messages 的四路 dispatch），React 调度器走任务队列 + 时间切片——都是在"异步消费者面对异构事件流"的场景下用"统一入口 + 类型分支"的模式。

与《Hyper 与 Tower：工业级 HTTP 栈》专栏第 10 章 http-body 的呼应——LangGraph 的 StreamPart 七种类型、http-body 的 Frame<T>（Data / Trailers）——都是"用带类型标签的联合体统一异构流式事件"，只是粒度不同。

与《Serde 元编程》专栏第 16 章 flatten 的呼应——LangGraph 的 _state_values 把 dict / BaseModel / dataclass 统一抽取字段值，Serde 的 flatten 把嵌套 struct 展平到父级——都是"用同一套遍历逻辑处理多种数据形态"。

与《Vite 设计与实现》专栏第 15 章 SSR 与模块运行器的呼应——模块运行器用并发请求去重避免重复加载，StreamMessagesHandler 用 seen 集合避免重复 emit——都是"用集合去重高频事件"，前者去重并发 fetch、后者去重已发出的消息。

与《vLLM 推理内核深度解析》专栏第 8 章 ModelRunner 的呼应——LangGraph 内部用紧凑的 StreamChunk 三元组、对外用灵活的 StreamPart TypedDict；vLLM 内部用紧凑的批量数组、对外用 Python 对象包装——都是"内部求紧凑、外部求友好"的两层设计。

13.13 十二条工程原则收束

把本章源码观察压缩成流式系统设计的 12 条原则：

① 类型标签 + 联合类型表达异构流（§13.3）——每个事件 type 字段、消费者精确分派。

② namespace 字段追踪层级（§13.3.2）——ns: tuple[str, ...] 让子图透明可见。

③ run_inline=True 保顺序（§13.6.4）——异步路径上 handler 在事件循环线程内联执行，不被派进线程池、不与其他 handler 并发。

④ UUID 兜底补消息 ID（§13.6.4 point 2）——节点返回的消息缺 ID 时 handler 补上、让 dedup 机制工作。

⑤ 4 路 dispatch 覆盖所有节点返回形态（§13.6.4 point 3）——BaseMessage / Sequence / dict / nested Sequence 都能找到消息。

⑥ 预加载 seen 支持增量消息（§13.6.6）——on_chain_start 时把 state 里老消息的 ID 先塞进 seen、只发真正新的。

⑦ __slots__ 优化高频对象（§13.5.2）——StreamProtocol 每层图都要创建，去掉 __dict__ 既省内存、属性访问也走槽位而非字典查找。

⑧ DuplexStream 闭包合并 fan-out（§13.5.4）——一个入口多个出口、每个出口独立 mode 过滤。

⑨ 三元组 wire format 最轻（§13.5.6）——StreamChunk 用 tuple 而非 dict/class、省 memory 省 CPU。

⑩ TAG_NOSTREAM 允许跳过流式（§13.6.5）——用户打 tag 表明"这个 LLM 不要流式"、handler 遵守。

⑪ 两个独立 TypedDict 代替 status enum（§13.9.4）——TaskPayload + TaskResultPayload 让字段类型精确。

⑫ TypeAliasType 一等公民 type alias（§13.10.6）——PEP 695 语法让 StreamPart 的泛型联合类型被 type checker 正确识别。

13.14 流式系统与可观测性——从 streaming 到 tracing

LangGraph 的 streaming 不只是 "给前端看个打字机效果"——它是可观测性的基石。

把 StreamMode 映射到 observability 术语：

  • messages → token-level tracing（每个 token 的 span）
  • tasks → function-level tracing（每个节点的 span）
  • checkpoints → state snapshot（任意时刻的完整状态）
  • updates → event log（状态变化事件流）
  • debug → distributed trace（跨层的完整链路）

一个完整的 AI 应用观测栈可以建立在 LangGraph 的 streaming 上：

# 以下 *_client / audit_log 均为示意用的占位对象
async for part in graph.astream(input, version="v2", stream_mode=["messages", "tasks", "checkpoints"]):
    match part["type"]:
        case "messages":
            # 发给 LangSmith / Braintrust / Arize 做 LLM tracing
            langsmith_client.log_token(part["data"])
        case "tasks":
            # 发给 Datadog / New Relic 做 APM
            dd_client.log_span(part["data"])
        case "checkpoints":
            # 发给自己的审计日志
            audit_log.record(part["data"])

streaming 因此不只是 UX 功能，也可以作为可观测性的数据来源——7 种模式、带 namespace、回调 handler，足以支撑这类用法。作为对照，《LangChain 设计与实现》专栏第 12 章讲的回调系统提供的是"事件通知"，而 LangGraph 在此之上又叠了一层带类型标签、带命名空间的流式协议——粒度更细，消费端的类型信息也更完整。

13.15 流式在生产环境的真实案例

案例 1：ChatGPT-style 前端——用 messages 模式、通过 SSE (Server-Sent Events) 推 token 给浏览器：

@app.post("/chat")
async def chat(req: ChatRequest):
    async def generate():
        async for part in graph.astream(req.input, stream_mode="messages", version="v2"):
            msg, meta = part["data"]
            yield f"data: {json.dumps({'content': msg.content})}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

用户在前端看到打字机效果——首字延迟取决于第一个 token 到达的时间，而不是整个响应生成完毕的时间；回答越长，两者差得越多。

案例 2：Multi-step Agent 的进度条——用 tasks 模式：

async for part in graph.astream(input, stream_mode=["tasks"], version="v2"):
    payload = part["data"]
    if "input" in payload:
        ui.show(f"Starting: {payload['name']}")
    else:
        ui.show(f"Done: {payload['name']}")

用户看到 "Starting: search_web → Done: search_web → Starting: analyze → ..."——透明的 agent 执行过程、而不是黑盒等待。

案例 3：分布式审计日志——用 checkpoints 模式、把每个 checkpoint 推到 Kafka：

async for part in graph.astream(input, stream_mode=["checkpoints"], version="v2"):
    cp = part["data"]
    await kafka.produce("langgraph-checkpoints", json.dumps(cp))

下游系统可以重放任意历史 checkpoint 复现问题——生产事故的 forensic 分析变得可能。

案例 4：开发调试——用 debug 模式、在 localhost 展示完整执行：

from pprint import pprint

async for part in graph.astream(input, stream_mode=["debug"], version="v2"):
    payload = part["data"]
    print(f"[{payload['timestamp']}] step={payload['step']} type={payload['type']}")
    pprint(payload["payload"])

开发者看到每一步的完整状态——比 print-debugging 的信息密度高得多。

四个案例——四种 stream_mode 组合——覆盖从用户可见到开发者可见的完整 UX/DX 谱系。LangGraph 的 7 种 mode 看似多、但每一种都在真实场景里有不可替代的角色。

13.16 从 streaming 看 LangGraph 的"渐进复杂度"哲学

LangGraph 的这套接口可以用 "渐进复杂度"（progressive disclosure of complexity）来概括：

  • Level 1——.invoke()、返回最终结果——和 LangChain LLMChain 一样简单
  • Level 2——.stream() 用默认模式（compile() 出来的图是 updates）——拿到每步谁改了什么、开始有 observability
  • Level 3——指定多 stream_mode——开始细分不同消费者
  • Level 4——version="v2" 加 StreamPart 类型——开始写类型安全的消费代码
  • Level 5——自定义 LangChain 回调 handler、按 ns 分流子图事件——嵌入到自己的 observability 栈

每一层的代码量和心智负担都是阶梯上升的——用户按需进入。一个 hackathon 原型用 .invoke() 就够、一个生产 agent 需要 v2 + checkpoints + messages、一个平台级 agent infrastructure 需要完整的自定义 handler 栈。

streaming 是 progressive disclosure 的第一跳——从 batch 思维到 streaming 思维的过渡。一旦跨过这一跳、后面所有 observability / customization 的高级特性都顺理成章。

LangGraph 在这一层提供了 7 种模式、v1/v2 双 API 和精细到 token 的 handler，让用户在不同阶段都能找到合适的抽象层。

13.16.1 一个具体的渐进场景——从 hackathon 到生产

用一个具体的 agent 项目演进来说明 LangGraph 各层抽象的用处：

Week 1（hackathon MVP）：

result = graph.invoke({"input": "search weather"})
print(result)

两行代码、一个 agent——够用、但没有任何 UX/observability。

Week 2（加前端）：

for message, metadata in graph.stream({"input": query}, stream_mode="messages"):
    yield message.content  # SSE 推到浏览器

（注意 messages 模式在 v1 下 yield 的是 (message, metadata) 二元组；若把 stream_mode 写成列表，外面还会再包一层 (mode, data)。）

打字机效果有了——用户等待的不再是整段回答，而是第一个 token。

Week 3（加监控）：

async for part in graph.astream(input, stream_mode=["messages", "tasks"], version="v2"):
    if part["type"] == "messages":
        yield_to_frontend(part["data"])
    elif part["type"] == "tasks":
        log_to_prometheus(part["data"])

业务可观测性——运营知道 agent 哪个节点慢、哪个 tool 常失败。

Week 6（接审计）：

async for part in graph.astream(input, stream_mode=["messages", "tasks", "checkpoints"], version="v2"):
    if part["type"] == "messages":
        yield_to_frontend(part["data"])
    elif part["type"] == "tasks":
        log_to_prometheus(part["data"])
    elif part["type"] == "checkpoints":
        await kafka_produce("agent-audit", part["data"])

每个 checkpoint 到审计系统——合规场景 ready、事后可以完整重放。

Week 12（专业化 observability）：

async for part in graph.astream(input, stream_mode=["messages"], version="v2"):
    # 交给自己的多个下游处理器：加密、多 backend 分发
    for handler in self.handlers:
        await handler.process(part)

企业级 observability 栈——几个不同系统各自从 LangGraph stream 取所需。

从 Week 1 的 MVP 到 Week 12 的企业生产——每一次演进只加几行代码、不推翻前面的工作。

13.17 本章的方法论贡献——如何读流式系统的源码

读完本章你应该有一种新的技能——如何评估任何 "streaming / event system" 库的质量。下面是一个 checklist：

① 事件类型有无明确标签？——直接 any data 还是带 type 字段？前者消费者要靠 context 推断、后者 type-safe。

② 事件层级有没有 namespace？——嵌套场景能不能追踪来源？没有的话、子图/子模块的事件会混成一团。

③ 顺序是否保证？——同一 channel 的事件是否按发生顺序到达？有没有 run_inline=True 或等价机制？

④ 有没有消息去重？——消息可能因重新发送、cleanup、retry 等重复触发——有没有 ID-based dedup？

⑤ 有没有背压？——producer 比 consumer 快时、buffer 会不会无限增长？

⑥ 有没有过滤 tag？——能不能对某些节点/事件说"我不要流式化这个"？

⑦ 生命周期清理配对完整吗？——on_start / on_end / on_error 都清理资源？

⑧ mapper 和 emitter 解耦吗？——事件生成逻辑和发送逻辑有没有分开？方便加新事件类型。

⑨ 类型系统有多精确？——StreamPart 用 TypedDict + type tag 可以精确类型窄化、vs 使用 dict[str, Any] 完全丢类型。

⑩ 版本策略——API 演进是否平滑？——能不能支持 v1 和 v2 并存？用户能渐进迁移吗？

LangGraph 的 stream 系统在这 10 条上大部分都能对上——唯一需要打问号的是第 ⑤ 条：如 §13.8.2 所说，它的缓冲队列是无界的，反压只来自异步生成器的 yield 点，而不是队列本身。下次你评估任何其他 streaming 库（Kafka Streams、RxJS、各家 AI 框架的事件系统）——用这 10 条对比，短板和强项都会很快显形。