LangGraph 框架指南 · 第 11 章

第 11 章流式输出

本章目标：系统掌握 stream 的五种模式——values / updates / messages / custom / debug，理解它们各自的粒度与适用场景，并学会用自定义流把图内部的事件推送出去。前几章已经多…

约 17 分钟6,816 字9 个代码示例

本章目标：系统掌握 stream 的五种模式——values / updates / messages / custom / debug，理解它们各自的粒度与适用场景，并学会用自定义流把图内部的事件推送出去。前几章已经多次借用 stream 做调试工具，本章把它转正。

11.1 为什么 invoke 不够了

到第 10 章为止，我们与图的交互都是 invoke：一锤子买卖，等全部跑完拿最终状态。这在两类场景下不可接受：

场景一：体验。批次评估 30 份简历要一两分钟。前端盯着"加载中"转圈一分钟，用户会以为系统挂了。期望的体验是：每评完一份就推送一条进度。

场景二：过程可见。报告生成节点内部，模型正在一个字一个字地写。invoke 模式下这些中间 token 全部被吞掉，最终一次性出现——而 LLM 应用的"打字机效果"恰恰依赖 token 级的推送。

invoke 的问题在于它是一个聚合函数：把整个执行过程折叠成一个最终值。stream 则把过程本身作为一等公民吐出来。

for chunk in graph.stream(input):
    do_something(chunk)     # 执行过程中，每个事件实时到达

用法上它与 invoke 完全同构（同一个编译产物，同一份输入），区别只是返回值从 dict 变成了一个事件迭代器。

11.2 五种 stream_mode 一览

stream 的核心参数是 stream_mode，决定"以什么粒度切事件"。五种模式各对应一个观察视角：

模式	粒度	每个事件长什么样	典型用途
`values`	节点级·全量	每个超步后的完整状态快照	观察状态演进过程
`updates`	节点级·增量	`{节点名: 补丁}`	进度推送、审计
`messages`	token 级	`(消息片段, 元数据)`	打字机效果
`custom`	业务级	你在节点里写的任意数据	业务事件推送
`debug`	一切	全部内部事件（含任务调度）	排查执行问题

选择思路：给用户看 → messages/custom；给自己看 → updates/debug；要全量快照 → values。

11.3 updates：进度推送的主力

updates 是第 5 章观察超步时用的模式（当时未点名）。每个节点完成，吐一个 {节点名: 该节点返回的补丁}：

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END

class ScreenState(TypedDict):
    resume_text: str
    score: Annotated[int, operator.add]
    log: Annotated[list[str], operator.add]

def hard_review(state: ScreenState) -> dict:
    return {"score": 35, "log": ["硬性条件 +35"]}

def soft_review(state: ScreenState) -> dict:
    return {"score": 30, "log": ["软性条件 +30"]}

builder = StateGraph(ScreenState)
builder.add_node("hard_review", hard_review)
builder.add_node("soft_review", soft_review)
builder.add_edge(START, "hard_review")
builder.add_edge("hard_review", "soft_review")
builder.add_edge("soft_review", END)
graph = builder.compile()

for chunk in graph.stream(
    {"resume_text": "李思远 ..."}, stream_mode="updates"
):
    print(chunk)

输出：

{'hard_review': {'score': 35, 'log': ['硬性条件 +35']}}
{'soft_review': {'score': 30, 'log': ['软性条件 +30']}}

注意两个细节：其一，事件里的值是补丁不是全量状态（log 只有本次追加的那条）；其二，并行节点完成时可能合并在同一个事件里（{'hard_review': ..., 'soft_review': ...}）——取决于它们是否同超步完成。

对进度条场景，updates 的消费方式朴素：前端订阅 SSE/WebSocket，收到 {'evaluate_3': {...}} 就知道第 3 份评完了。

11.4 values：状态演进的全量快照

values 每个超步吐一次合并后的完整状态——相当于回放"每个超步结束时如果你 invoke 一下会拿到什么"：

for chunk in graph.stream(
    {"resume_text": "李思远 ..."}, stream_mode="values"
):
    print(chunk)

输出：

{'resume_text': '李思远 ...'}                                    超步 0：初始输入
{'resume_text': '李思远 ...', 'score': 35, 'log': ['硬性条件 +35']}   超步 1
{'resume_text': '李思远 ...', 'score': 65, 'log': ['硬性条件 +35', '软性条件 +30']}   超步 2

（未启用持久化时初始输入是否单独成事件取决于版本行为，以实际输出为准。）values 是调试状态演进的首选：reducer 的合并轨迹、字段的逐步变化，一目了然。数据量大的状态别用它做生产推送——每个事件都是全量，带宽浪费。

11.5 messages：token 级推送

messages 模式专门为 LLM 节点服务：凡是节点内部通过 LangChain 模型产生的 token 流，都会穿透图结构直接吐出来：

import os
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from langchain_core.messages import AnyMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

load_dotenv()

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

llm = ChatOpenAI(model=os.getenv("LLM_MODEL", "gpt-4o-mini"), temperature=0)

def write_report(state: ChatState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(ChatState)
builder.add_node("write_report", write_report)
builder.add_edge(START, "write_report")
builder.add_edge("write_report", END)
graph = builder.compile()

from langchain_core.messages import HumanMessage

for msg_chunk, meta in graph.stream(
    {"messages": [HumanMessage("用三句话点评：候选人精通 Python 与高并发系统")]},
    stream_mode="messages",
):
    print(msg_chunk.content, end="", flush=True)
print()

终端上报告逐字浮现——打字机效果就这么简单。两个要点：

事件是二元组 (消息片段, 元数据)：片段带 content（本段文本）和 id（同一消息的所有片段共享 ID，靠它拼接）；元数据里藏着 langgraph_node 等字段，标注这个 token 产自哪个节点——多 Agent 场景下用它区分"现在是哪个 Agent 在说话"。
前提是节点内部用了标准 LangChain 模型接口（llm.invoke 的流式能力被引擎自动桥接）。节点里手写 requests.post 调裸 HTTP 接口，引擎无从拦截，messages 模式收不到任何东西。

11.6 custom：业务事件的直达通道

updates/messages 吐的都是"框架知道的事"（节点完成、token 产生）。但有一类事件只有你的业务代码知道：评估进度过半、某份简历命中了黑名单、成本已超预算。这类事件的推送通道叫自定义流——节点内部调用 get_stream_writer()，往里写什么，外面就收到什么：

from langgraph.config import get_stream_writer
import random

def evaluate_batch(state) -> dict:
    writer = get_stream_writer()          # 拿到当前执行上下文的写入器
    results = []
    resumes = state["resumes"]
    for i, resume in enumerate(resumes):
        score = random.randint(50, 95)    # 演示用伪评分
        results.append(score)
        # 业务事件：实时推送单份进度
        writer({"event": "progress", "index": i + 1, "total": len(resumes),
                "name": resume.split("：")[0], "score": score})
    return {"scores": results}

class BatchState(TypedDict):
    resumes: list[str]
    scores: list[int]

builder = StateGraph(BatchState)
builder.add_node("evaluate_batch", evaluate_batch)
builder.add_edge(START, "evaluate_batch")
builder.add_edge("evaluate_batch", END)
graph = builder.compile()

for chunk in graph.stream(
    {"resumes": ["李思远：...", "王敏：...", "赵子昂：..."]},
    stream_mode="custom",
):
    print(chunk)

输出：

{'event': 'progress', 'index': 1, 'total': 3, 'name': '李思远', 'score': 87}
{'event': 'progress', 'index': 2, 'total': 3, 'name': '王敏', 'score': 61}
{'event': 'progress', 'index': 3, 'total': 3, 'name': '赵子昂', 'score': 93}

三个值得注意的设计点：

custom 事件不进状态。writer 写的数据只走流通道，不污染图状态——它是"旁路广播"，与"状态补丁"是两条正交的线。想留痕就同时返回补丁，想广播就调 writer，互不干涉。
必须在节点执行期间调用。get_stream_writer() 依赖当前执行上下文（类似 contextvars），在节点函数体里调用才有效。拿到模块外存着回头再调，会得到错误或空操作。
事件结构自定义。惯例是带个 event 类型字段方便消费端分发，但引擎不约束结构。

11.7 多模式混合与 debug

需要同时观察多种粒度时，stream_mode 可以传列表，事件变成 (模式名, 载荷) 二元组：

for mode, chunk in graph.stream(
    input, stream_mode=["updates", "custom"]
):
    if mode == "updates":
        ...  # 节点进度
    elif mode == "custom":
        ...  # 业务事件

debug 模式则是"全都要"：任务调度、节点开始结束、状态变更等内部事件全部吐出。日常开发用不上，遇到"这图到底怎么跑的"级别的疑难杂症时是终极武器（事件结构与内部实现耦合较紧，当作诊断工具而非稳定接口依赖）。

另外两个实用参数顺带记录：subgraphs=True 把子图内部事件也展开（第 9 章提过的调试需求在此兑现）；异步版本 astream 的用法与同步版完全对称。

常见坑与排错

messages 模式收不到 token。按 11.5 的前提排查：节点内部是不是标准 LangChain 模型调用？是不是把 llm.invoke(...) 的结果存了变量却没走标准链路？手写 HTTP 调用的模型，流是穿透不出来的。
updates 的事件值当作全量状态用。补丁只有增量。要在事件里读某个字段的"最新完整值"，用 values 模式，或在消费端自己累积补丁。
custom 事件期望出现在最终状态里。writer 是广播不是写入，状态里永远不会出现 custom 事件的数据。需要"既广播又留痕"就两件事都做：writer 一份、返回补丁一份。
在节点外调用 get_stream_writer()。上下文丢失，写不进任何流。writer 必须在节点函数体（或其同步调用链）里获取和使用。同理，异步场景下要避免把 writer 存到跨 await 的全局变量里。
流被消费端拖慢。stream 的迭代是消费驱动的（尤其涉及 token 流的桥接时），消费端每个 chunk 做重计算（同步写库、复杂渲染逻辑）会背压到图的执行。重的消费逻辑改成入队 + 异步处理。
混合模式下忘了解包二元组。stream_mode=["updates", "custom"] 的事件是 (mode, chunk)，直接当 dict 用会到处 TypeError。单模式是裸载荷、多模式是二元组——切换时记得改解析代码。

动手练习

用 values 模式跑第 5 章的并行汇合 demo，观察每个超步后的完整状态快照，对照第 5 章的超步表格验证"补丁在超步边界合并"。
把第 8 章 ReAct 评估 demo 改成流式消费：updates 模式打印节点进度、messages 模式实时打印模型 token。混在一个循环里（stream_mode=["updates", "messages"]），体验一次"能看见思考过程"的 Agent。
给第 7 章批次评估的 evaluate 节点加自定义流：每评完一份推送 {"event": "progress", "done": i, "total": n}。用 random.randint 模拟分数即可，重点体会 custom 与 Send 的组合——N 个并行实例的 progress 事件交错到达，消费端要靠什么字段区分实例？

本章速查

API / 概念	一句话
`graph.stream(input, stream_mode=...)`	invoke 的过程化版本，返回事件迭代器
`values`	每超步后的完整状态快照，看状态演进
`updates`	每节点完成吐 `{节点名: 补丁}`，进度推送主力
`messages`	token 级二元组 `(片段, 元数据)`，打字机效果
`custom`	`get_stream_writer()` 广播业务事件，不进状态
`debug`	全部内部事件，诊断终极武器
多模式	`stream_mode` 传列表，事件变 `(模式, 载荷)` 二元组
`subgraphs=True`	展开子图内部事件
消费口诀	用户看 messages/custom，自查用 updates/debug