08 子 Agent 派工

QuibblerAgentQuibblerAgent 2026-11-13 约 17 分钟 15 次阅读

多 Agent 听起来像"人多力量大",真实动机却朴素得多:上下文隔离。本文是「从零手写 Agent」系列第 8 篇,用三十行代码给 Agent 装上子 Agent,让脏活的 token 留在分身的上下文里陪葬,主线只收结论。第 7 篇的压缩治标——垃圾进了主历史再清扫;本篇的隔离治本——脏活根本不进主历史。

本文按"动机 → 最小实现 → 并行派工实测 → 两条铁律 → 总结"组织。

1、动机:垃圾桶隔离

主 Agent 要调研十个文件,自己干意味着十次读文件的结果全进主历史——系列第 7 篇刚治理完的膨胀立刻复发,后续每轮循环都为这些"过程垃圾"付费。

派工模式的分工如下:

flowchart LR A[主 Agent] -->|完整任务描述| B[子 Agent] B -->|读文件与试错| C[子历史] B -->|只回传结论| A

解读:子 Agent 拿一份干净的历史去啃脏活,读文件、试错、中间结果全部留在它自己的上下文里;干完只把结论带回主线。主历史的膨胀被换成了子历史的廉价燃烧——多 Agent 的本质不是更聪明,是垃圾桶隔离。编码工具的子任务机制、各类多 Agent 框架,动机都是这一条。

换个角度算这笔账:主历史是"每轮重发"的计量基数,进主历史的每一个 token 都要乘上剩余轮数;子历史只烧一遍就扔。同样一万 token 的过程开销,放在主历史里是"一万乘轮数",放在子 Agent 里就只是一万——这就是"复利"与"陪葬"的区别。

1.1、什么活值得派

派工不是万灵药,先算账再派。任务画像与收益的对比如下:

任务画像 派活收益 建议
读十个文件出三行结论 高,过程 token 全隔离 派
读一两个文件的小事 低,转述成本反超收益 自己干
多路独立调研 高,还能并行省时间 并行派
强依赖主对话的纠偏 负,转述必然失真 自己干

解读:判据浓缩成一句话——过程与结论的 token 比值:比值大,隔离赚;比值接近一,派活只剩转述开销与失真风险。成本量级可以参考 Anthropic 多 Agent 研究报告:单个 Agent 的 token 消耗约为普通聊天的 4 倍,多 Agent 系统约 15 倍,而 token 消耗解释了其评测指标八成的方差——多 Agent 是拿 token 买质量与并行,值不值要看任务本身的价值。

拿本系列的代码库套一下这张表:调研"十几个 py 文件的调用关系"是典型的高比值任务——读几万 token 的源码,结论只是一张依赖清单,派;"看看 config 里超时设了几秒"是一眼的活,转述都比自己读贵,不派。判断发生在派活之前,比派出去再补救便宜一个数量级。

2、最小实现

对主 Agent 而言,子 Agent 就是一个普通工具。实现的关键是"每次调用都造一个全新 Agent":

def make_subagent_tool(name, description, tools_spec, tool_impls,
                       system, max_turns=6):
    def delegate_task(task: str) -> str:
        sub = Agent(tools_spec=tools_spec, tool_impls=tool_impls,
                    system=system, max_turns=max_turns, verbose=False)
        answer = sub.run(task)      # 全新实例:干净历史、独立消耗
        return (f"{answer}\n\n"
                f"[子agent消耗: 输入{sub.total_input}/输出{sub.total_output}]")

    return {
        "name": name,
        "description": description,
        "input_schema": {
            "type": "object",
            "properties": {
                "task": {"type": "string",
                         "description": "自包含的任务描述,"
                                        "子agent看不到你们的对话"},
            },
            "required": ["task"],
        },
    }, delegate_task

解读:函数返回"说明书加真函数"这对标准件,主 Agent 的工具列表里放入它,派活与调用计算器在协议上毫无区别——多 Agent 系统就是 Agent 的递归:一个工具会创建另一个 Agent。注意返回值尾部附上了子 Agent 的 token 账单,多 Agent 不是免费的,成本必须看得见。

一个容易忽略的细节是"全新实例"四个字:如果复用同一个子 Agent 实例跑第二个任务,上一个任务读过的文件内容还躺在它的历史里,跨任务的残留会污染下一个任务的判断。每次新建实例,等于把第 5 篇的 Agent 类当一次性用品——便宜、干净、可并行。

verbose=False 与子 Agent 的系统提示词也值得交代:子 Agent 的中间过程默认不进任何人的视野,它的系统提示词应当是一份"作业指引"——可用工具、产出格式、禁止事项,写得比主 Agent 的更窄更具体。子 Agent 不需要知道全局目标,只需要干好手里的活;知道得越少,转述失真的入口越少。

2.1、派工的类结构

函数版胜在直白,工程版还要管深度与账单。完整的派工类结构如下:

DELEGATE_SPEC = {
    "name": "delegate",
    "description": "派子 Agent 完成一项自包含的调研任务,回传结论。",
    "input_schema": {"type": "object",
                     "properties": {"task": {"type": "string"}},
                     "required": ["task"]},
}
class SubAgentPool:
    """子 Agent 工厂:统一创建、控制深度、归集账单。"""
    def __init__(self, tools_spec, tool_impls, system):
        self.base_spec, self.base_impls = tools_spec, tool_impls
        self.system = system
        self.bills = []                          # 逐笔记子 Agent 消耗
    def spawn(self, task, max_turns=6, depth=0):
        spec, impls = self.base_spec, dict(self.base_impls)
        if depth > 0:                            # 深度未尽才注入派活工具
            spec, impls = [*spec, DELEGATE_SPEC], \
                          {**impls, "delegate": self._delegator(depth)}
        sub = Agent(tools_spec=spec, tool_impls=impls,
                    system=self.system, max_turns=max_turns)
        answer = sub.run(task)                   # 全新实例:干净历史
        self.bills.append((task[:40], sub.total_input, sub.total_output))
        return answer
    def _delegator(self, depth):
        def delegate(task: str) -> str:          # 闭包携带剩余深度
            return self.spawn(task, depth=depth - 1)
        return delegate

解读:类把三件事收拢。创建——spawn 每次都造一个新 Agent,干净历史与独立消耗由此保证;深度——depth 大于零才注入派活工具,闭包逐层递减,套娃天然有底,第 4 节的铁律二在这里落地;账单——每笔消耗记进 bills,任务结束一屏看清谁烧的。主 Agent 拿到的派活工具,内核就是一行 pool.spawn。

3、并行派工实测

给主 Agent 一个三合一调研任务:定位核心循环、清点工具清单、统计文件数量。它自己没有任何文件工具,只有一个派活工具。真实运行轨迹如下:

第 1 轮:主 Agent 并行派出 3 个子任务
   调研核心循环位置 → 结论:run 方法,第 49 行起
   调研工具清单   → 结论:5 个工具
   统计 py 文件数  → 结论:12 个
第 2 轮:主 Agent 收工,汇总三行报告
主 Agent 账单:输入不足两千 token

解读:主 Agent 一轮并行点出三道"派活菜",每个任务描述都写得路径、背景、产出格式俱全;三个子 Agent 各自烧掉几千 token 读文件试错,主历史里只留下三段结论。同任务对比下,主 Agent 自己带文件工具干活的输入消耗是派活模式的好几倍——隔离省下的正是那些不进最终结论的过程内容。

值得专门看一眼主 Agent 写的三份任务描述:每份都自带工作目录背景与明确的产出格式,连统计任务都指定了按文件后缀计数的口径。派活质量的上限是主 Agent 的转述能力——这件事怎么强调都不过分,第 4 节的铁律一说的就是它。

子 Agent 失败的处理也要提前想好:超时、轮数耗尽、结论空洞时,账单照付、质量打折。池子的做法是不加掩饰地返回——结论原样带回、账单原样附上,让主 Agent 自己决定追问、换法还是认栽。把失败包装成成功,是多 Agent 系统里最难排查的一类事故:错误发生在子历史里,主历史里只看得到一个"看起来没问题的结论"。

3.1、执行侧真并行

"模型想并行"与"执行侧真并行"是两件事:第 5 篇的循环按顺序执行同一轮的多个调用,三个子 Agent 名义上并行、实际在排队。把执行层换成线程池即可兑现提速:

from concurrent.futures import ThreadPoolExecutor

def execute_round(self, calls):
    """同一轮的多个工具调用并发执行(IO 等待为主,线程即可)。"""
    with ThreadPoolExecutor(max_workers=4) as ex:
        futures = [ex.submit(self._execute, c) for c in calls]
        return [f.result() for f in futures]    # 结果顺序与调用对齐

解读:子 Agent 的工作几乎全是网络 IO 等待,线程池足够,不必上多进程;f.result() 按提交顺序收集,与调用块顺序对齐,工具配对协议不出错。实测三路并行把墙钟时间压到串行的四成左右(本项目口径,网络抖动下波动明显)。前提是线程安全:共享列表的 append 在 CPython 里由 GIL 保证原子,再复杂的共享结构就要加锁。用法细节见 concurrent.futures 文档。

3.2、账单要摊开

池子的 bills 摊开来看,派工的成本结构一目了然:

[派工账单](本项目实测口径,token)
  调研核心循环位置     输入 4210   输出 156
  调研工具清单         输入 3877   输出 132
  统计 py 文件数       输入 2953   输出  45
  合计                输入 11040  输出 333

解读:三个子 Agent 的输出都只有一两百 token——结论短,正是 1.1 节"值得派"的任务画像;输入四千上下,是读文件试错的正常开销。两种干法的主线差异汇总如下:

维度 主 Agent 自己干 派活模式
主历史内容 文件全文进历史 只有三段结论
主 Agent 输入 好几倍(实测口径) 不足两千 token
子 Agent 消耗 无 一万一千 token
后续每轮重发 随主历史膨胀 几乎不涨

一句话收束:隔离省的是主线每一轮的重发乘数,不是总 token——总额反而因转述与每个子 Agent 的系统提示词固定开销而上涨。这笔账看清楚,派不派活就不再是信仰问题。

4、两条铁律

派工模式有两条不成文就得撞墙的铁律。

第一是任务必须自包含:子 Agent 看不到主对话,任务描述里的"刚才那个文件"对它是黑洞。实测中不自包含的任务有三种下场:主 Agent 追问用户(最好)、猜一个路径转发(危险)、原样转发让子 Agent 再猜(最差)。主 Agent 转述任务时补全上下文,本身就是在做提示词工程。

自包含与否往往一线之差。对比如下:

反例:查一下刚才那个文件里核心循环在哪一行
正例:在 miniagent/agent.py 中找到 Agent 主循环所在的方法,
      给出方法名与起始行号,一行结论作答

解读:正例把四要素补齐——范围(哪个文件)、背景(找主循环)、产出格式(一行结论)、证据要求(方法名与行号)。写派活任务时按四要素过一遍,"刚才""那个"这类指代词就无处藏身。

顺带一个实测观察:主 Agent 被教育过一次之后学得很快——第一份含糊的任务描述被子 Agent 反问"哪个文件"之后,它后续写的任务明显更啰嗦也更完整。失败的教育价值比成功高,这条规律在第 6 篇的事故解剖里已经验证过一次。

第二是递归必须有深度上限:子 Agent 的工具里若也有派活工具,理论上可以无限套娃,每层扇出都是 token 的指数爆炸。给派活工具加深度参数,构造子 Agent 时只在剩余深度大于零时注入派活能力——与循环的轮数上限同级的安全阀。此外若追求真并行,注意"模型想并行"与"执行侧真并行"是两件事,需要线程池同时跑多个子 Agent 才能兑现提速。

深度上限在 2.1 节的 SubAgentPool 里就是 spawn 的 depth 参数:主 Agent 以 depth=2 起步,一层套娃后剩一,再套一层归零——第三层根本拿不到派活工具,指数爆炸在结构上就不可能发生。

5、总结

子 Agent 的最小实现是一个会创建 Agent 的工具,动机是上下文隔离而非集体智慧;任务自包含与递归深度上限是两条工程铁律,成本账单要随结果一起回传。什么任务值得派活?读得多、试错多、最终结论短的活最划算;读一两个文件的小事自己干更省。下一篇转向知识问题:RAG,给 Agent 一个查资料的外挂。多 Agent 模式的设计讨论可参考 Anthropic 多 Agent 研究报告,成品级的子代理机制对照可看 Claude Agent SDK 文档。

相关推荐

获取系统SDK版本、判断手机ROM
精选Code

获取系统SDK版本、判断手机ROM

Build获取系统SDK版本Android中部分API的使用,需要在特定的SDK版本之后才能使用,因此在兼容老版本SDK的时候,经常需要判断API的版本。各种Android版本的对应关系参考《Android各版本对应的SDK版本》判断手机ROM有时候需要判断手机系统的ROM,检测ROM是MIUI、EMUI还是Flyme,可以使用getprop命令,去系统build.prop文件(关于build.p

4.3k
AndroidStudio中各种中文乱码问题
精选Code

AndroidStudio中各种中文乱码问题

AndroidStudio中各种中文乱码问题1、编译Java错误信息乱码在出现这个Annotation processors must be explicitly declared now...问题的时候,正好也发现AndroidStudio Build Output错误信息都乱码。经常遇到各种问题,习惯了,现在遇到的问题,都是以后的答案。 1.1、修改项目build.gradle(无效)在整个p

2.8k
Multi-Agent 详解:多智能体协作架构与实践
精选Agent

Multi-Agent 详解:多智能体协作架构与实践

一个智能体配上一套工具,已经能完成相当多任务。但让它同时“当产品经理、写后端、跑测试、做代码评审”,很快就会露馅:上下文里塞满了互相干扰的指令,十几件工具的清单来回切换,任务做到一半忘了开头。解法看起来很自然——像人类团队一样分工:让多个智能体各司其职,由一个协调者调度,这就是 Multi-Agent(多智能体)系统。它也是 2026 年智能体工程里讨论度最高、误用率也最高的架构。本文按“架构模式…

144