摘要

在基于大模型构建的 Agent、工作流应用中，流式输出是提升用户体验、降低延迟感知的核心技术。
本文基于 LangGraph 最新官方文档，详细讲解 LangGraph Streaming v2 统一格式、7 种流模式、
状态流式、LLM 逐 Token 输出、自定义进度流、子图流式、异步兼容方案，
附带大量可直接运行的代码示例，适合 LLM 应用开发、Agent 工程化落地同学快速上手。

---

一、前言

在现在的 LLM 应用（聊天机器人、智能 Agent、自动化工作流）中，流式输出（Streaming） 已经不是加分项，而是标配能力。

它能做到：

• 不用等模型完整输出，边生成边展示
• 实时看到节点执行、状态变化、中间过程
• 大幅提升用户体验，降低延迟感知

LangGraph 作为构建复杂 LLM 应用的主流框架，提供了一套强大、统一、可扩展的流式系统。
本文基于官方最新文档，带你从零到一掌握：

• v2 统一流式格式
• 7 种 Stream Mode 完整用法
• 状态流 / LLM 打字机流 / 自定义流
• 子图流式、异步兼容、低版本 Python 避坑
• 生产级最佳实践

---

二、为什么一定要用 LangGraph Streaming

1. 降低用户等待焦虑：LLM 响应慢？流式输出直接解决体感问题
2. 执行过程透明化：Agent 每一步在干嘛，一目了然
3. 支持复杂交互：进度条、状态提示、中间结果推送
4. v2 格式彻底统一：单模式、多模式、子图结构完全一致
5. 类型安全：支持 IDE 自动提示，减少 Bug

---

三、核心：v2 统一流式格式

从 LangGraph ≥ 1.1 开始，强烈推荐使用 version="v2"。

所有流式返回结构统一为：

{
    "type": "values" | "updates" | "messages" | "custom" | ...,
    "ns": (),       # 子图命名空间
    "data": ...     # 真实数据
}

v1 与 v2 对比

• v1：单模式、多模式、子图返回结构不一样，容易乱
• v2：结构完全统一，只用 type 区分，代码更干净

---

四、7 种 Stream Mode

LangGraph 提供7 种流式模式，覆盖几乎所有工程场景：

模式	用途
values	每一步执行后的完整状态
updates	仅返回节点状态增量更新
messages	LLM 逐 Token 打字机输出
custom	自定义流：进度、状态、事件推送
checkpoints	检查点持久化事件（需记忆）
tasks	任务开始/结束事件
debug	最全调试信息

下面逐个实战。

---

五、最常用流式模式实战

1. updates：状态增量流

只返回节点修改的内容，轻量、高效。

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="updates",
    version="v2"
):
    if chunk["type"] == "updates":
        for node, state in chunk["data"].items():
            print(f"节点 {node} 更新: {state}")

2. values：完整状态流（调试用）

每一步都返回全量状态。

for chunk in graph.stream(
    ...,
    stream_mode="values",
    version="v2"
):
    print("完整状态:", chunk["data"])

3. messages：LLM 逐字输出（打字机效果）

聊天界面必备。

for chunk in graph.stream(
    ...,
    stream_mode="messages",
    version="v2"
):
    msg, meta = chunk["data"]
    print(msg.content, end="", flush=True)

还能精准过滤：

• 按 metadata["tags"] 过滤不同模型
• 按 metadata["langgraph_node"] 过滤指定节点

---

六、自定义流 custom：最强扩展能力

你可以在任意节点、任意工具里主动推送：

• 进度条
• 中间状态
• 自定义日志
• 第三方 API 流式结果

使用步骤

1. 获取流写入器
2. 主动发送数据
3. 开启 stream_mode="custom"

示例：

from langgraph.config import get_stream_writer

def generate_joke(state):
    writer = get_stream_writer()
    writer({"status": "正在思考笑话..."})
    return {"joke": "为什么冰淇淋去上学？为了得到圣代学历！"}

接收：

for chunk in graph.stream(
    ...,
    stream_mode="custom",
    version="v2"
):
    print(chunk["data"]["status"])

适用场景：

• 非 LangChain 模型接入
• 前端进度条
• 多步骤任务提示

---

七、子图流式输出

如果你用了子图（Subgraph），只需开启：

subgraphs=True

示例：

graph.stream(
    ...,
    subgraphs=True,
    stream_mode="updates",
    version="v2"
)

通过 chunk["ns"] 判断来源：

• ()：主图
• ("node_2:xxx",)：子图

---

八、同时开启多种流模式

v2 格式支持多模式同时流式：

stream_mode=["updates", "custom"]

遍历判断 type 即可：

for chunk in ...:
    if chunk["type"] == "updates":
        # 处理状态更新
    elif chunk["type"] == "custom":
        # 处理自定义事件

---

九、异步 & Python ❤️.11 避坑（重要）

Python < 3.11 异步不支持上下文自动传递，两个关键点：

1. 异步调用 LLM 必须手动传 config

async def call_model(state, config):
    await model.ainvoke(..., config)

2. 不能用 get_stream_writer()，改为参数注入

from langgraph.types import StreamWriter

async def generate_joke(state, writer: StreamWriter):
    writer({"msg": "异步自定义流"})

---

场景推荐方案

• 聊天机器人打字机 → messages
• Agent 执行步骤展示 → updates
• 进度条、中间提示 → custom
• 复杂嵌套图 → subgraphs=True
• 调试、回溯、断点 → checkpoints / debug
• 第三方模型流式 → custom 封装

---