AI Agent 开发（13）：流式输出与事件

第 12 篇给图装上了持久化：checkpointer、thread_id、时间旅行。到这里 LangGraph 的核心积木只剩最后一块没讲：所有图跑起来都是同一个体验——invoke 一声令下，屏幕沉默几十秒，然后一次性吐出最终 State。中间发生了什么、模型想了什么、工具查了什么、走到第几步，调用方一无所知。第 3 篇讲过模型级的 .stream() 打字机，但那只是单次调用的粒度；Agent 是多节点循环，需要的是整套图执行事件流。本篇是《AI Agent 开发》系列第 13 篇、LangGraph 核心模块收官篇，代码基于 langgraph 1.2.x 与 langchain-core 1.6.x，环境沿用第 2 篇的 uv 项目。

全文按"为什么必须流式 → stream_mode 全景 → 五种模式逐一过手 → 组合消费 → 工具与子图的边界 → 前端集成"组织，最后用一个"进度条 + 打字机"终端 Demo 练习收尾。

1、感知延迟与流式刚需

Chatbot 单轮调用等 3 秒可以接受，因为预期就是一问一答。Agent 完全不同：一次任务要走"模型决策 → 工具执行 → 再决策"好几圈，每圈一次模型调用 3～5 秒，五轮循环下来 30 秒起步是常态。用 invoke 等待，用户面对的是 30 秒无任何反馈的转圈，多数人在第 10 秒就认定程序挂了。

产品上的解法叫感知延迟管理：用户对有反馈的等待容忍度远高于无反馈的等待，体验由反馈密度决定，而不是总时长。同样是 30 秒，"正在查询订单 → 已找到 3 条记录 → 回答逐字上屏"的体感是它在干活，黑盒转圈的体感是它死了——两者的工程差距，只是有没有把执行过程流出来。

Agent 产品需要流出来的信息正好三类：过程（走到哪个节点）、业务进展（正在查哪张表）、内容增量（回答的 token）。LangGraph 把它们分别对应到 updates、custom、messages 三种流模式。改造成本也低得诱人：流式不需要动图，节点和边一行不改，只是把 invoke 换成 stream。

2、stream_mode 全景

图级流式的入口是 graph.stream(input, config, stream_mode=...)，异步版是 astream。它返回一个迭代器：图一边执行一边吐事件，循环退出即运行结束；invoke 拿到的最终 State，等价于把流消费完后的末尾快照，所以服务端默认走 stream 毫无损失。五种核心模式一张表说清：

模式	产出粒度	典型用途
`values`	每个超步后的完整 State 快照	前端整份渲染状态
`updates`	每个节点本次更新的增量	步骤条、进度条
`messages`	token 级 `AIMessageChunk` 增量	打字机效果
`custom`	节点内主动发出的自定义事件	业务过程提示
`debug`	全量运行信息	本地排障

另有两个模式需先挂上 checkpointer 才能用（第 12 篇）：checkpoints 流式输出每步 checkpoint，tasks 输出任务调度明细，按需再查。输出格式上，默认的 v1 格式最常用：单模式时迭代器直接产出该模式的载荷；传模式列表时产出 (mode, chunk) 元组；再叠 subgraphs=True 还要在前面加命名空间——层层嵌套是 v1 被诟病的地方，langgraph 1.1 起可传 version="v2" 把所有事件统一成带 type 字段的字典。本篇代码用默认 v1，两个版本语义一致。1.2 新增了类型化事件流 API：一次运行返回带多个类型化投影的 run 对象，各类事件独立迭代、可多消费者订阅，官方建议新项目优先评估——它的地基仍是本篇这套模式。细节见流式文档与 Pregel.stream 参考。

3、values 与 updates 模式

先搭一张贯穿全文的最小图：make_plan 纯 Python 生成提纲，write_report 调模型扩写。注意 write_report 用的是普通 invoke，这是伏笔：

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END

model = init_chat_model("openai:gpt-5")

class State(TypedDict):
    topic: str
    plan: str
    report: str

def make_plan(state: State):
    return {"plan": f"围绕《{state['topic']}》列三点提纲"}

def write_report(state: State):
    # 注意这里是普通 invoke，不是 stream
    resp = model.invoke(f"按提纲扩写成一段话：{state['plan']}")
    return {"report": resp.content}

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

两个节点两条边，State 三个字段。values 模式的消费与输出如下：

for snap in graph.stream({"topic": "流式输出"}, stream_mode="values"):
    print(snap)

输出共三条：

{'topic': '流式输出', 'plan': '', 'report': ''}
{'topic': '流式输出', 'plan': '围绕《流式输出》列三点提纲', 'report': ''}
{'topic': '流式输出', 'plan': '围绕《流式输出》列三点提纲', 'report': '流式输出把模型的生成过程……'}

规律：values 每个超步（superstep，图执行的一轮）产出一条完整 State 快照，第一条是输入快照，未写入的字段按类型补默认值。它适合前端整份渲染状态的场景，缺点是每次全量，消息历史一长又大又重复。只想看"谁更新了什么"，换 updates：

for upd in graph.stream({"topic": "流式输出"}, stream_mode="updates"):
    print(upd)

输出如下：

{'make_plan': {'plan': '围绕《流式输出》列三点提纲'}}
{'write_report': {'report': '流式输出把模型的生成过程……'}}

每条是"节点名 → 该节点本次 return 的字段"的字典，同一超步里并行执行的节点也各占一条。updates 是步骤条与进度条的标准数据源：拿到节点名就能点亮流程图上对应的格子。两台相机一台拍全景、一台拍变化，按前端要什么选。

4、messages 模式与打字机

第 3 节的图里 write_report 用的是 invoke，但把模式换成 messages，token 照样逐块流出来：

for msg, meta in graph.stream({"topic": "流式输出"}, stream_mode="messages"):
    if msg.content:
        print(msg.content, end="", flush=True)
print()

输出就是一段逐字上屏的正文：

流式输出把模型的生成过程切成增量事件，前端逐块上屏……

这是 messages 最反直觉也最省事的一点：节点代码一行不改，token 级流式开箱即得——LangGraph 运行时会自动接管节点里 LangChain 模型的流式输出，不管你写的是 invoke 还是 stream。每个事件是 (AIMessageChunk, metadata) 二元组：chunk 与第 3 篇打字机里的是同一种对象，全部拼接等价于完整回复；metadata 里的 langgraph_node 标明 token 产自哪个节点，tags 用来过滤——多模型图想只显示主回答节点时靠它。

两个实践提醒。其一，if msg.content 的判空不能省：模型决定调用工具时，流里会混进只携带 tool_call_chunks 的块，content 为空，直接上屏全是噪音。其二，用量统计仍挂在最后一个 chunk 上，OpenAI 系模型记得开 stream_usage=True（第 3 篇的老坑）。个别不支持原生流式的本地模型，官方另有把整段结果切块模拟的 fake stream 方案兜底。

5、custom 模式与业务事件

updates 只有节点名，用户听不懂"tools 节点执行完毕"，他们想知道的是"正在查询订单 #1024"。custom 模式让你在节点和工具内部主动发业务事件，当前写法是 get_stream_writer——旧的 dispatch_custom_event 是 callback 时代的产物，新代码别再用。给 make_plan 加两行（改完重新编译）：

from langgraph.config import get_stream_writer

def make_plan(state: State):
    writer = get_stream_writer()
    writer({"phase": "规划", "detail": "拆解提纲：概念、模式、边界"})
    return {"plan": f"围绕《{state['topic']}》列三点提纲"}

for ev in graph.stream({"topic": "流式输出"}, stream_mode="custom"):
    print(ev)

输出只有这一条，因为整张图只发了一个事件：

{'phase': '规划', 'detail': '拆解提纲：概念、模式、边界'}

writer() 接受任意 Python 对象，要发给前端时建议就用可 JSON 序列化的 dict。三个要点：自定义事件发在哪，业务语义就在哪——节点、@tool 工具函数、子图里都能发，第 7 节的工具过程感就靠它；其二，它不挑模型，非 LangChain 的 SDK 拿到的流式块也能从这里转发出去；其三，Python 3.11 以下的异步节点拿不到隐式 writer，要给节点签名显式加 writer: StreamWriter 参数注入。

6、模式组合与 debug 模式

真实产品往往进度条和打字机都要。stream_mode 传列表即可，v1 格式下迭代器产出 (mode, chunk) 元组，按 mode 分发：

for mode, chunk in graph.stream(
    {"topic": "流式输出"},
    stream_mode=["updates", "messages"],
):
    if mode == "updates":
        print(f"\n[步骤] {next(iter(chunk))} 完成")
    else:
        msg, _ = chunk
        if msg.content:
            print(msg.content, end="", flush=True)

输出如下：

[步骤] make_plan 完成
流式输出把模型的生成过程切成增量事件，前端逐块上屏……
[步骤] write_report 完成

细读顺序能发现一个关键事实：事件的到达顺序就是执行顺序。make_plan 的 updates 先到；write_report 开始执行，它的 token 逐块到达；它 return 之后，它的 updates 才到。updates 描述"节点完成时"，messages 描述"节点运行中"，前端一个画步骤条收尾、一个推打字机，时序天然对齐。这个分发循环换个外壳（SSE 帧）就是第 8 节的前端原型。

最后是 debug 模式，任务创建、节点完成、checkpoint、状态快照全量混流吐出：

for ev in graph.stream({"topic": "流式输出"}, stream_mode="debug"):
    print(ev)  # 数据量五种模式里最大，纯排障用

相当于把 LangSmith 的 trace 倒进终端，本地排障时一行代码换全量现场，产品环境别开。

7、工具调用的流式边界

工具环节容易产生误解，把边界画清。能流出来的是模型侧的一切：正文 token、思考、以及"它决定调用工具"这件事——messages 模式里工具调用的函数名与参数同样以 tool_call_chunks 增量到达，消费端按 id 聚合就是完整 tool_calls。工具执行本身没有流：那是你自己的 Python 代码，跑完才有返回值，中间不存在 token。想要"正在查询订单"的过程感，正确姿势是第 5 节的 writer，在工具函数内部发 custom 事件。

两个容易踩的坑。其一，子图：把 create_agent 这类 prebuilt 包成节点放进父图后，内层 token 默认不冒泡，消费端要 graph.stream(..., subgraphs=True) 才收得到；v1 格式下事件变成带命名空间前缀的元组，和多模式再叠加，解包层级相当劝退——这正是 1.1 推 v2、1.2 推事件流要治的病。其二，隔离：节点里若有不该给用户看的内部调用（比如历史摘要），给那个模型实例打 tags=["nostream"]，它就从 messages 流里退出了。第 9 篇对照表里提过 create_agent 会给消息打"最终回答 / 中间步骤"的语义标签，机制就落在这层 tags 上。

8、前端集成要点

通道选型一句话：token 和事件都是服务端单向推送，SSE 生而为此——纯 HTTP、穿代理友好、浏览器 EventSource 自带重连（Nginx 反代记得关 proxy_buffering）；只有客户端要在同一条连接里反向发消息（中途取消、interrupt 审批回传）才值得上 WebSocket。

后端姿势：FastAPI 里用 astream 消费图，把 (mode, chunk) 逐帧写成 SSE，帧里带类型与序号。两条纪律。第一，一次运行一条流：所有事件来自同一个迭代器、在单个循环里顺序写出，顺序天然有保证；最忌讳为每种模式各开一条请求再在前端拼时序——并发请求的到达顺序没有保证，打字机会被撕碎。第二，断线恢复靠 checkpoint，不靠重放：流是一次性的，断了不补发；但第 12 篇的 checkpointer 已把每个完成的超步按 thread_id 落库，重连时先 get_state 对齐进度、再订阅后续事件，前端不丢账。

流是一次性的，checkpoint 才是事实——所有重连设计都围绕"对齐状态 + 订阅增量"，别试图回放 token。

前端拿到帧后按类型分发：updates 点亮步骤条、messages 追加打字机、custom 弹业务提示，第 6 节的组合循环换个传输层就是产品原型。远程场景（LangGraph Platform 与自托管）的流协议细节，官方的 streaming-cookbook 值得整本翻一遍。

9、总结

收官盘点：图级流式的全部开关就是 graph.stream 的 stream_mode 参数——values 拍全景、updates 拍变化（进度条）、messages 出 token（打字机）、custom 发业务事件、debug 倒现场；多模式传列表，(mode, chunk) 按类型分发，到达顺序即执行顺序；工具执行本身不可流，过程感靠 writer 从工具内部发事件；前端单向推送用 SSE，断线恢复靠 checkpoint 而非重放。加上第 12 篇的持久化，LangGraph 核心模块到此闭环。

练习做一个终端 Demo，把第 9 篇的手写 ReAct 变成看得见的 Agent，测试题就用"北京和上海哪座城市人口密度更高"：

给第 9 篇的图接上 stream_mode 三模式组合消费
updates 每步打印 [步骤] model 或 [步骤] tools
messages 打字机只显示 model 节点的 token
人口工具里用 get_stream_writer 发"正在查询北京人口"
验收：终端持续滚动，没有超过 3 秒的沉默

进阶一步：用 astream 把分发循环包成 FastAPI 的 SSE 接口，curl -N 验证事件顺序；断开连接后凭 thread_id get_state，确认已完成的轮次都在。对于准备把 Agent 做成产品的读者，建议从第一天就默认 stream 而不是 invoke——流式不只是体验优化，它同时是调试通道、监控数据源和未来的前端协议。下一篇进入 RAG 模块，从 Embedding 与向量库讲起，Agent 的长期知识从这里开始。