Deep Agents 9. 异步 SubAgent、Agent Protocol 与事件流

1. 异步 SubAgent 解决什么问题

1.1 同步等待与后台任务

上一篇使用同步 task Tool 委派 Analyst 和 Reviewer。同步模式的优点是调用链短：Supervisor 等待 SubAgent 返回，校验结果，然后继续生成当前回答。几十秒内能够完成的计算、审核或格式转换通常适合这种方式。

下面这些任务则可能持续几分钟，甚至跨越一次 HTTP 请求的生命周期：

• 批量分析大量文件；
• 检索多个数据源并整理引用；
• 在 Sandbox 中安装依赖、构建和运行测试；
• 等待第三方系统导出、训练或审批；
• 生成报告后继续执行独立质量检查。

同步 SubAgent 会阻塞当前 Supervisor Run，并占用对应的 Worker 执行槽位。这里不能笼统地说“持续占用模型连接”：模型请求结束后，Runtime 可能正在执行 Tool、子图或外部请求，并不存在一条始终打开的模型 HTTP 连接。真正被占用的是当前 Graph Run、Worker Slot，以及当时仍保持的应用请求或 Stream。

异步 SubAgent 把长任务提交给 Agent Protocol Server。在远程 Thread 创建、Run 提交并取得标识后，Supervisor 获得 task_id，之后可以查询、补充要求、取消或列举任务。Run 此时可能仍在队列中；拿到 ID 不等于 Worker 已经开始执行。

task_id 是稳定标识，不是天然持久化保证。要在进程重启或跨请求后继续追踪，还需要同时满足：

1. Supervisor 的 async_tasks State 使用持久 Checkpointer；
2. 后续请求复用原 Supervisor thread_id；
3. 远程 Agent Server 持久化 Thread 和 Run；
4. 业务数据库保存租户、Supervisor 与远程任务映射。

1.2 同步与异步委派的差异

维度	同步 SubAgent	异步 SubAgent
调用入口	task	start_async_task
返回时机	子任务结束后	Thread 创建且 Run 提交后
执行位置	当前 Agent Runtime 中的 SubAgent Graph / Runnable	Agent Protocol Server 中注册的 Graph
当前 Supervisor Run	等待子任务	可以结束本轮并稍后恢复
后续操作	当前调用完成后结束	Check、Update、Cancel、List
上下文	一次委派的隔离上下文	远程 Thread 保存多次 Run 的历史
适合任务	当前回答必须依赖结果	长时、可并行或需要中途调整

异步不一定更好。创建远程 Thread、保存映射、处理队列、重试和孤儿任务都有成本。如果当前回答必须立即采用 SubAgent 结果，同步委派更简单。

1.3 它不是 Python async def

Python asyncio 解决单进程中的协作式并发；Async SubAgent 解决的是跨请求、跨进程的任务身份、状态与控制。二者可以组合，但不是同一个概念：

Python async/await       -> 当前进程怎样等待 I/O
Async SubAgent          -> 后台任务怎样启动、恢复、更新和取消
Event Streaming         -> 执行事件怎样传给观察者

后台任务与 Streaming 经常出现在同一产品中，但职责不同。异步任务可以只在完成后查询；同步 Graph 也可以持续输出 Token 和 State 更新。

1.4 Preview 与证据边界

本文固定以下环境：

组件	版本
Python	3.12.11
deepagents	0.6.12
langgraph	1.2.9
langgraph-sdk	0.4.2

AsyncSubAgentMiddleware 是 Preview 能力。包版本可以固定，Preview API 仍不具有长期兼容承诺。全文按五类依据描述结论：

标签	含义
官方当前文档	当前网页描述，可能晚于本文安装包
0.6.12 源码	锁定安装包的实际实现
本文离线测试	Fake SDK、Scripted Model 和 Graph 测试
本文协议实测	本地 Agent Server 的 HTTP 验证
产品建议	需要业务系统自行实现的治理能力

官方入口包括 Deep Agents Async Subagents 和 LangGraph Streaming。升级依赖时必须重新执行契约测试，不能仅根据当前网页推断旧版本行为。

2. Async SubAgent 架构与配置

2.1 生命周期不是一条同步调用链

一次完整任务包含以下动作：

1. Supervisor 判断任务适合后台执行；
2. start_async_task 创建远程 Thread 并提交第一个 Run；
3. Middleware 把远程 Thread ID 作为当前实现的 Task ID；
4. Supervisor 向用户报告完整 Task ID，不自动轮询；
5. Run 可能排队，Worker 获得执行槽位后才真正运行；
6. 用户之后触发 Check、Update 或 Cancel；
7. 成功时 Middleware 读取远程 Thread State 并提取结果。

“启动后停止”指停止自动 Check，不表示 Supervisor 必须立即终止整条回答。它仍可以说明任务已提交、如何查看和取消，也可以处理不依赖任务结果的轻量内容。

2.2 控制平面、执行平面与事件平面

生产系统至少有三类平面：

• 控制平面：Supervisor、五个 Async Tool、Checkpointer 和 Task Ownership DB；
• 执行平面：Agent Server、Run Queue、Worker、Remote Thread 和 Artifact Store；
• 事件平面：Agent Server Stream、应用事件网关、浏览器订阅与事件回放。

父 State 是任务索引，不是唯一事实来源。缓存的 running 可能已经过期；远程 Server 也不应独自承担租户所有权和产品状态。业务数据库需要建立自己的长期任务模型，把 Preview TypedDict 隔离在适配器之后。

2.3 AsyncSubAgent 配置字段

在 deepagents==0.6.12 中，AsyncSubAgent 是包含五个字段的 TypedDict：

字段	必需	作用
name	是	Supervisor 选择的逻辑 Agent 类型
description	是	能力、使用条件和返回边界
graph_id	是	Agent Server 注册的 Graph / Assistant ID
url	否	HTTP 服务地址；省略时走同部署 ASGI
headers	否	自托管服务需要的附加静态 Header

本文真实配置位于 p09_async_streaming/async_config.py：

ASYNC_SUBAGENTS: list[AsyncSubAgent] = [
    {
        "name": "research-service",
        "description": (
            "执行预计超过一分钟的多轮资料检索与证据整理；"
            "返回报告路径、引用和未解决问题，不处理简单查询。"
        ),
        "graph_id": "research_agent",
        "url": "http://127.0.0.1:2024",
    }
]

name 与 graph_id 不要求相同。Description 是模型路由规则，不是宣传语，应说明何时使用、何时不要使用、返回什么以及资源特征。

headers 是构造时的静态字典，不适合直接固化长期 Secret，也不能解决每个租户的短期 Token 刷新。生产认证应由服务身份、动态凭证提供器或应用网关完成。

2.4 HTTP 与 ASGI Transport

Transport	配置	适用拓扑	主要边界
ASGI	省略 url	Graph 注册在同一个 langgraph.json 的同部署场景	进程内路由，无额外网络认证，但仍有独立 Thread
HTTP	显式 url	Supervisor 与 Worker 独立部署	TLS、服务认证、重试、熔断与租户授权

官方当前文档推荐同部署时使用 ASGI。锁定的 0.6.12 还有一个更窄的同步入口限制：_ClientCache.get_sync() 在 url=None 时明确抛出 ASGI transport (url=None) requires async invocation。因此：

• 完整 Agent 使用 ainvoke() / astream() 时可以走异步 Client；
• 本文直接调用同步 .func 的底层测试不能用无 URL ASGI；
• 这不是 ASGI Transport 的通用缺陷，而是 0.6.12 同步 Client 入口的源码契约。

对应离线测试固定了这条错误信息，升级后若实现改变，测试会提醒文章同步更新。

2.5 Middleware 自身提供五个 Tool

AsyncSubAgentMiddleware 自身提供五个控制 Tool，不代表整个 Deep Agent 只有五个 Tool：

Tool	输入	作用
start_async_task	description、subagent_type	创建 Thread 并提交初始 Run
check_async_task	task_id	获取当前 Run 状态，成功时提取结果
update_async_task	task_id、message	在同一 Thread 创建新 Run
cancel_async_task	task_id	请求取消当前 Run
list_async_tasks	可选 status_filter	查看当前 Supervisor Thread 追踪的任务

同步 task、文件 Tool、自定义 Tool 仍可能同时存在。模型路由测试不仅要看是否选择异步 Tool，还应验证：简单任务不启动、启动后不自动轮询、查询使用完整 ID、多个任务选对目标、未知 ID 不伪造、更新和取消不越权。

2.6 创建 Supervisor 与恢复线程

def build_agent(*, model=None):
    return create_deep_agent(
        model=model if model is not None else build_model(),
        subagents=ASYNC_SUBAGENTS,
        checkpointer=InMemorySaver(),
        system_prompt=(
            "长时间研究任务必须调用 start_async_task；启动后立即向用户返回 task_id，"
            "不要在同一轮自动调用 check_async_task。用户明确要求查询时才检查任务。"
        ),
    )

InMemorySaver 只适合本地验证。即使配置了 Checkpointer，后续调用不复用 Thread ID 也无法恢复 async_tasks：

config = {"configurable": {"thread_id": "supervisor-thread-001"}}
result = agent.invoke(input, config=config)

生产环境应替换成持久 Checkpointer，同时在业务数据库保存 Task Ownership。State 中的租户字段不能代替服务端认证结果。

3. Task、Thread、Run 与状态控制

3.1 三类 ID 的关系

在 deepagents==0.6.12 Preview 实现中：

Middleware task_id == remote thread_id
run_id             == 当前一次远程执行

首次启动创建 Thread 和 Initial Run；Update 复用 Thread 并创建新 Run。因此 Task ID 保持不变，Run ID 会变化。Preview 未来可能把 Task 与 Thread 解耦，业务库不应只保存一个 UUID，而应保存：

• Middleware Task ID；
• Remote Thread ID；
• 当前与历史 Run ID；
• Graph / Assistant ID；
• Deployment ID 和版本。

3.2 async_tasks 是 Preview State Schema

以下字段来自 0.6.12 的 AsyncTask TypedDict，不应直接固化成永久数据库 Schema：

字段	含义
task_id	Middleware 暴露的任务 ID
agent_name	Async SubAgent 逻辑名称
thread_id	远程 Thread
run_id	当前 Run
status	最近一次已知状态字符串
created_at	创建时间
last_checked_at	最近查询时间
last_updated_at	最近状态变化或更新时间

_tasks_reducer 的锁定版实现只是：复制现有字典，然后 dict.update(update)。同 Task ID 后写覆盖前写，没有删除标记；并发更新如何排序取决于 Graph 执行和 Reducer 应用顺序。业务清理、并发冲突和数据库乐观锁需要自行设计。

3.3 Start：获得 ID 不等于开始执行

start_async_task 依次完成：

1. 校验 subagent_type；
2. threads.create()；
3. runs.create()；
4. 以 Thread ID 构造 Task 记录；
5. 返回包含 ToolMessage 和 async_tasks 增量的 Command。

0.6.12 在提交后把缓存状态写成 running，但 Agent Server 可能仍在等待 Worker Slot。产品 UI 可以增加 submitted -> queued -> running，但要注明 queued 是平台或业务层状态，不是凭空从 Middleware TypedDict 推断出来的。

3.4 Check：结果提取不是协议通用规范

当前实现先查询 Run。只有状态为 success 时才读取 Thread State，并执行下面的结果规则：

1. 读取 values[“messages”]；
2. 有消息时取最后一条；
3. 最后一条是字典时读取其 content，否则执行 str(last)；
4. 没有消息时返回 (completed with no output messages)。

它没有检查最后一条一定是 AIMessage。因此最后一条 ToolMessage、空字符串或多模态 Content Blocks 都会按原值进入结果；只有自定义 custom_result 而没有 messages 的 Graph 也会得到“无输出消息”。这只是 0.6.12 Middleware 的提取约定，不是 Agent Protocol 对所有 Graph 的统一结果规范。

本文用六组测试覆盖普通 AI 内容、Tool 内容、空内容、多模态列表、空消息和自定义 State。远程 Graph 应主动提供 Middleware 能识别的最终 messages。

3.5 Update：Interrupt 不是事务回滚

update_async_task 在相同 Thread 上提交新消息和新 Run，并在 0.6.12 中传入：

multitask_strategy="interrupt"

旧 Run 被请求中断，新 Run 基于远程 Thread 已提交的历史继续。需要同时认识四个边界：

• Interrupt 是运行控制，不是数据库事务回滚；
• 已完成的外部副作用不会自动撤销；
• 取消信号到达前，旧 Run 可能继续短时间；
• 新 Run 只能看到已提交到 Thread Checkpoint 的状态。

发送通知、上传报告、扣减配额等动作必须使用业务幂等键和补偿机制。

3.6 Cancel：只在远端请求成功后更新本地状态

锁定实现先调用 runs.cancel()，成功后才返回本地 cancelled State 增量。如果网络调用失败，Tool 返回错误字符串，不会谎称父 State 已取消。本文增加了取消失败测试。

远端返回取消成功仍不代表外部副作用回滚。生产系统需要定期对账本地任务、远程 Run 和外部工作项。

3.7 List、缓存与版本差异

这里存在一个必须公开的文档与源码差异：

依据	直接使用缓存的终态
官方当前 Async Subagents 页面	success、error、cancelled
deepagents==0.6.12 _TERMINAL_STATUSES	success、error、cancelled、timeout、interrupted

本文以锁定运行时为代码基线，因此保留五种缓存终态的参数化测试，同时明确它是私有常量和 Preview 行为。升级后必须重新核对。

还有一个容易忽略的细节：list_async_tasks 的公开 status_filter Schema 只接受 running/success/error/cancelled/all，没有 timeout/interrupted。内部缓存集合、公开过滤枚举和 Agent Server Run 状态并不是同一套契约。

产品 UI 可使用 submitted/queued/running/updating/success/error/cancelled/deadline_exceeded 等更丰富状态，但应通过显式适配器映射，不能把 UI 枚举反向假装成 Middleware 原生状态。

4. Agent Protocol 的本地验证

4.1 注册一个确定性远程 Graph

协议测试需要隔离模型随机性。remote_agent.py 注册了一个不调用 LLM 的真实 LangGraph：

def research(state: MessagesState) -> dict[str, list[AIMessage]]:
    request = state["messages"][-1].text
    return {
        "messages": [
            AIMessage(
                content=(
                    "research_agent completed: "
                    f"{request}; artifact_reference=/reports/research.md"
                )
            )
        ]
    }

artifact_reference 只是协议测试字符串，不证明文件已创建、当前租户有权读取或 Supervisor 能访问该路径。真实工件应保存所有权、校验值、访问策略和生命周期。

langgraph.json 注册 Graph：

{
  "dependencies": ["."],
  "graphs": {
    "research_agent": "./p09_async_streaming/remote_agent.py:graph"
  },
  "env": ".env"
}

4.2 启动本地 Agent Server

cd source/_posts/deepagent/examples

.venv_deepagent/bin/langgraph dev \
  --config p09_async_streaming/langgraph.json \
  --host 127.0.0.1 --port 2024 \
  --no-browser --no-reload

本次实测启动日志显示：

langgraph-api=0.10.3
Using langgraph_runtime_inmem
Using auth of type=noop
Starting 1 background workers

因此只能得出“本文这次本地开发配置使用内存 Runtime 和 No-op Auth”，不能把它写成所有 CLI 版本和配置的永久默认行为。该服务只能用于本机验证，不能原样暴露到公网。

4.3 底层 Tool 契约与真实 HTTP

real_protocol_check.py 调用真实 Middleware Tool，但使用 StructuredTool.func 进入底层函数：

start = tools["start_async_task"].func(
    description="整理 Deep Agents 的异步任务资料",
    subagent_type="research-service",
    runtime=build_runtime(state, tool_call_id="start-call-001"),
)
task_id = only_task_id(start)
state = apply_command(state, start)

remote_status = wait_for_remote_run(state["async_tasks"][task_id])
check = tools["check_async_task"].func(
    task_id=task_id,
    runtime=build_runtime(state, tool_call_id="check-call-001"),
)

这是锁定版本的底层协议契约测试，不是业务代码推荐入口。.func 绕过模型 Tool Calling、ToolNode 的完整输入验证、Middleware Hook、完整 Trace、Streaming 和 Agent State Reducer 应用。正常业务应让 Supervisor 通过 Agent Loop 调用 Tool。

测试中的 TEST_WAIT_TIMEOUT_SECONDS=5.0 是集成测试等待上限，不是远程 Run Timeout，也不是业务 Deadline。轮询不消耗 Supervisor 的 LLM Tool 轮次，但仍产生 SDK 请求、连接、日志和 Agent Server 负载；生产应优先使用事件、Webhook 或后台调度器。

重新实测的脱敏结果：

remote_status=success
task_status=success
task_id_equals_thread_id=True
result={"status": "success", "thread_id": "<UUID>",
        "result": "research_agent completed: 整理 Deep Agents 的异步任务资料; artifact_reference=/reports/research.md"}

Server 日志还显示 Run 排队约 844ms、实际节点执行约 17ms。这直接说明“Run 已提交”和“Worker 已开始”之间可能存在等待。

4.4 Fake SDK、Scripted Model 与测试边界

层次	验证内容	不能证明什么
.func + Fake SDK	五个 Tool、State、状态和错误分支	Agent Loop、HTTP
.func + 本地 Server	Thread、Run、HTTP 和结果收集	模型路由、生产认证
Scripted Model + 完整 Agent	Tool Calling、ToolMessage、Reducer、跨轮 Thread 恢复	随机模型路由质量
在线模型 + 本地 Server	真实模型选择 Start / Check	远程实时事件转发、生产可靠性

新增的完整 Agent 测试分成两轮：第一轮模型只调用 start_async_task 并停止；远端完成后，第二轮用户明确查询，模型才调用 check_async_task。它验证同一 Supervisor Thread 中的 async_tasks 恢复，不把 .func 当成 Agent E2E。

5. LangGraph v2 Event Streaming

5.1 本地 Supervisor 与远程 Run 是两条流

Supervisor 的 stream() 只自动包含当前 Graph 的模型、Tool 和嵌套子图事件。远程 Async SubAgent 是 Agent Server 中另一条 Thread / Run，不会自动被拼进 Supervisor Namespace。

推荐拓扑是：

Supervisor Stream -----------+
                             +-> Application Event Gateway -> Browser
Remote Agent Server Stream --+

两条流使用 Task、Thread、Run、Trace 和 Deployment ID 关联。浏览器不应仅凭 Task ID 直接持有 Agent Server Service Token；应用事件网关负责租户授权、短期订阅凭证、断线续传、事件回放和服务地址隐藏。

5.2 v2 StreamPart 信封

本文使用 LangGraph 1.2.9，并显式传入 version=”v2”。v2 需要 LangGraph 1.1+，统一返回：

{
    "type": "updates",
    "ns": ("research:<task-id>",),
    "data": {...},
}

• type：Stream Mode；
• ns：根图或嵌套子图路径；
• data：对应模式的载荷。

低于支持版本或没有传 version=”v2” 时，单模式、多模式和子图组合的返回形状不同，前端不能混用解析器。

5.3 Stream Mode 与 Checkpointer 前提

Mode	内容	用途与边界
messages	Message Chunk 与模型元数据	文本、Tool Call 参数增量
updates	节点产生的 State 增量	步骤和 Tool 结果
values	每一步完整 State	小 State 调试，带宽较高
custom	Stream Writer 自定义数据	阶段与确定性业务进度
checkpoints	Checkpoint 事件	需要 Checkpointer
tasks	Task 开始和完成事件	需要 Checkpointer 和 thread_id
debug	Checkpoint、Task 和更多内部数据	仅受控排错环境

本文为 tasks 增加了真正可运行的示例：

graph = builder.compile(checkpointer=InMemorySaver(), name="task-stream")
graph.stream(
    {"topic": topic, "result": ""},
    config={"configurable": {"thread_id": "p09-task-stream"}},
    stream_mode=["tasks"],
    version="v2",
)

测试断言它产生同一个节点的 Start 和 Finish 两个 tasks StreamPart。debug 可能包含完整 State、Prompt、Tool 参数和文件内容，不应直接转发给终端用户。

5.4 messages 不是 Token 计数

messages 的每个 StreamPart 可能包含：

• 文本 Message Chunk；
• Tool Call 名称或参数 Chunk；
• 模型和节点元数据；
• 空内容或结束 Chunk；
• 多次模型调用中的不同消息片段。

Tool 参数需要按 Tool Call ID 和 Chunk 顺序累积，在结束状态到达后再解析完整 JSON。事件数量不能当作文本 Token 数，也不能直接衡量回答长度。

5.5 updates、values 与 Reducer

values 反复发送完整 State；updates 只发送节点增量。长消息和大报告场景应优先使用 updates。

前端不能对所有字段执行浅层 dict.update()。Messages、Todo 和自定义 Channel 都可能有专用 Reducer。若客户端无法复刻服务端 Reducer，应在重连后读取官方 Snapshot，再消费新的增量，而不是自己猜完整 State。

5.6 custom 业务进度与 Python 版本

def research_node(state: ResearchState) -> dict[str, str]:
    writer = get_stream_writer()
    writer({"stage": "researching", "topic": state["topic"]})
    return {"result": state["topic"].upper()}

Python 3.11+ 的异步上下文可以按当前方式使用。Python 3.10 及以下的异步代码存在 ContextVar 传播限制，需要显式传递 Writer 或配置。

completed=38 这类进度只有在确定性计数器生成时才可信。模型输出的百分比最多是估算，不能驱动计费、SLA 或完成判定。

5.7 Namespace、顺序和重连

开启 subgraphs=True 后，子图事件带非空 Namespace，根图使用 ()。同一个节点可能执行多次，必须保留完整 Namespace 与调用身份。

事件顺序至少要定义到 Task / Run 范围。多个 Worker 和事件源很难产生真正的全局单调 seq，可组合使用：

• Event ID；
• Task / Run ID；
• 源分区与源内序号；
• 逻辑时间；
• Snapshot 版本。

SSE 或 WebSocket 到达顺序不是全局业务顺序。重连时需要最后确认位置、重复事件幂等处理和 Snapshot / Delta 边界。

5.8 背压与前端状态机

模型 Chunk 可能快于浏览器渲染。服务端可以按 30～100ms 合并文本 Chunk，对可覆盖进度只保留最新值；审批、终态等不可丢事件进入持久队列。慢客户端超过缓冲上限后断开，通过 Snapshot 恢复。

产品状态机可以设计为：

submitted -> queued -> running -> success
                            +---> error
                            +---> cancelled
             running -> updating -> running

queued/updating 属于产品或平台映射层。Run Timeout、Tool Timeout、业务 Deadline 与 Sandbox TTL 也必须分别展示，不能都压成一个 timeout。

6. 生产安全与可靠性

6.1 Task ID 不是授权凭据

服务端执行 Check、Update 或 Cancel 的可信流程应是：

1. 认证层取得真实用户和租户；
2. 查询业务数据库的 Task Ownership；
3. 校验 Task、Remote Thread、Deployment 和租户映射；
4. 生成服务间身份并调用 Agent Server；
5. 记录审计结果。

不能信任模型在自然语言中声称“这个 Task 属于当前用户”，也不能让客户端 Header 覆盖服务端认证身份。Runtime Context 中的租户字段只用于传递上下文，不是唯一授权来源。

6.2 应用级幂等

网络超时不表示 Thread 没有创建。业务幂等键应由应用控制，例如：

tenant_id + supervisor_thread_id + application_operation_id

tool_call_id 可作为 Trace 关联字段，但模型重新生成 Tool Call 或上层重试时它可能变化，不能作为唯一业务幂等来源。服务端应先保存 Launch Intent，再创建远程任务；重复请求返回已有映射。

6.3 四种时间边界

边界	含义	负责组件
Run Timeout	平台停止一次 Run	Agent Server
Tool Timeout	单个外部操作上限	Tool / Runtime
业务 Deadline	用户可接受的完成时间	应用调度层
Sandbox TTL	执行环境生命周期	Sandbox 平台

此外还需要孤儿任务扫描、Worker 心跳、工件保留策略和清理失败告警。超时后保存阶段性结果，并对外返回脱敏摘要。

6.4 更新、取消与副作用

可重试的远程节点必须假定自己可能重复执行。发送通知、创建工单、上传文件和扣减配额都应使用业务幂等键，并记录外部系统回执。Cancel 后由对账任务确认远程 Run、外部工作项和本地 Task 是否一致。

6.5 认证与网络

HTTP Transport 至少需要 TLS、短期服务身份、超时、有限重试、熔断和敏感日志过滤。动态租户凭证不要长期存进 AsyncSubAgent.headers。浏览器通过应用事件网关订阅，不直接接触 Agent Server Service Token。

6.6 可观测性与容量

日志和 Trace 建议携带：

• Tenant、Authenticated User；
• Supervisor Thread / Run；
• Tool Call、Application Operation；
• Async Task、Remote Thread / Run；
• Deployment、Assistant / Graph；
• Revision / Commit SHA；
• Model ID、Worker Region、Trace ID。

容量监控还应覆盖 Queue Depth、Queue Wait Time、Run Start Delay、Worker Utilization、Concurrent Runs 和事件网关缓冲。只有请求延迟而没有队列指标，无法解释“Task 已创建但很久没有开始”。

7. 测试、排错与在线验收

7.1 离线和协议测试结果

运行：

cd source/_posts/deepagent/examples

.venv_deepagent/bin/python -m compileall -q p09_async_streaming
.venv_deepagent/bin/python -m ruff check p09_async_streaming
.venv_deepagent/bin/python -m pytest -o addopts='' p09_async_streaming -q

结果：

All checks passed!
..............................                                           [100%]
30 passed

30 个 pytest Test Item 覆盖：

• 配置、Description、重复 Name 和五个 Tool；
• ASGI 同步底层入口限制；
• Start Task 记录、完整 ID 与时间戳；
• Running、Success、Update、Cancel 和取消失败；
• 五种锁定版缓存终态；
• 六种最终结果提取边界；
• Reducer 覆盖与无删除标记；
• 未知 Task ID 和 ToolMessage Call ID；
• v2 type/ns/data、Namespace、Custom 事件；
• 带 Checkpointer 的 tasks Start / Finish；
• 确定性远程 Graph；
• Scripted Model 完整 Agent 两轮 Start / Check。

7.2 常见故障定位

现象	优先检查
Start 后模型不断 Check	System Prompt、路由评估和完整 Agent Trace
任务创建后长时间不执行	Queue Depth、Worker Slot、Run Start Delay
无 URL 同步调用报 ASGI 错误	是否误用 0.6.12 同步 .func；正常部署是否使用异步 Agent 入口
下一轮找不到 Task	Checkpointer、Supervisor thread_id、Reducer
Check Success 但没有结果	Thread State 是否有 messages；最后消息内容类型
Cancel 后外部任务仍执行	Cancel 是否协作式；外部系统是否支持撤销或补偿
List 状态与 Server 不同	缓存终态、过滤顺序、网络失败回退
页面没有 tasks 事件	Graph 是否配置 Checkpointer 和 config.thread_id
图片或工件路径不可访问	是否只是引用；Ownership、ACL、签名 URL 和 TTL

7.3 在线 Supervisor 记录

保存的在线记录使用第三方 OpenAI-compatible Gateway。gpt-5.5 是该端点返回的模型 ID / Gateway 别名，不应当作所有 Provider 都可直接使用的公共模型名。结果文件保留原字段名：

{
  "scene": "p09_async_agent_protocol",
  "status": "passed",
  "model": "gpt-5.5",
  "latency_ms": 25423,
  "tool_calls": ["start_async_task", "check_async_task"],
  "subagent_type": "research-service",
  "remote_status": "success",
  "task_id_equals_thread_id": true,
  "stream_message_events": 149,
  "usage": {"input_tokens": 49898, "output_tokens": 310, "total_tokens": 50208},
  "artifact": "/reports/research.md"
}

该记录能证明：

• 在线 Supervisor 选择了 Start；
• 后续明确查询时选择了 Check；
• 两轮复用了 Supervisor Thread 和 Task ID；
• 本地 Agent Protocol Run 成功；
• Supervisor 本地 messages Stream 有事件。

它不能证明：

• 149 是 149 个文本 Token；它们是 messages 模式 StreamPart / Chunk；
• 远程 Worker Token 和 Custom Progress 已实时转发到浏览器；
• Artifact 文件真实存在并通过 ACL；
• 断线续传、多任务合流和事件顺序已经完成；
• 生产认证和持久化可用。

49,898 Input Tokens 是多次 Supervisor 模型请求的累计 Usage，不是 Agent Protocol 本身需要近五万 Token。结果没有保存每次模型调用的独立 Usage 或 Span，无法可靠拆成 Start 与 Check 两行，本文不伪造分项数字。

高输入成本可能来自 Deep Agents Harness Prompt、完整 Tool Schema、消息历史重复和 Provider 统计方式。优化方向包括：限制 Supervisor Tool 集合、使用专用 Prompt、由确定性应用逻辑处理简单刷新、启用合适的 Prompt Cache，并在未来记录每次模型请求的独立 Usage。

supervisor_message_stream_parts 字段用于避免把本地消息流误解为远程事件或 Token 数。在线运行应保存分调用 Usage、Provider / Endpoint 类型、SDK 版本和远程 Stream 独立断言。

7.4 示例文件

p09_async_streaming/
├── async_config.py          # AsyncSubAgent 与 Supervisor
├── remote_agent.py          # 确定性远程 Graph
├── langgraph.json           # Agent Server 注册
├── real_protocol_check.py   # 真实 HTTP 底层契约
├── fake_protocol.py         # SDK 协议替身
├── protocol_harness.py      # Tool Runtime 与 State 支架
├── scripted_model.py        # 完整 Agent 的确定性模型
├── event_stream.py          # v2 与 tasks Stream
├── live_demo.py             # 在线 Supervisor
├── live_result.json         # 历史脱敏记录
└── test_async_config.py     # 30 个回归 Test Item

8. 总结

异步 SubAgent 的核心不是把函数改成 async def，而是建立可恢复的任务控制面。在 deepagents==0.6.12 中，远程 Thread 暂时兼任 Task ID，Run 表示一次执行，Supervisor 的 async_tasks 保存 Preview 任务索引，五个 Tool 负责启动、查询、更新、取消和列举。

这些对象之间没有自动提供完整的生产保证：Task ID 的持久性依赖两端存储，Interrupt 和 Cancel 不回滚外部副作用，State 缓存不等于远程事实，产品 UI 状态也不能直接冒充 Middleware 状态。

Event Streaming 是另一份协议。LangGraph v2 统一 type/ns/data，但本地 Supervisor Stream 与远程 Async Run Stream 仍是两条链路，需要事件网关、租户授权、关联 ID、Snapshot、去重和背压才能合并成可靠的用户体验。

本文用 30 个离线 Test Item、真实 HTTP Agent Server 和在线 Supervisor 记录分别验证不同层次，并明确每项证据不能证明什么。对于 Preview API，这种“源码契约 + 分层测试 + 诚实边界”比一句“已经测试通过”更重要。