事件流式传输（Event streaming）

使用带有消息、状态、子图、输出和扩展类型化投影的方式流式传输 LangGraph 运行。

事件流式传输是大多数 LangGraph 应用程序代码推荐的进程内流式传输模型。它返回一个运行流对象，可以同时以多种方式进行消费。

快速入门

typescript

const stream = await graph.streamEvents(
  { messages: [{ role: "user", content: "What is 42 * 17?" }] },
  { version: "v3" }
);

for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

const finalState = await stream.output;

若要针对部署在 Agent Server 后面的图进行流式传输，请参阅 LangSmith Streaming API。

各部分如何协同工作

流式传输技术栈有两个主要层：

Streaming：从 Pregel 引擎发出原始图执行事件。
Event streaming：规范化这些事件，通过流转换器运行它们，并公开类型化的投影。

Pregel 引擎运行图步骤，发出原始 Pregel 事件（updates、values、messages、custom、checkpoints、tasks、debug），发送至事件路由器（Event router，负责将每个事件路由通过转换器管道），级联通过流转换器（Stream transformers：ValuesTransformer、MessagesTransformer、...、自定义转换器），生成 Event Stream（面向应用程序代码的投影事件）。

事件路由器是两层之间的桥梁。它接收规范化的 Pregel 事件，并将每个事件传递给已注册的流转换器。内置转换器创建标准投影，例如 stream.messages、stream.values、stream.subgraphs 和 stream.output。自定义转换器可以在 stream.extensions 下添加特定于应用程序的投影。

事件流式传输提供的内容

运行流在一个底层事件流上公开了类型化投影：

投影	用途
`stream`	遍历每个协议事件。
`stream.messages`	流式传输聊天模型消息和 Token 增量。
`stream.values`	遍历状态快照并等待最终值。
`stream.output`	等待最终输出。
`stream.subgraphs`	发现并观察嵌套图执行。
`stream.interrupts`	检查人机协同中断有效载荷。
`stream.interrupted`	检查运行是否因人类输入而暂停。
`stream.extensions`	消费自定义流转换器投影。

多个使用者可以并发读取这些投影。读取 stream.messages 不会消耗 stream.values、stream.subgraphs 或 stream.output 所需的事件。

事件流式传输位于流式传输（Streaming）之上，后者通过 updates、values、messages、custom、checkpoints、tasks 和 debug 等 stream_mode 模式暴露原始图执行事件。当你需要对这些模式进行底层访问时，请使用流式传输；当应用程序代码受益于类型化投影时，请使用事件流式传输。

流式传输消息（Stream messages）

使用 stream.messages 获取聊天模型输出：

typescript

const stream = await graph.streamEvents(input, { version: "v3" });

for await (const message of stream.messages) {
  const text = await message.text;
  const usage = await message.usage;

  console.log(text);
  console.log(usage);
}

message.text 既是异步可迭代对象，又是类似 Promise 的值。遍历它可逐 Token 获取输出，或 await 它获取完整文本。

流式传输子图（Stream subgraphs）

使用 stream.subgraphs 观察嵌套图工作，而无需解析命名空间字符串：

typescript

const stream = await graph.streamEvents(input, { version: "v3" });

for await (const subgraph of stream.subgraphs) {
  console.log(subgraph.name, subgraph.path);

  for await (const message of subgraph.messages) {
    console.log(await message.text);
  }
}

subgraph.graph_name 是已编译图或智能体的 name。从工具调度的命名智能体（例如通过 Deep Agents task 工具调用的 create_agent(name=...)）会在此处以该名称显示，并且打开该范围的 lifecycle 事件带有一个 cause，链接回发起调度的工具调用。有关详细信息，请参阅生命周期。

有关特定产品的流式传输，请参阅 Deep Agents 流式传输了解子代理流，以及 LangChain 智能体流式传输了解工具调用和中间件事件。

流式传输状态（Stream state）

使用 stream.values 在每个步骤之后流式传输完整的状态快照：

typescript

const stream = await graph.streamEvents(input, { version: "v3" });

for await (const snapshot of stream.values) {
  console.log(snapshot);
}

const finalState = await stream.output;

流式传输多个投影

在 JavaScript 中需要多个投影时使用并发使用者：

typescript

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      console.log(await message.text);
    }
  })(),
  (async () => {
    for await (const subgraph of stream.subgraphs) {
      console.log(subgraph.path);
    }
  })(),
]);

中断后恢复

当图因人类输入而暂停时，检查 stream.interrupted 和 stream.interrupts，然后再次使用 Command 调用 streamEvents(..., { version: "v3" }) 恢复。

恢复需要使用检查点保存器编译图，并携带包含线程 ID 的配置——请参阅持久化。

typescript

import { Command } from "@langchain/langgraph";

let stream = await graph.streamEvents(input, { version: "v3" });

for await (const message of stream.messages) {
  console.log(await message.text);
}

if (stream.interrupted) {
  console.log(stream.interrupts);
}

stream = await graph.streamEvents(
  new Command({ resume: { decisions: [{ type: "approve" }] } }),
  { version: "v3" }
);
const finalState = await stream.output;

流式传输所有协议事件

当你需要原始协议事件流时，请直接使用运行对象本身：

typescript

const stream = await graph.streamEvents(
  { messages: [{ role: "user", content: "What is 42 * 17?" }] },
  { version: "v3" }
);

for await (const event of stream) {
  const namespace = event.params.namespace;
  console.log(namespace, event.method, event.params.data);
}

每个事件都是包装了特定于通道的有效载荷的 ProtocolEvent 信封。转换器的 process(event) 接收到的也是相同的形式。

typescript

interface ProtocolEvent {
  readonly seq: number;         // 在一次运行中严格递增；用于排序
  readonly method: string;      // 通道名称："messages", "values", "updates", "custom", "tools", "lifecycle", ...
  readonly params: {
    readonly namespace: string[];  // 从根图开始的 "<name>:<runtime_id>" 段路径；[] 是根
    readonly timestamp: number;    // 挂钟毫秒数；可能漂移，不要依赖于排序
    readonly node?: string;        // 发出此事件的图节点，适用时提供
    readonly data: unknown;        // 特定于通道的有效载荷；形状取决于 `method`
  };
}

namespace 是从根图到发出事件的作用域的路径。根为空数组 []。每个子级执行添加一个 "name:runtime_id" 段，因此子图内部的嵌套工具调用类似于 ["researcher:6f4d", "tools:91ac"]。: 之前的名称是稳定的图或节点名称；后缀是每次调用的运行时 ID。当你只关心特定的子树时，可以自行按命名空间过滤原始事件——stream.subgraphs 已经为嵌套图执行完成了这项工作。

通道与事件生命周期

原始事件在通道上流动。通道名称显示为事件的 method；每个通道发出特定的事件形态。

通道	用途
`values`	完整图状态快照。
`updates`	逐节点状态增量。
`messages`	以内容块为中心的聊天模型输出。
`tools`	工具调用启动、流式输出、完成和错误事件。
`lifecycle`	运行、子图和子代理状态更改。
`checkpoints`	用于分支和时间旅行的轻量级检查点信封。
`input`	人机协同输入请求和响应。
`tasks`	Pregel 任务创建和结果事件。
`custom`	来自图代码的用户定义载荷。
`custom:<name>`	应用程序定义的流转换器输出。

类型化投影（stream.messages、stream.values 等）是从这些通道构建的。当你直接遍历运行对象时，通道名称显示为原始事件上的 method 字段。

Messages

messages 通道将输出建模为内容块。数据的 event 字段是以下之一：

message-start
content-block-start
content-block-delta
content-block-finish
message-finish

内容块具有显式边界：一个块启动，发出零个或多个增量，并在同一消息中的下一个块开始之前完成。这使得 Token 流式传输、推理块、工具调用块和多模态内容变得明确，而不需要特定于提供商的格式。message-finish 可能包括 Token 使用情况；不可恢复的模型调用失败将作为消息错误事件到达。

每个 content-block-delta 事件在其 index 处携带块的一个增量：

增量 `type`	对块的影响
`text-delta`	将其 `text` 追加到块的 `text` 中。
`reasoning-delta`	将其 `reasoning` 追加到块的 `reasoning` 中。
`data-delta`	将其 `data` 追加到块的 `base64` 中。
`tool-call-delta`	将其 `args` 追加到块的 `args` 中。
`block-delta`	在块上设置 `fields` 中的每个键，替换早期的值。

content-block-finish 携带完整的块。忽略任何你无法识别的增量类型，并继续读取流。

直接消费原始内容块事件而不是使用 stream.messages 投影：

typescript

for await (const event of stream) {
  if (event.method !== "messages") continue;

  const data = event.params.data;
  if (data.event !== "content-block-delta") continue;

  const block = data.delta ?? {};
  if (block.type === "text-delta") {
    process.stdout.write(block.text ?? "");
  } else if (block.type === "reasoning-delta") {
    process.stdout.write(`[thinking]${block.reasoning ?? ""}`);
  } else if (block.type === "tool-call-delta") {
    process.stdout.write(`[args]${block.args ?? ""}`);
  }
}

Tools

tools 通道公开工具执行情况。数据的 event 字段是以下之一：

tool-started
tool-output-delta
tool-finished
tool-error

工具事件通过工具调用 ID 进行关联，因此工具执行可以连接回其在 messages 通道上的原始工具调用内容块。

Lifecycle

lifecycle 通道跟踪根运行、子图和子代理状态。数据的 event 字段是以下之一：

started
running
completed
failed
interrupted

除了 event，生命周期数据可能包含可选的 graph_name、error 和描述子作用域启动原因的 cause（父工具调用、扇出发送、边转换）。

构建你自己的投影

流转换器（Stream transformers）是事件流式传输中的投影层。它们观察协议事件，保留自己的状态，并公开运行的派生视图——例如工具活动、Token 总数、进度事件、工件或针对另一个协议的消息。StreamChannel 是转换器用于发布这些视图的投影原语。

内置投影（stream.messages、stream.values、stream.subgraphs、stream.output）和特定于产品的投影（LangChain 的 stream.tool_calls，Deep Agents 的 stream.subagents）本身就是使用相同契约的转换器。用户转换器通过编译时或调用时注册堆叠在顶部，其投影出现在 stream.extensions 下。

当现有投影不符合应用程序所需的形状时，编写自定义投影。

转换器如何工作

事件流式传输始于来自 LangGraph Pregel 引擎的流式输出。运行时将这些块规范化为协议事件，然后流处理程序通过一系列流转换器路由每个事件。

流处理程序是单个流的中心调度程序。对于每个协议事件，它：

按顺序调用每个已注册转换器的 process(event) 钩子。
将命名的 StreamChannel 推送连接回协议事件流。
将事件存储在运行流中，除非转换器将其抑制。
在运行结束时对每个转换器调用 finalize() 或 fail()。

转换器是观察性的。它们不会回调图运行时。相反，它们消费事件并将派生值推送到 StreamChannel、Promise 或其他投影对象中。

转换器形状

转换器实现 StreamTransformer 接口：

typescript

interface StreamTransformer<TProjection = unknown> {
  init(): TProjection;
  process(event: ProtocolEvent): boolean;
  finalize?(): void | PromiseLike<void>;
  fail?(err: unknown): void;
}

init() 创建投影对象。用户转换器投影出现在 stream.extensions 下。
process() 观察每个协议事件。请参阅流式传输所有协议事件了解 ProtocolEvent 形状。仅当你故意想要抑制原始事件时返回 false。
finalize() 在成功的流之后关闭或解析非通道投影。
fail() 将错误传播到非通道投影。

声明所需的流模式

required_stream_modes 控制底层图在流期间发出哪些 Pregel 流模式。运行时获取每个已注册转换器的 required_stream_modes 的并集，并将该并集作为 stream_mode 参数传递给图的 .stream() 调用。没有转换器请求的模式绝不会被发出——声明 ("custom",) 才会使 custom 事件流经运行。

process() 接收图发出的每个事件，并负责按 event["method"] 进行过滤。声明开启上游发射；它不会缩小 process() 看到的内容。有效值是 Pregel 流模式："messages"、"tools"、"custom"、"values"、"updates"、"checkpoints"、"tasks"、"debug"。每个转换器必须声明它作用的每种模式——被省略的模式不会被图发出，也绝不会到达 process()。

StreamChannel

StreamChannel 是转换器用于流式传输值的投影原语。它始终在 stream.extensions.<name> 上公开可迭代流。构造函数参数决定每个 push() 是否也会作为 custom:<name> 事件流入运行的主事件流——即投影的值在遍历原始协议事件时是否显示。

需求	用法
仅限副通道投影	`new StreamChannel<T>()`
同时将每次推送流入主事件流	`new StreamChannel<T>(name)`

命名通道的载荷必须是可序列化的，因为每个推送的值也会成为主流中的 custom:<name> 协议事件。将 Promise、异步可迭代对象、类实例和其他进程内句柄保留在未命名通道中。

流处理程序拥有通道生命周期。一旦 init() 返回通道，处理程序在运行结束时为你关闭或失败它。转换器只推送值。

示例：命名通道

将字符串名称传递给 StreamChannel 以通过 stream.extensions 公开流式投影，并且将每个推送的值作为 custom:<name> 协议事件转发到运行的主事件流中：

typescript

import { StreamChannel } from "@langchain/langgraph";

const toolActivityTransformer = () => {
  const activity = new StreamChannel<{
    name: string;
    status: "started" | "finished" | "error";
  }>("toolActivity");

  return {
    init: () => ({ toolActivity: activity }),
    process(event) {
      if (event.method === "tools") {
        const data = event.params.data as { tool_name?: string; event?: string };
        if (data.tool_name && data.event) {
          activity.push({
            name: data.tool_name,
            status: data.event === "tool-error" ? "error" : "started",
          });
        }
      }
      return true;
    },
  };
};

示例：未命名通道

没有名称的通道仅是副通道投影——可在 stream.extensions 上访问，但对遍历原始事件的消费者不可见。对于持有无法在主事件流上序列化的进程内句柄（Promise、异步可迭代对象、类实例）的投影，这是正确的选择。

以下示例将未命名通道与 get_stream_writer 配对，允许图节点发出 custom 通道事件，然后转换器将这些事件排出到投影中：

typescript

import { StreamChannel } from "@langchain/langgraph";

const customTransformer = () => {
  const custom = new StreamChannel<unknown>();

  return {
    init: () => ({ custom }),
    process(event) {
      if (event.method === "custom") {
        custom.push(event.params.data);
      }
      return true;
    },
  };
};

示例：最终值投影

当投影不应流入主事件流时，请使用未命名流、Promise 或其他进程内对象：

typescript

const statsTransformer = () => {
  let totalTokens = 0;
  let resolveTotal!: (value: number) => void;
  const totalTokensPromise = new Promise<number>((resolve) => {
    resolveTotal = resolve;
  });

  return {
    init: () => ({ totalTokens: totalTokensPromise }),
    process(event) {
      if (event.method === "messages") {
        const data = event.params.data as { usage?: { output_tokens?: number } };
        totalTokens += data.usage?.output_tokens ?? 0;
      }
      return true;
    },
    finalize: () => resolveTotal(totalTokens),
  };
};

在调用时或编译时注册

在调用时传递转换器以进行本地实验：

typescript

const stream = await graph.streamEvents(input, {
  version: "v3",
  transformers: [statsTransformer, toolActivityTransformer],
});

当该图的每次运行都应该产生投影时，将转换器编译到图中：

typescript

const graph = builder.compile({
  transformers: [statsTransformer, toolActivityTransformer],
});

事件流式传输（Event streaming） ​

快速入门 ​

各部分如何协同工作 ​

事件流式传输提供的内容 ​

流式传输消息（Stream messages） ​

流式传输子图（Stream subgraphs） ​

流式传输状态（Stream state） ​

流式传输多个投影 ​

中断后恢复 ​

流式传输所有协议事件 ​

通道与事件生命周期 ​

Messages ​

Tools ​

Lifecycle ​

构建你自己的投影 ​

转换器如何工作 ​

转换器形状 ​

声明所需的流模式 ​

StreamChannel ​

示例：命名通道 ​

示例：未命名通道 ​

示例：最终值投影 ​

在调用时或编译时注册 ​

相关内容 ​

事件流式传输（Event streaming）

快速入门

各部分如何协同工作

事件流式传输提供的内容

流式传输消息（Stream messages）

流式传输子图（Stream subgraphs）

流式传输状态（Stream state）

流式传输多个投影

中断后恢复

流式传输所有协议事件

通道与事件生命周期

Messages

Tools

Lifecycle

构建你自己的投影

转换器如何工作

转换器形状

声明所需的流模式

StreamChannel

示例：命名通道

示例：未命名通道

示例：最终值投影

在调用时或编译时注册

相关内容