第13章 流式输出与调试

13.1 引言

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

LangGraph 1.1.6 引入了 v2 Stream API，它将所有流式数据统一为带类型标签的 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 定义在 langgraph/types.py 中，是一个字面量联合类型：

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

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

StreamMode	输出内容	触发时机	典型消费者
values	完整状态快照	每个超步结束后	前端展示完整状态
updates	节点名 + 该节点的输出	每个节点执行后	调试、日志
custom	用户自定义数据	节点内调用 StreamWriter	进度条、中间结果
messages	LLM 消息 token	LLM 流式生成时	打字机效果
checkpoints	检查点快照	检查点创建时	持久化监控
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
for part in graph.stream(input, version="v2"):
    if part["type"] == "values":
        print(f"State: {part['data']}")
    elif part["type"] == "messages":
        msg, meta = part["data"]
        print(f"Token: {msg.content}")

13.3 StreamPart 类型体系

13.3.1 v2 API 的核心创新

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

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

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