第 07 章

流式处理

让图的每一步执行，都变成用户能感知到的实时反馈

六种流模式，覆盖从状态更新到逐 Token 输出的全部场景

等太久了

用户向智能体提问后，盯着一个空白的加载圈看了三分钟——什么反馈都没有

没有流式处理

用户发送请求后，界面一直转圈
智能体内部跑了 5 个节点，用户完全不知道
LLM 正在逐 token 生成回复，用户看到的还是一片空白
工具调用了 30 秒，没有进度条

有流式处理

每个节点执行完，立即推送状态更新
LLM 的每个 token 实时出现在屏幕上
工具执行过程中，持续报告搜索进度
用户随时知道"它在干什么、进行到哪了"

流式处理解决的核心问题：让图的执行过程从"黑盒等待"变成"透明进度"。LangGraph 提供了六种流模式，分别对应不同的实时更新需求。

开始之前

前置要求

已完成第 04 章 Quickstart，能构建和运行基本图
了解 TypeScript 的 async/await 和 for await...of 异步迭代
了解 LLM 的 token 概念

读完后你应该能

区分六种流模式并选择正确的模式
用 updates 和 values 流式输出图状态
用 messages 模式实现逐 token 输出
用 custom 模式从节点内部推送自定义数据
用 tools 模式追踪工具执行进度
组合多种流模式同时接收不同类型的事件

暂不涉及

前端 UI 框架中集成流式输出
LangGraph Platform 的服务端流式部署
WebSocket / SSE 协议细节
多用户并发流式管理

六种流模式

LangGraph 的 `stream()` 方法支持六种流模式，每种关注不同维度的执行信息

模式	说明	典型用途
values	每步之后的完整状态快照	调试整体状态变化
updates	每步之后的状态增量更新	追踪单个节点的输出
messages	LLM 调用的逐 token 输出	实时显示 AI 回复
custom	节点内通过 `config.writer()` 发送的自定义数据	推送进度条、日志等
tools	工具执行的生命周期事件	追踪工具调用开始/结束/错误
debug	执行过程中的全部可用信息	完整调试

// 基本用法——所有流模式的调用方式相同
for await (const chunk of await graph.stream(inputs, {
  streamMode: "updates",  // 替换为任意模式名
})) {
  console.log(chunk);
}

选择原则：需要看状态变化用 values/updates，需要逐字输出用 messages，需要自定义进度用 custom，需要追踪工具用 tools，需要全部信息用 debug。

状态流

用 `updates` 和 `values` 模式追踪图的每一步状态变化

先定义一个简单的两步图：refineTopic 润色主题，generateJoke 生成笑话。

思考题：你的图有多个步骤，你希望每完成一步就能看到进度，而不是等全部跑完才出结果。graph.stream() 有两种模式："values" 每次给完整状态，"updates" 只给增量。该用哪个？区别在哪？

// 定义图
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  topic: z.string(),
  joke: z.string(),
});

const graph = new StateGraph(State)
  .addNode("refineTopic", (state) => {
    return { topic: state.topic + " and cats" };
  })
  .addNode("generateJoke", (state) => {
    return { joke: `This is a joke about ${state.topic}` };
  })
  .addEdge(START, "refineTopic")
  .addEdge("refineTopic", "generateJoke")
  .addEdge("generateJoke", END)
  .compile();

updates 增量更新

for await (const chunk of await graph.stream(
  { topic: "ice cream" },
  { streamMode: "updates" }
)) {
  for (const [name, state] of Object.entries(chunk)) {
    console.log(`Node ${name}:`, state);
  }
}
// Node refineTopic: { topic: "ice cream and cats" }
// Node generateJoke: { joke: "This is a joke about ice cream and cats" }

values 完整快照

for await (const chunk of await graph.stream(
  { topic: "ice cream" },
  { streamMode: "values" }
)) {
  console.log(chunk);
}
// { topic: "ice cream" }                      ← 初始
// { topic: "ice cream and cats" }             ← refineTopic 后
// { topic: "ice cream and cats", joke: "..." } ← 最终

区别：updates 只输出当前节点的改动，体积小，适合增量渲染；values 输出完整状态快照，适合需要全貌的场景。

📄 完整代码：07_stream_values_updates.ts

逐 Token 流

用 `messages` 模式实现 LLM 逐字输出——和 ChatGPT 一样的打字效果

当图中有 LLM 调用时，messages 模式会流式返回每一个 token。每个 chunk 是一个二元组：[messageChunk, metadata]。

思考题：你希望 AI 的回复像 ChatGPT 一样逐字打出，而不是等全部生成完才一次性显示。怎么获取 LLM 的 token 级别流？

import { ChatOpenAI } from "@langchain/openai";

const model = new ChatOpenAI({ model: "gpt-4.1-mini" });

const callModel: GraphNode<typeof MyState> = async (state) => {
  const response = await model.invoke([
    { role: "user", content: `Generate a joke about ${state.topic}` },
  ]);
  return { joke: response.content };
};

// messages 模式——逐 token 接收
for await (const [messageChunk, metadata] of await graph.stream(
  { topic: "ice cream" },
  { streamMode: "messages" }
)) {
  if (messageChunk.content) {
    process.stdout.write(messageChunk.content + "|");
  }
}

输出效果：
Why| did| the| ice| cream| go| to| school|?|
Because| it| wanted| to| be| a| smart| cookie!|

关键理解：messages 模式的每个 chunk 是 [messageChunk, metadata] 元组。messageChunk.content 包含当前 token 文本；metadata 包含来源节点、标签等元信息。

📄 完整代码：07_stream_messages.ts

过滤与排除

用标签 (tags) 筛选特定 LLM 调用，用 `nostream` 标签排除不需要流式输出的部分

当一个图中有多个 LLM 调用时，你可能只想流式输出其中一个。LangGraph 提供两种控制手段。

思考题：你的图有两个 LLM 节点：主对话 LLM（用户要看到逐字输出）和内部分类器 LLM（用户不需要看到）。怎么在 messages 流中只展示主 LLM 的输出？

按标签过滤——只流式特定 LLM

// 给不同 LLM 实例打标签
const jokeModel = new ChatOpenAI({ model: "gpt-4.1-mini", tags: ['joke'] });
const poemModel = new ChatOpenAI({ model: "gpt-4.1-mini", tags: ['poem'] });

// 只接收带 "joke" 标签的 LLM token
for await (const [msg, metadata] of await graph.stream(
  { topic: "cats" },
  { streamMode: "messages" }
)) {
  if (metadata.tags?.includes("joke")) {
    process.stdout.write(msg.content);
  }
}

排除特定 LLM 的流式输出

// 内部辅助 LLM 不需要流式输出
const internalModel = new ChatAnthropic({
  model: "claude-haiku-4-5-20251001",
}).withConfig({ tags: ["nostream"] });

// "nostream" 标签的 LLM 不会出现在 messages 流中

标签过滤 `tags`

在客户端用 metadata.tags 条件判断，只处理关心的 LLM 调用。适合有多个 LLM 节点、只想展示部分输出的场景。

排除 `nostream`

给 LLM 加上 nostream 标签，它就不会出现在 messages 流中。适合内部辅助调用（如分类、路由）不需要展示给用户的场景。

注意：nostream 是在 LLM 实例上设置的，不是在 stream() 调用时设置的。一旦标记，所有使用该实例的节点都不会产生流式 token。

📄 完整代码：07_tag_filtering.ts

自定义数据流

用 `config.writer()` 从节点内部推送任意数据，用 `streamMode: "custom"` 接收

状态更新和 token 流覆盖了大多数场景，但有时你需要推送更灵活的数据——比如中间计算结果、进度百分比、日志信息。custom 模式就是为此设计的。

思考题：你的分析任务有多个阶段，前端需要展示进度条。但你不想让这些"进度信息"污染 State。能不能在不修改状态的情况下，推送自定义数据给前端？

import { StateGraph, StateSchema, GraphNode, START, LangGraphRunnableConfig } from "@langchain/langgraph";

const analyzeNode: GraphNode<typeof State> = async (state, config) => {
  // 用 writer 发送自定义数据——不改变状态，只推送信息
  config.writer({ stage: "loading_data", progress: 0.2 });
  const data = await loadData();

  config.writer({ stage: "analyzing", progress: 0.6 });
  const result = await analyze(data);

  config.writer({ stage: "done", progress: 1.0 });
  return { answer: result };
};

// 设置 streamMode: "custom" 接收自定义数据
for await (const chunk of await graph.stream(inputs, {
  streamMode: "custom"
})) {
  console.log(chunk);
}
// { stage: "loading_data", progress: 0.2 }
// { stage: "analyzing", progress: 0.6 }
// { stage: "done", progress: 1.0 }

📄 完整代码：07_stream_custom.ts

核心区别：return 更新图的状态（影响后续节点），config.writer() 推送自定义数据（不影响状态，仅用于通知客户端）。两者互不干扰。

工具进度流

用 `tools` 模式追踪工具执行的完整生命周期

当图中的智能体调用工具时，你可能需要实时知道"哪个工具开始了、执行到哪了、有没有出错"。tools 模式提供了完整的生命周期事件。

事件	触发时机	包含数据
`on_tool_start`	工具调用开始	name, input, toolCallId
`on_tool_event`	工具产生中间数据	name, data, toolCallId
`on_tool_end`	工具返回最终结果	name, output, toolCallId
`on_tool_error`	工具抛出错误	name, error, toolCallId

流式进度的工具——用 async generator 定义

普通工具只在结束时返回结果。如果你需要中间进度，可以用 async generator（async function*）来定义工具，每 yield 一次就产生一个 on_tool_event。

思考题：你的搜索工具执行时间较长，你想在工具执行过程中向前端报告进度（"正在搜索..."、"处理结果..."）。怎么让工具在执行中推送中间状态？

普通工具

一次性返回完整结果。用户在等待期间看不到任何进度。

const search = tool(
  async ({ query }) => {
    // 用户只能干等 30 秒
    const results = await api.search(query);
    return JSON.stringify(results);
  },
  { name: "search", ... }
);

流式工具

每完成一部分就 yield 一次，用户能看到实时进度。

const search = tool(
  async function* ({ query }) {
    for (const batch of batches) {
      const r = await api.search(batch);
      yield { progress, partial: r };
    }
    return finalResults;
  },
  { name: "search", ... }
);

async generator 工具的工作方式：每次 yield 会触发 on_tool_event，最终的 return 会触发 on_tool_end。如果中途抛出异常，则触发 on_tool_error。

📄 完整代码：07_stream_tools.ts

完整代码示例

一个逐航司搜索航班的流式工具，以及消费其事件的客户端代码

思考题：你的搜索工具执行时间较长，你想在工具执行过程中向前端报告进度（"正在搜索..."、"处理结果..."）。怎么让工具在执行中推送中间状态？

// 1. 定义流式工具——用 async generator 逐航司搜索
import { tool } from "@langchain/core/tools";
import { z } from "zod";

const searchFlights = tool(
  async function* (input) {
    const airlines = ["United", "Delta", "American", "JetBlue"];
    const completed: string[] = [];

    for (let i = 0; i < airlines.length; i++) {
      await new Promise((r) => setTimeout(r, 500));  // 模拟网络延迟
      completed.push(airlines[i]);
      yield {                        // 每次 yield → on_tool_event
        message: `Searching ${airlines[i]}...`,
        progress: (i + 1) / airlines.length,
        completed,
      };
    }

    return JSON.stringify({            // return → on_tool_end
      flights: [
        { airline: "United", price: 450, duration: "5h 30m" },
        { airline: "Delta", price: 520, duration: "5h 15m" },
      ],
    });
  },
  {
    name: "search_flights",
    description: "Search for available flights to a destination.",
    schema: z.object({
      destination: z.string(),
      date: z.string(),
    }),
  }
);

// 2. 消费工具事件——注意 streamMode 是数组形式
for await (const [mode, chunk] of await graph.stream(
  { messages: [{ role: "user", content: "Find flights to Tokyo" }] },
  { streamMode: ["updates", "tools"] }    // 同时接收状态更新和工具事件
)) {
  if (mode === "tools") {
    switch (chunk.event) {
      case "on_tool_start":
        console.log(`Tool started: ${chunk.name}`); break;
      case "on_tool_event":
        console.log(`  Progress: ${Math.round(chunk.data.progress * 100)}% — ${chunk.data.message}`); break;
      case "on_tool_end":
        console.log(`Tool finished: ${chunk.name}`, chunk.output); break;
      case "on_tool_error":
        console.error(`Tool failed: ${chunk.name}`, chunk.error); break;
    }
  }
}

📄 完整代码：07_stream_tools.ts

子图与调试

流式输出子图内部状态，以及用 debug 模式获取完整执行信息

子图输出流

默认情况下，stream() 只输出父图的节点更新。如果你需要看到子图内部每个节点的变化，加上 subgraphs: true。

// 开启子图流式输出
for await (const chunk of await graph.stream(
  { foo: "foo" },
  { subgraphs: true, streamMode: "updates" }
)) {
  console.log(chunk);
  // 会额外输出子图内部节点的更新
}

Debug 模式

debug 模式会输出执行过程中的所有可用信息——包括状态变化、节点名称、时间戳等。适合开发调试阶段使用。

// 获取完整调试信息
for await (const chunk of await graph.stream(
  { topic: "ice cream" },
  { streamMode: "debug" }
)) {
  console.log(chunk);
  // 输出包含：节点名、时间戳、输入输出状态、执行耗时等全部信息
}

subgraphs: true

输出包含子图内部每个节点的更新
输出格式中会包含子图标识信息
适合需要精细追踪子图执行的场景

debug 模式

输出最完整的执行信息
包含时间戳、节点名、输入输出等
适合开发调试，生产环境一般不用

多模式组合

同时使用多种流模式——用数组语法组合，输出 `[mode, chunk]` 元组

实际项目中，你可能同时需要状态更新和自定义进度数据。LangGraph 允许你把多个流模式组合在一起。

思考题：你既想看每步的状态更新（updates），又想看自定义进度（custom）。两种数据混在一起怎么区分？

// 多模式组合——streamMode 传数组
for await (const [mode, chunk] of await graph.stream(inputs, {
  streamMode: ["updates", "custom"],  // 同时订阅两种模式
})) {
  if (mode === "updates") {
    console.log("State update:", chunk);
  } else if (mode === "custom") {
    console.log("Custom data:", chunk);
  }
}

常用组合

组合	用途
`["updates", "custom"]`	状态更新 + 自定义进度
`["updates", "tools"]`	状态更新 + 工具事件
`["messages", "custom"]`	逐 token + 自定义数据

输出格式变化

单模式：chunk 直接就是数据
多模式：chunk 变成 [mode, data] 元组，需要先判断 mode

易错点：单模式时 chunk 就是数据本身；多模式时 chunk 变成 [mode, data] 元组。忘记解构会导致代码报错。建议始终用 const [mode, chunk] 解构。

📄 完整代码：07_stream_multi_mode.ts

排错指南

流式处理中的常见问题与修复方案

stream 返回空 / 没有输出

原因：图没有编译，或 streamMode 拼写错误。

修复：确认使用了 .compile() 编译后的图；检查 streamMode 是否为六种模式之一。

messages 模式没有 token 输出

原因：图中的 LLM 可能被标记了 nostream，或没有使用 LangChain 的 Chat Model。

修复：检查 LLM 实例的 tags 配置；确认使用的是 ChatOpenAI 等兼容模型。

custom 模式收不到数据

原因：节点中忘记调用 config.writer()，或没有把 config 参数传入。

修复：确认节点签名为 async (state, config)，且在节点内调用了 config.writer(...)。

多模式解构报错

原因：多模式的输出是 [mode, chunk] 元组，但代码按单模式的方式处理了。

修复：多模式时必须用 const [mode, chunk] 解构。单模式用 const chunk 直接接收。

快速自检清单：图编译了吗？streamMode 拼写对吗？LLM 是 LangChain 兼容的 Chat Model 吗？节点签名包含 config 参数吗？多模式时解构正确吗？

动手练习

给第 04 章的计算器智能体加上流式输出

现在轮到你了。拿出第 04 章的计算器智能体，给它加上流式能力。

练习 1：状态流

用 updates 模式运行计算器图
观察每次工具调用后状态如何变化
把每次更新的节点名和输出打印出来

练习 2：逐 Token 流

切换到 messages 模式
实现逐 token 打印 LLM 回复
用 process.stdout.write() 而非 console.log() 实现无缝输出

练习 3：组合流

同时使用 ["updates", "messages"]
状态更新时打印 [STATE] 前缀
token 输出时直接打印到控制台

提示：从最简单的 updates 模式开始，确认能看到每个节点的输出后再切换模式。不要一上来就组合三种模式。

参考答案

给计算器智能体添加流式输出的一种参考写法

// 练习 1：updates 模式
for await (const chunk of await graph.stream(
  { messages: [{ role: "user", content: "3 * 12 + 5 等于多少？" }] },
  { streamMode: "updates", configurable: { thread_id: "calc-1" } }
)) {
  for (const [nodeName, update] of Object.entries(chunk)) {
    console.log(`[STATE] ${nodeName}:`, JSON.stringify(update, null, 2));
  }
}

// 练习 2：messages 模式
for await (const [messageChunk, metadata] of await graph.stream(
  { messages: [{ role: "user", content: "3 * 12 + 5 等于多少？" }] },
  { streamMode: "messages", configurable: { thread_id: "calc-2" } }
)) {
  if (messageChunk.content) {
    process.stdout.write(messageChunk.content);
  }
}
console.log();  // 最后换行

// 练习 3：组合模式
for await (const [mode, chunk] of await graph.stream(
  { messages: [{ role: "user", content: "3 * 12 + 5 等于多少？" }] },
  { streamMode: ["updates", "messages"], configurable: { thread_id: "calc-3" } }
)) {
  if (mode === "updates") {
    for (const [name, update] of Object.entries(chunk)) {
      console.log(`[STATE] ${name}:`, update);
    }
  } else if (mode === "messages") {
    const [msgChunk] = chunk;
    if (msgChunk.content) process.stdout.write(msgChunk.content);
  }
}

关键理解：单模式返回 chunk，多模式返回 [mode, chunk]。这个区别在三个练习中体现得很清楚——练习 1 和 2 是单模式，练习 3 是多模式。

本章回顾

六种流模式，从状态更新到逐 Token 输出

核心要点

六种流模式各有分工：values/updates 看状态，messages 逐 token，custom 推自定义数据，tools 追踪工具，debug 全量信息
所有模式共享同一调用方式：graph.stream(inputs, { streamMode })
多模式组合时输出变成 [mode, chunk] 元组
async generator 工具可以在执行过程中 yield 中间进度
config.writer() 推送自定义数据，不影响图状态

模式选择速查

你想做什么	用什么模式
看每个节点的输出	`updates`
看完整状态变化	`values`
逐字显示 AI 回复	`messages`
推送自定义进度	`custom`
追踪工具执行	`tools`
完整调试	`debug`

核心结论：流式处理把"黑盒等待"变成了"透明进度"。选择正确的流模式，让你的智能体在每一步都能给用户有意义的反馈。这不仅是技术选择，更是用户体验设计。

延伸阅读

继续本课程

下一章：中断与人机交互 — 在任意节点暂停执行，等待人类审批后恢复
时间旅行 — 重放和分叉过去的检查点

官方文档

LangGraph.js 官方文档 — 流式处理完整 API 参考
Streaming How-to — 各模式的详细用法和示例