流式传输 (Streaming)

流式传输来自 Agent 运行的实时更新

TIP

对于新应用，我们推荐使用事件流式传输 (Event streaming)——这是 LangChain v1.3 中引入的类型化投影（typed-projection）API。事件流式传输为每个投影（消息、状态值、工具调用、子图）提供单独的迭代器，因此你可以独立消费它们，而无需基于 stream_mode 数据块进行条件分支判断。

LangChain 实现了一套流式传输系统来展示实时更新。

流式传输对于提升基于 LLM 构建的应用的响应速度至关重要。通过渐进式展示输出（甚至在完整响应准备就绪之前），流式传输能够显著改善用户体验（UX），尤其是在应对 LLM 的固有延迟时。

概述

LangChain 的流式传输系统允许你将来自 Agent 运行的实时反馈呈现在应用程序中。

通过 LangChain 流式传输可以实现：

• 流式传输 Agent 进度——在每个 Agent 步骤之后获取状态更新。
• 流式传输 LLM Token——在语言模型生成 Token 时对其进行流式传输。
• 流式传输思考 / 推理 Token——在模型生成推理过程时将其呈现出来。
• 流式传输自定义更新——发出用户自定义的信号（例如 "Fetched 10/100 records"）。
• 流式传输多种模式——可从 updates（Agent 进度）、messages（LLM Token + 元数据）或 custom（任意用户数据）中进行选择。

有关其他端到端示例，请参阅下方的常见模式部分。

支持的流式模式

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

模式	描述
updates	在每个 Agent 步骤之后流式传输状态更新。如果在同一步骤中进行了多次更新（例如运行了多个节点），这些更新将分别流式传输。
messages	从调用 LLM 的任何图节点流式传输 (token, metadata) 元组。
custom	使用 stream writer 从图节点内部流式传输自定义数据。

Agent 进度

要流式传输 Agent 进度，请使用带有 streamMode: "updates" 的 stream 方法。这会在每个 Agent 步骤之后发出一个事件。

例如，如果你有一个调用了一次工具的 Agent，你应该会看到以下更新：

• LLM 节点：带有工具调用请求的 AIMessage
• 工具节点：带有执行结果的 ToolMessage
• LLM 节点：最终的 AI 响应

通过 configurable 传递 thread_id，以便对话被设置检查点，后续轮次可以恢复相同的历史记录。thread_id 与 streamMode 相互独立；你还可以在其旁边传递 context，用于工具从 runtime.context 读取每次运行的数据。

Google

import { createAgent, tool } from "langchain";
import { uuid7 } from "langsmith";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";

const getWeather = tool(
  async ({ city }) => {
    return `The weather in ${city} is always sunny!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({
      city: z.string(),
    }),
  },
);

const agent = createAgent({
  model: "google:gemini-3.6-flash",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});

const config = { configurable: { thread_id: uuid7() } };

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "what is the weather in sf" }] },
  { ...config, version: "v3" },
);
await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const token of message.text) {
        process.stdout.write(token);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(`\nTool call: ${call.name}(${JSON.stringify(call.input)})`);
      console.log(`Tool result: ${await call.output}`);
    }
  })(),
]);

const finalState = await stream.output;
// Tool call: get_weather({"city":"San Francisco"})
// Tool result: [object ToolMessage]
// According to the data I have, the weather in San Francisco is always sunny! Would you like current conditions or a short forecast for today or the next few days?

OpenAI

import { createAgent, tool } from "langchain";
import { uuid7 } from "langsmith";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";

const getWeather = tool(
  async ({ city }) => {
    return `The weather in ${city} is always sunny!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({
      city: z.string(),
    }),
  },
);

const agent = createAgent({
  model: "openai:gpt-5.5",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});

const config = { configurable: { thread_id: uuid7() } };

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "what is the weather in sf" }] },
  { ...config, version: "v3" },
);
await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const token of message.text) {
        process.stdout.write(token);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(`\nTool call: ${call.name}(${JSON.stringify(call.input)})`);
      console.log(`Tool result: ${await call.output}`);
    }
  })(),
]);

const finalState = await stream.output;
// Tool call: get_weather({"city":"San Francisco"})
// Tool result: [object ToolMessage]
// According to the data I have, the weather in San Francisco is always sunny! Would you like current conditions or a short forecast for today or the next few days?

Anthropic

import { createAgent, tool } from "langchain";
import { uuid7 } from "langsmith";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";

const getWeather = tool(
  async ({ city }) => {
    return `The weather in ${city} is always sunny!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({
      city: z.string(),
    }),
  },
);

const agent = createAgent({
  model: "anthropic:claude-sonnet-5",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});

const config = { configurable: { thread_id: uuid7() } };

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "what is the weather in sf" }] },
  { ...config, version: "v3" },
);
await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const token of message.text) {
        process.stdout.write(token);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(`\nTool call: ${call.name}(${JSON.stringify(call.input)})`);
      console.log(`Tool result: ${await call.output}`);
    }
  })(),
]);

const finalState = await stream.output;
// Tool call: get_weather({"city":"San Francisco"})
// Tool result: [object ToolMessage]
// According to the data I have, the weather in San Francisco is always sunny! Would you like current conditions or a short forecast for today or the next few days?

OpenRouter

import { createAgent, tool } from "langchain";
import { uuid7 } from "langsmith";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";

const getWeather = tool(
  async ({ city }) => {
    return `The weather in ${city} is always sunny!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({
      city: z.string(),
    }),
  },
);

const agent = createAgent({
  model: "openrouter:z-ai/glm-5.2",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});

const config = { configurable: { thread_id: uuid7() } };

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "what is the weather in sf" }] },
  { ...config, version: "v3" },
);
await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const token of message.text) {
        process.stdout.write(token);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(`\nTool call: ${call.name}(${JSON.stringify(call.input)})`);
      console.log(`Tool result: ${await call.output}`);
    }
  })(),
]);

const finalState = await stream.output;
// Tool call: get_weather({"city":"San Francisco"})
// Tool result: [object ToolMessage]
// According to the data I have, the weather in San Francisco is always sunny! Would you like current conditions or a short forecast for today or the next few days?

Fireworks

import { createAgent, tool } from "langchain";
import { uuid7 } from "langsmith";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";

const getWeather = tool(
  async ({ city }) => {
    return `The weather in ${city} is always sunny!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({
      city: z.string(),
    }),
  },
);

const agent = createAgent({
  model: "fireworks:accounts/fireworks/models/glm-5p2",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});

const config = { configurable: { thread_id: uuid7() } };

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "what is the weather in sf" }] },
  { ...config, version: "v3" },
);
await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const token of message.text) {
        process.stdout.write(token);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(`\nTool call: ${call.name}(${JSON.stringify(call.input)})`);
      console.log(`Tool result: ${await call.output}`);
    }
  })(),
]);

const finalState = await stream.output;
// Tool call: get_weather({"city":"San Francisco"})
// Tool result: [object ToolMessage]
// According to the data I have, the weather in San Francisco is always sunny! Would you like current conditions or a short forecast for today or the next few days?

Baseten

import { createAgent, tool } from "langchain";
import { uuid7 } from "langsmith";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";

const getWeather = tool(
  async ({ city }) => {
    return `The weather in ${city} is always sunny!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({
      city: z.string(),
    }),
  },
);

const agent = createAgent({
  model: "baseten:zai-org/GLM-5.2",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});

const config = { configurable: { thread_id: uuid7() } };

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "what is the weather in sf" }] },
  { ...config, version: "v3" },
);
await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const token of message.text) {
        process.stdout.write(token);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(`\nTool call: ${call.name}(${JSON.stringify(call.input)})`);
      console.log(`Tool result: ${await call.output}`);
    }
  })(),
]);

const finalState = await stream.output;
// Tool call: get_weather({"city":"San Francisco"})
// Tool result: [object ToolMessage]
// According to the data I have, the weather in San Francisco is always sunny! Would you like current conditions or a short forecast for today or the next few days?

Ollama

import { createAgent, tool } from "langchain";
import { uuid7 } from "langsmith";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";

const getWeather = tool(
  async ({ city }) => {
    return `The weather in ${city} is always sunny!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({
      city: z.string(),
    }),
  },
);

const agent = createAgent({
  model: "ollama:north-mini-code-1.0",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});

const config = { configurable: { thread_id: uuid7() } };

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "what is the weather in sf" }] },
  { ...config, version: "v3" },
);
await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const token of message.text) {
        process.stdout.write(token);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(`\nTool call: ${call.name}(${JSON.stringify(call.input)})`);
      console.log(`Tool result: ${await call.output}`);
    }
  })(),
]);

const finalState = await stream.output;
// Tool call: get_weather({"city":"San Francisco"})
// Tool result: [object ToolMessage]
// According to the data I have, the weather in San Francisco is always sunny! Would you like current conditions or a short forecast for today or the next few days?

查看示例 Trace

打开此示例的公开 LangSmith 运行记录。

NOTE

使用 thread_id 持久化对话历史记录需要为 Agent 配置一个 checkpointer。在 LangSmith 部署中，checkpointer 会自动预配。在本地，请显式传递一个，例如 createAgent({ ..., checkpointer: new MemorySaver() })。本页其余代码片段为简洁起见省略了 thread_id，但你在生产环境中应当传递它。

LLM Token

要在大语言模型（LLM）生成 Token 时实时对其进行流式传输，请使用 streamMode: "messages"：

import z from "zod";
import { createAgent, tool } from "langchain";

const getWeather = tool(
    async ({ city }) => {
        return `The weather in ${city} is always sunny!`;
    },
    {
        name: "get_weather",
        description: "Get weather for a given city.",
        schema: z.object({
        city: z.string(),
        }),
    }
);

const agent = createAgent({
    model: "gpt-5.4-mini",
    tools: [getWeather],
});

for await (const [token, metadata] of await agent.stream(
    { messages: [{ role: "user", content: "what is the weather in sf" }] },
    { streamMode: "messages" }
)) {
    console.log(`node: ${metadata.langgraph_node}`);
    console.log(`content: ${JSON.stringify(token.contentBlocks, null, 2)}`);
}

NOTE

将 Agent 作为节点包装在父级 StateGraph 中？ createAgent 返回一个 ReactAgent 包装器；在将其作为节点添加时请传递 agent.graph。同时使用 subgraphs: true，以使消息块包含子图命名空间。请参阅子图输出 (Subgraph outputs)。

自定义更新

要在工具执行时流式传输来自工具的更新，可以使用配置中的 writer 参数。

import z from "zod";
import { tool, createAgent } from "langchain";
import { LangGraphRunnableConfig } from "@langchain/langgraph";

const getWeather = tool(
    async (input, config: LangGraphRunnableConfig) => {
        // 流式传输任意数据
        config.writer?.(`Looking up data for city: ${input.city}`);
        // ... 获取城市数据
        config.writer?.(`Acquired data for city: ${input.city}`);
        return `It's always sunny in ${input.city}!`;
    },
    {
        name: "get_weather",
        description: "Get weather for a given city.",
        schema: z.object({
        city: z.string().describe("The city to get weather for."),
        }),
    }
);

const agent = createAgent({
    model: "gpt-5.4-mini",
    tools: [getWeather],
});

for await (const chunk of await agent.stream(
    { messages: [{ role: "user", content: "what is the weather in sf" }] },
    { streamMode: "custom" }
)) {
    console.log(chunk);
}

title="Output"

Looking up data for city: San Francisco
Acquired data for city: San Francisco

NOTE

如果你向工具添加了 writer 参数，那么在不提供 writer 函数的情况下，你将无法在 LangGraph 执行上下文之外调用该工具。

流式传输多种模式

你可以通过将 streamMode 作为数组传递来指定多种流式传输模式：streamMode: ["updates", "messages", "custom"]。

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

import z from "zod";
import { tool, createAgent } from "langchain";
import { LangGraphRunnableConfig } from "@langchain/langgraph";

const getWeather = tool(
    async (input, config: LangGraphRunnableConfig) => {
        // 流式传输任意数据
        config.writer?.(`Looking up data for city: ${input.city}`);
        // ... 获取城市数据
        config.writer?.(`Acquired data for city: ${input.city}`);
        return `It's always sunny in ${input.city}!`;
    },
    {
        name: "get_weather",
        description: "Get weather for a given city.",
        schema: z.object({
        city: z.string().describe("The city to get weather for."),
        }),
    }
);

const agent = createAgent({
    model: "gpt-5.4-mini",
    tools: [getWeather],
});

for await (const [streamMode, chunk] of await agent.stream(
    { messages: [{ role: "user", content: "what is the weather in sf" }] },
    { streamMode: ["updates", "messages", "custom"] }
)) {
    console.log(`${streamMode}: ${JSON.stringify(chunk, null, 2)}`);
}

常见模式

以下示例展示了流式传输的常见用例。

流式传输思考 / 推理 Token

某些模型在生成最终答案之前会执行内部推理。你可以通过在标准内容块 (Standard content blocks)中筛选 type 为 "reasoning" 的内容块，在这些思考 / 推理 Token 生成时对其进行流式传输。

NOTE

必须在模型上启用推理输出。

有关配置详情，请参阅推理部分以及你的提供商集成页面。

要快速检查模型的推理支持情况，请参阅 models.dev。

要从 Agent 中流式传输思考 Token，请使用 streamMode: "messages" 并筛选推理内容块。在模型支持的情况下，使用启用了扩展思考（extended thinking）的模型实例（例如 ChatAnthropic）：

import z from "zod";
import { createAgent, tool } from "langchain";
import { ChatAnthropic } from "@langchain/anthropic";

const getWeather = tool(
  async ({ city }) => {
    return `It's always sunny in ${city}!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({ city: z.string() }),
  },
);

const agent = createAgent({
  model: new ChatAnthropic({
    model: "claude-sonnet-4-6",
    thinking: { type: "enabled", budget_tokens: 5000 },
  }),
  tools: [getWeather],
});

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "What is the weather in SF?" }] },
  { version: "v3" }, 
);
for await (const message of stream.messages) {
  for await (const token of message.reasoning) {
    process.stdout.write(`[thinking] ${token}`);
  }
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

查看示例 Trace

打开此示例的公开 LangSmith 运行记录。

title="Output"

[thinking] The user is asking about the weather in San Francisco. I have a tool
[thinking]  available to get this information. Let me call the get_weather tool
[thinking]  with "San Francisco" as the city parameter.
The weather in San Francisco is: It's always sunny in San Francisco!

无论使用哪种模型提供商，这种方式的工作原理都是相同的——LangChain 通过 content_blocks 属性将各提供商专属的格式（Anthropic 的 thinking 块、OpenAI 的 reasoning 摘要等）统一规范化为标准的 "reasoning" 内容块类型。

要直接从对话模型（不使用 Agent）流式传输推理 Token，请参阅使用对话模型进行流式传输。

禁用流式传输

在某些应用中，你可能需要禁用特定模型的独立 Token 流式传输。这在以下场景中非常有用：

• 在多 Agent (Multi-agent) 系统中工作，以控制哪些 Agent 流式传输其输出
• 将支持流式传输的模型与不支持的模型混合使用
• 部署到 LangSmith 并希望阻止某些模型的输出被流式传输到客户端

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

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

const model = new ChatOpenAI({
  model: "gpt-5.5",
  streaming: false,  
});

TIP

部署到 LangSmith 时，在任何你不希望将其输出流式传输到客户端的模型上设置 streaming=False。这是在部署前在你的图代码中配置的。

NOTE

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

更多详情请参阅 LangGraph 流式传输指南。

相关资源

• 前端流式传输 (Frontend streaming)——使用 useStream 构建 React UI 以进行实时 Agent 交互
• 对话模型流式传输 (Streaming with chat models)——无需使用 Agent 或图，直接从对话模型流式传输 Token
• 对话模型推理 (Reasoning with chat models)——配置并获取来自对话模型的推理输出
• 标准内容块 (Standard content blocks)——了解用于推理、文本和其他内容类型的规范化内容块格式
• 人机协同流式传输 (Streaming with human-in-the-loop)——在流式传输 Agent 进度的同时处理人工审核中断
• LangGraph 流式传输 (LangGraph streaming)——高级流式传输选项，包括 values、debug 模式和子图流式传输