第 11 章 · 流式输出与子智能体事件

stream 支持 updates/messages/custom 等模式；对委派子任务还能拿到 stream.subagents 的独立事件流，把每个子任务作为单独通道展示。

机制与设计选择

UI 层消费 typed events，不同模式对应不同组件。Deep Agents 在标准流之上增加 subagents 事件，让每个被委派任务有独立句柄；这些事件默认隐藏，需要显式订阅。

示例

import os
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model=os.environ["MODEL_ID"],
    api_key=os.environ["DEEPAGENT_API_KEY"],
    base_url=os.environ.get("DEEPAGENT_BASE_URL", "https://api.openai.com/v1"),
)

import asyncio
from deepagents import create_deep_agent

agent = create_deep_agent(
    model=model,
    system_prompt="Answer in one line.",
)

async def run():
    async for chunk in agent.astream(
        {"messages": [{"role": "user", "content": "Say: ready."}]},
        config={"configurable": {"thread_id": "ch11"}},
        stream_mode=["messages", "custom"],
        version="v2",
    ):
        if chunk["type"] == "messages":
            token, meta = chunk["data"]
            if token.content:
                print(token.content, end="|", flush=True)

asyncio.run(run())

运行与验收

需要 API Key。验收：终端流式打印出 “ready”（或等价的单行确认），无异常。

容易踩的坑

typed 事件不等于把全部 state 发给前端，敏感字段要先过滤；子智能体事件默认隐藏，需要时再订阅。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. messages 模式给什么？

2. 子任务事件默认可见吗？

3. 自定义进度用什么模式？

4. 敏感字段应如何处理？

动手实践

1. 订阅 updates 模式并渲染进度。
2. 为子任务订阅 subagents 事件并单独渲染。
3. 设计断线重连后的去重策略。

官方参考

• overview
• context-engineering

核验日期：2026-09-17，运行环境：Deep Agents 0.7.15。