以编程方式使用文档

本页面介绍了 LangGraph 的流式输出模式 API。它通过以下流式输出模式公开图执行： updates, values, messages, custom, checkpoints, tasks和 debug。当您需要直接访问图运行时事件或特定的流式输出模式输出时使用它。

快速开始

基本用法

LangGraph 图公开了 stream 方法以迭代器形式产生流式输出。

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

流式输出模式

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

模式描述
values每步后的完整状态。
updates每步后的状态更新。同一步骤中的多个更新会单独流式传输。
messagesLLM调用的2元组（LLM token，元数据）。
custom通过 writer config参数从节点发出的自定义数据。
tools工具调用生命周期事件（on_tool_start, on_tool_event, on_tool_end, on_tool_error).
debug图形执行过程中的所有可用信息。

图形状态

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

  • * updates 流式传输 **updates** 到图形每步之后的状态。
  • * values 流式传输 **完整值** ，即图形每步之后的状态。
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

使用此选项仅流式传输 **state updates** ，即每步后节点返回的状态更新。流式输出包括节点名称和更新内容。

    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

使用此选项流式传输 **完整状态** ，即图形每步之后的状态。

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

LLM tokens

使用 messages 流式模式来流式传输大语言模型（LLM）输出 **逐token** ，包括图形中的节点、工具、子图形或任务。

的流式输出来自 messages 模式 是一个元组 [message_chunk, metadata] where:

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

> 如果你的 LLM 没有 LangChain 集成版本，你可以通过以下方式流式传输其输出 custom 模式。参见 与任何 LLM 配合使用 查看详情。

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。

// 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

  // 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 mode.

这在以下情况下很有用：

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

按节点筛选

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

// 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

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

自定义数据

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

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

node

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

tool

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

工具进度

使用 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 值被用作工具的最终结果。

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

在服务端消费工具事件

传递 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

中的 useStream 钩子提供 @langchain/langgraph-sdk/react 暴露了一个 toolProgress 数组当你包含 "tools" 在流模式中。每个条目是一个 ToolProgress 对象，用于跟踪正在运行的工具的当前状态：

字段描述
name工具名称
state当前生命周期状态： "starting", "running", "completed", or "error"
toolCallId来自 LLM 的工具调用 ID
input工具的输入参数
data最近从 on_tool_event
result最终结果，在 on_tool_end
error错误，在 on_tool_error
 
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 (

      {stream.messages.map((msg) => (

      ))}


      {activeTools.map((tool) => (

      ))}

  );
}

Extended example: travel planning agent with tool progress

这个示例展示了一个完整的代理，包含流式搜索进度到 React UI 的异步生成器工具。

代理定义：

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

  model: new ChatOpenAI({ model: "gpt-4o-mini" }),
  tools: [searchFlights, checkHotels],
  checkpointer: new MemorySaver(),
});

带进度卡的 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 (

      {stream.messages.map((msg) => (
        {msg.content}
      ))}

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

        return (

            <strong>{tool.name}</strong>
            {data?.message && <p>{data.message}</p>}
            {data?.progress != null && (



            )}
            {data?.completed?.map((step, i) => (
              &#10003; {step}
            ))}

        );
      })}

  );
}

tools vs custom 流模式

两种流模式都可以显示工具进度，但它们用途不同：

  • * **tools**—自动发出结构化的生命周期事件（on_tool_start, on_tool_event, on_tool_end, on_tool_error），你的工具无需修改代码，只需使用 async function*。该 useStream hook 提供响应式 toolProgress 数组开箱即用。
  • * **custom**—让您完全控制发出的数据以及使用 config.writer()的时间。当您需要不映射到工具生命周期的自由格式数据，或想从节点（不仅仅是工具）流式传输时使用。

子图输出

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

输出将作为元组流式传输 [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);
}

Extended example: streaming from subgraphs

  // 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);
  }
  
  [[], {'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'}}]

注意 我们不仅接收节点更新，还接收命名空间，这些命名空间告诉我们正在从哪个图（或子图）流式传输。

<a id="debug"></a> ### 调试

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

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

同时使用多个模式

您可以传递一个数组作为 streamMode 参数来一次流式传输多个模式。

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

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

高级

适用于任何 LLM

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

这使您可以集成原始 LLM 客户端或提供自己流式接口的外部服务，使 LangGraph 在自定义设置中具有极高的灵活性。

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

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

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

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

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

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