尚拙
Search K

Appearance

事件流式传输 (Event streaming) [Beta] ​

从 Deep Agents 中实时流式传输子智能体、消息、工具调用与最终输出

本页面专门介绍特定于 Deep Agents 的流式传输特性——最核心的是通过 stream.subagents 监听来自被委派子智能体的流式事件。有关通用的智能体流式传输（stream.messages、stream.values、工具调用与自定义更新），请参阅 LangChain 事件流式传输。

流式传输子智能体 (Stream subagents) ​

Deep Agents 在 LangGraph 流式传输的基础之上添加了专门的子智能体投影层。当你想为每个被委派的 task 调用获取一个独立的流句柄时，请使用 stream.subagents。该投影非常轻量：它优先发现子智能体任务，只有当你在子智能体句柄上显式访问它们时，才会按需打开消息、工具调用和状态值流。

每个句柄的 name 对应子智能体配置的名称：即主协调器在调用 task 工具时传入的 subagent_type。Deep Agents 将该名称绑定到被委派的运行实例，因此你在子智能体规格说明中定义的标签即为你在流中进行过滤和路由的标识符。

ts
const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "请为我写一首关于大海的俳句" }] },
  { version: "v3" },
);

const subagentNames: string[] = [];
for await (const subagent of stream.subagents) {
  console.log("子智能体名称:", subagent.name);
  console.log("任务输入:", await subagent.taskInput);

  for await (const message of subagent.messages) {
    console.log("子智能体生成内容:", await message.text);
  }

  subagentNames.push(subagent.name);
}

await stream.output;

子智能体流的核心字段 ​

每个子智能体流暴露与父级运行相同种类的投影对象，例如消息、工具调用、嵌套子智能体和最终输出。有关通用的父级流式模型，请参阅 LangChain 事件流式传输。

TypeScript 采用小驼峰命名规范（如 toolCalls 和 taskInput）。每个子智能体流均可暴露 .messages、.toolCalls、.values、.subagents 和 .output：

字段名称描述说明
name子智能体名称，取自协调器在 task 调用中传入的 subagent_type。
messages子智能体内部发出的消息事件流。
subagents嵌套的下一级子智能体调用流。
output子智能体的最终状态，或被委派任务的完成信号 Promise。
taskInput传递给 task 工具的任务 Prompt 提示词 Promise。
toolCalls限定在该子智能体作用域内的工具调用流。

跟踪子智能体生命周期 ​

当仅需在前端 UI 上展示哪些子智能体已启动、已完成或执行失败时，使用 stream.subagents 即可。除非你访问具体某个子智能体上的消息或状态投影，否则无需订阅内部的消息流。

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

let running = 0;
let completed = 0;
let failed = 0;
const watchers: Promise<void>[] = [];

for await (const subagent of stream.subagents) {
  running += 1;
  console.log(`${subagent.name}: 已启动`);

  watchers.push(
    subagent.output.then(
      () => {
        running -= 1;
        completed += 1;
        console.log(`${subagent.name}: 执行完毕`);
      },
      () => {
        running -= 1;
        failed += 1;
        console.log(`${subagent.name}: 执行失败`);
      },
    ),
  );
}

await Promise.all(watchers);
console.log({ running, completed, failed });

查看示例 Trace

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

流式传输消息 (Stream messages) ​

Deep Agents 会同时发出主协调器智能体的消息以及被委派子智能体的消息。使用 stream.messages 消费顶层主协调器的消息，使用 subagent.messages 消费各个子智能体内部的消息：

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

const coordinatorMessages: string[] = [];
for await (const message of stream.messages) {
  const text = await message.text;
  console.log("[协调器]", text);
  coordinatorMessages.push(text);
}

for await (const subagent of stream.subagents) {
  for await (const message of subagent.messages) {
    console.log(`[子智能体: ${subagent.name}]`, await message.text);
  }
}

await stream.output;

流式传输工具调用 (Stream tool calls) ​

Deep Agents 在智能体树的各个层级分别暴露工具调用。使用顶层的 stream.toolCalls 监听主协调器的工具调用，使用各 subagent.toolCalls 监听被委派子任务中的工具调用：

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

for await (const call of stream.toolCalls) {
  console.log("[协调器工具调用]", call.name, call.input);
  console.log(await call.status);
}

for await (const subagent of stream.subagents) {
  for await (const call of subagent.toolCalls) {
    console.log(`[子智能体 ${subagent.name} 工具调用]`, call.name, call.input);

    const status = await call.status;
    if (status === "finished") {
      console.log("工具输出:", await call.output);
    } else if (status === "error") {
      console.error("工具报错:", await call.error);
    }
  }
}

查看示例 Trace

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

监听嵌套子智能体 (Stream nested work) ​

你可以递归进入子智能体流，以观察多层嵌套的子智能体、消息和工具调用：

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

for await (const subagent of stream.subagents) {
  console.log(`子智能体 ${subagent.name}: 已启动`);

  for await (const toolCall of subagent.toolCalls) {
    console.log(`${toolCall.name}(${JSON.stringify(toolCall.input)})`);
  }

  for await (const nested of subagent.subagents) {
    console.log(`嵌套下级子智能体 ${nested.name}: 已启动`);
  }
}

查看示例 Trace

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

并发消费流式数据 (Consume concurrently) ​

主协调器与子智能体的输出往往是交错发生的。在构建前端实时交互界面时，使用并发消费模式能获得最佳体验：

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 subagent of stream.subagents) {
      void (async () => {
        for await (const message of subagent.messages) {
          console.log(`[${subagent.name}]`, await message.text);
        }
      })();
    }
  })(),
]);

查看示例 Trace

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

若要获取严格的全局时间到达顺序，可直接遍历底层原生协议事件，并通过 namespace 区分来源：

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

for await (const event of stream) {
  if (event.method !== "messages") continue;

  const data = event.params.data;
  if (data.event !== "content-block-delta") continue;

  const block = data.delta ?? {};
  if (block.type === "text-delta") {
    const isSubagent = event.params.namespace.some((seg) =>
      seg.startsWith("tools:"),
    );
    const source = isSubagent ? "subagent" : "coordinator";
    console.log(`[${source}] ${block.text}`);
  }
}

子智能体流 vs. 子图流 (Subagents vs. subgraphs) ​

stream.subgraphs 展示的是底层图执行架构的内部拓扑；而 stream.subagents 呈现的是业务产品层面的 Deep Agents 任务委派抽象。面向最终用户的客户端 UI 应当使用 stream.subagents，因为它屏蔽了底层图节点的技术细节，直接对应子智能体的用户心智模型。

相关参考 ​