资讯动态

Code Agent 解剖(08):一个任务太复杂,怎么拆给子 agent 做?

发布时间:2026/8/24 21:40:21 来源:尧图企业网站定制
主 agent 为什么需要子 agent主 agent 的上下文是有限的。一次复杂任务——比如找出项目里所有涉及认证的代码整理成报告——可能需要读几十个文件、跑很多次搜索。这些探索步骤会快速消耗上下文而且和最终的生成报告任务混在一起增加了压缩和失真的风险。更根本的问题是探索性工作只读、找信息和生成性工作写代码、修改文件的风险等级完全不同。如果能把探索任务单独剥离放在一个只读沙箱里跑主 agent 只拿回结果好处很明显探索过程不污染主 agent 的上下文子 agent 用轻量模型降低成本子 agent 不能写文件即使被诱导也无法造成副作用MyCodeAgent 的 Task 工具就是这个机制的实现。结论先放这主 agent 调 Task 工具 ↓ TaskTool.run() 校验参数构造 TaskRequest ↓ _DeferredSubagentLauncher.launch() 代理层惰性初始化真正的 launcher ↓ SubagentLauncher.launch() 真正的执行入口subagents.py ├─ 1. 查找 RuntimeProfile工具白名单、步数、token 预算、模型 ├─ 2. _select_llm()优先用 LIGHT_LLM没配则回退到主模型 ├─ 3. _create_child_trace()创建独立 trace 日志 ├─ 4. 向父 trace 发射 subagent_requested/started 事件 ├─ 5. _SubagentRuntimeHost创建独立沙箱history/context/tool 全新 ├─ 6. _render_request()把任务描述 结构化上下文拼成 prompt ├─ 7. RuntimeRunner(host).run(prompt) ← 完整 ReAct 循环 ├─ 8. _child_metrics()从 trace 事件里提取 terminal_reason/tool_usage/token_usage └─ 9. ExploreResult.from_json()解析并校验子 agent 输出的 JSON ↓ SubagentLaunchResult → TaskTool 打包成标准信封主 agent 拿到摘要继续工作子 agent 跑的是完整的 ReAct 循环和主 agent 用的是同一个RuntimeRunner。不同的是它运行在一个受约束的沙箱里工具、步数、token 预算都被RuntimeProfile严格限定。调用链中的三层 launch代码里有三个launch方法容易混淆位置类型作用task.pyTaskLauncher.launchProtocol 接口只是接口声明无实现...host.py_DeferredSubagentLauncher.launch代理层惰性初始化转发给真正的 launchersubagents.pySubagentLauncher.launch真正实现创建沙箱、运行子循环、解析结果这三层分离的原因TaskTool注册时SubagentLauncher还不能创建CodeAgent未初始化完代理层解决了这个时序问题Protocol接口让TaskTool不直接依赖SubagentLauncher两者解耦。TaskRequest 和 SubagentRequest 的关系两者是字段完全相同的镜像 dataclass。# task.py # subagents.pyclassTaskRequest:classSubagentRequest:profile_name:strprofile_name:strtask:strtask:strmodel_choice:str|Nonemodel_choice:str|Nonestructured_context:dictstructured_context:dictparent_session_id:str|Noneparent_session_id:str|Noneparent_run_id:str|Noneparent_run_id:str|Nonetask.py不能直接 importsubagents.py会循环依赖所以在本文件定义了一个镜像。_DeferredSubagentLauncher.launch(request)把TaskRequest直接传给SubagentLauncher.launch(request: SubagentRequest)Python 运行时只看字段值不做类型校验两者完全兼容。SubagentLauncher.launch()第一件事是RUNTIME_PROFILES.get(request.profile_name)profile_nameexplore就对应EXPLORE_PROFILE沙箱约束从这里开始生效。Task 工具入口、参数校验、完整调用链# tools/builtin/task.py — TaskTool.run()简化defrun(self,parameters):descriptionparameters.get(description)# 任务的简短描述给主 agent 看的标签promptparameters.get(prompt)# 自包含的探索指令子 agent 执行的完整任务profileparameters.get(subagent_type)# 当前只支持 exploremodelparameters.get(model,light)# light轻量模型或 main# 参数校验通过后构造 TaskRequest 委派出去launchedself._launcher.launch(TaskRequest(profile_nameexplore,# ← 这个字符串决定后面走哪条路taskf{description}\n\n{prompt},model_choicemodel,))self._launcher是什么在host.py里注册 TaskTool 时传入的# runtime/host.py — _initialize_runtime_components()self.tool_registry.register_tool(TaskTool(project_root...,launcherself._DeferredSubagentLauncher(self._get_subagent_launcher),# ↑ 这就是 self._launcher 的真实身份))所以self._launcher.launch(request)实际走的完整调用链是TaskTool.run() self._launcher.launch(TaskRequest(profile_nameexplore, ...)) ↓ [host.py _DeferredSubagentLauncher.launch] self._get_launcher() ← 惰性初始化首次调用才真正创建 SubagentLauncher ↓ [host.py _get_subagent_launcher] create_subagent_launcher(host) ← factory.py把主 agent 的 llm/registry 传进去 → SubagentLauncher(main_llm, tool_registry, ...) SubagentLauncher.launch(request) ← subagents.py真正的实现 ↓ profile RUNTIME_PROFILES.get(request.profile_name) # ↑ explore → EXPLORE_PROFILE工具白名单、步数、token 预算全在这里 ↓ 创建沙箱、跑 ReAct 循环、解析结果profile_nameexplore这个字符串就是从 TaskTool 到EXPLORE_PROFILE的连接点——SubagentLauncher.launch()第一件事就是用它在RUNTIME_PROFILES字典里查对应的 profile。prompt必须是自包含的——子 agent 看不到主 agent 的历史只有这一段指令。模型写 Task 调用时必须把所有背景信息显式写进prompt不能依赖隐式上下文传递。RuntimeProfile沙箱约束的定义子 agent 的所有约束都在RuntimeProfile里声明# runtime/subagents.pyEXPLORE_PROFILERuntimeProfile(nameexplore,system_promptEXPLORE_SYSTEM_PROMPT,# 固定系统提示词要求返回 JSONtool_allowlist{Read,Grep,Glob},# 只读工具不能写文件、不能执行命令max_steps12,# 最多 12 步防止无限循环context_token_budget16_000,# 单次上下文窗口上限主 agent 是 128ktotal_token_budget32_000,# 累计 token 上限超出强制终止model_choicelight,# 默认用轻量模型降低成本result_contractExploreResult,# 必须返回符合此合约的 JSON)RuntimeProfile.__post_init__有硬约束Task 工具和 Edit/Bash 不允许出现在tool_allowlist里ifself.recursive_subagentsorTaskinself.tool_allowlist:raiseValueError(formal subagent profiles cannot recurse)forbidden{Edit,Bash}ifforbiddenself.tool_allowlist:raiseValueError(formal subagent profiles must be strictly read-only)为什么要在 profile 里写死而不是运行时动态控制Profile 是声明性的约束不依赖运行时状态。主 agent 的历史、当前步数、context 压缩状态都不会影响子 agent 能用哪些工具——子 agent 的边界在代码里固定好了无法被提示词诱导绕过。SubagentLauncher.launch()沙箱创建与运行细节第一步选模型和 trace# subagents.py — SubagentLauncher.launch()llm,model_choiceself._select_llm(requested_model)# _select_llm优先用 light_llm从 LIGHT_LLM_* 环境变量创建# 没配 LIGHT_LLM_MODEL_ID 则回退到主 agent 的 main_llmchild_traceself._create_child_trace()# 独立的 trace JSONL 文件session_id 以 child- 开头# 和主 agent trace 分离但 launch() 里发射的父子事件通过 parent_session_id 关联第二步创建沙箱 _SubagentRuntimeHost_SubagentRuntimeHost和主 agent 的CodeAgent结构完全对齐——都有history_manager、context_engine、tool_executor、tool_orchestrator——但所有状态全部新建主 agent 的历史不会流入子 agent。RuntimeRunner通过鸭子类型复用它只依赖 host 上的属性不要求继承任何基类所以_SubagentRuntimeHost不需要继承CodeAgent只要有同名属性就能跑同一套循环。沙箱里的关键约束# 1. config用 profile 预算覆盖关键字段self.configConfig.from_env().model_copy(update{context_window:profile.context_token_budget,# explore16000远小于主 agent 的 128k})self.max_stepsprofile.max_steps# explore12硬性步数上限self.max_total_tokensprofile.total_token_budget# explore32000超出强制终止# 2. registry只含白名单工具build_registry 过滤# 主 agent 的所有工具中只有 Read/Grep/Glob 会进子 agent 的 registry# 3. ContextBuilder系统提示词固定为 profile.system_prompt要求返回 JSON# 不加载 MCP 工具提示词和 Skills不暴露非白名单工具的 schema# 4. 上下文压缩用 _summarize_child_messages纯截断拼接不启 LLM# 原因子 agent 预算小启 LLM 压缩太贵简单截断足够# 5. 权限双重保障permission_contextPermissionContext(runtime_modereadonly_subagent)# RiskClassifier 在此模式下对 Edit/Bash/Task 直接 DENY# 即使这些工具意外出现在 registry 里也执行不了——白名单 权限门双重拦截# 6. completion_verifier_StructuredResultCompletionVerifier子 agent 专用# 检查输出是否符合 result_contract JSON 格式不符合则触发 FAIL 反馈让子 agent 重输第三步拼 prompt 并运行# _render_request把任务文本和结构化上下文拼成 prompt# 输出格式task 文本 \n\nStructured context:\n JSON# 这是子 agent 唯一能看到的上下文——它看不到主 agent 的历史prompt_render_request(request)# 和主 agent 完全相同的 RuntimeRunner跑完整 ReAct 循环# 子 agent 的系统提示词要求它最终输出一个 JSON 对象raw_resultRuntimeRunner(host).run(prompt)第四步提取指标解析结果# _child_metrics遍历 child_trace.events统计三类信息# terminal_reason子 agent 怎么结束的completed/max_steps/token_budget 等# tool_usage每个工具调用了几次# token_usage累计消耗 token 数terminal_reason,tool_usage,token_usage_child_metrics(child_trace.events)# 子 agent 必须以 completed/completed_unverified 结束其他原因都当失败ifterminal_reasonnotin{completed,completed_unverified}:raiseValueError(fchild terminal reason:{terminal_reason})# ExploreResult.from_json 严格校验# - status 只能是 completed/partial# - summary 不能为空# - 有 Markdown fence直接抛异常# 不合法则 launch() 捕获异常返回 statusFAILEDstructuredExploreResult.from_json(raw_result,tool_usagetool_usage,...)结构化结果合约子 agent 的系统提示词要求它只返回 JSON不返回 MarkdownYou are an Explore Agent. Inspect the repository with read-only tools and return exactly one JSON object: {status:completed|partial,summary:...,findings:[...], evidence:[relative/path.py:line],unresolved_questions:[...]}. Do not use markdown fences.RuntimeRunner运行结束后launch()解析这段 JSONraw_resultRuntimeRunner(host).run(prompt)structuredExploreResult.from_json(raw_result,tool_usagetool_usage,terminal_reasonterminal_reason,)ExploreResult.from_json()严格校验status只能是completed或partialsummary必须有内容否则抛异常launch()捕获后返回statusFAILED。为什么要结构化 JSON 而不是自然语言主 agent 拿到子 agent 的结果后需要机器可靠地提取summary给模型看的摘要、findings具体发现、evidence代码位置证据。自然语言需要主 agent 再解析一次增加了失真的可能性也没法做格式校验。JSON 合约把子 agent 必须提供什么写死在代码里。轻量模型与 LIGHT_LLM子 agent 默认model_choicelight_select_llm()会尝试用light_llmdef_select_llm(self,requested_model):ifrequested_modellight:ifself.light_llmisNone:self.light_llm_create_light_llm()# 从 LIGHT_LLM_* 环境变量创建ifself.light_llmisnotNone:returnself.light_llm,lightreturnself.main_llm,main# 没配轻量模型就回退到主模型LIGHT_LLM_*环境变量在.env里配置LIGHT_LLM_PROVIDER、LIGHT_LLM_MODEL_ID等没配就回退到主 agent 的模型。探索任务通常只需要读代码、搜索、总结轻量模型足够胜任成本可以降一个量级。主 agent 拿到结果后看到什么TaskTool.run()最终返回标准信封returnself.success_result(data{status:completed,profile:explore,result:{summary:找到了 3 处认证相关代码...,findings:[auth/login.py:45 — JWT 验证,...],evidence:[auth/login.py:45,...],},},textresult.summary,# ← 给模型看的摘要直接进 observationextra_stats{tool_calls:8,token_usage:12000,model:light,},)text字段是摘要直接作为 observation 追加到主 agent 的历史里。模型看到这段摘要决定下一步该做什么。完整的data里有findings和evidence模型可以在后续步骤里引用具体的代码位置。当前的限制Task 工具目前只支持subagent_typeexplore传入其他值直接返回参数错误。代码里有VERIFICATION_PROFILE但没有对应的 Task 入口——验证子 agent 只由主循环的完成门直接调用--enable-verification-agent开启后。多轮对话中每次 Task 调用都是独立的——子 agent 没有跨次调用的记忆每次都是新的_SubagentRuntimeHost。如果需要分多次探索、逐步积累需要主 agent 自己把上次的发现写进下次调用的prompt。设计亮点复用同一个 RuntimeRunner子 agent 和主 agent 跑完全一样的 ReAct 循环不是简化版是用 profile 约束了能力边界沙箱是声明性的profile 里写死工具白名单运行时无法绕过不依赖提示词的守规矩结构化合约JSON 结果让主 agent 可靠地提取信息不需要再解析自然语言轻量模型降成本探索任务不需要最强模型配置 LIGHT_LLM 可以省一大笔开销独立 trace子 agent 有自己的 trace 文件父子事件通过parent_session_id关联可以单独分析子任务执行情况小结机制作用TaskTool参数校验 委派入口RuntimeProfile声明沙箱约束工具、步数、token、模型SubagentLauncher创建沙箱、选模型、运行子循环、解析结果_SubagentRuntimeHost独立的 history/context/tool与主 agent 完全隔离ExploreResultJSON 结构化合约保证结果可机器解析readonly_subagent权限模式双重保障白名单 权限分类器双重拦截写操作关于本系列的源码本系列所有分析均基于开源项目 MyCodeAgent。源码里已经按照本系列文章的讲解顺序在关键位置加入了配套注释——读文章时可以对照代码也可以直接克隆下来自己跑、改、扩展基于它开发你自己的 agent。gitclone https://github.com/chendongqi/MyCodeAgentcdMyCodeAgentcp.env.example .env# 填入你的 LLM API keyuvsyncuv run python main.py欢迎访问 PrimeSkills —— 一个精心策划的 AI Agent 与技能市场所有内容均经过真实企业级工作流验证。没有噱头只有真正有效的东西。更多实用知识和有趣产品欢迎访问我的个人主页

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价