流式传输

对于新应用，我们推荐事件流——这是 LangGraph v1.2 中引入的类型化投影 API。事件流为每个投影（消息、值、子图、输出）提供独立的迭代器，这样您就可以独立消费它们，而无需基于 stream_mode 块进行分支处理。

本页面介绍 LangGraph 的流模式 API。它通过 updates、values、messages、custom、checkpoints、tasks 和 debug 等流模式暴露图执行过程。当您需要直接访问图运行时事件或特定流模式输出时，请使用此 API。

开始使用

基本用法

LangGraph 图暴露了 stream 方法，以迭代器的形式生成流式输出。

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

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

流模式

向 stream 方法传递以下一个或多个流模式（作为列表）：

模式	描述
values	每一步后的完整状态。
updates	每一步后的状态更新。同一步中的多个更新会分别流式传输。
messages	来自 LLM 调用的 (LLM token, metadata) 2 元组。
custom	节点通过 `writer` 配置参数发出的自定义数据。
tools	工具调用生命周期事件 (`on_tool_start`, `on_tool_event`, `on_tool_end`, `on_tool_error`)。
debug	图执行过程中的所有可用信息。

图状态

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

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

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
values

使用此模式仅流式传输节点在每一步后返回的 状态更新。流式输出包含节点名称及其更新内容。

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);
  }
}

使用此模式在每一步后流式传输图的 完整状态。

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

LLM tokens

使用 messages 流模式从图的任何部分（包括节点、工具、子图或任务）逐个 token 流式传输大型语言模型（LLM）的输出。 messages 模式的流式输出是一个元组 [message_chunk, metadata]，其中：

message_chunk：来自 LLM 的 token 或消息段。
metadata：一个包含图节点和 LLM 调用详细信息的字典。

如果您的 LLM 作为 LangChain 集成不可用，您仍然可以使用 custom 模式流式传输其输出。详情请参见与任何 LLM 配合使用。

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。

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 + "|");
  }
}

扩展示例：按标签过滤

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

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

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

const callModel: GraphNode<typeof State> = async (state) => {
  const topic = state.topic;
  console.log("Writing joke...");

  const jokeResponse = await jokeModel.invoke([
    { role: "user", content: `Write a joke about ${topic}` }
  ]);

  console.log("\n\nWriting poem...");
  const poemResponse = await poemModel.invoke([
    { role: "user", content: `Write a short poem about ${topic}` }
  ]);

  return {
    joke: jokeResponse.content,
    poem: poemResponse.content
  };
};

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

// 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 + "|");
  }
}

从流中省略消息

使用 nostream 标签可完全排除 LLM 输出。标记为 nostream 的调用仍然会运行并产生输出；只是它们的 token 不会在 messages 模式中发出。这在以下情况下很有用：

您需要 LLM 输出用于内部处理（例如结构化输出），但不希望将其流式传输到客户端
您通过不同的通道（例如自定义 UI 消息）流式传输相同的内容，并希望避免在 messages 流中出现重复输出

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

const streamModel = new ChatAnthropic({ model: "claude-haiku-4-5-20251001" });
const internalModel = new ChatAnthropic({
  model: "claude-haiku-4-5-20251001",
}).withConfig({
  tags: ["nostream"],
});

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

const contentToText = (content: unknown): string => {
  if (typeof content === "string") {
    return content;
  }
  if (Array.isArray(content)) {
    return content
      .map((block) => {
        if (
          typeof block === "object" &&
          block !== null &&
          "text" in block &&
          typeof (block as { text?: unknown }).text === "string"
        ) {
          return (block as { text: string }).text;
        }
        return "";
      })
      .filter(Boolean)
      .join("\n");
  }
  return "";
};

const writeAnswer = async (state: typeof State.State) => {
  const r = await streamModel.invoke([
    { role: "user", content: `简要回复关于 ${state.topic} 的内容` },
  ]);
  return { answer: contentToText(r.content) };
};

const internalNotes = async (state: typeof State.State) => {
  // 由于 nostream 标签，此模型的令牌在 streamMode: "messages" 中会被省略
  const r = await internalModel.invoke([
    { role: "user", content: `关于 ${state.topic} 的内部笔记` },
  ]);
  return { notes: contentToText(r.content) };
};

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

const stream = await graph.streamEvents(
  { topic: "AI", answer: "", notes: "" },
  { version: "v3" },
);

按节点过滤

要仅从特定节点流式传输 token，请使用 stream_mode="messages" 并通过流式元数据中的 langgraph_node 字段过滤输出：

// "messages" 流模式返回一个 [messageChunk, metadata] 元组
// 其中 messageChunk 是 LLM 流式传输的 token，metadata 是一个字典
// 包含调用 LLM 的图节点信息及其他信息
for await (const [msg, metadata] of await graph.stream(
  inputs,
  { streamMode: "messages" }
)) {
  // 通过元数据中的 langgraph_node 字段过滤流式传输的 token
  // 以仅包含来自指定节点的 token
  if (msg.content && metadata.langgraph_node === "some_node_name") {
    // ...
  }
}

扩展示例：从特定节点流式传输 LLM token

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

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

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

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

const writePoem: GraphNode<typeof State> = async (state) => {
  const topic = state.topic;
  const poemResponse = await model.invoke([
    { role: "user", content: `Write a short poem about ${topic}` }
  ]);
  return { poem: poemResponse.content };
};

const graph = new StateGraph(State)
  .addNode("writeJoke", writeJoke)
  .addNode("writePoem", writePoem)
  // 并发写笑话和诗
  .addEdge(START, "writeJoke")
  .addEdge(START, "writePoem")
  .compile();

// "messages" 流模式返回一个 [messageChunk, metadata] 元组
// 其中 messageChunk 是 LLM 流式传输的 token，metadata 是一个字典
// 包含调用 LLM 的图节点信息及其他信息
for await (const [msg, metadata] of await graph.stream(
  { topic: "cats" },
  { streamMode: "messages" }
)) {
  // 通过元数据中的 langgraph_node 字段过滤流式传输的 token
  // 以仅包含来自 writePoem 节点的 token
  if (msg.content && metadata.langgraph_node === "writePoem") {
    console.log(msg.content + "|");
  }
}

自定义数据

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

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

节点
工具

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);
}

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: "查询数据库。",
    schema: z.object({
      query: z.string().describe("要执行的查询。"),
    }),
  }
);

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

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

工具进度

使用 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 function*)。每个 yield 会将中间数据发送到流中，而 return 的值则作为工具的最终结果。

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 都会向流发出一个 on_tool_event
      yield {
        message: `Searching ${airlines[i]}...`,
        progress: (i + 1) / airlines.length,
        completed,
      };
    }

    // 返回的值将成为工具结果 (ToolMessage.content)
    return JSON.stringify({
      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(),
    }),
  }
);

现有的返回 Promise 的工具完全兼容。它们发出 on_tool_start 和 on_tool_end 事件，但不会发出 on_tool_event 事件。

在服务器端消费工具事件

将 streamMode: ["tools"]（或与其他模式组合）传递给 graph.stream()：

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}`, chunk.input);
        break;
      case "on_tool_event":
        console.log(`Tool progress: ${chunk.name}`, chunk.data);
        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;
    }
  }
}

在 React 中使用 `useStream` 显示工具进度

来自 @langchain/langgraph-sdk/react 的 useStream hook 在您的流模式中包含 "tools" 时，会暴露一个 toolProgress 数组。每个条目都是一个 ToolProgress 对象，用于跟踪正在运行的工具的当前状态：

字段	描述
`name`	工具名称
`state`	当前生命周期状态：`"starting"`、`"running"`、`"completed"` 或 `"error"`
`toolCallId`	来自 LLM 的工具调用 ID
`input`	工具的输入参数
`data`	来自 `on_tool_event` 的最新产生数据
`result`	最终结果，在 `on_tool_end` 时设置
`error`	错误，在 `on_tool_error` 时设置

import { useStream } from "@langchain/langgraph-sdk/react";

function Chat() {
  const stream = useStream({
    assistantId: "my-agent",
    streamMode: ["values", "tools"],
  });

  // 过滤出正在运行的工具
  const activeTools = stream.toolProgress.filter(
    (t) => t.state === "starting" || t.state === "running"
  );

  return (
    <div>
      {stream.messages.map((msg) => (
        <MessageBubble key={msg.id} message={msg} />
      ))}

      {/* 为正在运行的工具显示进度卡片 */}
      {activeTools.map((tool) => (
        <ToolProgressCard
          key={tool.toolCallId ?? tool.name}
          name={tool.name}
          state={tool.state}
          data={tool.data}
        />
      ))}
    </div>
  );
}

扩展示例：带有工具进度的旅行规划代理

此示例展示了一个完整的代理，其异步生成器工具可将搜索进度流式传输到 React UI。代理定义：

import { tool } from "@langchain/core/tools";
import { ChatOpenAI } from "@langchain/openai";
import { createAgent } from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph-checkpoint-memory";
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, 600));
      completed.push(`${airlines[i]}: checked`);
      yield {
        message: `Searching ${airlines[i]}...`,
        progress: (i + 1) / airlines.length,
        completed,
      };
    }

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

const checkHotels = tool(
  async function* (input) {
    const hotels = ["Grand Hyatt", "Marriott", "Hilton"];
    const completed: string[] = [];

    for (let i = 0; i < hotels.length; i++) {
      await new Promise((r) => setTimeout(r, 400));
      completed.push(`${hotels[i]}: available`);
      yield {
        message: `Checking ${hotels[i]}...`,
        progress: (i + 1) / hotels.length,
        completed,
      };
    }

    return JSON.stringify({
      hotels: [
        { name: "Grand Hyatt", price: 250, rating: 4.5 },
        { name: "Marriott", price: 180, rating: 4.2 },
      ],
    });
  },
  {
    name: "check_hotels",
    description: "Check hotel availability.",
    schema: z.object({
      city: z.string(),
      check_in: z.string(),
      nights: z.number(),
    }),
  }
);

export const agent = createAgent({
  model: new ChatOpenAI({ model: "gpt-5.4-mini" }),
  tools: [searchFlights, checkHotels],
  checkpointer: new MemorySaver(),
});

带有进度卡片的 React 组件：

import { useStream } from "@langchain/langgraph-sdk/react";

function TravelPlanner() {
  const stream = useStream<typeof agent>({
    assistantId: "travel-agent",
    streamMode: ["values", "tools"],
  });

  const activeTools = stream.toolProgress.filter(
    (t) => t.state === "starting" || t.state === "running"
  );

  return (
    <div>
      {stream.messages.map((msg) => (
        <div key={msg.id}>{msg.content}</div>
      ))}

      {activeTools.map((tool) => {
        const data = tool.data as {
          message?: string;
          progress?: number;
          completed?: string[];
        } | undefined;

        return (
          <div key={tool.toolCallId ?? tool.name}>
            <strong>{tool.name}</strong>
            {data?.message && <p>{data.message}</p>}
            {data?.progress != null && (
              <div style={{ width: "100%", background: "#eee" }}>
                <div
                  style={{
                    width: `${data.progress * 100}%`,
                    background: "#4CAF50",
                    height: 8,
                    transition: "width 0.3s ease",
                  }}
                />
              </div>
            )}
            {data?.completed?.map((step, i) => (
              <div key={i}>&#10003; {step}</div>
            ))}
          </div>
        );
      })}
    </div>
  );
}

`tools` 与 `custom` 流模式

这两种流模式都可以展示工具进度，但它们服务于不同的目的：

tools——自动发出结构化的生命周期事件（on_tool_start、on_tool_event、on_tool_end、on_tool_error），除了在您的工具中使用 async function* 外无需更改代码。useStream hook 提供了开箱即用的响应式 toolProgress 数组。
custom——让您使用 config.writer() 完全控制发出什么数据以及何时发出。当您需要不映射到工具生命周期的自由格式数据，或者您想从节点（而不仅仅是工具）流式传输时，请使用此模式。

子图输出

要将子图的输出包含在流式输出中，您可以在父图的 .stream() 方法中设置 subgraphs: true。这将同时流式传输来自父图和任何子图的输出。输出将以元组 [namespace, data] 的形式流式传输，其中 namespace 是一个包含调用子图的节点路径的元组，例如 ["parent_node:<task_id>", "child_node:<task_id>"]。

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

这适用于每个 streamMode，包括 "messages"。createAgent 返回一个 ReactAgent 包装器；将其作为节点添加时传递 agent.graph，这样父图会将其视为子图。使用 subgraphs: true 时，消息块是 [namespace, [token, metadata]]，这样您就可以知道是哪个子图发出的。

import { createAgent } from "langchain";
import { END, START, StateGraph } from "@langchain/langgraph";

const agent = createAgent({ model, tools, stateSchema: State });

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

for await (const [ns, data] of await graph.stream(
    { messages: [{ role: "user", content: "..." }] },
    {
        streamMode: "messages",
        subgraphs: true,
    }
)) {
    const [token, metadata] = data;
    console.log(ns, token, metadata);
}

扩展示例：从子图流式传输

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

// 定义子图
const SubgraphState = new StateSchema({
  foo: z.string(), // 注意：此键与父图状态共享
  bar: z.string(),
});

const subgraphBuilder = new StateGraph(SubgraphState)
  .addNode("subgraphNode1", (state) => {
    return { bar: "bar" };
  })
  .addNode("subgraphNode2", (state) => {
    return { foo: state.foo + state.bar };
  })
  .addEdge(START, "subgraphNode1")
  .addEdge("subgraphNode1", "subgraphNode2");
const subgraph = subgraphBuilder.compile();

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

::::::python

Root: {'node_1': {'foo': 'hi! foo'}}
Subgraph ('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',): {'subgraph_node_1': {'bar': 'bar'}}
Subgraph ('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',): {'subgraph_node_2': {'foo': 'hi! foobar'}}
Root: {'node_2': {'foo': 'hi! foobar'}}

::::::js

[[], {'node1': {'foo': 'hi! foo'}}]
[['node2:dfddc4ba-c3c5-6887-5012-a243b5b377c2'], {'subgraphNode1': {'bar': 'bar'}}]
[['node2:dfddc4ba-c3c5-6887-5012-a243b5b377c2'], {'subgraphNode2': {'foo': 'hi! foobar'}}]
[[], {'node2': {'foo': 'hi! foobar'}}]

:::注意，我们收到的不仅是节点更新，还有命名空间，它告诉我们正在从哪个图（或子图）进行流式传输。

:::python

检查点

使用 checkpoints 流式模式在图执行时接收检查点事件。每个检查点事件的格式与 get_state() 的输出相同。需要一个检查点存储器。

from langgraph.checkpoint.memory import MemorySaver

graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .add_edge("generate_joke", END)
    .compile(checkpointer=MemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

for chunk in graph.stream(
    {"topic": "ice cream"},
    config=config,
    stream_mode="checkpoints",
    version="v2",
):
    if chunk["type"] == "checkpoints":
        print(chunk["data"])

任务

使用 tasks 流式模式在图执行时接收任务开始和结束事件。任务事件包含有关哪个节点正在运行、其结果以及任何错误的信息。需要一个检查点存储器。

from langgraph.checkpoint.memory import MemorySaver

graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .add_edge("generate_joke", END)
    .compile(checkpointer=MemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

for chunk in graph.stream(
    {"topic": "ice cream"},
    config=config,
    stream_mode="tasks",
    version="v2",
):
    if chunk["type"] == "tasks":
        print(chunk["data"])

:::

调试

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

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="debug",
    version="v2",
):
    if chunk["type"] == "debug":
        print(chunk["data"])

::: :::js

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

::: :::python

debug 模式将 checkpoints 和 tasks 事件与额外的元数据结合在一起。如果你只需要调试信息的一个子集，请直接使用 checkpoints 或 tasks。

:::

同时使用多种模式

:::python 你可以将列表作为 stream_mode 参数传递，以同时流式传输多种模式。使用 version="v2" 时，每个数据块都是一个 StreamPart 字典。使用 chunk["type"] 来区分不同的模式：

for chunk in graph.stream(inputs, stream_mode=["updates", "custom"], version="v2"):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node `{node_name}` updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Custom event: {chunk['data']}")

for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
    print(chunk)

::: :::js 你可以将数组作为 streamMode 参数传递，以同时流式传输多种模式。流式输出将是 [mode, chunk] 的元组，其中 mode 是流模式的名称，chunk 是该模式流式传输的数据。

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

:::

高级用法

与任意 LLM 一起使用

:::python 你可以使用 stream_mode="custom" 从任何 LLM API 流式传输数据——即使该 API 没有实现 LangChain 聊天模型接口。这使你可以集成原始的 LLM 客户端或提供自己流式接口的外部服务，使得 LangGraph 对于自定义设置非常灵活。

from langgraph.config import get_stream_writer

def call_arbitrary_model(state):
    """调用任意模型并流式输出结果的示例节点"""
    # 获取流写入器以发送自定义数据
    writer = get_stream_writer()
    # 假设你有一个生成数据块的流式客户端
    # 使用你的自定义流式客户端生成 LLM 令牌
    for chunk in your_custom_streaming_client(state["topic"]):
        # 使用写入器向流发送自定义数据
        writer({"custom_llm_chunk": chunk})
    return {"result": "completed"}

graph = (
    StateGraph(State)
    .add_node(call_arbitrary_model)
    # 根据需要添加其他节点和边
    .compile()
)
# 设置 stream_mode="custom" 以在流中接收自定义数据
for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="custom",
    version="v2",
):
    if chunk["type"] == "custom":
        # 数据块将包含从 LLM 流式传输的自定义数据
        print(chunk["data"])

::: :::js 你可以使用 streamMode: "custom" 从任何 LLM API 流式传输数据——即使该 API 没有实现 LangChain 聊天模型接口。这使你可以集成原始的 LLM 客户端或提供自己流式接口的外部服务，使得 LangGraph 对于自定义设置非常灵活。

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 令牌
  for await (const chunk of yourCustomStreamingClient(state.topic)) {
    // 使用写入器向流发送自定义数据
    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);
}

:::

扩展示例：流式传输任意聊天模型

python

import operator
import json

from typing import TypedDict
from typing_extensions import Annotated
from langgraph.graph import StateGraph, START

from openai import AsyncOpenAI

openai_client = AsyncOpenAI()
model_name = "gpt-5.4-mini"


async def stream_tokens(model_name: str, messages: list[dict]):
    response = await openai_client.chat.completions.create(
        messages=messages, model=model_name, stream=True
    )
    role = None
    async for chunk in response:
        delta = chunk.choices[0].delta

        if delta.role is not None:
            role = delta.role

        if delta.content:
            yield {"role": role, "content": delta.content}


# 这是我们的工具
async def get_items(place: str) -> str:
    """使用此工具列出你被询问的地方可能找到的物品。"""
    writer = get_stream_writer()
    response = ""
    async for msg_chunk in stream_tokens(
        model_name,
        [
            {
                "role": "user",
                "content": (
                    "你能告诉我我在以下地方可能找到什么样的物品吗: '{place}'。 "
                    "列出至少3件这样的物品，用逗号分隔。 "
                    "并包括对每件物品的简要描述。"
                ),
            }
        ],
    ):
        response += msg_chunk["content"]
        writer(msg_chunk)

    return response


class State(TypedDict):
    messages: Annotated[list[dict], operator.add]


# 这是调用工具的图节点
async def call_tool(state: State):
    ai_message = state["messages"][-1]
    tool_call = ai_message["tool_calls"][-1]

    function_name = tool_call["function"]["name"]
    if function_name != "get_items":
        raise ValueError(f"Tool {function_name} not supported")

    function_arguments = tool_call["function"]["arguments"]
    arguments = json.loads(function_arguments)

    function_response = await get_items(**arguments)
    tool_message = {
        "tool_call_id": tool_call["id"],
        "role": "tool",
        "name": function_name,
        "content": function_response,
    }
    return {"messages": [tool_message]}


graph = (
    StateGraph(State)
    .add_node(call_tool)
    .add_edge(START, "call_tool")
    .compile()
)

让我们使用一个包含工具调用的 @[AIMessage] 来调用该图：

inputs = {
    "messages": [
        {
            "content": None,
            "role": "assistant",
            "tool_calls": [
                {
                    "id": "1",
                    "function": {
                        "arguments": '{"place":"bedroom"}',
                        "name": "get_items",
                    },
                    "type": "function",
                }
            ],
        }
    ]
}

async for chunk in graph.astream(
    inputs,
    stream_mode="custom",
    version="v2",
):
    if chunk["type"] == "custom":
        print(chunk["data"]["content"], end="|", flush=True)

:::

import { StateGraph, StateSchema, MessagesValue, GraphNode, START, LangGraphRunnableConfig } from "@langchain/langgraph";
import { tool } from "@langchain/core/tools";
import * as z from "zod";
import OpenAI from "openai";

const openaiClient = new OpenAI();
const modelName = "gpt-5.4-mini";

async function* streamTokens(modelName: string, messages: any[]) {
  const response = await openaiClient.chat.completions.create({
    messages,
    model: modelName,
    stream: true,
  });

  let role: string | null = null;
  for await (const chunk of response) {
    const delta = chunk.choices[0]?.delta;

    if (delta?.role) {
      role = delta.role;
    }

    if (delta?.content) {
      yield { role, content: delta.content };
    }
  }
}

// 这是我们的工具
const getItems = tool(
  async (input, config: LangGraphRunnableConfig) => {
    let response = "";
    for await (const msgChunk of streamTokens(
      modelName,
      [
        {
          role: "user",
          content: `你能告诉我我在以下地方可能找到什么样的物品吗: '${input.place}'. 列出至少3件这样的物品，用逗号分隔。并包括对每件物品的简要描述。`,
        },
      ]
    )) {
      response += msgChunk.content;
      config.writer?.(msgChunk);
    }
    return response;
  },
  {
    name: "get_items",
    description: "使用此工具列出你被询问的地方可能找到的物品。",
    schema: z.object({
      place: z.string().describe("要查询物品的地方。"),
    }),
  }
);

const State = new StateSchema({
  messages: MessagesValue,
});

const callTool: GraphNode<typeof State> = async (state) => {
  const aiMessage = state.messages.at(-1);
  const toolCall = aiMessage.tool_calls?.at(-1);

  const functionName = toolCall?.function?.name;
  if (functionName !== "get_items") {
    throw new Error(`Tool ${functionName} not supported`);
  }

  const functionArguments = toolCall?.function?.arguments;
  const args = JSON.parse(functionArguments);

  const functionResponse = await getItems.invoke(args);
  const toolMessage = {
    tool_call_id: toolCall.id,
    role: "tool",
    name: functionName,
    content: functionResponse,
  };
  return { messages: [toolMessage] };
};

const graph = new StateGraph(State)
  // 这是调用工具的图节点
  .addNode("callTool", callTool)
  .addEdge(START, "callTool")
  .compile();

让我们使用一个包含工具调用的 @[AIMessage] 来调用该图：

const inputs = {
  messages: [
    {
      content: null,
      role: "assistant",
      tool_calls: [
        {
          id: "1",
          function: {
            arguments: '{"place":"bedroom"}',
            name: "get_items",
          },
          type: "function",
        }
      ],
    }
  ]
};

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

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

如果你的应用程序混合使用了支持流式传输和不支持流式传输的模型，你可能需要明确地为不支持流式传输的模型禁用它。在初始化模型时设置 streaming: false。

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

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

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

---

<div className="source-links">
<Callout icon="terminal-2">
    [Connect these docs](/use-these-docs) to Claude, VSCode, and more via MCP for real-time answers.
</Callout>
<Callout icon="edit">
    [Edit this page on GitHub](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/streaming.mdx) or [file an issue](https://github.com/langchain-ai/docs/issues/new/choose).
</Callout>
</div>

Get started

Capabilities

Production

Frontend

LangGraph APIs

开始使用

基本用法

流模式

图状态

LLM tokens

按 LLM 调用过滤

从流中省略消息

按节点过滤

自定义数据

工具进度

定义流式传输进度的工具

在服务器端消费工具事件

在 React 中使用 `useStream` 显示工具进度

`tools` 与 `custom` 流模式

子图输出

检查点

任务

调试

同时使用多种模式

高级用法

与任意 LLM 一起使用

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

​开始使用

​基本用法

​流模式

​图状态

​LLM tokens

​按 LLM 调用过滤

​从流中省略消息

​按节点过滤

​自定义数据

​工具进度

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

​在服务器端消费工具事件

​在 React 中使用 useStream 显示工具进度

​tools 与 custom 流模式

​子图输出

​检查点

​任务

​调试

​同时使用多种模式

​高级用法

​与任意 LLM 一起使用

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

开始使用

基本用法

流模式

图状态

LLM tokens

按 LLM 调用过滤

从流中省略消息

按节点过滤

自定义数据

工具进度

定义流式传输进度的工具

在服务器端消费工具事件

在 React 中使用 `useStream` 显示工具进度

`tools` 与 `custom` 流模式

子图输出

检查点

任务

调试

同时使用多种模式

高级用法

与任意 LLM 一起使用

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