事件流

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

深度智能体建立在 LangGraph 的事件流模型之上，并添加了一流的 subagents 投影。当你需要面向应用的流来获取协调器智能体、委派的子智能体、嵌套消息、工具调用和最终状态时使用它。

查看流式输出 cookbook获取可运行的示例和详细参考文档链接。

有关底层 Pregel 流模式的信息，请参阅深度智能体流式输出文档。

流式输出子智能体

深度智能体在 LangGraph 流式输出之上添加了子智能体投影。当你想要每个委派的 task 调用一个流句柄时，使用 run.subagents。投影首先发现子智能体任务，然后在你访问子智能体句柄上的投影时打开消息、工具调用和值流。

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

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

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

子智能体流字段

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

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

字段	说明
name	子智能体名称。
messages	子智能体发出的消息。
subagents	嵌套的子智能体调用。
output	最终子智能体状态，或委派任务的完成信号。
task_input	传递给 task 工具的提示词或输入。
path	子智能体运行的命名空间路径。
status	生命周期状态，如 started、running、completed、failed 或 interrupted。
tool_calls	限定在子智能体范围内的工具调用。

跟踪子智能体生命周期

当你只需要显示哪些子智能体已启动和完成时，使用 run.subagents。除非你访问个别子智能体上的投影，否则不需要订阅消息或值流。

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

running = 0
completed = 0
failed = 0

for subagent in run.subagents:
    running += 1
    print(f"{subagent.name}: 已启动")

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

流式输出消息和工具

深度智能体可以从协调器智能体和委派的子智能体发出消息和工具调用。对协调器流使用 run.messages 或 run.tool_calls，然后在每个子智能体上访问相同的投影。

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

for message in run.messages:
    print("[协调器]", str(message.text))

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

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

并发消费

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

当你想在一个同步循环中遍历多个投影时，使用 run.interleave(...)：

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

for name, item in run.interleave("messages", "subagents"):
    if name == "messages":
        print("[协调器]", str(item.text))
    else:
        for message in item.messages:
            print(f"[{item.name}]", str(message.text))

相关内容

• 流式输出 cookbook 展示可运行的事件流示例。
• LangChain 事件流 涵盖通用的智能体消息和工具调用流概念。
• 子智能体前端流式输出 展示将协调器消息与子智能体卡片分离的 UI 模式。
• LangGraph 事件流 涵盖底层图流模型。

---

连接这些文档到 Claude、VSCode 等工具，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面 或提交 Issue。