《Deep Agents 实战笔记》第 6 篇 | 基于 deepagents (Python) 官方文档 原文：https://docs.langchain.com/oss/python/deepagents/async-subagents 、https://docs.langchain.com/oss/python/deepagents/dynamic-subagents

上一篇的同步子代理有个先天限制：主代理派出任务后会阻塞等待，子代理跑十分钟，主代理就干等十分钟，用户面对一个沉默的界面。这一篇讲两剂解药：异步子代理让主代理"派出去就不管"，动态子代理让代理在代码里批量编排子代理。从"派一个等一个"到"派出去就不管"，再到"用代码编排一切"，这是子代理的三段进化。

一、异步子代理：派出去就不管

异步子代理让主控代理（supervisor）启动立即返回的后台任务：start_async_task 拿到任务 ID 就返回，子代理在后台跑，主控代理继续和用户聊天。任务耗时长、可并行、需要中途改方向——这三种场景就是为它准备的。

配置方式是把 AsyncSubAgent 规格传给 subagents 参数：

from deepagents import AsyncSubAgent, create_deep_agent

async_subagents = [
    AsyncSubAgent(
        name="researcher",
        description="Research agent for information gathering and synthesis",
        graph_id="researcher",
        # 不设置 url → 使用 ASGI 传输（部署在同一个服务内）
    ),
    AsyncSubAgent(
        name="coder",
        description="Coding agent for code generation and review",
        graph_id="coder",
        # url="https://coder-deployment.langsmith.dev"  # 可选：远程时使用 HTTP 传输
    ),
]

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    subagents=async_subagents,
)

和同步子代理的字典定义不同，异步子代理指向的是 Agent Protocol 服务器上的一个图（graph_id）：不设 url 走 ASGI 传输（进程内函数调用，零网络延迟，官方推荐的默认方式）；设了 url 就走 HTTP，连接远程服务器——适合子代理需要独立扩缩容或由另一个团队维护的情况。部署拓扑也由此分出三档：单一部署（所有图注册在同一个 langgraph.json 里，ASGI 传输，官方推荐的起点）、拆分部署（主控一台服务器、子代理另一台）、混合部署（部分同机、部分远程，url 按需设置即可）。自托管服务器要自定义认证时，用规格里的 headers 字段附加请求头。

配置好后，AsyncSubAgentMiddleware 会给主控代理装上五个管理工具：

工具用途返回
start_async_task启动新的后台任务任务 ID（立即返回）
check_async_task获取任务当前状态和结果状态 + 结果（若已完成）
update_async_task向正在运行的任务发送新指令确认 + 更新后的状态
cancel_async_task停止正在运行的任务确认
list_async_tasks列出所有被跟踪任务及其实时状态所有任务的摘要

对主控代理的 LLM 来说，这就是五个普通工具，中间件自动处理线程创建、运行管理和状态持久化。一次典型交互的时序如下：

sequenceDiagram
    participant U as 用户
    participant S as 主控代理
    participant P as Agent Protocol 服务器

    U->>S: 调研主题 X
    S->>P: start_async_task(researcher)
    P-->>S: task_id: abc123（立即返回）
    S-->>U: 已启动任务 abc123
    Note over P: 子代理后台工作中
    U->>S: 顺便帮我查下别的
    S-->>U: （主控代理继续正常对话）
    Note over U,P: 子代理完成<br/>自动通知主控代理
    P-->>S: 通知：abc123 success
    S-->>U: 调研结果出来了，摘要如下

正在渲染图表…

注意最后那段：在 Agent Protocol 服务器上运行时，子代理完成会自动更新主控代理——完成、失败或需要输入时都会通知，不需要主动轮询。空闲的主控代理会被子代理直接唤醒（同部署场景），否则在下一次运行开始时收到更新。这个通知功能要求 deepagents>=0.7.26，目前是 beta。

通知以 HumanMessage 形式返回，在 additional_kwargs["deepagents_notification"] 里携带 TaskEvent（含 task_id、agent_name、status、run_id），你的应用可以把它和普通用户消息区分展示：

for message in state["messages"]:
    if event := message.additional_kwargs.get("deepagents_notification"):
        print(event["agent_name"], event["status"])  # success, error, timeout, or waiting

二、任务生命周期：一张状态图看懂

五个工具操作的是同一个后台任务，它的生命周期用状态图表达最清楚：

stateDiagram-v2
    [*] --> Running: start_async_task
    Running --> Success: 完成，通知主控代理
    Running --> Error: 出错，通知主控代理
    Running --> Timeout: 超时
    Running --> Waiting: interrupt 或 HITL 审批
    Waiting --> Running: 应用在子代理线程上 resume
    Running --> Cancelled: cancel_async_task
    Success --> [*]
    Error --> [*]
    Timeout --> [*]
    Cancelled --> [*]

正在渲染图表…

几个状态值得展开：

waiting（等待输入）。子代理触发 interrupt() 或人机协同审批时会暂停在这个状态，check 和 list 会报告它并附带挂起的中断请求。关键纪律是：主控代理不能代替用户回答或批准——那个请求是面向用户的。你的应用需要在子代理自己的线程上回答（任务 ID 就是子代理的线程 ID）：

from langgraph_sdk import get_client

async def answer_waiting_subagent(task_id: str, graph_id: str, answer: object) -> None:
    client = get_client()  # 若子代理在另一个部署上，传入 url=...
    thread = await client.threads.get(task_id)  # 任务 ID 就是子代理的线程 ID
    interrupt = next(iter(thread["interrupts"].values()))[0]
    await client.runs.create(
        task_id,
        graph_id,
        command={"resume": {interrupt["id"]: answer}},
    )

如果子代理需要的是主控代理（而非用户）提供的信息，官方建议让它把问题写在结果里，主控代理再用 update_async_task 回复。另外注意：更新一个 waiting 状态的任务会丢弃它挂起的审批请求；取消这类任务只是标记状态，因为它根本没有正在运行的 run。

update 的真实语义。update_async_task 不是"发个消息"那么简单——它会在同一线程上创建一个带 interrupt 策略的新运行，上一次运行被中断，子代理带着完整对话历史和新增指令重新开始，任务 ID 不变。这是"中途调整方向"的实现方式。

三、为什么任务清单不怕上下文压缩

这是异步子代理设计里最容易被忽略、也最能体现工程功力的一处：任务元数据存储在主控代理图上一个独立的状态通道（async_tasks）里，与消息历史分开。

为什么要多此一举？因为深度代理会在上下文窗口填满时对消息历史做总结压缩（第 7 篇的主题）。如果任务 ID 只存在于工具消息里，几轮压缩之后就被总结没了——主控代理会"忘记"自己派过哪些任务，list_async_tasks 变成空手套白狼。专用通道保证了即使消息历史被压缩了 N 轮，主控代理依然记得自己的所有后台任务。每个被跟踪的任务记录任务 ID、代理名称、线程 ID、运行 ID、状态和时间戳。

配套的还有一个贴心设计（同样要求 >=0.7.26）：start_async_task 创建的每个子代理线程都带 metadata={"parent_thread_id": ...}，即主控代理的线程 ID。你可以用它反查某个主控代理的所有子代理线程，子代理也能在图工厂函数里读出父线程——比如复用主控代理的线程级沙箱。

四、轮询排错：官方故障排查四则

异步模式解放了主控代理，但也引入了新的行为问题。官方排查指南列了四种高频翻车，每一个都很有代表性：

**1. 启动后立刻疯狂轮询。**症状是主控代理 start 完循环调用 check，把异步执行硬生生变成了阻塞执行——异步的意义全无。中间件注入的系统提示词规则本应防止这种情况，但如果模型还是忍不住，就在主控代理的系统提示词里强化：

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    system_prompt="""...your instructions...

    After launching an async subagent, ALWAYS return control to the user.
    Never call check_async_task immediately after launch.""",
    subagents=async_subagents,
)

**2. 报告过期状态。**主控代理引用对话历史里较早的任务状态，而不是发起一次全新的 check。中间件提示词会告诉模型"对话历史中的任务状态永远是过期的"——这句值得抄进你自己的提示词里。仍出现就加明确指令：报告状态前必须先调 check 或 list。

**3. 任务 ID 被截断改写。**主控代理把 abc123def456... 缩写成 "abc123..."，check 和 cancel 全部失败。这通常是特定模型的毛病，换模型或在系统提示词里加"始终展示完整 task_id，绝不截断或缩写"。

**4. 子代理排队不跑。**任务启动后长时间挂起，多半是本地 langgraph dev 的工作线程池耗尽了——每个活跃运行占一个槽位，1 个主控 + 3 个并发子代理就需要 4 个槽位。加大线程池：

langgraph dev --n-jobs-per-worker 10

五、动态子代理：用代码编排子代理

异步解决的是"等待"问题，动态子代理解决的是编排问题。回顾一下同步/异步模式下的委派：主代理的 LLM 每次决定调用一个 task 工具，要审查 50 个文件就得发 50 次工具调用——模型逐次决策，既慢又贵还可能漏。

动态子代理换了个思路：给代理挂上解释器后，解释器暴露一个内置的 task() 全局函数，代理可以编写 JavaScript 编排脚本，用循环、分支和并行批处理把工作分发到多个子代理上，再在代码里汇总结果。一次 task() 调用运行完整的子代理循环并返回其结果：

const review = await task({
  description: "Review src/auth/login.ts for auth issues. Cite line numbers.",
  subagentType: "reviewer",
  responseSchema: issuesSchema, // 结构化输出 schema（可选）
});

// 传入 responseSchema 后，结果已经是带类型的值，无需 JSON.parse
const critical = review.issues.filter((issue) => issue.severity === "high");

启用只需两步：安装 QuickJS 解释器包，再加中间件：

pip install -U "deepagents[quickjs]"

from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    subagents=[{
        "name": "reviewer",
        "description": "Reviews code for security issues, citing lines and severity",
        "system_prompt": "You are a security-focused code reviewer. Report issues with line numbers and severity.",
    }],
    middleware=[CodeInterpreterMiddleware()],
)

解释器要求 langchain-quickjs>=0.2.0 和 Python 3.11+；整个动态子代理运行时目前是 beta。想强制走常规 task 工具路径，传 CodeInterpreterMiddleware(subagents=False) 即可。

这套机制的学术背景也值得一提：官方明确说它支持递归语言模型（RLM）工作流——把工作集保存在解释器变量里，选取切片，用 task() 调用子代理处理，再综合结果。换句话说，解释器不只是"批处理脚本工具"，它是代理递归分解自己的工作空间的基础设施。

**一个奇技淫巧：把请求表述为 "workflow"。**内置解释器系统提示词会把 "workflow" 这个词视为"应该从代码中编排工作"的信号。想主动触发动态编排，就在请求里带上这个词：

result = agent.invoke({
    "messages": [{"role": "user", "content": "Run a workflow that reviews every file in src/routes/ and summarizes the top risks."}]
})

如果只是一次简单直接的委派，就用平实的表述，别乱加 "workflow"。

六、五种编排模式

动态编排的策略不是配置出来的，而是代理根据任务形态写代码写出来的。官方文档给了五种常见形态，共享同一套编排方式：把工作保存在 JS 变量里，用 task() 分派，在代码中组合结果。

1. 分类后处理（Classify and act）：先分类，再按类别路由给专门化子代理。适合工单分诊、错误日志、用户反馈。

2. 扇出与综合（Fan-out and synthesize）：同一种工作并行分派到许多条目，再合并各路结果。目录级代码审查、批量文档分析都是典型。这里有个组合技：解释器内要先"发现文件"得靠程序化工具调用（PTC），需要在中间件上显式开允许列表，比如 CodeInterpreterMiddleware(ptc=["glob"])，然后代理就能在代码里 await tools.glob(...) 拿到文件清单再扇出。

3. 对抗式验证（Adversarial verification）：两遍模式。第一遍产生发现，第二遍把每个发现交给独立验证者，只有通过确认的才保留——误报代价高的安全审计、合规检查必备。

4. 生成与过滤（Generate and filter）：多个子代理对同一问题各自生成独立方案，代码里比较评分，只留最好的。适合架构方案、重构策略这类"先探索再决策"的任务。

5. 锦标赛（Tournament）：裁判子代理对变体两两对决，胜者淘汰赛晋级。适合主观标准下的优化和风格挑选。拓扑长这样：

flowchart LR
    A1["变体 1"] --> J1{"裁判对决 1"}
    A2["变体 2"] --> J1
    A3["变体 3"] --> J2{"裁判对决 2"}
    A4["变体 4"] --> J2
    J1 -->|"胜者"| JF{"决赛"}
    J2 -->|"胜者"| JF
    A5["变体 5 轮空直进"] --> JF
    JF --> W["获胜版本"]
    style J1 fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
    style J2 fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
    style JF fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
    style W fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a

正在渲染图表…

锦标赛的核心代码是一个淘汰赛循环，裁判通过结构化输出返回选择：

let bracket = await Promise.all(
  [1, 2, 3, 4, 5].map((n) =>
    task({ description: `Rewrite processOrder for readability (variant ${n}).`, subagentType: "writer" }),
  ),
);

while (bracket.length > 1) {
  const winners = [];
  for (let i = 0; i < bracket.length; i += 2) {
    if (bracket[i + 1] === undefined) { winners.push(bracket[i]); break; }
    const { winner } = await task({
      description: `Pick the more readable:\n\nA:\n${bracket[i]}\n\nB:\n${bracket[i + 1]}`,
      subagentType: "judge",
      responseSchema: pickSchema, // -> { winner: "A" | "B" }
    });
    winners.push(winner === "A" ? bracket[i] : bracket[i + 1]);
  }
  bracket = winners;
}
bracket[0]; // 获胜的重写版本

顺带一提：想最快体验动态子代理，可以试试 dcode（LangChain 的终端编码代理），它默认启用解释器，动态子代理开箱即用，还会在面板里按调度阶段实时展示子代理。

七、实战作业：3 个子代理对抗评审

现在把本篇的两个技术点拧在一起，做一个"3 个子代理对抗评审"demo：1 个审计子代理找漏洞，每个发现再派 3 个独立验证子代理背靠背验证，多数投票（≥2 票）才确认——这正是对抗式验证模式的"3 人陪审团"变体。

flowchart TB
    T["目标代码 payments 模块"] --> R["reviewer<br/>审计找漏洞"]
    R --> F1["发现 1"]
    R --> F2["发现 2"]
    F1 --> V1
    F2 --> V1
    subgraph JURY["每个发现派 3 个独立验证者"]
        V1["verifier 1"]
        V2["verifier 2"]
        V3["verifier 3"]
    end
    V1 --> MV{"多数投票<br/>≥2 票确认"}
    V2 --> MV
    V3 --> MV
    MV --> OK["确认漏洞清单"]
    MV --> NO["剔除误报"]
    style R fill:#fdeef0,color:#701d2a,stroke:#c94a5e
    style JURY fill:#f4effb,color:#4a2170,stroke:#8b52c9
    style MV fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
    style OK fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a

正在渲染图表…

**第一步：配置代理。**定义 reviewer 和 verifier 两个专门化子代理，挂上解释器中间件。给 verifier 的系统提示词里写上"Be skeptical"很关键——验证者要带怀疑立场，否则会顺着报告放水：

from deepagents import create_deep_agent
from langchain_quickjs import CodeInterpreterMiddleware

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    subagents=[
        {
            "name": "reviewer",
            "description": "Finds potential security vulnerabilities in code",
            "system_prompt": "You are a security auditor. Find potential vulnerabilities and report each with file, line, and description.",
        },
        {
            "name": "verifier",
            "description": "Independently verifies whether a reported vulnerability is real",
            "system_prompt": "You are a security verification specialist. Given a reported vulnerability, independently verify whether it is exploitable. Be skeptical. Only confirm real issues.",
        },
    ],
    middleware=[CodeInterpreterMiddleware()],
)

第二步：用 "workflow" 触发动态编排。

result = agent.invoke({
    "messages": [{"role": "user", "content": "Run a workflow that audits the payments module for vulnerabilities, then adversarially verifies every finding with 3 independent verifiers and keeps only majority-confirmed issues."}]
})

**第三步：理解代理会写出的编排脚本。**大致长这样（官方对抗式验证示例的 3 人陪审团扩展）：

// 第一遍：审计
const { findings } = await task({
  description: "Audit the payments module for vulnerabilities.",
  subagentType: "reviewer",
  responseSchema: findingsSchema, // -> { findings: [{ id, file, line, description }] }
});

// 第二遍：每个发现派 3 个独立验证者，多数投票
const confirmed = [];
for (const f of findings) {
  const verdicts = await Promise.all(
    [1, 2, 3].map((n) =>
      task({
        description: `Verify finding #${f.id} at ${f.file}:${f.line} (${f.description}). Confirm or refute. Review pass ${n}.`,
        subagentType: "verifier",
        responseSchema: verdictSchema, // -> { confirmed: boolean, reason: string }
      }),
    ),
  );
  const yes = verdicts.filter((v) => v?.confirmed).length;
  if (yes >= 2) confirmed.push({ ...f, votes: `${yes}/3` });
}
confirmed;

跑起来后建议开着 LangSmith 看 trace：你会发现第一遍 1 个 reviewer run，第二遍每个发现 3 个 verifier run，全部通过线程 ID 关联——"误报被过滤掉"这件事第一次变得可审计。

两个进阶方向留给读者：把验证者改成异步子代理（AsyncSubAgent），审计结果出来后主控代理继续干别的，验证完成自动通知；或者在 3 个 verifier 的描述里注入不同的审查视角（利用条件、运行时环境、供应链），让"陪审团"真正意见独立，而不是三个复读机。

本篇踩坑点

  • 主控代理忍不住轮询：启动后立刻循环 check，异步变同步。系统提示词里明确"启动后必须把控制权还给用户"；
  • 报告过期状态：对话历史里的任务状态永远不可信，报告前必须重新 check 或 list；
  • 任务 ID 被截断：部分模型会缩写长 ID 导致 check/cancel 失败，提示词里禁止截断，必要时换模型；
  • 本地跑并发子代理卡住：langgraph dev 默认线程池太小，用 --n-jobs-per-worker 加大（1 主控 + N 子代理 ≥ N+1 个槽位）；
  • waiting 状态别越权代答：主控代理不能替用户批准中断请求，必须在子代理线程上由应用层 resume；
  • 忘了 "workflow" 触发词：动态编排是隐式触发的，想让代理写代码批量派发，请求里就带上这个词；反之简单委派别乱用；
  • 解释器里发现文件要开 PTC：默认关闭，CodeInterpreterMiddleware(ptc=["glob"]) 显式允许列表才生效。

下一篇预告：《07 | 上下文工程实战：长任务不崩的秘密》——子代理越多，上下文管理越重要。系统提示、运行时 context、自动总结压缩、大结果落盘四板斧，以及官方给出的上下文工程 Do & Don't 对比案例。

参考资料：