尚拙
Search K

Appearance

事件流（Event streaming） ​

流式传输来自 LangChain 智能体运行的实时更新

LangChain 智能体基于 LangGraph 构建，因此它们支持相同的流式处理技术栈，并为消息、工具调用、状态和自定义更新提供了面向智能体的投影（projections）。

对于大多数应用和前端使用场景，建议通过 stream_events(..., version="v3") 使用事件流（Event Streaming）。事件流会返回一个带有类型化投影的运行对象，因此可以独立消费每个投影，而无需解析流模式（stream-mode）元组。

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

可流式传输的内容 ​

投影用途
for event in stream包含完整封包并可访问每个通道的原始协议事件。
stream.messages模型消息流，每次 LLM 调用产生一个。
message.text消息的文本增量与最终文本。
message.reasoning支持公开思考内容的模型的思考增量。
message.toolCalls工具调用参数块与已完成的工具调用。
message.output模型调用完成后的最终消息对象。
message.usage提供商返回时的 Token 用量元数据。
stream.values智能体状态快照。
stream.output智能体最终状态。
stream.subgraphs嵌套图运行（子智能体与普通子图）。
stream.extensions自定义转换器投影。
stream.toolCalls工具执行生命周期、输入、输出增量、最终输出以及错误。

stream.messages 生成消息流。每个消息流都公开了 .text、.reasoning、.toolCalls、.output 和 .usage。异步投影可以通过迭代获取实时增量，也可以通过 await 获取最终值。

智能体消息 ​

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

ts
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 消息，包括特定提供商的内容块（content blocks）。在 TypeScript 中，如果只需要 Token 数量或其他用量元数据，请使用 message.usage；在 Python 中，则从 message.output.usage_metadata 读取用量。

思考内容 ​

思考内容采用与文本内容相同的结构，但仅在所选模型输出思考块（reasoning blocks）时可用。

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

有关模型配置的详细信息，请参阅思考指南以及所用提供商的集成页面。

工具调用 ​

有两个实用的工具调用投影：

  • message.tool_calls 在模型生成工具调用期间流式传输工具调用参数块。
  • stream.tool_calls 在工具调用开始后流式传输工具执行的生命周期。
ts
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);
    }
  })(),
]);

流式传输子智能体 ​

当一个 createAgent 调用另一个具名的 createAgent（通常通过包装工具）时，内部智能体的事件会在嵌套命名空间中流转。传递给 createAgent 的 name 会在流中标识该内部智能体，以便你可以按智能体进行过滤和标记。

具名子智能体会呈现在专用的 stream.subagents 投影上。每个句柄都会公开内部智能体自身的 .messages、.toolCalls 和 .output，以及 .name（传入的 name=）、.cause（派发子智能体的工具调用）和嵌套的 .subagents。因为这里只会出现具名的 createAgent 运行，所以无需手动过滤掉普通子图。

ts
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.5",
  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.5",
  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.subagents) {
  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");
}
//Output: "weather_agent: The weather in Boston is sunny!"

从工具中调用的普通 StateGraph 子图也会出现在 stream.subgraphs 上——在 .compile(name=...) 上设置 name= 即可在 subagent.graph_name 中获取标签。

stream.subagents 是具名 createAgent 子智能体的专注视图，而 stream.subgraphs 则涵盖了每一个嵌套图。请根据你的 UI 需求选用。

状态与最终输出 ​

使用 stream.values 获取状态快照，使用 stream.output 获取最终智能体状态。

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

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

const finalState = await stream.output;

多投影 ​

在 JavaScript 中需要同时消费多个投影时，请使用并发消费者：

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

如需访问未作为类型化投影公开的通道，或检查完整的事件封包，可以迭代原始协议事件：

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

自定义更新 ​

当应用程序需要非内置投影（例如检索进度、工件或特定领域事件）时，可以使用自定义流转换器（stream transformers）。

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

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

在中间件上注册转换器 ​

NOTE

通过中间件注册转换器需要 langchain@1.4.3 或更高版本。

中间件可以在声明 Hook 和工具的同时声明流转换器工厂。各语言之间的工厂结构有所不同：

以工厂元组形式向 createMiddleware 传递 streamTransformers。每个工厂的形态均为 () => StreamTransformer<any>（零参数），且在每个作用域内调用一次。每次调用返回一个全新的转换器可以保持各个子图的相互隔离。

ts
import { createAgent, createMiddleware } from "langchain";

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

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

在编译时，createAgent 会将中间件注册的工厂与其自身 streamTransformers 选项中传入的内容进行合并。编译后图上的最终顺序为：

  1. 内置的 ToolCallTransformer。
  2. 中间件注册的工厂，按中间件顺序排列。
  3. 调用方通过 createAgent 提供的 streamTransformers。

这样可以保证内置的工具调用投影排在消费者转换器之前，并赋予调用方提供的条目最终决定权。

有关转换器契约的详细信息，请参阅构建自己的投影。

相关内容 ​