数据截至 (上游 commit 25aa2735dabb)
子 agent:三种形态、隔离的上下文与远程任务
30 秒导读: 主 agent 的上下文窗口是最贵的资源。子 agent 就是「把一大坨探索赶进别人的上下文里干完,只把一句结论拿回来」。Deep Agents 把这件事做成三种 spec:声明式
SubAgent、直接给编译好 runnable 的CompiledSubAgent、跑在远端 LangGraph 部署上的AsyncSubAgent。本章讲清三者怎么分流、声明式那种为什么会被父层重新组栈、task工具内部到底做了哪几步、以及父子之间的 state 是怎么被裁剪的。
引用约定: 本章正文里的
path:line默认相对克隆里的 Python 包目录libs/deepagents/deepagents/——例如graph.py:647的完整路径是libs/deepagents/deepagents/graph.py:647。两个例外一律相对克隆根: 以libs/或examples/开头的引用(它们落在包目录之外),以及末尾 §11 代码地图的整张表。
前置章节:装配流水线 讲了 create_deep_agent 的整体组栈顺序,文件系统与权限 讲了 permissions 的语义。本章只讲「委派」这一条线。
1. 为什么要子 agent(先讲动机,不看代码)
一句话:上下文窗口是内存,子 agent 是把临时变量丢进函数作用域里算完就释放。
主线程里的每一条 tool 结果都是永久成本——它会一直留在 messages 里,后面每一轮都要重新发给模型。可一次「把整个仓库翻一遍找出所有用到某 API 的地方」会产生几十条 grep 结果,而你真正想要的只有最后那句「有 3 处,分别在 A/B/C」。
于是有了这条划分:
| 什么样的活该委派 | 为什么 |
|---|---|
| 探索型(搜索、翻文件、试错) | 中间过程量大,结论极短 |
| 可并行(研究三个人物、准备两份议程) | 彼此不依赖,可以同时开三个上下文 |
| 只要结果、不要过程 | 中间步骤对主线程没有复用价值 |
| 需要另一套工具/权限/模型 | 换一套装备比在主 agent 上开关工具更干净 |
反过来,官方工具描述也明说了什么时候不该委派:任务很小、需要看到中间推理、拆开反而增加延迟——这些"何时用/何时不用"的指引现在直接写在 TASK_TOOL_DESCRIPTION 里(middleware/subagents.py:285-296;旧版独立的 TASK_SYSTEM_PROMPT 已并入工具描述)。
关键约束(理解全章的钥匙): 子 agent 是一次性、单向的——父层只发一段 description 过去,子 agent 只回一条最终消息,中途不能对话。这条约束被反复写进工具描述里(middleware/subagents.py:292:"Each invocation is stateless")。正因为不能对话,父层才必须在 description 里把话说全。
用起来长这样:
# 示意,非源码
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-5",
subagents=[
{
"name": "researcher", # 父层用这个名字调 task()
"description": "深挖一个主题并给出摘要", # 模型靠这句决定要不要委派
"system_prompt": "你是研究员,返回结构化摘要。",
"tools": [web_search], # 不写就继承父层工具
}
],
)
主 agent 拿到的不是"researcher 这个 agent",而是一个叫 task 的普通工具;调用时填 subagent_type="researcher" 就行(middleware/subagents.py:272-283,TaskToolSchema)。
2. 三种形态与分流判据
这节讲:同样是 subagents=[...] 这一个参数,里面可以塞三种完全不同的东西,Deep Agents 靠字段存在性把它们分开。
2.1 三种形态对照
| 形态 | 你提供什么 | 谁来编译 | 识别字段 | 定义位置 |
|---|---|---|---|---|
SubAgent(声明式) | prompt / tools / model / middleware / skills / permissions / interrupt_on / response_format | 父层重新组栈后,由 create_sub_agent 编译 | 既无 graph_id 也无 runnable | middleware/subagents.py:36(SubAgent) |
CompiledSubAgent | 一个已编译好的 Runnable | 你自己,父层不碰 | 有 runnable | middleware/subagents.py:167(CompiledSubAgent) |
AsyncSubAgent(远程) | graph_id,可选 url / headers | 远端 LangGraph 部署 | 有 graph_id | middleware/async_subagents.py:34(AsyncSubAgent) |
三者的差别不只是"怎么造",更是运行语义不同:
| 形态 | 阻塞? | 结果怎么回来 | 归哪个中间件管 |
|---|---|---|---|
SubAgent | 阻塞,一次 invoke 打完 | 直接变成 ToolMessage | SubAgentMiddleware |
CompiledSubAgent | 阻塞 | 同上 | SubAgentMiddleware |
AsyncSubAgent | 不阻塞,立刻返回 task_id | 后续靠 check_async_task 拉 | AsyncSubAgentMiddleware |
2.2 分流判据:两个 in 判断
分流发生在 create_deep_agent 内部,就是两行朴素的字典键检查(graph.py:647-655):
for spec in subagents or []:
if "graph_id" in spec: # AsyncSubAgent
async_subagents.append(cast("AsyncSubAgent", spec))
continue
if "runnable" in spec: # CompiledSubAgent —— 原样收下
inline_subagents.append(spec)
else: # SubAgent —— 下面几十 行都在重组它
...
注意顺序:graph_id 先判,runnable 后判,都不满足才走声明式分支。所以同时带 graph_id 和 runnable 的 spec 会被当成远程的(graph.py:648-654)。
2.3 一张图看完整条分流
从上往下读,每个判断命中即停:
create_deep_agent(subagents=[...])
│
├─ 有 graph_id ? ──是──▶ async_subagents ──▶ AsyncSubAgentMiddleware
│ (5 个远程任务工具)
├─ 有 runnable ? ──是──▶ inline_subagents(原样,不重组栈)
│
└─ 否则:声明式 ────────▶ 重新组栈 ──▶ inline_subagents
│
(没人叫 general-purpose 且 profile 没禁用)
把默认 general-purpose 插到队首
│
inline_subagents 非空 ──▶ SubAgentMiddleware
(注入 task 工具)
两条要记住的边界:
inline_subagents为空 → 主 agent 根本没有task工具(graph.py:827-840,if inline_subagents:)。只配了远程子 agent 时,主 agent 有 5 个 async 工具但没有task。SubAgentMiddleware被列进_REQUIRED_MIDDLEWARE(graph.py:238-248),harness profile 的excluded_middleware不许把它剔掉——剔了task工具就哑了,所以那里宁可抛ValueError。
3. 声明式 子 agent 会被父层「重新组栈」
这节讲:为什么写三行 spec 就能得到一个功能齐全的子 agent——因为父层替你把整套中间件又搭了一遍。
3.1 思路:子 agent 不是"缩水版主 agent",是"另一台完整的 agent"
一个只会调 web_search 的子 agent 是没用的——它不会读写文件、不会记 todo、上下文满了不会压缩。所以 Deep Agents 的做法是:为每一个声明式 spec 单独建一套默认中间件栈,然后才把用户自己写的 middleware 拼上去。
默认三件套(graph.py:667-675):
| 中间件 | 给子 agent 带来什么 |
|---|---|
FilesystemMiddleware(...) | 文件工具族 + 权限闸门 |
create_summarization_middleware(subagent_model, backend) | 子 agent 自己的上下文压缩 |
PatchToolCallsMiddleware() | 工具调用修补 |
注意 create_summarization_middleware 传的是 subagent_model 而不是父模型——压缩阈值按子 agent 自己的模型算。
有 skills 就再追加一个 SkillsMiddleware(graph.py:676-678);子 agent 栈里同样没有 TodoListMiddleware(它已从默认栈整体移除)。
3.2 每个子 agent 单独走一遍 harness profile 流程
这是最容易被忽略的一点:子 agent 换了模型,profile 也跟着换。
spec.get("model", model)
│
▼
_harness_profile_for_model(subagent_model, _subagent_spec) graph.py:657-661
│
├─▶ tool_description_overrides → 喂给 FilesystemMiddleware / 工具描述改写
├─▶ materialize_extra_middleware() → 追加 profile 自带中间件 graph.py:682
├─▶ excluded_middleware → _apply_excluded_middleware(跑两遍) graph.py:693-709
└─▶ excluded_tools → 尾部挂 _ToolExclusionMiddleware graph.py:717-718
_apply_excluded_middleware 跑两遍(用户中间件拼进来之前一遍、之后一遍),是为了让用户塞进来的中间件也逃不过 profile 的排除名单(graph.py:693-709)。
用户中间件的拼接用 _apply_custom_middleware,并且把 core_names 定在profile 尾巴追加之前(graph.py:680,_subagent_core_names)——同名的就地替换,新名字插在核心段末尾、profile 尾巴之前(graph.py:201-235)。
3.3 继承 vs 覆盖:一张表定生死
这是使用者最需要背下来的一张表。判据全在源码里明摆着:
| 配置项 | 源码判据 | 不写 | 写了 |
|---|---|---|---|
model | graph.py:657 spec.get("model", model) | 继承父模型 | 覆盖,并连带换 harness profile |
permissions | graph.py:664 spec.get("permissions", permissions) | 继承父规则 | 整体替换,不是合并(middleware/subagents.py:116-125) |
interrupt_on | graph.py:720 spec.get("interrupt_on", interrupt_on) | 继承父配置 | 覆盖;随后再与权限推导出的闸门合并(graph.py:721-724) |
tools | graph.py:657 spec.get("tools") if "tools" in spec else tools | 继承父工具集 | 以 spec 为准;末值再经 or [] 兜底(graph.py:737) |
middleware | graph.py:699-703 _apply_custom_middleware | 只有默认四件套 | 同名替换 / 新名插入核心段尾 |
skills | graph.py:676-678 | 不装 SkillsMiddleware | 单独建一个 SkillsMiddleware |
system_prompt | graph.py:740 _apply_profile_prompt | 必填字段,没有默认 | 再叠 harness profile 的前后缀 |
三条最容易踩的坑:
permissions是替换语义。给子 agent 写一条"只读/tmp"的规则,父层那些规则就全部消失了,而不是叠加。tools不写才继承。想要"父工具 + 一个额外工具",必须自己把父工具列全。CompiledSubAgent什么都不继承——不继承interrupt_on(graph.py:493-503的 docstring)、不继承state_schema(graph.py:517-530)。你自己编译的,你自己负责。
最后所有东西合成一个 processed_spec,进 inline_subagents(graph.py:734-743)。
4. 默认的 general-purpose 子 agent
这节讲:为什么你什么都不配,主 agent 也有个 task 工具可用。
4.1 自动注入的条件
create_deep_agent 会自动补一个叫 general-purpose 的子 agent,但有两道闸门(graph.py:750-751):
gp_profile = _profile.general_purpose_subagent or GeneralPurposeSubagentProfile()
if gp_profile.enabled is not False and not any(
spec["name"] == GENERAL_PURPOSE_SUBAGENT["name"] for spec in inline_subagents
):
翻成人话,两个条件都满足才注入:
| 闸门 | 含义 |
|---|---|
gp_profile.enabled is not False | harness profile 没显式禁用。enabled 是三态:None 继承/默认开、True 强制开、False 关(profiles/harness/harness_profiles.py:97-111) |
没人已经叫 general-purpose | 你自己写一个同名 spec,就是"覆盖默认"的正规姿势 |
写成 is not False 而不是 if gp_profile.enabled 是有讲究的:None 表示"没表态",要按开处理;只有明写 False 才关。
先处理调用者的 subagents、再决定要不要补默认的,这个顺序是刻意的——否则会先把 GP 那一整套中间件建出来再丢掉,而 extra_middleware 里可能有工厂函数,白跑一次有副作用(graph.py:745-749 的注释)。
4.2 GP 的栈和普通子 agent 有什么不一样
同样是三件套 + skills + profile 尾巴,但用的是父层的 profile 和父层的 model(graph.py:752-760)。真正的差别在中间件继承:
_gp_inheritable = [m for m in (middleware or []) if m.name in _gp_original_name_to_index]
gp_middleware = _apply_custom_middleware(gp_middleware, _gp_inheritable)
graph.py:768-778:主 agent 的自定义中间件里,只有那些名字撞上 GP 默认槽位的才会被继承下去。理由写在注释里——覆盖默认槽位的(比如你换掉了 FilesystemMiddleware)应该跟过去,纯属主 agent 专用的那些不该跟。
其余细节:
- 描述和 prompt 可被 profile 覆盖;GP 专属 prompt 优先于
profile.base_system_prompt,只有 profile 的 suffix 会再叠上去(graph.py:796-806)。 - 工具用主 agent 那份改 写过描述的
_tools(graph.py:793)。 - 最后
insert(0, ...)插到队首(graph.py:814),所以它在工具描述的 available agents 列表里排第一。
4.3 关掉之后会怎样
GeneralPurposeSubagentProfile.enabled = False 且你没配任何同步子 agent → inline_subagents 为空 → task 工具整个不出现。这条在 profile 的 docstring 里明写了(profiles/harness/harness_profiles.py:107-110)。远程 AsyncSubAgent 不算数,它们走另一套工具。
5. task 工具解剖
这节讲:那个唯一暴露给模型的 task 工具,内部五步分别在干什么。
5.1 模型看到的是什么
模型看到的接口极简,只有两个字符串参数(middleware/subagents.py:272-283,TaskToolSchema):
| 参数 | 作用 |
|---|---|
description | 完整的任务说明。因为不能追问,这里必须一次说全 |
subagent_type | 选哪个子 agent,必须是 工具描述里列出的名字之一 |
{available_agents} 占位符是把静态 prompt 和动态配置缝起来的针脚。TASK_TOOL_DESCRIPTION 里留了一个 {available_agents}(middleware/subagents.py:288),构建时用 - 名字: 描述 的列表填进去(middleware/subagents.py:466-472):
subagent_description_str = "\n".join(f"- {s['name']}: {s['description']}" for s in compiled_subagents)
if task_description is None:
description = TASK_TOOL_DESCRIPTION.format(available_agents=subagent_description_str)
elif "{available_agents}" in task_description:
description = task_description.format(available_agents=subagent_description_str)
else:
description = task_description
三分支的意思:不给自定义描述就用内置模板;给了且带占位符就替换;给了但没带占位符,那模型就看不见有哪些子 agent 了——这正是 graph.py:832-836 那段注释在警告的事。
同一份名字列表还会被追加到主 agent 的 system prompt(middleware/subagents.py:697-700),所以"有哪些子 agent"在工具描述和系统提示里各出现一次;什么时候该委派、"尽量并 行开多个 task"这类指引则在 TASK_TOOL_DESCRIPTION 的 usage notes 里(middleware/subagents.py:290-296)。
5.2 一次 task 调用的五步
_build_task_tool(middleware/subagents.py:402-605)是个大闭包,把四个内部函数和一批预编译产物关在一起。一次调用的流向:
父 agent 调 task(description, subagent_type)
│
① 名字校验 —— 不在册就 return 一句错误字符串(不抛异常) subagents.py:547-549
│
② _select_subagent —— config 里带 response_format 就现编,否则用预编译的
│ subagents.py:514-527
③ _validate_and_prepare_state subagents.py:529-540
│ 父 state ─ 去掉 _EXCLUDED_STATE_KEYS ─ 去掉 private_state_keys
│ └─▶ messages 换成 [HumanMessage(description)]
│
④ subagent.invoke(隔离后的 state) ← 外面包 _subagent_tracing_context
│ subagents.py:565-567
⑤ _return_command_with_state_update subagents.py:474-512
结构化输出 → JSON 串;否则回溯取最后一条非空 AIMessage 文本
其余 state 键并回父层;messages 只回一条 ToolMessage
第①步返回字符串而不是抛异常——模型选错名字是可恢复错误,把可用名字列出来让它重试比炸掉整条链好(middleware/subagents.py:547-549)。
5.3 懒编译:_compile_spec
_compile_spec(middleware/subagents.py:423-460)统一处理两种同步形态:
- 有
runnable:不重新编译,只用with_config打上lc_agent_name/run_name。注释说得很清楚——用with_config而非改属性,是为了不动原对象,好让同一个 runnable 注册到多个名字下(注释在middleware/subagents.py:434-436,对应的调用体在437-443)。 - 没有
runnable:调create_sub_agent(spec, state_schema=..., response_format=...)现编。
构建工具时先把所有 spec 编一遍存成 subagent_graphs(middleware/subagents.py:462-464),运行期直接查表。只有需要动态 response_format 时才走"现编一个"的慢路径——这就是 _select_subagent 的全部逻辑(middleware/subagents.py:519-523)。
create_sub_agent(middleware/subagents.py:333-386)本身很薄:
| 它做的事 | 位置 |
|---|---|
强校验 model / tools 必须存在,否则 ValueError | subagents.py:358-363 |
有 interrupt_on 就在中间件尾部追加 HumanInTheLoopMiddleware | subagents.py:370-372 |
动态 response_format 优先于 spec 里的 | subagents.py:374 |
有 state_schema 才传给 create_agent(否则用默认) | subagents.py:382-383 |
5.4 结果怎么变回一条消息
_return_command_with_state_update(middleware/subagents.py:474-512)按这个优先级取内容:
structured_response非None→ JSON 序列化。分三种情况:pydantic 走model_dump_json()、dataclass 走dataclasses.asdict、其余走json.dumps(middleware/subagents.py:487-492)。- 否则从后往前找最后一条文本非空的
AIMessage(middleware/subagents.py:495-502)。
第 2 条的"从后往前 + 跳过空消息"是个真实的踩坑记录,注释直说:Anthropic 偶尔会在最后一次成功工具调用后再吐一条空的 end_turn AIMessage,不跳过就会给父层一条空 ToolMessage。
另外,result 里除去 _EXCLUDED_STATE_KEYS 的其余字段会原样并回父层 state(middleware/subagents.py:484)——所以子 agent 写的文件、改的中间件 state,父层是看得见的。真正隔离的只是对话历史。
6. 状态隔离:什么该过去、什么该回来
这节讲:父子之间那道 state 过滤网,由两层名单构成。
6.1 第一层:硬编码的 _EXCLUDED_STATE_KEYS
三个键,进出双向都 挡(middleware/subagents.py:252-269):
| 键 | 为什么挡 |
|---|---|
messages | 出:只回一条最终 ToolMessage;入:换成一条 HumanMessage(description) |
todos | 没有定义好的 reducer,父子 todo 合并没有明确语义 |
structured_response | 同上,而且它已经被转成 ToolMessage 内容了 |
6.2 第二层:动态算出来的 private_state_keys
这层是"中间件的私房钱"。有些中间件会往 state 上挂只给自己用的字段(用 PrivateStateAttr 标注)。这类字段漏进子 agent 有两个坏处:一是子 agent 的同名中间件会读到父层的内部游标,二是它本来就不是公共契约。
算法在 middleware/_state.py:13-22(private_state_field_names):遍历传进来的所有 state schema,用 get_type_hints(..., include_extras=True) 取带注解的类型,凡是 Annotated[..., PrivateStateAttr] 的字段名收进一个 frozenset。_has_marker(middleware/_state.py)会递归下钻泛型参数,所以嵌在 NotRequired[...] 里的标记也能认出来。
装配收尾时才算(graph.py:894-898):
private_state_keys = private_state_field_names(
*(mw.state_schema for mw in deepagent_middleware if getattr(mw, "state_schema", None) is not None)
)
if sub_agent_middleware is not None:
sub_agent_middleware.private_state_keys = private_state_keys
为什么必须最后算? 因为要等整条主栈(含用户中间件、profile 中间件、MemoryMiddleware 等)全部定型,才知道有哪些私有字段。但 SubAgentMiddleware 在 graph.py:827-840 早就实例化了——于是 private_state_keys 被写成一个带 setter 的 property,赋值时把 task 工具整个重建一遍(middleware/subagents.py:711-720),让新的名单进到闭包里。这是本章最"绕"但也最值得看的一处设计。
过滤动作本身就两行(middleware/subagents.py:537-538):
subagent_state = {k: v for k, v in runtime.state.items() if k not in _EXCLUDED_STATE_KEYS}
subagent_state = {k: v for k, v in subagent_state.items() if k not in private_state_keys}
顺带一提,SubAgentMiddleware.subagent_names 是个公开的 frozenset(middleware/subagents.py:686-688),存在的理由写在 docstring 里:让流式输出的消费方能直接拿到子 agent 名单,不必去扒 task 工具的闭包。
7. 结构化输出透传与 tracing 标记
这节讲两个小而实用的旁路机制。
7.1 每次调用换一个 response_format
有时同一个子 agent,这次要它返回 Findings,下次要它返回 Summary。Deep Agents 不让你为此建两个子 agent,而是走 config 通 道:
调用方在 RunnableConfig["configurable"] 里塞
"__deepagents_subagent_response_format": <schema>
│ 常量:SUBAGENT_RESPONSE_FORMAT_CONFIG_KEY subagents.py:32
▼
_get_subagent_response_format(runtime) 读出来 subagents.py:388-401
▼
_select_subagent 发现非 None ──▶ _compile_spec(现编一个) subagents.py:519-523
▼
create_sub_agent(response_format=...) 覆盖 spec 自带的 subagents.py:374
限制很明确:CompiledSubAgent 不支持。_compile_spec 在 runnable 分支上直接抛 ValueError,提示"动态 schema 需要一个 raw SubAgent spec"(middleware/subagents.py:430-432)。道理也直白——运行期没法给一个已编译的图换输出 schema。
_get_subagent_response_format 写得很防御:config 不是 dict、configurable 不是 dict、值是 None,统统返回 None 走快路径(middleware/subagents.py:390-401)。
7.2 让 LangSmith 分得清主线程和子线程
_subagent_tracing_context(middleware/subagents.py:309-332)是个 contextmanager,只干一件事:把 langsmith tracing context 的 metadata 里加上 ls_agent_type="subagent",对齐 LangChain 给根 agent 打 "root" 的做法。
写法上有个细节值得学:它先 get_tracing_context() 拿到当前全部字段,再 {**current, "metadata": merged_metadata} 整体透传回去(middleware/subagents.py:320-327)。注释解释了动机——只改 metadata,其他字段(parent、client、tags)原样带过,这样 langsmith 以后加新字段也不会被这层包装吞掉。
配置侧同样只做加法。task 里只塞一个 {"configurable": {"ls_agent_type": "subagent"}}(middleware/subagents.py:565),父层的 callbacks / tags / configurable 靠 langgraph 的 ensure_config 自动逐键合并——注释点名了 langgraph#7926 和 deepagents#3634,并说明手动转发反而会双计(比如 tags 重复)。
8. 异步 / 远程子 agent
这节讲:当子 agent 跑在另一台机器上、而且一跑就是十分钟时,模型该怎么和它打交道。
8.1 换了一套心智模型
同步子 agent 是"函数调用",异步子 agent 是"提交作业 + 轮询"。因此工具从 1 个变成 5 个(middleware/async_subagents.py:815-839,_build_async_subagent_tools):
| 工具 | 干什么 | 实现 |
|---|---|---|
start_async_task | 建 thread + 建 run,立刻返回 task_id | async_subagents.py:245-341 |
check_async_task | 拉一次状态;成功了顺便取结果 | async_subagents.py:407-475 |
update_async_task | 在同一 thread 上开新 run,打断旧的 | async_subagents.py:476-579 |
cancel_async_task | 取消当前 run | async_subagents.py:580-658 |
list_async_tasks | 批量拉活状态 | async_subagents.py:729-814 |
ASYNC_TASK_TOOL_DESCRIPTION(middleware/async_subagents.py:173-183,旧名 ASYNC_TASK_SYSTEM_PROMPT 已随机制重构改并入工具描述)里有几条硬规矩,基本都在防同一个模型坏习惯——开完就轮询:
- 启动后立刻把 task_id 报给用户就收手,不准马上 check。
- 只有用户问进度或结果时才用
check_async_task。 - 想改指令用
update_async_task,多个任务可以并行挂着。
8.2 任务台账:state 里的一张字典
远程任务的 id 必须活得比一次对话久(上下文压缩会吃掉消息),所以它落在 agent state 里:
AsyncTask(async_subagents.py:80-120)是一条记录:task_id(=thread_id)、agent_name、run_id、status,外加三个 ISO-8601 时间戳(创建 / 上次查询 / 上次变化)。AsyncSubAgentState(async_subagents.py:132-136)把async_tasks: dict[str, AsyncTask]挂上去,reducer 是_tasks_reducer。_tasks_reducer(async_subagents.py:122-131)三行,dict(existing)复制后update——按 task_id 覆盖式合并,不删除。所以并发返回的多个任务更新不会互相冲掉。
status 特意声明成 str 而不是 Literal,docstring 给了理由:LangGraph SDK 的 Run.status 就是 str,收窄成 Literal 会逼得每个 SDK 边界都写 cast(async_subagents.py:96-101)。
8.3 客户端缓存 _ClientCache
每次调工具都新建一个 HTTP 客户端太浪费,于是有 _ClientCache(async_subagents.py:199-236)。
缓存键不是 agent 名字,而是 (url, 冻结后的 headers)(async_subagents.py:207-209):
def _cache_key(self, spec):
return (spec.get("url"), frozenset(_resolve_headers(spec).items()))
这样指向同一个部署的多个子 agent 共用一个连接;headers 换了(比如换了鉴权)就自然分开。frozenset 让 dict 变成可哈希的键,同时消掉 header 顺序的影响。
sync 与 async 两个池分开存(_sync / _async),因为 SDK 的 get_sync_client / get_client 返回的是两个不同类型。
有一条不对称的硬规则:get_sync 在 url is None 时直接抛 ValueError(async_subagents.py:214-216),而 get_async 不抛。原因是 url=None 走的是 ASGI 进程内传输,那条路只有异步实现。所以「本地 ASGI 子 agent 只能在异步调用下用」。
_resolve_headers(async_subagents.py:186-198)默认补一个 x-auth-scheme: langsmith,自建服务器一般会忽略它;要覆盖就在 spec 的 headers 里显式写(示例里就写成了 "custom",见 examples/async-subagent-server/supervisor.py:53,相对克隆根)。
8.4 状态轮询与终态短路
list_async_tasks 要为每条任务拉一次真实状态,但已经结束的任务不该再拉。于是有:
_TERMINAL_STATUSES = frozenset({"cancelled", "success", "error", "timeout", "interrupted"})
async_subagents.py:659-660。_fetch_live_status(async_subagents.py:663-680)第一件事就是查这个集合,命中直接返回缓存值,连客户端都不建。
第二个设计是失败不打断:拉状态出错时 logger.warning 之后返回缓存状态(async_subagents.py:663-680),而不是让整个 list_async_tasks 失败 。一个远端抖动不应该毁掉一次全量列表。
异步版本 _afetch_live_status(async_subagents.py:682-699)逻辑相同,但列表工具会用 asyncio.gather 把所有任务的状态并发拉取(async_subagents.py:773),然后 zip(..., strict=True) 对回去。同步版只能挨个来(async_subagents.py:743)。
过滤与拉状态的顺序也有讲究:_filter_tasks(async_subagents.py:706-728)按缓存状态过滤,拉活状态发生在过滤之后。docstring 明写了这个取舍——好处是不用为被过滤掉的任务付网络钱,代价是 status_filter="running" 可能漏掉一个刚从别处变回 running 的任务。
8.5 时间戳的语义差别
last_checked_at 和 last_updated_at 长得像,含义完全不同(async_subagents.py:372、746、779):
last_updated_at = now if task["status"] != result["status"] else task["last_updated_at"]
last_checked_at:每次查询都刷新——"我什么时候看过它"。last_updated_at:只有状态真的变了才刷新——"它什么时候变过"。update_async_task发新消息时也会刷(async_subagents.py:508-517)。
这样一条 running 了 20 分钟的任务,即使被查了 10 次,last_updated_at 仍然停在启动那一刻。
8.6 update 的语义:打断并重开
update_async_task 不是"追加一条消息",而是在同一个 thread 上用 multitask_strategy="interrupt" 开一个新 run(async_subagents.py:499-503):
thread(不变) ── run#1 ──[interrupt]
└────── run#2 ← 看得见 run#1 的完整历史 + 新消息
task_id 不变(= thread_id),run_id 换新,status 重置为 "running"
因为 thread 没变,子 agent 能看到原任务和之前的产出;因为 run 换了,父层记录里的 run_id 必须同步更新——否则后续 check 会去查一个已被打断的 run(async_subagents.py:508-517)。
9. 巧妙之处(可以直接借走的技术)
-
按字段存在性分流,而不是按类型标签。 三种 spec 都是
TypedDict,运行期没有类型信息,"graph_id" in spec这种判据既零成本又对手写 dict 友好(graph.py:648-654)。 -
setter 触发工具重建。
private_state_keys要等主栈全部定型才能算出来,但工具早就建好了。用 property setter 重建工具,比"把整个装配顺序倒过来"简单得多(middleware/subagents.py:711-720+graph.py:894-898)。 -
回溯取最后一条非空
AIMessage。 三行代码封掉了一个真实的供应商行为差异(Anthropic 的空end_turn尾消息),注释还留了现场(middleware/subagents.py:495-502)。 -
with_config而非改属性。 让同一个 runnable 能以不同名字注册多次,避免共享实例被互相污染(middleware/subagents.py:434-443)。 -
缓存键选
(url, headers)而非 agent 名。 同部署多 agent 自动共享连接池,是"选对键"带来的免费收益(async_subagents.py:241-243)。 -
终态短路 + 失败降级。
_TERMINAL_STATUSES省掉无谓请求,拉取失败退回缓存值而不是抛错——两条一起才让list_async_tasks在部分远端不可用时仍然可用(async_subagents.py:659-699)。 -
不转发父 config,只加一个标记。 靠 langgraph 的逐键合并拿到父上下文,手动转发反而会双计(
middleware/subagents.py:558-565的注释)。
10. 边界与局限
- 同步子 agent 只有一轮。 父层发一段 description,子 agent 回一条消息,中途无法追问。要多轮就得上远程
AsyncSubAgent的update_async_task。 CompiledSubAgent是黑盒。 不继承interrupt_on、不继承state_schema、不能用动态response_format(graph.py:498-503、524-530,middleware/subagents.py:430-432)。permissions是替换不是合并。 给子 agent 写规则会整体顶掉父层规则,容易在无意间放宽或收紧(middleware/subagents.py:116-125)。- 子 agent 不能再生子 agent。 子栈里只建了三件套 + skills(
graph.py:667-678),没有SubAgentMiddleware,所以委派只有一层。 url=None的远程子 agent 只能异步调。 同步路径直接抛ValueError(async_subagents.py:214-216)。- 异步任务的状态永远可能过期。 台账里的
status是上次查询的快照;prompt 反复强调必须重新查,恰恰说明这是个结构性问题而非可以修掉的 bug(async_subagents.py:178-183)。 - 列表工具按缓存状态过滤。
status_filter命中的是旧状态,不是活状态(async_subagents.py:706-728的 docstring 自己承认了)。
11. 代码地图
本表路径一律相对克隆根。
| 主题 | 文件路径 | 关键符号 |
|---|---|---|
| 三形态分流 | libs/deepagents/deepagents/graph.py | create_deep_agent("graph_id" in spec / "runnable" in spec 分支) |
| 声明式子 agent 重新组栈 | libs/deepagents/deepagents/graph.py | _apply_custom_middleware、_apply_excluded_middleware、_harness_profile_for_model |
| 默认 GP 子 agent | libs/deepagents/deepagents/graph.py | GENERAL_PURPOSE_SUBAGENT、gp_profile.enabled |
| GP 开关与覆盖 | libs/deepagents/deepagents/profiles/harness/harness_profiles.py | GeneralPurposeSubagentProfile |
| 子 agent spec 类型 | libs/deepagents/deepagents/middleware/subagents.py | SubAgent、CompiledSubAgent |
task 工具构建 | libs/deepagents/deepagents/middleware/subagents.py | _build_task_tool、TaskToolSchema、TASK_TOOL_DESCRIPTION |
| 懒编译与选路 | libs/deepagents/deepagents/middleware/subagents.py | _compile_spec、_select_subagent、create_sub_agent |
| state 过滤与回填 | libs/deepagents/deepagents/middleware/subagents.py | _EXCLUDED_STATE_KEYS、_validate_and_prepare_state、_return_command_with_state_update |
| 私有 state 字段计算 | libs/deepagents/deepagents/middleware/_state.py | private_state_field_names、_has_marker |
| 中间件本体 | libs/deepagents/deepagents/middleware/subagents.py | SubAgentMiddleware、private_state_keys |
| 结构化输出透传 | libs/deepagents/deepagents/middleware/subagents.py | SUBAGENT_RESPONSE_FORMAT_CONFIG_KEY、_get_subagent_response_format |
| tracing 标记 | libs/deepagents/deepagents/middleware/subagents.py | _subagent_tracing_context |
| 远程 spec 与台账 | libs/deepagents/deepagents/middleware/async_subagents.py | AsyncSubAgent、AsyncTask、AsyncSubAgentState、_tasks_reducer |
| 远程五工具 | libs/deepagents/deepagents/middleware/async_subagents.py | _build_start_tool、_build_check_tool、_build_update_tool、_build_cancel_tool、_build_list_tasks_tool |
| 客户端缓存 | libs/deepagents/deepagents/middleware/async_subagents.py | _ClientCache、_cache_key、_resolve_headers |
| 状态轮询 | libs/deepagents/deepagents/middleware/async_subagents.py | _fetch_live_status、_afetch_live_status、_TERMINAL_STATUSES、_filter_tasks |
| 可运行示例 | examples/async-subagent-server/supervisor.py | async_subagents、create_deep_agent(subagents=...) |
相邻章节: 组栈全流程见 装配流水线;子 agent 的文件工具与权限闸门见 文件系统中间件;为什么要压缩上下文见 上下文工程;skills 字段的语义见 技能、记忆与自评。