事件流（Event Streaming）

为什么需要它

你有没有试过让 Agent 执行复杂任务，却只能干等最终结果？

流式传输子代理

用 stream.subagents 获取每个委托子代理的独立流句柄

Deep Agents 在 LangGraph 流式传输基础上增加了子代理投影（projection）。当你希望每个委托的 task 调用都有独立的流句柄时，使用 stream.subagents。

这个投影是轻量级的：它首先发现子代理任务，只有当你访问子代理句柄上的消息、工具调用和值流时，才会打开这些流。

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "Write me a haiku about the sea" }] },
  { version: "v3" }  // 使用 v3 版本的事件协议
);

// subagents 是异步迭代器，每当有新子代理被委托时 yield 一个句柄
for await (const subagent of stream.subagents) {
  console.log(subagent.name);            // 子代理名称，用于 UI 标识
  console.log(await subagent.taskInput); // 协调器传给 task 工具的 prompt

  // 惰性求值：只有迭代 messages 时才会打开消息流
  for await (const message of subagent.messages) {
    console.log(await message.text);     // 子代理生成的文本，逐 token 到达
  }
}

子代理流字段

每个子代理流暴露与父运行相同类型的投影

每个子代理流句柄提供以下字段，让你按需访问不同层级的数据：

跟踪生命周期

用 output Promise 跟踪每个子代理的启动与完成

当你只需要显示哪些子代理已启动和已完成时，使用 stream.subagents 的 output 字段。无需订阅消息或值流，就能实时统计进度。

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

let running = 0;     // 正在运行的子代理计数
let completed = 0;  // 已成功完成的子代理计数
let failed = 0;     // 失败的子代理计数
const watchers: Promise<void>[] = [];  // 收集所有 output Promise

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

  // output 是 Promise：resolve=成功，reject=失败
  watchers.push(
    subagent.output.then(
      () => {
        running -= 1;
        completed += 1;
        console.log(`${subagent.name}: completed`);
      },
      () => {
        running -= 1;
        failed += 1;
        console.log(`${subagent.name}: failed`);
      }
    )
  );
}

// 等待所有子代理完成后再打印最终统计
await Promise.all(watchers);
console.log({ running, completed, failed });

流式传输消息

用 stream.messages 和 subagent.messages 分别获取不同层级的消息

Deep Agents 可以从协调器代理和委托的子代理分别发出消息。两者的流句柄不同：

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

// 协调器消息：只包含顶层协调器发出的内容
for await (const message of stream.messages) {
  console.log("[coordinator]", await message.text);
}

// 子代理消息：每个子代理有独立的消息流
for await (const subagent of stream.subagents) {
  for await (const message of subagent.messages) {
    console.log(`[${subagent.name}]`, await message.text);
  }
}

流式传输工具

在代理树的每一层级暴露工具调用的实时状态

Deep Agents 在代理树的每一层级暴露工具调用。你可以分别获取协调器和每个子代理的工具调用，并实时查看执行状态。

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

// 协调器层级的工具调用
for await (const call of stream.toolCalls) {
  console.log("[coordinator tool]", call.name, call.input);
  console.log(await call.status);  // "running" | "finished" | "error"
}

// 每个子代理的工具调用，独立于协调器
for await (const subagent of stream.subagents) {
  for await (const call of subagent.toolCalls) {
    console.log(`[${subagent.name} tool]`, call.name, call.input);

    // 必须先检查 status，再按需访问 output 或 error
    const status = await call.status;
    if (status === "finished") {
      console.log(await call.output);   // 工具执行成功，获取返回值
    } else if (status === "error") {
      console.error(await call.error);  // 工具执行出错，获取错误信息
    }
  }
}

嵌套子代理流

递归进入子代理流，观察多层嵌套的工具调用和子代理

你可以递归进入子代理流来观察嵌套的子代理、消息和工具调用。这对于展示深层任务委派链特别有用。

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

for await (const subagent of stream.subagents) {
  console.log(`subagent ${subagent.name}: started`);

  // 第一层子代理的工具调用
  for await (const toolCall of subagent.toolCalls) {
    console.log(`${toolCall.name}(${JSON.stringify(toolCall.input)})`);

    const status = await toolCall.status;
    if (status === "finished") {
      console.log(await toolCall.output);
    } else if (status === "error") {
      console.error(await toolCall.error);
    }
  }

  // 递归进入嵌套子代理，可继续向下遍历任意深度
  for await (const nested of subagent.subagents) {
    console.log(`nested subagent ${nested.name}: started`);
  }
}

并发消费

用 Promise.all 并发消费协调器和子代理的流，实现实时 UI 更新

协调器和子代理的输出通常会交错出现。当你需要实时的 UI 更新时，必须并发地消费各投影。

方式一：用 Promise.all 并发迭代

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

// 并发消费：协调器和子代理的事件交错到达，必须同时迭代
await Promise.all([
  // 协调器消息流
  (async () => {
    for await (const message of stream.messages) {
      console.log("[coordinator]", await message.text);
    }
  })(),
  // 子代理消息流：每个子代理启动一个独立的异步消费协程
  (async () => {
    for await (const subagent of stream.subagents) {
      // void 表示不等待内部协程，允许多个子代理并发消费
      void (async () => {
        for await (const message of subagent.messages) {
          console.log(`[${subagent.name}]`, await message.text);
        }
      })();
    }
  })(),
]);

原始协议事件

用 namespace 精确区分协调器和子代理的内容

当你需要在协调器和所有子代理之间获取精确的到达顺序时，请迭代原始协议事件，并使用 namespace 来标识来源。

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

// 迭代原始协议事件，获取严格的时间顺序
for await (const event of stream) {
  // 只处理消息类型的事件
  if (event.method !== "messages") continue;

  const data = event.params.data;
  // 只处理文本增量事件（每个 token 触发一次）
  if (data.event !== "content-block-delta") continue;

  const block = data.delta ?? {};
  if (block.type === "text-delta") {
    // namespace 中含 "tools:" 前缀说明事件来自子代理
    const isSubagent = event.params.namespace.some(
      (seg) => seg.startsWith("tools:")
    );
    const source = isSubagent ? "subagent" : "coordinator";
    console.log(`[${source}] ${block.text}`);
  }
}

子代理与子图

stream.subagents 与 stream.subgraphs 的区别

本章回顾

到这里，你已经掌握了 Deep Agents 事件流的完整用法

延伸阅读