事件流式传输

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

本页涵盖 Deep Agents 特有的流式传输相关事项——最重要的是通过 stream.subagents 流式输出被委派的子智能体。有关通用的智能体流式传输（stream.messages、stream.values、工具调用、自定义更新），请参阅 LangChain 事件流式传输。

流式输出子智能体

Deep Agents 在 LangGraph 流式传输之上增加了一个子智能体投影。当你希望每次被委派的 task 调用对应一个流句柄时，请使用 stream.subagents。该投影是轻量级的：它先发现子智能体任务，只有在子智能体句柄上访问消息、工具调用和值流时才会打开它们。

每个句柄的 name 就是子智能体配置的名称：即协调器在调用 task 工具时传入的 subagent_type。Deep Agents 会将该名称绑定到被委派的运行上，因此你在子智能体规格中定义的同一个标签，就是你在流中进行筛选和路由所依据的标签。

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 运行记录。

子智能体流字段

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

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

字段	描述
name	子智能体名称，取自协调器在其 task 调用中选择的 subagent_type。
messages	子智能体发出的消息。
subagents	嵌套的子智能体调用。
output	子智能体的最终状态，或被委派任务的完成信号。
path	子智能体流的命名空间路径。
status	生命周期状态，例如 started、completed、failed 或 interrupted。
tool_calls	限定在该子智能体范围内的工具调用。

跟踪子智能体生命周期

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

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 运行记录。

流式输出消息

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

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 运行记录。

流式输出工具调用

Deep Agents 在智能体树的每一层都暴露工具调用。协调器工具请使用顶层的 stream.tool_calls，被委派的工作请使用各个 subagent.tool_calls。

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 运行记录。

流式输出嵌套工作

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

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 运行记录。

并发消费

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

在异步代码中并发消费时，请把 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 运行记录。

当你需要跨协调器和所有子智能体的精确到达顺序时，请遍历原始协议事件，并使用 namespace 来标识来源：

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 运行记录。

子智能体与子图

stream.subgraphs 展示图的执行结构。stream.subagents 展示产品层面的 Deep Agents 任务委派。面向用户的 UI 请使用 stream.subagents，因为它隐藏了内部图节点，并直接暴露子智能体这一概念。

相关

• LangChain 事件流式传输涵盖通用的智能体消息和工具调用流式传输概念。
• 子智能体前端流式传输展示了将协调器消息与子智能体卡片区分开的 UI 模式。
• LangGraph 事件流式传输涵盖底层的图流式传输模型。

---