流式传输（Streaming）

TIP

对于新应用程序，我们建议使用事件流式传输（Event streaming）——这是在 LangGraph v1.2 中引入的类型化投影 API。事件流式传输为每个投影（消息、状态值、子图、输出）提供单独的迭代器，因此你可以独立使用它们，而不必在 stream_mode 数据块上做类型分支判断。

本页面介绍 LangGraph 的流模式（stream-mode）API。它通过 updates、values、messages、custom、checkpoints、tasks 和 debug 等流模式公开图执行。当你需要直接访问图运行时事件或特定的流模式输出时，请使用它。

入门

基本用法

LangGraph 图公开了 stream 方法，以将流式输出作为迭代器产生。

typescript

for await (const chunk of await graph.stream(inputs, {
  streamMode: "updates",
})) {
  console.log(chunk);
}

TIP

使用 LangSmith 调试流式传输事件、检查逐 Token 的 LLM 输出并监控延迟。按照跟踪快速入门进行设置。

流模式（Stream modes）

将以下一个或多个流模式作为数组传递给 stream 方法：

模式	说明
values	每个步骤之后的完整状态。
updates	每个步骤之后的状态更新。同一步骤中的多个更新单独流式传输。
messages	来自 LLM 调用的（LLM Token, 元数据）二元元组。
custom	通过 `writer` 配置参数从节点发出的自定义数据。
tools	工具调用生命周期事件（`on_tool_start`、`on_tool_event`、`on_tool_end`、`on_tool_error`）。
debug	贯穿整个图执行过程的所有可用信息。

图状态（Graph state）

使用流模式 updates 和 values 在图执行时流式传输图的状态。

updates 流式传输图的每个步骤之后对状态的更新。
values 流式传输图的每个步骤之后状态的完整值。

typescript

import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod/v4";

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

使用它可以在每个步骤之后仅流式传输节点返回的状态更新。流式输出包括节点名称以及更新内容。

typescript

for await (const chunk of await graph.stream(
  { topic: "ice cream" },
  { streamMode: "updates" }
)) {
  for (const [nodeName, state] of Object.entries(chunk)) {
    console.log(`Node ${nodeName} updated:`, state);
  }
}

values

使用它可以在每个步骤之后流式传输图的完整状态。

typescript

for await (const chunk of await graph.stream(
  { topic: "ice cream" },
  { streamMode: "values" }
)) {
  console.log(`topic: ${chunk.topic}, joke: ${chunk.joke}`);
}

LLM Token

使用 messages 流模式从图的任何部分（包括节点、工具、子图或任务）逐 Token 流式传输大语言模型（LLM）的输出。

来自 messages 模式的流式输出是元组 [message_chunk, metadata]，其中：

message_chunk：来自 LLM 的 Token 或消息片段。
metadata：包含有关图节点和 LLM 调用的详细信息的字典。

如果你的 LLM 无法作为 LangChain 集成使用，你可以改用 custom 模式流式传输其输出。有关详细信息，请参阅与任意 LLM 配合使用。

typescript

import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, StateSchema, GraphNode, START } from "@langchain/langgraph";
import * as z from "zod";

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

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

const callModel: GraphNode<typeof MyState> = async (state) => {
  // 调用 LLM 生成关于某个主题的笑话
  // 请注意，即使 LLM 使用 .invoke 而不是 .stream 运行，也会发出消息事件
  const modelResponse = await model.invoke([
    { role: "user", content: `Generate a joke about ${state.topic}` },
  ]);
  return { joke: modelResponse.content };
};

const graph = new StateGraph(MyState)
  .addNode("callModel", callModel)
  .addEdge(START, "callModel")
  .compile();

// "messages" 流模式返回元组 [messageChunk, metadata] 的迭代器
// 其中 messageChunk 是 LLM 流式传输的 Token，metadata 是包含调用 LLM 的图节点信息和其他信息的字典
for await (const [messageChunk, metadata] of await graph.stream(
  { topic: "ice cream" },
  { streamMode: "messages" }
)) {
  if (messageChunk.content) {
    console.log(messageChunk.content + "|");
  }
}

按 LLM 调用过滤

你可以将 tags 与 LLM 调用相关联，以按 LLM 调用过滤流式传输的 Token。

typescript

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

// model1 标记为 "joke"
const model1 = new ChatOpenAI({
  model: "gpt-5.4-mini",
  tags: ['joke']
});
// model2 标记为 "poem"
const model2 = new ChatOpenAI({
  model: "gpt-5.4-mini",
  tags: ['poem']
});

const graph = // ... 定义使用这些 LLM 的图

// streamMode 设置为 "messages" 以流式传输 LLM Token
// metadata 包含有关 LLM 调用的信息，包括 tags
for await (const [msg, metadata] of await graph.stream(
  { topic: "cats" },
  { streamMode: "messages" }
)) {
  // 按元数据中的 tags 字段过滤流式传输的 Token，以仅包含带有 "joke" 标签的 LLM 调用的 Token
  if (metadata.tags?.includes("joke")) {
    console.log(msg.content + "|");
  }
}

自定义数据（Custom data）

若要从 LangGraph 节点或工具内部发送自定义用户定义数据，请遵循以下步骤：

使用来自 LangGraphRunnableConfig 的 writer 参数发出自定义数据。
调用 .stream() 时设置 streamMode: "custom" 以在流中获取自定义数据。你可以组合多种模式（例如 ["updates", "custom"]），但至少有一个必须是 "custom"。

节点（node）

typescript

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

const State = new StateSchema({
  query: z.string(),
  answer: z.string(),
});

const node: GraphNode<typeof State> = async (state, config) => {
  // 使用 writer 发出自定义键值对（例如进度更新）
  config.writer({ custom_key: "Generating custom data inside node" });
  return { answer: "some data" };
};

const graph = new StateGraph(State)
  .addNode("node", node)
  .addEdge(START, "node")
  .compile();

const inputs = { query: "example" };

// 设置 streamMode: "custom" 以在流中接收自定义数据
for await (const chunk of await graph.stream(inputs, { streamMode: "custom" })) {
  console.log(chunk);
}

工具（tool）

typescript

import { tool } from "@langchain/core/tools";
import { LangGraphRunnableConfig } from "@langchain/langgraph";
import * as z from "zod";

const queryDatabase = tool(
  async (input, config: LangGraphRunnableConfig) => {
    // 使用 writer 发出自定义键值对（例如进度更新）
    config.writer({ data: "Retrieved 0/100 records", type: "progress" });
    // 执行查询
    // 发出另一个自定义键值对
    config.writer({ data: "Retrieved 100/100 records", type: "progress" });
    return "some-answer";
  },
  {
    name: "query_database",
    description: "Query the database.",
    schema: z.object({
      query: z.string().describe("The query to execute."),
    }),
  }
);

const graph = // ... 定义使用此工具的图

// 设置 streamMode: "custom" 以在流中接收自定义数据
for await (const chunk of await graph.stream(inputs, { streamMode: "custom" })) {
  console.log(chunk);
}

工具进度（Tool progress）

使用 tools 流模式接收工具执行的实时生命周期事件。这对于在工具运行时在 UI 中显示进度指示器、中间结果和错误状态非常有用。

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`

定义流式传输进度的工具

要发出 on_tool_event 事件，请将工具函数定义为异步生成器（async generator）（async function*）。每个 yield 向流发送中间数据，return 值用作工具的最终结果。

typescript

import { tool } from "@langchain/core/tools";
import { z } from "zod/v4";

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 { completed: [...completed], progress: (i + 1) / airlines.length };
    }

    return `Found flights on ${completed.join(", ")}`;
  },
  {
    name: "search_flights",
    description: "Search for available flights across airlines.",
    schema: z.object({ destination: z.string() }),
  }
);

子图输出（Subgraph outputs）

要在流式输出中包含来自子图的输出，你可以在父图的 .stream() 方法中设置 subgraphs: true。这将同时流式传输来自父图和任何子图的输出。

输出将作为元组 [namespace, data] 进行流式传输，其中 namespace 是带有调用子图的节点路径的元组，例如 ["parent_node:<task_id>", "child_node:<task_id>"]。

typescript

for await (const chunk of await graph.stream(
  { foo: "foo" },
  {
    // 设置 subgraphs: true 以流式传输来自子图的输出
    subgraphs: true,
    streamMode: "updates",
  }
)) {
  console.log(chunk);
}

INFO

这适用于每种 streamMode，包括 "messages"。createAgent 返回一个 ReactAgent 包装器；在将其作为节点添加时传递 agent.graph，以便父级将其视为子图。设置 subgraphs: true 后，消息块为 [namespace, [token, metadata]]，因此你可以判断哪个子图发出了它们。

调试（Debug）

使用 debug 流模式在整个图执行过程中流式传输尽可能多的信息。流式输出包括节点名称以及完整状态。

typescript

for await (const chunk of await graph.stream(
  { topic: "ice cream" },
  { streamMode: "debug" }
)) {
  console.log(chunk);
}

一次使用多种模式

你可以传递一个数组作为 streamMode 参数以同时流式传输多种模式。

流式输出将是 [mode, chunk] 的元组，其中 mode 是流模式的名称，chunk 是该模式流式传输的数据。

typescript

for await (const [mode, chunk] of await graph.stream(inputs, {
  streamMode: ["updates", "custom"],
})) {
  console.log(chunk);
}

进阶（Advanced）

与任意 LLM 配合使用

你可以使用 streamMode: "custom" 从任何 LLM API 流式传输数据——即使该 API 未实现 LangChain 聊天模型接口。

这使你能够集成原始 LLM 客户端或提供自己的流式传输接口的外部服务，使 LangGraph 在自定义设置中具有高度灵活性。

typescript

import { StateGraph, GraphNode, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({ result: z.string() });

const callArbitraryModel: GraphNode<typeof State> = async (state, config) => {
  // 调用任意模型并流式传输输出的示例节点
  // 假设你有一个产生数据块的流式客户端
  // 使用自定义流式客户端生成 LLM Token
  for await (const chunk of yourCustomStreamingClient(state.topic)) {
    // 使用 writer 向流发送自定义数据
    config.writer({ custom_llm_chunk: chunk });
  }
  return { result: "completed" };
};

const graph = new StateGraph(State)
  .addNode("callArbitraryModel", callArbitraryModel)
  .compile();

// 设置 streamMode: "custom" 以在流中接收自定义数据
for await (const chunk of await graph.stream(
  { topic: "cats" },
  { streamMode: "custom" }
)) {
  // 该块将包含从 llm 流式传输的自定义数据
  console.log(chunk);
}

针对特定聊天模型禁用流式传输

如果你的应用程序混合了支持流式传输的模型和不支持流式传输的模型，你可能需要为不支持的模型显式禁用流式传输。

在初始化模型时设置 streaming: false。

typescript

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

const model = new ChatOpenAI({
  model: "gpt-5.5",
  // 设置 streaming: false 以禁用聊天模型的流式传输
  streaming: false,
});

INFO

并非所有聊天模型集成都支持 streaming 参数。如果你的模型不支持它，请改用 disableStreaming: true。此参数可通过基类在所有聊天模型上使用。

流式传输（Streaming） ​

入门 ​

基本用法 ​

流模式（Stream modes） ​

图状态（Graph state） ​

LLM Token ​

按 LLM 调用过滤 ​

自定义数据（Custom data） ​

工具进度（Tool progress） ​

定义流式传输进度的工具 ​

子图输出（Subgraph outputs） ​

调试（Debug） ​

一次使用多种模式 ​

进阶（Advanced） ​

与任意 LLM 配合使用 ​

针对特定聊天模型禁用流式传输 ​

流式传输（Streaming）

入门

基本用法

流模式（Stream modes）

图状态（Graph state）

LLM Token

按 LLM 调用过滤

自定义数据（Custom data）

工具进度（Tool progress）

定义流式传输进度的工具

子图输出（Subgraph outputs）

调试（Debug）

一次使用多种模式

进阶（Advanced）

与任意 LLM 配合使用

针对特定聊天模型禁用流式传输