事件流

本页介绍 Deep Agents 特有的流式处理关注点——最重要的是，通过 stream.subagents 从委派的子代理进行流式传输。关于一般代理流式处理（stream.messages、stream.values、工具调用、自定义更新），请参阅 LangChain 事件流。

流式传输子代理

Deep Agents 在 LangGraph 流式传输之上增加了子代理投影。当你希望为每个委派的 task 调用获得一个流句柄时，请使用 stream.subagents。这种投影是轻量级的：它首先发现子代理任务，消息、工具调用和值流仅在你在子代理句柄上访问它们时才会被打开。每个句柄的 name 是子代理配置的名称：即协调器在调用 task 工具时传递的 subagent_type。Deep Agents 将该名称绑定到委派的运行，因此你在子代理规范中定义的相同标签就是你在流中用于过滤和路由的依据。

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

子代理流字段

每个子代理流暴露与父运行相同类型的投影，例如消息、工具调用、嵌套子代理和最终输出。关于通用的父运行流式处理模型，请参阅 LangChain 事件流。 TypeScript 使用 camelCase 投影名称，例如 toolCalls 和 taskInput。每个子代理流可以暴露 .messages、.toolCalls、.values、.subagents 和 .output。

字段	描述
`name`	子代理名称，取自协调器在其 `task` 调用中选择的 `subagent_type`。
`messages`	子代理发出的消息。
`subagents`	嵌套的子代理调用。
`output`	最终的子代理状态，或委派任务的完成信号。
`taskInput`	传递给任务工具的提示的 Promise。
`toolCalls`	作用域限定于该子代理的工具调用。

追踪子代理生命周期

当你只需要显示哪些子代理已启动和完成时，请使用 stream.subagents。除非你在单个子代理上访问这些投影，否则无需订阅消息或值流。

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

流式传输消息

Deep Agents 可以从协调器代理和委派的子代理发出消息。对于顶级消息使用 stream.messages，对于每个委派的子代理使用 subagent.messages。

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

const coordinatorMessages: string[] = [];
for await (const message of stream.messages) {
  console.log("[协调器]", await message.text); // 打印协调器消息文本
  coordinatorMessages.push(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 在代理树的每个层级暴露工具调用。对于协调器工具使用顶层的 stream.tool_calls，对于委派的工作使用每个 subagent.tool_calls。

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

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

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

流式传输嵌套工作

你可以递归进入子代理流以观察嵌套的子代理、消息和工具调用。

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

const subagentNames: string[] = [];
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)})`);

    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.name}: 已启动`);
  }

  subagentNames.push(subagent.name);
}

并发消费

协调器和子代理的输出通常是交错的。当你需要实时 UI 更新时，并发消费投影。在 JavaScript 中使用并发消费者：

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

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

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

const textDeltas: string[] = [];
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 ? "子代理" : "协调器";
    console.log(`[${source}] ${block.text}`);
    textDeltas.push(block.text);
  }
}

子代理与子图

stream.subgraphs 显示图的执行结构。stream.subagents 显示产品级别的 Deep Agents 任务委派。对于面向用户的 UI，请使用 stream.subagents，因为它隐藏了内部图节点并直接暴露了子代理概念。

Get started

Deployment

Execution environment

Context management

Delegation

Steering

Frontend

Protocols

流式传输子代理

子代理流字段

追踪子代理生命周期

流式传输消息

流式传输工具调用

流式传输嵌套工作

并发消费

子代理与子图

相关文档

​流式传输子代理

​子代理流字段

​追踪子代理生命周期

​流式传输消息

​流式传输工具调用

​流式传输嵌套工作

​并发消费

​子代理与子图

​相关文档

流式传输子代理

子代理流字段

追踪子代理生命周期

流式传输消息

流式传输工具调用

流式传输嵌套工作

并发消费

子代理与子图

相关文档