流式 - AI 开发中文文档

对于新应用，我们推荐事件流——在 LangGraph v1.2 中引入的 typed-projection API。事件流为每个投影（消息、值、子图、输出）提供单独的迭代器，因此您可以独立消费它们，而不是在 stream_mode 块上分支。

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

入门

基本用法

:::python LangGraph 图表公开了 @[stream][Pregel.stream]（同步）和 @[astream][Pregel.astream]（异步）方法来生成流式输出作为迭代器。将一个或多个流式模式传递给它们，以控制您接收的数据。

for chunk in graph.stream(
    {"topic": "ice cream"},
    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"Status: {chunk['data']['status']}")

Output

Status: thinking of a joke...
Node generate_joke updated: {'joke': 'Why did the ice cream go to school? To get a sundae education!'}

Full example

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.config import get_stream_writer


class State(TypedDict):
    topic: str
    joke: str


def generate_joke(state: State):
    writer = get_stream_writer()
    writer({"status": "thinking of a joke..."})
    return {"joke": f"Why did the {state['topic']} go to school? To get a sundae education!"}

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

for chunk in graph.stream(
    {"topic": "ice cream"},
    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"Status: {chunk['data']['status']}")

Output

Status: thinking of a joke...
Node generate_joke updated: {'joke': 'Why did the ice cream go to school? To get a sundae education!'}

::: :::js LangGraph 图表公开了 @[stream][Pregel.stream] 方法来生成流式输出作为迭代器。

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

:::

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

:::python

流式输出格式 (v2)

需要 LangGraph 1.1 或更高版本。本页面上的所有示例都使用 version="v2"。

将 version="v2" 传递给 stream() 或 astream() 以获取统一输出格式。每个块都是一个 StreamPart 字典，具有一致的形状——无论流式模式、模式数量或子图设置如何：

{
    "type": "values" | "updates" | "messages" | "custom" | "checkpoints" | "tasks" | "debug",
    "ns": (),           # namespace tuple, populated for subgraph events
    "data": ...,        # the actual payload (type varies by stream mode)
}

每个流式模式都有一个对应的 TypedDict，包含 @[ValuesStreamPart]、@[UpdatesStreamPart]、@[MessagesStreamPart]、@[CustomStreamPart]、@[CheckpointStreamPart]、@[TasksStreamPart]、@[DebugStreamPart]。您可以从 langgraph.types 导入这些类型。联合类型 @[StreamPart] 在 part["type"] 上是分离的联合，能够在编辑器和类型检查器中实现完整的类型缩小。在 v1（默认）中，输出格式根据您的流式选项变化（单模式返回原始数据，多模式返回 (mode, data) 元组，子图返回 (namespace, data) 元组）。在 v2 中，格式始终相同：

for chunk in graph.stream(inputs, stream_mode="updates", version="v2"):
    print(chunk["type"])  # "updates"
    print(chunk["ns"])    # ()
    print(chunk["data"])  # {"node_name": {"key": "value"}}

for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)  # {"node_name": {"key": "value"}}

v2 格式还启用了类型缩小，这意味着您可以按 chunk["type"] 过滤块并获取正确的有效负载类型。每个分支将 part["data"] 缩小为该模式的特定类型：

for part in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["values", "updates", "messages", "custom"],
    version="v2",
):
    if part["type"] == "values":
        # ValuesStreamPart — full state snapshot after each step
        print(f"State: topic={part['data']['topic']}")
    elif part["type"] == "updates":
        # UpdatesStreamPart — only the changed keys from each node
        for node_name, state in part["data"].items():
            print(f"Node `{node_name}` updated: {state}")
    elif part["type"] == "messages":
        # MessagesStreamPart — (message_chunk, metadata) from LLM calls
        msg, metadata = part["data"]
        print(msg.content, end="", flush=True)
    elif part["type"] == "custom":
        # CustomStreamPart — arbitrary data from get_stream_writer()
        print(f"Progress: {part['data']['progress']}%")

:::

流式模式

:::python 将以下一个或多个流式模式作为列表传递给 @[stream][CompiledStateGraph.stream] 或 @[astream][CompiledStateGraph.astream] 方法：

模式	类型	描述
values	@[`ValuesStreamPart`]	每个步骤后的完整状态。
updates	@[`UpdatesStreamPart`]	每个步骤后的状态更新。同一步骤中的多个更新将分别流式传输。
messages	@[`MessagesStreamPart`]	LLM 调用的 (LLM token, 元数据) 2 元组。
custom	@[`CustomStreamPart`]	通过 @[`get_stream_writer`] 从节点发出的自定义数据。
checkpoints	@[`CheckpointStreamPart`]	检查点事件（与 `get_state()` 相同格式）。需要检查点器。
tasks	@[`TasksStreamPart`]	需要检查点器的任务开始/完成事件，包含结果和错误。
debug	@[`DebugStreamPart`]	所有可用信息——结合 `checkpoints` 和 `tasks` 并带有额外元数据。
:::

:::js 将以下一个或多个流式模式作为列表传递给 @[stream][CompiledStateGraph.stream] 方法：

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

图状态

使用流式模式 updates 和 values 来流式传输图在执行过程中的状态。

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

:::python

from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
  topic: str
  joke: str


def refine_topic(state: State):
    return {"topic": state["topic"] + " and cats"}


def generate_joke(state: State):
    return {"joke": f"This is a joke about {state['topic']}"}

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()
)

::: :::js

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

使用此方法来流式传输节点在每一步后返回的 状态更新。流式输出的内容包括节点名称以及更新。:::python

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="updates",  
    version="v2",  
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node `{node_name}` updated: {state}")

Output

Node `refine_topic` updated: {'topic': 'ice cream and cats'}
Node `generate_joke` updated: {'joke': 'This is a joke about ice cream and cats'}

::::::js

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

:::

使用此方法来流式传输图的 完整状态。:::python

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

Output

topic: ice cream, joke:
topic: ice cream and cats, joke:
topic: ice cream and cats, joke: This is a joke about ice cream and cats

:::

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）输出。 :::python 从 messages 模式流式传输的输出是一个元组 (message_chunk, metadata)，其中：

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

如果您的 LLM 不可用为 LangChain 集成，您可以使用 custom 模式来流式传输其输出。有关详细信息，请参阅与任何 LLM 一起使用。

对于 Python < 3.11 的异步代码需要手动配置 当使用 Python < 3.11 与异步代码时，您必须显式地将 @[RunnableConfig] 传递给 ainvoke() 以启用正确的流式传输。有关详细信息，请参阅使用 Python < 3.11 的异步或升级到 Python 3.11+。

from dataclasses import dataclass

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START


@dataclass
class MyState:
    topic: str
    joke: str = ""


model = init_chat_model(model="gpt-5.4-mini")

def call_model(state: MyState):
    """Call the LLM to generate a joke about a topic"""
    # Note that message events are emitted even when the LLM is run using .invoke rather than .stream
    model_response = model.invoke(  
        [
            {"role": "user", "content": f"Generate a joke about {state.topic}"}
        ]
    )
    return {"joke": model_response.content}

graph = (
    StateGraph(MyState)
    .add_node(call_model)
    .add_edge(START, "call_model")
    .compile()
)

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        message_chunk, metadata = chunk["data"]
        if message_chunk.content:
            print(message_chunk.content, end="|", flush=True)

::: :::js 从 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) => {
  // Call the LLM to generate a joke about a topic
  // Note that message events are emitted even when the LLM is run using .invoke rather than .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();

// The "messages" stream mode returns an iterator of tuples [messageChunk, metadata]
// where messageChunk is the token streamed by the LLM and metadata is a dictionary
// with information about the graph node where the LLM was called and other information
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。 :::python

from langchain.chat_models import init_chat_model

# model_1 is tagged with "joke"
model_1 = init_chat_model(model="gpt-5.4-mini", tags=['joke'])
# model_2 is tagged with "poem"
model_2 = init_chat_model(model="gpt-5.4-mini", tags=['poem'])

graph = ... # define a graph that uses these LLMs

# The stream_mode is set to "messages" to stream LLM tokens
# The metadata contains information about the LLM invocation, including the tags
async for chunk in graph.astream(
    {"topic": "cats"},
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the tags field in the metadata to only include
        # the tokens from the LLM invocation with the "joke" tag
        if metadata["tags"] == ["joke"]:
            print(msg.content, end="|", flush=True)

::: :::js

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

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

const graph = // ... define a graph that uses these LLMs

// The streamMode is set to "messages" to stream LLM tokens
// The metadata contains information about the LLM invocation, including the tags
for await (const [msg, metadata] of await graph.stream(
  { topic: "cats" },
  { streamMode: "messages" }
)) {
  // Filter the streamed tokens by the tags field in the metadata to only include
  // the tokens from the LLM invocation with the "joke" tag
  if (metadata.tags?.includes("joke")) {
    console.log(msg.content + "|");
  }
}

:::

Extended example: filtering by tags

:::python

from typing import TypedDict

from langchain.chat_models import init_chat_model
from langgraph.graph import START, StateGraph

# The joke_model is tagged with "joke"
joke_model = init_chat_model(model="gpt-5.4-mini", tags=["joke"])
# The poem_model is tagged with "poem"
poem_model = init_chat_model(model="gpt-5.4-mini", tags=["poem"])


class State(TypedDict):
      topic: str
      joke: str
      poem: str


async def call_model(state, config):
      topic = state["topic"]
      print("Writing joke...")
      # Note: Passing the config through explicitly is required for python < 3.11
      # Since context var support wasn't added before then: https://docs.python.org/3/library/asyncio-task.html#creating-tasks
      # The config is passed through explicitly to ensure the context vars are propagated correctly
      # This is required for Python < 3.11 when using async code. Please see the async section for more details
      joke_response = await joke_model.ainvoke(
            [{"role": "user", "content": f"Write a joke about {topic}"}],
            config,
      )
      print("\n\nWriting poem...")
      poem_response = await poem_model.ainvoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}],
            config,
      )
      return {"joke": joke_response.content, "poem": poem_response.content}


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

# The stream_mode is set to "messages" to stream LLM tokens
# The metadata contains information about the LLM invocation, including the tags
async for chunk in graph.astream(
      {"topic": "cats"},
      stream_mode="messages",
      version="v2",
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        if metadata["tags"] == ["joke"]:
            print(msg.content, end="|", flush=True)

::::::js

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

// The jokeModel is tagged with "joke"
const jokeModel = new ChatOpenAI({
  model: "gpt-5.4-mini",
  tags: ["joke"]
});
// The poemModel is tagged with "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();

// The streamMode is set to "messages" to stream LLM tokens
// The metadata contains information about the LLM invocation, including the tags
for await (const [msg, metadata] of await graph.stream(
  { topic: "cats" },
  { streamMode: "messages" }
)) {
  // Filter the streamed tokens by the tags field in the metadata to only include
  // the tokens from the LLM invocation with the "joke" tag
  if (metadata.tags?.includes("joke")) {
    console.log(msg.content + "|");
  }
}

:::

忽略流中的消息

使用 nostream 标签完全排除 LLM 输出。带有 nostream 标签的调用仍然会运行并产生输出；它们的 Token 仅在 messages 模式下不会被发出。这很有用，当：

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

:::python

from typing import Any, TypedDict

from langchain_anthropic import ChatAnthropic
from langgraph.graph import START, StateGraph

stream_model = ChatAnthropic(model_name="claude-haiku-4-5-20251001")
internal_model = ChatAnthropic(model_name="claude-haiku-4-5-20251001").with_config(
    {"tags": ["nostream"]}
)


class State(TypedDict):
    topic: str
    answer: str
    notes: str


def answer(state: State) -> dict[str, Any]:
    r = stream_model.invoke(
        [{"role": "user", "content": f"Reply briefly about {state['topic']}"}]
    )
    return {"answer": r.content}


def internal_notes(state: State) -> dict[str, Any]:
    # Tokens from this model are omitted from stream_mode="messages" because of nostream
    r = internal_model.invoke(
        [{"role": "user", "content": f"Private notes on {state['topic']}"}]
    )
    return {"notes": r.content}


graph = (
    StateGraph(State)
    .add_node("write_answer", answer)
    .add_node("internal_notes", internal_notes)
    .add_edge(START, "write_answer")
    .add_edge("write_answer", "internal_notes")
    .compile()
)

initial_state: State = {"topic": "AI", "answer": "", "notes": ""}
stream = graph.stream_events(initial_state, version="v3")

View example trace

为此示例打开一个公开的 LangSmith 运行。

::: :::js

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: `Reply briefly about ${state.topic}` },
  ]);
  return { answer: contentToText(r.content) };
};

const internalNotes = async (state: typeof State.State) => {
  // Tokens from this model are omitted from streamMode: "messages" because of nostream
  const r = await internalModel.invoke([
    { role: "user", content: `Private notes on ${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" },
);

View example trace

为此示例打开一个公开的 LangSmith 运行。

:::

按节点筛选

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

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    inputs,
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the langgraph_node field in the metadata
        # to only include the tokens from the specified node
        if msg.content and metadata["langgraph_node"] == "some_node_name":
            ...

::: :::js

// The "messages" stream mode returns a tuple of [messageChunk, metadata]
// where messageChunk is the token streamed by the LLM and metadata is a dictionary
// with information about the graph node where the LLM was called and other information
for await (const [msg, metadata] of await graph.stream(
  inputs,
  { streamMode: "messages" }
)) {
  // Filter the streamed tokens by the langgraph_node field in the metadata
  // to only include the tokens from the specified node
  if (msg.content && metadata.langgraph_node === "some_node_name") {
    // ...
  }
}

:::

Extended example: streaming LLM tokens from specific nodes

:::python

from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-5.4-mini")


class State(TypedDict):
      topic: str
      joke: str
      poem: str


def write_joke(state: State):
      topic = state["topic"]
      joke_response = model.invoke(
            [{"role": "user", "content": f"Write a joke about {topic}"}]
      )
      return {"joke": joke_response.content}


def write_poem(state: State):
      topic = state["topic"]
      poem_response = model.invoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}]
      )
      return {"poem": poem_response.content}


graph = (
      StateGraph(State)
      .add_node(write_joke)
      .add_node(write_poem)
      # write both the joke and the poem concurrently
      .add_edge(START, "write_joke")
      .add_edge(START, "write_poem")
      .compile()
)

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the langgraph_node field in the metadata
        # to only include the tokens from the write_poem node
        if msg.content and metadata["langgraph_node"] == "write_poem":
            print(msg.content, end="|", flush=True)

::::::js

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)
  // write both the joke and the poem concurrently
  .addEdge(START, "writeJoke")
  .addEdge(START, "writePoem")
  .compile();

// The "messages" stream mode returns a tuple of [messageChunk, metadata]
// where messageChunk is the token streamed by the LLM and metadata is a dictionary
// with information about the graph node where the LLM was called and other information
for await (const [msg, metadata] of await graph.stream(
  { topic: "cats" },
  { streamMode: "messages" }
)) {
  // Filter the streamed tokens by the langgraph_node field in the metadata
  // to only include the tokens from the writePoem node
  if (msg.content && metadata.langgraph_node === "writePoem") {
    console.log(msg.content + "|");
  }
}

:::

自定义数据

:::python 要在 LangGraph 节点或工具内部发送 自定义用户定义数据，请按照以下步骤操作：

使用 @[get_stream_writer] 访问流式写入器并发出自定义数据。
在调用 .stream() 或 .astream() 时设置 stream_mode="custom" 以获取流中的自定义数据。您可以组合多种模式（例如，["updates", "custom"]），但至少必须有一个是 "custom"。

Python 版本低于 3.11 的异步代码中不能使用 @[get_stream_writer] 在 Python 版本低于 3.11 的异步代码中，@[get_stream_writer] 将无法工作。相反，向您的节点或工具添加一个 writer 参数，并手动传递它。参见 Python 版本低于 3.11 的异步使用示例获取使用示例。

node
tool

from typing import TypedDict
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START

class State(TypedDict):
    query: str
    answer: str

def node(state: State):
    # Get the stream writer to send custom data
    writer = get_stream_writer()
    # Emit a custom key-value pair (e.g., progress update)
    writer({"custom_key": "Generating custom data inside node"})
    return {"answer": "some data"}

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

inputs = {"query": "example"}

# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(inputs, stream_mode="custom", version="v2"):
    if chunk["type"] == "custom":
        print(f"Custom event: {chunk['data']['custom_key']}")

from langchain.tools import tool
from langgraph.config import get_stream_writer

@tool
def query_database(query: str) -> str:
    """Query the database."""
    # Access the stream writer to send custom data
    writer = get_stream_writer()  
    # Emit a custom key-value pair (e.g., progress update)
    writer({"data": "Retrieved 0/100 records", "type": "progress"})  
    # perform query
    # Emit another custom key-value pair
    writer({"data": "Retrieved 100/100 records", "type": "progress"})
    return "some-answer"


graph = ... # define a graph that uses this tool

# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(inputs, stream_mode="custom", version="v2"):
    if chunk["type"] == "custom":
        print(f"{chunk['data']['type']}: {chunk['data']['data']}")

::: :::js 要在 LangGraph 节点或工具内部发送自定义用户定义数据，请按照以下步骤操作：

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

node
tool

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) => {
  // Use the writer to emit a custom key-value pair (e.g., progress update)
  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" };

// Set streamMode: "custom" to receive the custom data in the stream
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) => {
    // Use the writer to emit a custom key-value pair (e.g., progress update)
    config.writer({ data: "Retrieved 0/100 records", type: "progress" });
    // perform query
    // Emit another custom key-value pair
    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 = // ... define a graph that uses this tool

// Set streamMode: "custom" to receive the custom data in the stream
for await (const chunk of await graph.stream(inputs, { streamMode: "custom" })) {
  console.log(chunk);
}

::: :::js

工具进度

使用 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]);

      // Each yield emits an on_tool_event to the stream
      yield {
        message: `Searching ${airlines[i]}...`,
        progress: (i + 1) / airlines.length,
        completed,
      };
    }

    // The return value becomes the tool result (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;
    }
  }
}

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

@langchain/langgraph-sdk/react 的 @[useStream] 钩子当您在流模式中包含 "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"],
  });

  // Filter for actively running 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} />
      ))}

      {/* Show progress cards for running tools */}
      {activeTools.map((tool) => (
        <ToolProgressCard
          key={tool.toolCallId ?? tool.name}
          name={tool.name}
          state={tool.state}
          data={tool.data}
        />
      ))}
    </div>
  );
}

Extended example: travel planning agent with tool progress

此示例展示了一个完整的智能体，其中包含异步生成器工具，将搜索进度流式传输到 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 钩子提供了开箱即用的反应式 toolProgress 数组。
custom—让您完全控制使用 config.writer() 发出什么数据以及何时发出。当您需要非结构化数据，它不映射到工具生命周期，或者当您希望从节点（而不仅仅是工具）流式传输时使用此模式。

:::

子图输出

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

v2 (LangGraph 1.1 or later)
v1 (default)

使用 version="v2"，子图事件使用相同的 StreamPart 格式。ns 字段标识源：

for chunk in graph.stream(
    {"foo": "foo"},
    subgraphs=True,  
    stream_mode="updates",
    version="v2", 
):
    print(chunk["type"])  # "updates"
    print(chunk["ns"])    # () for root, ("node_name:<task_id>",) for subgraph
    print(chunk["data"])  # {"node_name": {"key": "value"}}

for chunk in graph.stream(
    {"foo": "foo"},
    # Set subgraphs=True to stream outputs from subgraphs
    subgraphs=True,  
    stream_mode="updates",
):
    print(chunk)

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

for await (const chunk of await graph.stream(
  { foo: "foo" },
  {
    // Set subgraphs: true to stream outputs from subgraphs
    subgraphs: true,
    streamMode: "updates",
  }
)) {
  console.log(chunk);
}

::: :::python

这适用于每个 stream_mode，包括 "messages"。像 @[create_agent] 这样的智能体构建器返回一个 编译后的图，因此将一个作为节点添加，它就变成了子图。没有 subgraphs=True，父图上的 stream_mode="messages" 不会发出来自内部智能体 LLM 调用的 Token 段。直接调用 agent.stream(...) 会，这就是为什么这通常只在包装后出现。

from langchain.agents import create_agent
from langgraph.graph import END, START, StateGraph

graph = (
    StateGraph(State)
    .add_node("agent", create_agent(model, tools, state_schema=State))
    .add_edge(START, "agent")
    .add_edge("agent", END)
    .compile()
)

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "..."}]},
    stream_mode="messages",
    subgraphs=True,  
    version="v2",
):
    print(chunk["type"])  # "messages"
    print(chunk["ns"])    # () for root, ("agent:<task_id>",) for subgraph
    print(chunk["data"])  # (token, metadata)

::: :::js

这适用于每一个 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);
}

:::

Extended example: streaming from subgraphs

:::python

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

# Define subgraph
class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    bar: str

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    return {"foo": state["foo"] + state["bar"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

for chunk in graph.stream(
    {"foo": "foo"},
    stream_mode="updates",
    # Set subgraphs=True to stream outputs from subgraphs
    subgraphs=True,  
    version="v2",  
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            print(f"Subgraph {chunk['ns']}: {chunk['data']}")
        else:
            print(f"Root: {chunk['data']}")

::::::js

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

// Define subgraph
const SubgraphState = new StateSchema({
  foo: z.string(), // note that this key is shared with the parent graph state
  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();

// Define parent graph
const ParentState = new StateSchema({
  foo: z.string(),
});

const builder = new StateGraph(ParentState)
  .addNode("node1", (state) => {
    return { foo: "hi! " + state.foo };
  })
  .addNode("node2", subgraph)
  .addEdge(START, "node1")
  .addEdge("node1", "node2");
const graph = builder.compile();

for await (const chunk of await graph.stream(
  { foo: "foo" },
  {
    streamMode: "updates",
    // Set subgraphs: true to stream outputs from subgraphs
    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() 的输出格式相同。需要 checkpointer。

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 流式模式接收图执行过程中的任务开始和结束事件。任务事件包括关于正在运行的节点、其结果以及任何错误的信息。需要 checkpointer。

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):
    """Example node that calls an arbitrary model and streams the output"""
    # Get the stream writer to send custom data
    writer = get_stream_writer()  
    # Assume you have a streaming client that yields chunks
    # Generate LLM tokens using your custom streaming client
    for chunk in your_custom_streaming_client(state["topic"]):
        # Use the writer to send custom data to the stream
        writer({"custom_llm_chunk": chunk})  
    return {"result": "completed"}

graph = (
    StateGraph(State)
    .add_node(call_arbitrary_model)
    # Add other nodes and edges as needed
    .compile()
)
# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="custom",  
    version="v2",  
):
    if chunk["type"] == "custom":
        # The chunk data will contain the custom data streamed from the 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) => {
  // Example node that calls an arbitrary model and streams the output
  // Assume you have a streaming client that yields chunks
  // Generate LLM tokens using your custom streaming client
  for await (const chunk of yourCustomStreamingClient(state.topic)) {
    // Use the writer to send custom data to the stream
    config.writer({ custom_llm_chunk: chunk });
  }
  return { result: "completed" };
};

const graph = new StateGraph(State)
  .addNode("callArbitraryModel", callArbitraryModel)
  // Add other nodes and edges as needed
  .compile();

// Set streamMode: "custom" to receive the custom data in the stream
for await (const chunk of await graph.stream(
  { topic: "cats" },
  { streamMode: "custom" }
)) {
  // The chunk will contain the custom data streamed from the llm
  console.log(chunk);
}

:::

Extended example: streaming arbitrary chat model

:::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}


# this is our tool
async def get_items(place: str) -> str:
    """Use this tool to list items one might find in a place you're asked about."""
    writer = get_stream_writer()
    response = ""
    async for msg_chunk in stream_tokens(
        model_name,
        [
            {
                "role": "user",
                "content": (
                    "Can you tell me what kind of items "
                    f"i might find in the following place: '{place}'. "
                    "List at least 3 such items separating them by a comma. "
                    "And include a brief description of each item."
                ),
            }
        ],
    ):
        response += msg_chunk["content"]
        writer(msg_chunk)

    return response


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


# this is the tool-calling graph node
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)

::::::js

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

// this is our tool
const getItems = tool(
  async (input, config: LangGraphRunnableConfig) => {
    let response = "";
    for await (const msgChunk of streamTokens(
      modelName,
      [
        {
          role: "user",
          content: `Can you tell me what kind of items i might find in the following place: '${input.place}'. List at least 3 such items separating them by a comma. And include a brief description of each item.`,
        },
      ]
    )) {
      response += msgChunk.content;
      config.writer?.(msgChunk);
    }
    return response;
  },
  {
    name: "get_items",
    description: "Use this tool to list items one might find in a place you're asked about.",
    schema: z.object({
      place: z.string().describe("The place to look up items for."),
    }),
  }
);

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)
  // this is the tool-calling graph node
  .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 + "|");
}

:::

禁用特定对话模型的流式输出

如果您的应用程序混合了支持流式输出的模型和不支持的模型，您可能需要显式禁用不支持流式输出的模型的流式输出。 :::python 在初始化模型时设置 streaming=False。

init_chat_model
Chat model interface

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "claude-sonnet-4-6",
    # Set streaming=False to disable streaming for the chat model
    streaming=False
)

from langchain_openai import ChatOpenAI

# Set streaming=False to disable streaming for the chat model
model = ChatOpenAI(model="gpt-5.5", streaming=False)

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

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

const model = new ChatOpenAI({
  model: "gpt-5.5",
  // Set streaming: false to disable streaming for the chat model
  streaming: false,
});

::: :::python

并非所有对话模型集成都支持 streaming 参数。如果您的模型不支持它，请使用 disable_streaming=True 代替。此参数通过基类在所有对话模型中可用。

::: :::js

并非所有对话模型集成都支持 streaming 参数。如果您的模型不支持它，请使用 disableStreaming: true 代替。此参数通过基类在所有对话模型中可用。

::: :::python

迁移到 v2

v2 流式输出格式（本页全程使用）提供了一个统一的输出格式。以下是主要差异和迁移方法的总结：

场景	v1（默认）	v2（`version="v2"`）
单流模式	原始数据（dict）	`StreamPart` dict 包含 `type`, `ns`, `data`
多流模式	`(mode, data)` 元组	相同 `StreamPart` dict，在 `chunk["type"]` 上过滤
子图流式输出	`(namespace, data)` 元组	相同 `StreamPart` dict，检查 `chunk["ns"]`
多模式+子图	`(namespace, mode, data)` 三元组	相同 `StreamPart` dict
`invoke()` 返回类型	普通dict（状态）	`GraphOutput` 包含 `.value` 和 `.interrupts`
流中断位置	状态dict中的 `__interrupt__` 键	`interrupts` 字段在 `values` 流部分
调用中断位置	结果dict中的 `__interrupt__` 键	`.interrupts` 属性在 `GraphOutput`
Pydantic/数据类输出	返回普通dict	强制转换为模型/数据类实例

v2 调用格式

当您将 version="v2" 传递给 invoke() 或 ainvoke() 时，它将返回一个 @[GraphOutput] 对象，包含 .value 和 .interrupts 属性：

from langgraph.types import GraphOutput

result = graph.invoke(inputs, version="v2")

assert isinstance(result, GraphOutput)
result.value       # your output — dict, Pydantic model, or dataclass
result.interrupts  # tuple[Interrupt, ...], empty if none occurred

对于默认 "values" 之外的任何流模式，invoke(..., stream_mode="updates", version="v2") 返回 list[StreamPart] 而不是 list[tuple]。

GraphOutput（result["key"], "key" in result, result["__interrupt__"]）的字典式访问仍然适用于向后兼容，但已弃用，将在未来版本中移除。迁移到 result.value 和 result.interrupts。

这将状态与中断元数据分离。在 v1 中，中断嵌套在返回的 dict 下 __interrupt__：

config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(inputs, config=config, version="v2")

if result.interrupts:
    print(result.interrupts[0].value)
    graph.invoke(Command(resume=True), config=config, version="v2")

config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(inputs, config=config)

if "__interrupt__" in result:
    print(result["__interrupt__"][0].value)
    graph.invoke(Command(resume=True), config=config)

Pydantic 和数据类状态强制转换

当您的图状态是 Pydantic 模型或数据类时，v2 values 模式会自动将输出转换为正确的类型：

from pydantic import BaseModel
from typing import Annotated
import operator

class MyState(BaseModel):
    value: str
    items: Annotated[list[str], operator.add]

# With version="v2", chunk["data"] is a MyState instance
for chunk in graph.stream(
    {"value": "x", "items": []}, stream_mode="values", version="v2"
):
    print(type(chunk["data"]))  # <class 'MyState'>

Python < 3.11 时的异步

在 Python 版本 < 3.11 中，异步任务不支持 context 参数。这限制了 LangGraph 自动传播上下文的能力，并从两个关键方面影响 LangGraph 的流式机制：

您必须将 RunnableConfig 显式传递给异步 LLM 调用（例如，ainvoke()），因为回调不会自动传播。
您不能在异步节点或工具中使用 @[get_stream_writer]——您必须直接传递一个 writer 参数。

Extended example: async LLM call with manual config

from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain.chat_models import init_chat_model

model = init_chat_model(model="gpt-5.4-mini")

class State(TypedDict):
    topic: str
    joke: str

# Accept config as an argument in the async node function
async def call_model(state, config):
    topic = state["topic"]
    print("Generating joke...")
    # Pass config to model.ainvoke() to ensure proper context propagation
    joke_response = await model.ainvoke(  
        [{"role": "user", "content": f"Write a joke about {topic}"}],
        config,
    )
    return {"joke": joke_response.content}

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

# Set stream_mode="messages" to stream LLM tokens
async for chunk in graph.astream(
    {"topic": "ice cream"},
    stream_mode="messages",  
    version="v2",  
):
    if chunk["type"] == "messages":
        message_chunk, metadata = chunk["data"]
        if message_chunk.content:
            print(message_chunk.content, end="|", flush=True)

Extended example: async custom streaming with stream writer

from typing import TypedDict
from langgraph.types import StreamWriter

class State(TypedDict):
      topic: str
      joke: str

# Add writer as an argument in the function signature of the async node or tool
# LangGraph will automatically pass the stream writer to the function
async def generate_joke(state: State, writer: StreamWriter):  
      writer({"custom_key": "Streaming custom data while generating a joke"})
      return {"joke": f"This is a joke about {state['topic']}"}

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

# Set stream_mode="custom" to receive the custom data in the stream  
async for chunk in graph.astream(
      {"topic": "ice cream"},
      stream_mode="custom",
      version="v2",
):
      if chunk["type"] == "custom":
          print(chunk["data"])

:::

官方原文：src/oss/langgraph/streaming.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。

​入门

​基本用法

​流式输出格式 (v2)

​流式模式

​图状态

​LLM token

​根据 LLM 调用进行过滤

​忽略流中的消息

View example trace

View example trace

​按节点筛选

​自定义数据

​工具进度

​定义流式进度的工具

​服务器端消费工具事件

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

​tools 与 custom 流模式

​子图输出

​检查点

​任务

​调试

​同时使用多个模式

​高级

​与任何 LLM 一起使用

​禁用特定对话模型的流式输出

​迁移到 v2

​v2 调用格式

​Pydantic 和数据类状态强制转换

​Python < 3.11 时的异步

入门

基本用法

流式输出格式 (v2)

流式模式

图状态

LLM token

根据 LLM 调用进行过滤

忽略流中的消息

按节点筛选

自定义数据

工具进度

定义流式进度的工具

服务器端消费工具事件

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

`tools` 与 `custom` 流模式

子图输出

检查点

任务

调试

同时使用多个模式

高级

与任何 LLM 一起使用

禁用特定对话模型的流式输出

迁移到 v2

v2 调用格式

Pydantic 和数据类状态强制转换

Python < 3.11 时的异步