流式处理

📚 学习目标

学完这篇文章后，你将能够：

理解为什么流式处理对 AI 应用至关重要
掌握 LangGraphJS 提供的七种流式模式
区分 stream 和 streamEvents 的使用场景

前置知识

在开始学习之前，建议先阅读：

01-chatbots

1 为什么需要流式处理？

AI 模型的推理通常很慢。如果用户必须等待整个回答生成完才能看到内容，体验会非常糟糕。

流式处理 (Streaming) 允许我们：

降低感知延迟：用户几乎立刻就能看到第一个字。
提升透明度：展示 AI 的思考过程（如正在调用什么工具）。
增强容错性：如果生成出错，可以立刻停止。

LangGraphJS 提供了极致灵活的流式支持。

2 核心流式模式

LangGraphJS 的 stream 方法支持多种模式。

const stream = await app.stream(inputs, { streamMode: "..." });

1. Values 模式 (默认)

返回每个步骤执行后的完整状态。

// 适合：更新整个页面状态
for await (const chunk of await app.stream(inputs, { streamMode: "values" })) {
  console.log(chunk); // { messages: [...], otherState: ... }
}

2. Updates 模式

只返回每个步骤新产生/更新的状态部分。

// 适合：增量更新
for await (const chunk of await app.stream(inputs, { streamMode: "updates" })) {
  console.log(chunk); // { llmNode: { messages: [AIMessage(...)] } }
}

3. Messages 模式 (最常用)

只返回 LLM 生成的 Message Chunk。这是实现类似 ChatGPT "打字机效果" 的最佳方式。

// 适合：聊天界面
for await (const chunk of await app.stream(inputs, { streamMode: "messages" })) {
  // chunk 是 [BaseMessageChunk]
  process.stdout.write(chunk[0].content);
}

更贴近真实项目的写法通常会区分不同 chunk 类型：

import { isAIMessageChunk } from '@langchain/core/messages';

const stream = await app.stream(inputs, { streamMode: 'messages' });
for await (const [msg] of stream) {
  if (isAIMessageChunk(msg)) {
    process.stdout.write(String(msg.content));
  }
}

3 高级流式模式

4. Custom 模式

如果你想在节点内部发送自定义数据（非 State），可以使用 writer。

const myNode = async (state, config) => {
  const { writer } = config;
  writer({ type: "progress", value: 50 }); // 发送自定义事件
  return { ... };
};

for await (const chunk of await app.stream(inputs, { streamMode: "custom" })) {
  console.log(chunk); // { type: "progress", value: 50 }
}

5. Debug 模式

返回最详细的执行日志，包括每个节点的输入输出、时间戳等。

6. Events 模式 (`streamEvents`)

这是最底层的 API，它暴露了 LangChain 内部的所有事件（如 on_chat_model_stream, on_tool_start）。

for await (const event of app.streamEvents(inputs, { version: 'v2' })) {
  // 只演示一个常见事件：模型 token 流
  if (event.event === 'on_chat_model_stream') {
    process.stdout.write(String(event.data.chunk.content));
  }
}

什么时候用 streamEvents？

你要在 UI 上展示“工具开始/结束”“检索中”“已拿到结果”等状态
你要做更精细的观测（token 计数、耗时、每个节点的事件）

7. Checkpoints 模式

如果你启用了持久化（checkpointer），可以用 checkpoints 模式读取执行中的检查点信息：

for await (const cp of await app.stream(inputs, { streamMode: 'checkpoints' })) {
  console.log('checkpoint:', cp);
}

8. Tasks 模式

用于观察节点/子图级别的任务事件：

for await (const task of await app.stream(inputs, { streamMode: 'tasks' })) {
  console.log('task:', task);
}

💡 提示

tasks 模式适合做“执行进度条”或任务监控面板。

3.1 streamEvents 进阶用法

你可以捕获更多事件类型，如工具调用开始/结束：

for await (const event of app.streamEvents(inputs, { version: 'v2' })) {
  if (event.event === 'on_tool_start') {
    console.log('tool start:', event.name);
  }
  if (event.event === 'on_tool_end') {
    console.log('tool end:', event.name);
  }
}

ℹ️ 说明

streamEvents 是最底层的事件流，适合做监控/埋点。

4 选型指南：我该用哪个？

目标	推荐	理由
聊天 UI 打字机效果	`streamMode: "messages"`	最直接、最轻量
状态面板/调试面板	`streamMode: "values"`	你能拿到完整 state
后台日志/埋点/观测	`streamEvents(v2)`	事件最丰富
自定义进度条/阶段提示	`streamMode: "custom"`	节点可写入自定义事件

💡 练习题

选择题：如果你正在构建一个聊天机器人，想要在 UI 上实现打字机效果，应该优先选择哪种模式？
- A. Values
- B. Updates
- C. Messages
- D. Debug
点击查看答案
选 C（Messages）。
实战题：修改你的 Graph，使其支持 streamMode: "custom"，并在节点处理开始时发送一个 "Thinking..." 的自定义事件。

点击查看答案
在节点中使用 config.writer({ type: 'thinking' })。
思考题：什么时候应该用 streamEvents 而不是 stream？

点击查看答案
当你需要工具开始/结束事件或更细粒度的埋点时。
操作题：为执行过程输出 checkpoints 事件。

点击查看答案
启用 checkpointer 后使用 streamMode: 'checkpoints'。
思考题：为什么 Updates 模式比 Values 模式更省流量？

点击查看答案
Updates 只返回变化部分，避免重复传输完整状态。

✅ 总结

本章要点：

Messages 模式是做聊天机器人的首选。
Updates 模式适合做状态监控。
Events 模式提供了最大的灵活性，但 API 也最复杂。

下一步：如何让 AI 主动使用工具？学习工具调用。

登录以继续阅读

解锁完整文档、代码示例及更多高级功能。

立即登录

流式处理

登录以继续阅读

On this page