第 20 讲 · 流式输出体系与 ReAct Agent 拆解

学习目标

掌握各 StreamMode（values/updates/messages/custom/debug 等）的产出与实现。
了解 v1/v2 与 v3 两代流式架构的区别。
用 create_react_agent / ToolNode 把全课机制串成一个完整 Agent。

一、流式输出：一次 run 的多种"视角"

第 9 讲说过 stream 支持多种模式。它们由 PregelLoop 在执行过程中 emit （_loop.py 的 _emit 1357–1391 行 / output_writes 1393–1443 行）。

StreamMode	何时发出	内容	典型用途
`values`	每步结束/中断	全量 state 快照	展示当前完整状态
`updates`	节点写完通道	`{node: 本步更新}`	增量进度、节点级日志
`messages`	LLM callback	token chunk / 完整 AIMessage	打字机效果
`custom`	节点调 `stream_writer`	任意 payload	自定义进度条、中间产物
`tasks`	任务开始/结束	task payload	调试任务调度
`debug`	重映射 checkpoints+tasks	带 step/时间戳的详细事件	深度调试（第 2 讲推荐）
`checkpoints`	checkpoint 持久化时	同 get_state 格式	观察存档

可同时要多个：app.stream(inputs, stream_mode=["updates", "messages"])。

custom：节点主动推送进度

from langgraph.config import get_stream_writer

def long_node(state):
    writer = get_stream_writer()
    for i in range(3):
        writer({"progress": i})   # 实时推给 stream_mode="custom"
    return {...}

这是给前端做"自定义进度"的标准做法，不依赖框架内置事件。

二、两代流式架构

v1/v2：PregelLoop 直接 emit

graph.stream(..., stream_mode=...) 直接由 loop 产出 chunk（dict 或 StreamPart）。 messages 模式靠 _messages.py 的 StreamMessagesHandler 挂在 LLM callback 上抓 token。这是目前最常用的接口。

v3：Pull 驱动 + Transformer 管道（实验性）

graph.stream_events(..., version="v3")，源码在 libs/langgraph/langgraph/stream/。它把 v2 的 StreamPart 转成协议事件，经一组 Transformer 投影成多个命名流（run.values/ run.messages/run.updates/...）：

flowchart LR
    P[Pregel.stream v2] --> SP[StreamPart]
    SP --> CV[convert_to_protocol_event]
    CV --> MUX[StreamMux 事件总线]
    MUX --> T1[ValuesTransformer → run.values]
    MUX --> T2[MessagesTransformer → run.messages]
    MUX --> T3[UpdatesTransformer → run.updates]
    MUX --> T4[...]

StreamMux（_mux.py）是事件总线 + transformer 链。
每个 Transformer（transformers.py）声明它需要哪些底层 stream_mode，并投影出一个命名流。
优势：调用方可以"按需订阅 + 多流合并（interleave）"，更适合复杂 UI。
装配入口 _pregel_stream_v3（main.py 3520–3573 行），内部仍调 self.stream(..., version="v2")。

选择建议：常规场景用 v1/v2 stream；需要多投影/精细事件控制再考虑 v3（注意实验性）。

三、ReAct Agent 拆解：把全课串起来

libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py，create_react_agent。它用本课所有机制搭了一个开箱即用的工具循环 Agent。

注：create_react_agent 已标记 deprecated，推荐迁移到 langchain.agents.create_agent，但其源码仍是理解"如何用 LangGraph 搭 Agent"的最佳范例。

图拓扑

flowchart TD
    START --> AG[agent: 调 LLM]
    AG -->|should_continue| C{有 tool_calls?}
    C -->|无| END
    C -->|有| T[tools: ToolNode]
    T --> AG

AgentState（57–63 行）：messages: Annotated[list, add_messages]（第 7 讲 reducer）。
call_model（661–721 行）：agent 节点，调 LLM，写回 messages。
should_continue（831–859 行）：条件边——无 tool_calls → END；有 → 路由 tools（第 7/8 讲）。
tools → agent 的回边形成循环（第 8 讲：DAG 做不到，Pregel 可以）。

ToolNode：并行执行工具

prebuilt/tool_node.py，ToolNode（622 行起）。一次可并行执行多个 tool_calls （sync executor.map / async asyncio.gather，793–824 行）。工具还能返回 Command 做"改 state / 跳转"（第 18 讲）。

v1 vs v2：工具并行的两种粒度

v1：should_continue 返回 "tools" → 单个 ToolNode 在一个超步内并行跑所有 tool_calls。
v2：返回 [Send("tools", ToolCallWithContext(...))]（第 19 讲）→ 每个 tool_call 成为独立 PUSH 任务，由 Pregel 并行调度。好处：可按 tool_call 粒度做 checkpoint / interrupt（第 18 讲危险操作确认）。

这就把第 19 讲的 Send 和第 18 讲的中断，落到了一个真实 Agent 上。

一个 Agent 用到的全课机制

机制	本课	在 ReAct 中的体现
add_messages reducer	第 7 讲	`messages` 通道合并
条件边/循环	第 7/8 讲	should_continue + tools→agent 回边
超步并行	第 11/12 讲	ToolNode 并行多工具
Send fan-out	第 19 讲	v2 按 tool 粒度扇出
checkpoint	第 15-17 讲	多轮对话记忆
interrupt	第 18 讲	工具调用前人审
流式	本讲	messages 模式输出 token

使用场景

快速搭 Agent：create_react_agent(model, tools, checkpointer=...) 一行起步。
打字机效果：stream_mode="messages" 输出 LLM token。
危险操作确认：v2 + interrupt，在执行某工具前暂停人审。
自定义进度：get_stream_writer() + stream_mode="custom"。
复杂前端：多流订阅用 v3 stream_events。

动手实验

用 create_react_agent + 一个简单工具搭 Agent，stream_mode="messages" 看 token 流。
同一 Agent 加 checkpointer + thread_id，多轮对话验证记忆（第 15 讲）。
用 version="v2" + interrupt_before=["tools"]，在工具执行前暂停，Command(resume=...) 放行。
用 stream_mode="debug" 跑一次，对照第 9-12 讲观察超步/任务事件。

阅读作业

精读 prebuilt/chat_agent_executor.py 的 create_react_agent / should_continue / call_model。
精读 prebuilt/tool_node.py 的 ToolNode（622–739 行）与 tools_condition（1582–1659 行）。
浏览 _loop.py 的 _emit / output_writes 与 stream/transformers.py。

小结

StreamMode 提供一次 run 的多种视角；v1/v2 由 loop 直接 emit，v3 用 Transformer 管道投影多命名流。
create_react_agent = add_messages + 条件边循环 + ToolNode 并行 + checkpoint + interrupt + 流式，是全课机制的集大成范例。
v2 用 Send 把工具并行细化到 tool_call 粒度，支撑按工具中断。

课程总结

二十讲走完，你应该建立起这样一条主线：

StateGraph（建图）→ compile（翻译成 Pregel）→ 超步循环（Plan/Execute/Update） → 通道版本驱动触发 → 写缓冲 + apply_writes → checkpoint 持久化 → 续跑/时间旅行/人审 → Send/函数式动态并行 → 流式输出 → ReAct Agent 集大成。

再回看第 1 讲的六个核心概念（超步、通道、版本触发、写缓冲、检查点、动态图），现在每一个你都能在源码里指出它的实现位置。这就是"读懂源码机制"的目标。

接下来建议：挑一个真实需求（带工具 + 记忆 + 人审的 Agent），从零用 StateGraph 实现一遍，遇到问题回到对应讲次的源码定位——把"读过"变成"用过"。