事件流式传输 - Docs by LangChain中文

LangChain 代理构建在 LangGraph 之上，因此支持同一套流式传输栈，并为代理场景提供面向消息、工具调用、状态和自定义更新的 projections。对于大多数应用和前端用例，请通过 stream_events(..., version="v3") 使用 Event Streaming。Event Streaming 会返回带类型化 projections 的运行对象，因此可以独立消费每个 projection，而不需要解析 stream-mode 元组。

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

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

const agent = createAgent({
  model: "gpt-5-nano",
  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 delta of message.text) {
    process.stdout.write(delta);
  }
}

const finalState = await stream.output;

可以流式传输什么

Projection	用途
`for event in stream`	带完整 envelope 的原始协议事件，并可访问每个 channel。
`stream.messages`	模型消息流，每次 LLM 调用一个。
`message.text`	消息的文本 deltas 和最终文本。
`message.reasoning`	暴露 reasoning content 的模型的 reasoning deltas。
`message.toolCalls`	工具调用参数 chunks 和最终工具调用。
`message.output`	模型调用完成后的最终消息对象。
`message.usage`	provider 返回 token usage metadata 时的用量元数据。
`stream.values`	代理状态快照。
`stream.output`	最终代理状态。
`stream.subgraphs`	嵌套图运行（sub-agents 和普通 subgraphs）。
`stream.extensions`	自定义 transformer projections。
`stream.toolCalls`	工具执行生命周期、输入、输出 deltas、最终输出和错误。

stream.messages 会产生消息流。每个消息流暴露 .text、.reasoning、.toolCalls、.output 和 .usage。异步 projections 可以迭代以获取实时 deltas，也可以 await 以获取最终值。

代理消息

当你想获取每次 LLM 调用的模型输出时，使用 stream.messages。

const stream = await agent.streamEvents(input, { version: "v3" });

for await (const message of stream.messages) {
  process.stdout.write(`[${message.node}] `);
  for await (const delta of message.text) {
    process.stdout.write(delta);
  }

  const fullMessage = await message.output;
  console.log(fullMessage.content);

  const usage = await message.usage;
  if (usage) {
    console.log(usage);
  }
}

message.output 会给出最终 AI 消息，包括 provider 特定的 content blocks。在 TypeScript 中，当你只需要 token 计数或其他 usage metadata 时使用 message.usage；在 Python 中，从 message.output.usage_metadata 读取用量。

Reasoning content

Reasoning content 使用与文本内容相同的形状，但只有所选模型发出 reasoning blocks 时才可用。

const stream = await agent.streamEvents(input, { version: "v3" });

for await (const message of stream.messages) {
  for await (const delta of message.reasoning) {
    process.stdout.write(`[thinking] ${delta}`);
  }

  for await (const delta of message.text) {
    process.stdout.write(delta);
  }
}

模型配置详情请参阅 reasoning guide 和你的 provider 集成页。

工具调用

有两个有用的工具调用 projections：

message.tool_calls 会在模型生成工具调用时流式传输工具调用参数 chunks。
stream.tool_calls 会在工具调用开始后流式传输工具执行生命周期。

const stream = await agent.streamEvents(input, { version: "v3" });

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const chunk of message.toolCalls) {
        console.log("tool call chunk", chunk);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(call.name, call.input);
      console.log(await call.output, await call.error);
    }
  })(),
]);

流式传输 sub-agents

当一个 create_agent 调用另一个具名 create_agent（通常通过包装工具）时，内部代理的事件会流入嵌套 namespace。传给 create_agent 的 name= 会在流中标识该内部代理，因此你可以按代理筛选和标记。具名 sub-agents 会作为 handles 出现在 stream.subgraphs 上，与普通 subgraphs 一起出现。每个 handle 会暴露内部代理的 .messages、.values、.toolCalls 和 .output；可通过 subagent.name（你传入的 name=）筛选并操作特定代理。

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

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

const weatherAgent = createAgent({
  model: "openai:gpt-5.4",
  tools: [getWeather],
  name: "weather_agent",
});

const callWeather = tool(
  async ({ query }) => {
    const result = await weatherAgent.invoke({
      messages: [{ role: "user", content: query }],
    });
    return result.messages.at(-1)?.text ?? "";
  },
  { name: "call_weather", schema: z.object({ query: z.string() }) }
);

const supervisor = createAgent({
  model: "openai:gpt-5.4",
  tools: [callWeather],
  name: "supervisor",
});

const stream = await supervisor.streamEvents(
  { messages: [{ role: "user", content: "What's the weather in Boston?" }] },
  { version: "v3" }
);

for await (const subagent of stream.subgraphs) {
  if (subagent.name !== "weather_agent") continue;
  process.stdout.write(`${subagent.name}: `);
  for await (const message of subagent.messages) {
    for await (const token of message.text) {
      process.stdout.write(token);
    }
  }
  process.stdout.write("\n");
}

从工具调用的普通 StateGraph subgraphs 也会出现在 stream.subgraphs 上，请在 .compile(name=...) 上设置 name=，以便在 subagent.graph_name 中获得标签。具名 sub-agents 与普通 subgraphs 共享 stream.subgraphs projection；你在循环中编写的筛选逻辑会将它们区分开。

状态和最终输出

使用 stream.values 获取状态快照，使用 stream.output 获取最终代理状态。

const stream = await agent.streamEvents(input, { version: "v3" });

for await (const snapshot of stream.values) {
  console.log(snapshot);
}

const finalState = await stream.output;

多个 projections

在 JavaScript 中需要多个 projections 时，使用并发消费者：

const stream = await agent.streamEvents(input, { version: "v3" });

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      console.log(await message.text);
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(call.name, call.input);
    }
  })(),
]);

要访问未作为类型化 projections 暴露的 channels，或检查完整 event envelope，请迭代原始协议事件：

for await (const event of stream) {
  console.log(event.method, event.params.namespace, event.params.data);
}

自定义更新

当应用需要内置之外的 projection 时，使用自定义 stream transformers，例如检索进度、artifacts 或领域特定事件。

const stream = await agent.streamEvents(input, {
  version: "v3",
  transformers: [toolActivityTransformer],
});

for await (const activity of stream.extensions.toolActivity) {
  console.log(activity);
}

在 middleware 上注册 transformers

Middleware 注册的 transformers 需要 langchain@1.4.3 或更高版本。

Middleware 可以在 hooks 和 tools 旁声明 stream transformer factories。不同语言中的 factory 形状不同：将 streamTransformers 作为 factories 元组传给 createMiddleware。每个 factory 的形状为 () => StreamTransformer<any>（零参数），并且每个 scope 调用一次。每次调用返回一个新的 transformer 可以让每个 subgraph 保持隔离。

import { createAgent, createMiddleware } from "langchain";

const toolActivityMiddleware = createMiddleware({
  name: "ToolActivityMiddleware",
  streamTransformers: [toolActivityTransformer],
});

const agent = createAgent({
  model: "gpt-5-nano",
  tools: [getWeather],
  middleware: [toolActivityMiddleware],
});

编译时，createAgent 会将 middleware 注册的 factories 与传给自身 streamTransformers 选项的任何内容合并。编译后图上的最终顺序为：

内置 ToolCallTransformer。
Middleware 注册的 factories，按 middleware 顺序排列。
调用方从 createAgent 提供的 streamTransformers。

这样会让内置工具调用 projection 排在消费者 transformers 之前，并让调用方提供的条目拥有最终优先权。 Transformer contract 请参阅 Build your own projection。

​可以流式传输什么

​代理消息

​Reasoning content

​工具调用

​流式传输 sub-agents

​状态和最终输出

​多个 projections

​自定义更新

​在 middleware 上注册 transformers

​相关

可以流式传输什么

代理消息

Reasoning content

工具调用

流式传输 sub-agents

状态和最终输出

多个 projections

自定义更新

在 middleware 上注册 transformers

相关