事件流

从 Deep Agents 中流式输出子智能体、消息、工具调用和最终结果。

本页面涵盖 Deep Agents 特定的流式处理问题——最重要的是通过 stream.subagents 从委托的智能体进行流式处理。对于通用的智能体流式处理（stream.messages、stream.values、工具调用、自定义更新），请参阅 LangChain 流式处理。

流式处理子智能体

Deep Agents 在 LangGraph 流式处理的基础上添加了子智能体投影。当你希望每个委托的 task 调用都有一个流式处理句柄时，请使用 stream.subagents。该投影非常轻量级：它首先发现子智能体任务，并且只有在你在子智能体句柄上访问消息、工具调用和值流时才会打开这些流。

每个句柄的 name 是子智能体的配置名称：协调器在调用 task 工具时传递的 subagent_type。Deep Agents 将该名称绑定到委托的运行，因此你在子智能体规范中定义的相同标签就是你流式处理中过滤和路由的标签。

:::python

stream = agent.stream_events(
    {
        "messages": [{"role": "user", "content": "Write me a haiku about the sea"}],
    },
    version="v3",
)

subagent_names: list[str] = []
for subagent in stream.subagents:
    print(subagent.name, subagent.path, subagent.status)

    for message in subagent.messages:
        print(message.text)

    subagent_names.append(subagent.name)

View example trace

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

:::

:::js

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "Write me a haiku about the sea" }] },
  { 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 流式处理。

:::python Python 使用 snake_case 投影名称，例如 tool_calls。每个子智能体流式处理可以暴露 .messages、.tool_calls、.values、.subagents 和 .output。 :::

:::js TypeScript 使用 camelCase 投影名称，例如 toolCalls 和 taskInput。每个子智能体流式处理可以暴露 .messages、.toolCalls、.values、.subagents 和 .output。 :::

字段	描述
name	子智能体名称，来自协调器在其 task 调用中选择的 subagent_type。
messages	子智能体发出的消息。
subagents	嵌套子智能体调用。
output	最终子智能体状态，或委托任务的完成信号。
:::python
path	子智能体流式处理的命名空间路径。
status	生命周期状态，例如 started、completed、failed 或 interrupted。
tool_calls	限定在子智能体范围内的工具调用。
::::::js
taskInput	传递给任务工具的提示词的 Promise。
toolCalls	限定在子智能体范围内的工具调用。
:::

跟踪子智能体生命周期

当你只需要显示哪些子智能体已启动和已结束时，请使用 stream.subagents。除非你在单个子智能体上访问这些投影，否则你不需要订阅消息或值流。

:::python

stream = agent.stream_events(input, version="v3")

running = 0
completed = 0
failed = 0

for subagent in stream.subagents:
    running += 1
    print(f"{subagent.name}: started")

    try:
        _ = subagent.output
        running -= 1
        completed += 1
        print(f"{subagent.name}: completed")
    except Exception:
        running -= 1
        failed += 1
        print(f"{subagent.name}: failed")

View example trace

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

:::

:::js

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}: started`);

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

View example trace

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

:::

流式处理消息

Deep Agents 可以从协调器智能体和委托的子智能体发出消息。使用 stream.messages 表示顶层消息，使用 subagent.messages 表示每个委托的子智能体。

:::python

stream = agent.stream_events(input, version="v3")

coordinator_messages: list[str] = []
for message in stream.messages:
    print("[coordinator]", message.text)
    coordinator_messages.append(message.text)

for subagent in stream.subagents:
    for message in subagent.messages:
        print(f"[{subagent.name}]", message.text)

View example trace

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

:::

:::js

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("[coordinator]", 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;

:::

流式处理工具调用

Deep Agents 在智能体树的每个级别都暴露工具调用。使用顶层 stream.tool_calls 表示协调器工具，使用每个 subagent.tool_calls 表示委托的工作。

:::python

stream = agent.stream_events(input, version="v3")

coordinator_tool_names: list[str] = []
for call in stream.tool_calls:
    print("[coordinator tool]", call.tool_name, call.input)
    print(call.completed, call.error)
    coordinator_tool_names.append(call.tool_name)

for subagent in stream.subagents:
    for call in subagent.tool_calls:
        print(f"[{subagent.name} tool]", call.tool_name, call.input)
        for delta in call.output_deltas:
            print(delta, end="", flush=True)

        if call.completed and call.error is None:
            print(call.output)
        elif call.error is not None:
            print(call.error)

View example trace

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

:::

:::js

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

const coordinatorToolNames: string[] = [];
for await (const call of stream.toolCalls) {
  console.log("[coordinator tool]", 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} tool]`, 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);
    }
  }
}

View example trace

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

:::

流式处理嵌套工作

你可以递归进入子智能体流式处理以观察嵌套子智能体、消息和工具调用。

:::python

stream = agent.stream_events(input, version="v3")

subagent_names: list[str] = []
for subagent in stream.subagents:
    print(f"subagent {subagent.name}: {subagent.status}")

    for tool_call in subagent.tool_calls:
        print(f"{tool_call.tool_name}({tool_call.input})")
        for delta in tool_call.output_deltas:
            print(delta, end="", flush=True)

    for nested in subagent.subagents:
        print(f"nested subagent {nested.name}: {nested.status}")

    subagent_names.append(subagent.name)

View example trace

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

:::

:::js

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

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

  subagentNames.push(subagent.name);
}

View example trace

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

:::

并发消费

协调器和子智能体的输出经常交错。当你需要实时 UI 更新时，请并发消费投影。

:::python 在异步代码中进行并发消费，请使用 astream_events 与 asyncio.gather：

import asyncio

stream = await agent.astream_events(input, version="v3")

async def consume_coordinator():
    async for message in stream.messages:
        print("[coordinator]", await message.text)

async def consume_subagents():
    async for subagent in stream.subagents:
        async for message in subagent.messages:
            print(f"[{subagent.name}]", await message.text)

await asyncio.gather(consume_coordinator(), consume_subagents())

在同步代码中，请使用 stream.interleave(...) 代替：

stream = agent.stream_events(input, version="v3")

for name, item in stream.interleave("messages", "subagents"):
    if name == "messages":
        print("[coordinator]", item.text)
    else:
        for message in item.messages:
            print(f"[{item.name}]", message.text)

View example trace

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

:::

:::js 在 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);
        }
      })();
    }
  })(),
]);

View example trace

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

:::

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

:::python

stream = agent.stream_events(input, version="v3")

text_deltas: list[str] = []
for event in stream:
    if event.get("method") != "messages":
        continue

    payload = event["params"]["data"][0]
    if not isinstance(payload, dict):
        continue
    if payload.get("event") != "content-block-delta":
        continue

    block = payload.get("delta") or {}
    if block.get("type") == "text-delta":
        source = "subagent" if event["params"]["namespace"] else "coordinator"
        print(f"[{source}] {block['text']}")
        text_deltas.append(block["text"])

View example trace

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

:::

:::js

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 ? "subagent" : "coordinator";
    console.log(`[${source}] ${block.text}`);
    textDeltas.push(block.text);
  }
}

View example trace

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

:::

子智能体与子图

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

相关

• LangChain 流式处理 涵盖了通用智能体消息和工具调用流式处理概念。
• 子智能体前端流式处理 显示将协调器消息与子智能体卡片分离的 UI 模式。
• LangGraph 流式处理 涵盖了底层的图流式处理模型。

---

官方原文：src/oss/deepagents/event-streaming.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。