以编程方式使用文档

本页面介绍 Deep Agents 特有的流式传输问题——最重要的是，通过以下方式从委托的子代理进行流式传输 stream.subagents。有关常规代理流式传输（stream.messages, stream.values、工具调用、自定义更新），请参阅 LangChain 事件流.

流式传输子代理

Deep Agents 在 LangGraph 流式传输之上添加了子代理投影。当您希望每个委托 stream.subagents 使用一个流句柄时，请使用 task 调用。该投影是轻量级的：它首先发现子代理任务，只有在您访问子代理句柄上的这些流时，才会打开消息、工具调用和值流。

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

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

for subagent in stream.subagents:
    print(subagent.name, subagent.path, subagent.status)

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

子代理流字段

每个子代理流公开与父运行相同类型的投影，例如消息、工具调用、嵌套子代理和最终输出。有关常规父运行流式传输模型，请参阅 LangChain 事件流.

Python 使用 snake_大小写投影名称，例如 tool_calls。每个子代理流可以公开 .messages, .tool_calls, .values, .subagents和 .output.

字段描述
name子代理名称，取自 subagent_type 协调器在其 task 调用中选择的内容。
messages子代理发出的消息。
subagents嵌套子代理调用。
output子代理的最终状态，或委托任务的完成信号。
path子代理流的命名空间路径。
status生命周期状态，例如 started, completed, failed, or 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")

流式传输消息

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

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

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

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

流式传输工具调用

Deep Agents 在代理树的每个层级公开工具调用。使用顶级 stream.tool_calls 用于协调器工具和每个 subagent.tool_calls 用于委托工作。

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

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

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)

流式处理嵌套工作

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

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

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

并发消费

协调器和子代理输出经常交织。当您需要实时UI更新时，并发消费投影。

对于异步代码中的并发消费，请使用 astream_events 与 asyncio.gather:

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(...) instead:

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)

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

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

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']}")

子代理与子图

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

相关