流式处理

对于新应用，我们推荐使用事件流——这是在 LangChain v1.3 中引入的类型化投影 API。事件流为每个投影（消息、值、工具调用、子图）提供独立的迭代器，因此您可以独立消费它们，而不是根据 stream_mode 数据块进行分支处理。

LangChain 实现了一个流式处理系统，用于实时反馈更新。流式处理对于提升基于大语言模型构建的应用的响应能力至关重要。通过逐步显示输出，即使在完整响应准备就绪之前，流式处理也能显著改善用户体验（UX），尤其是在处理大语言模型延迟的情况下。

概述

LangChain 的流式处理系统让您可以将智能体运行的实时反馈呈现在您的应用中。 LangChain 流式处理可以实现以下功能：

流式传输智能体进度—在每个智能体步骤后获取状态更新。
流式传输大语言模型 token—在语言模型生成 token 时进行流式传输。
流式传输思考/推理 token—实时呈现模型推理过程。
流式传输自定义更新—发出用户定义的信号（例如，"Fetched 10/100 records"）。
多模式流式传输—可选择 updates（智能体进度）、messages（大语言模型 token + 元数据）或 custom（任意用户数据）。

更多端到端示例请参阅下方的常见模式部分。

支持的流模式

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

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

智能体进度

要流式传输智能体进度，请使用带 streamMode: "updates" 的 stream 方法。这会在每个智能体步骤后发出一个事件。例如，如果您有一个调用一次工具的智能体，您应该会看到以下更新：

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

通过 configurable 传递一个 thread_id，以便对会话进行检查点保存，后续轮次可以恢复相同的对话历史。thread_id 独立于 streamMode；您也可以在传递 thread_id 的同时传递 context，用于工具从 runtime.context 读取的每次运行数据。

import { createAgent, tool } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";
    
const getWeather = tool(
  async ({ city }) => {
    return `${city}的天气总是晴朗！`;
  },
  {
    name: "get_weather",
    description: "获取给定城市的天气。",
    schema: z.object({
      city: z.string(),
    }),
  },
);
    
const agent = createAgent({
  model: "google-genai:gemini-3.6-flash",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});
    
const config = { configurable: { thread_id: crypto.randomUUID() } };
    
const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "旧金山的天气怎么样" }] },
  { ...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(`\n工具调用：${call.name}(${JSON.stringify(call.input)})`);
      console.log(`工具结果：${await call.output}`);
    }
  })(),
]);
    
const finalState = await stream.output;
// 工具调用：get_weather({"city":"旧金山"})
// 工具结果：[object ToolMessage]
// 根据我掌握的数据，旧金山的天气总是晴朗！您想了解当前状况还是今天或未来几天的简短预报？

import { createAgent, tool } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";
    
const getWeather = tool(
  async ({ city }) => {
    return `${city}的天气总是晴朗！`;
  },
  {
    name: "get_weather",
    description: "获取给定城市的天气。",
    schema: z.object({
      city: z.string(),
    }),
  },
);
    
const agent = createAgent({
  model: "openai:gpt-5.5",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});
    
const config = { configurable: { thread_id: crypto.randomUUID() } };
    
const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "旧金山的天气怎么样" }] },
  { ...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(`\n工具调用：${call.name}(${JSON.stringify(call.input)})`);
      console.log(`工具结果：${await call.output}`);
    }
  })(),
]);
    
const finalState = await stream.output;
// 工具调用：get_weather({"city":"旧金山"})
// 工具结果：[object ToolMessage]
// 根据我掌握的数据，旧金山的天气总是晴朗！您想了解当前状况还是今天或未来几天的简短预报？

import { createAgent, tool } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";
    
const getWeather = tool(
  async ({ city }) => {
    return `${city}的天气总是晴朗！`;
  },
  {
    name: "get_weather",
    description: "获取给定城市的天气。",
    schema: z.object({
      city: z.string(),
    }),
  },
);
    
const agent = createAgent({
  model: "anthropic:claude-sonnet-4-6",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});
    
const config = { configurable: { thread_id: crypto.randomUUID() } };
    
const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "旧金山的天气怎么样" }] },
  { ...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(`\n工具调用：${call.name}(${JSON.stringify(call.input)})`);
      console.log(`工具结果：${await call.output}`);
    }
  })(),
]);
    
const finalState = await stream.output;
// 工具调用：get_weather({"city":"旧金山"})
// 工具结果：[object ToolMessage]
// 根据我掌握的数据，旧金山的天气总是晴朗！您想了解当前状况还是今天或未来几天的简短预报？

import { createAgent, tool } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";
    
const getWeather = tool(
  async ({ city }) => {
    return `${city}的天气总是晴朗！`;
  },
  {
    name: "get_weather",
    description: "获取给定城市的天气。",
    schema: z.object({
      city: z.string(),
    }),
  },
);
    
const agent = createAgent({
  model: "openrouter:openrouter:z-ai/glm-5.2",
  tools: [getWeather],
  checkpointer: new MemorySaver(),
});
    
const config = { configurable: { thread_id: crypto.randomUUID() } };
    
const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "旧金山的天气怎么样" }] },
  { ...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(`\n工具调用：${call.name}(${JSON.stringify(call.input)})`);
      console.log(`工具结果：${await call.output}`);
    }
  })(),
]);
    
const finalState = await stream.output;
// 工具调用：get_weather({"city":"旧金山"})
// 工具结果：[object ToolMessage]
// 根据我掌握的数据，旧金山的天气总是晴朗！您想了解当前状况还是今天或未来几天的简短预报？

import { createAgent, tool } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";
    
const getWeather = tool(
  async ({ city }) => {
    return `${city}的天气总是晴朗！`;
  },
  {
    name: "get_weather",
    description: "获取给定城市的天气。",
    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: crypto.randomUUID() } };
    
const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "旧金山的天气怎么样" }] },
  { ...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(`\n工具调用：${call.name}(${JSON.stringify(call.input)})`);
      console.log(`工具结果：${await call.output}`);
    }
  })(),
]);
    
const finalState = await stream.output;
// 工具调用：get_weather({"city":"旧金山"})
// 工具结果：[object ToolMessage]
// 根据我掌握的数据，旧金山的天气总是晴朗！您想了解当前状况还是今天或未来几天的简短预报？

import { createAgent, tool } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";
    
const getWeather = tool(
  async ({ city }) => {
    return `${city}的天气总是晴朗！`;
  },
  {
    name: "get_weather",
    description: "获取给定城市的天气。",
    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: crypto.randomUUID() } };
    
const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "旧金山的天气怎么样" }] },
  { ...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(`\n工具调用：${call.name}(${JSON.stringify(call.input)})`);
      console.log(`工具结果：${await call.output}`);
    }
  })(),
]);
    
const finalState = await stream.output;
// 工具调用：get_weather({"city":"旧金山"})
// 工具结果：[object ToolMessage]
// 根据我掌握的数据，旧金山的天气总是晴朗！您想了解当前状况还是今天或未来几天的简短预报？

import { createAgent, tool } from "langchain";
import { MemorySaver } from "@langchain/langgraph";
import z from "zod";
    
const getWeather = tool(
  async ({ city }) => {
    return `${city}的天气总是晴朗！`;
  },
  {
    name: "get_weather",
    description: "获取给定城市的天气。",
    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: crypto.randomUUID() } };
    
const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "旧金山的天气怎么样" }] },
  { ...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(`\n工具调用：${call.name}(${JSON.stringify(call.input)})`);
      console.log(`工具结果：${await call.output}`);
    }
  })(),
]);
    
const finalState = await stream.output;
// 工具调用：get_weather({"city":"旧金山"})
// 工具结果：[object ToolMessage]
// 根据我掌握的数据，旧金山的天气总是晴朗！您想了解当前状况还是今天或未来几天的简短预报？

使用 thread_id 持久化会话历史需要为智能体配置一个检查点存储器。在 LangSmith 部署上，检查点存储器是自动配置的。在本地运行时，需要显式传递一个，例如 createAgent({ ..., checkpointer: new MemorySaver() })。为简洁起见，本页剩余的代码片段省略了 thread_id，但在生产环境中您应该传递它。

大语言模型 token

要流式传输大语言模型生成时的 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: "获取给定城市的天气。",
        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)}`);
}

将智能体作为节点封装在父 StateGraph 中？ createAgent 返回一个 ReactAgent 包装器；添加为节点时请使用 agent.graph。使用 subgraphs: true，以便消息块包含子图命名空间。参见子图输出。

自定义更新

要在工具执行时流式传输其更新，您可以使用配置中的 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: "获取给定城市的天气。",
        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);
}

Output

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

如果您为工具添加了 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: "获取给定城市的天气。",
        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

一些模型在生成最终答案前会进行内部推理。您可以通过过滤 type 为 "reasoning" 的标准内容块来流式传输这些思考/推理 token。

必须在模型上启用推理输出。配置详情请参阅推理部分和您的提供商集成页面。要快速检查模型的推理支持，请查看 models.dev。

要从智能体流式传输思考 token，请使用 streamMode: "messages" 并过滤推理内容块。当模型支持时，使用启用了扩展思考的模型实例（例如 ChatAnthropic）：

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

const getWeather = tool(
  async ({ city }) => {
    return `${city} 总是阳光明媚！`;
  },
  {
    name: "get_weather",
    description: "获取给定城市的天气信息。",
    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: "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);
  }
}

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" 内容块类型。要直接从聊天模型（不使用智能体）流式传输推理 token，请参阅使用聊天模型进行流式传输。

禁用流式传输

在某些应用中，您可能需要禁用特定模型的单个 token 流式传输。这在以下情况下很有用：

处理多智能体系统时，控制哪些智能体流式传输其输出。
混合使用支持流式传输和不支持流式传输的模型。
部署到 LangSmith 时，希望阻止某些模型输出流式传输到客户端。

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

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

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

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

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

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

Get started

Core components

Middleware

Frontend

Advanced usage

Agent development

Production

概述

支持的流模式

智能体进度

大语言模型 token

自定义更新

多模式流式传输

常见模式

流式传输思考/推理 token

禁用流式传输

相关内容

​概述

​支持的流模式

​智能体进度

​大语言模型 token

​自定义更新

​多模式流式传输

​常见模式

​流式传输思考/推理 token

​禁用流式传输

​相关内容

概述

支持的流模式

智能体进度

大语言模型 token

自定义更新

多模式流式传输

常见模式

流式传输思考/推理 token

禁用流式传输

相关内容