《AI Agent 开发入门》13.1 角色与分工

13.1 角色与分工

第 12 章把单个 Agent 放进图里跑，但整张图里还是同一个 Agent——同一个 system 提示、同一套工具、同一段上下文。当任务横跨多个领域时，这个「全能选手」会开始力不从心：一个既要查资料、又要做数学、又要写报告的系统提示，往往哪一样都做不好；一个塞满几十个工具的注册表，让模型选错工具的概率陡增。多 Agent 的第一课不是「怎么让它们聊天」，而是怎么分角色。

本节把 minagent 推进到 minagent/multi/：定义 Worker（带专用 system 与工具子集的 Agent）与 Supervisor（派发与汇总），并用假模型真跑一次「拆成两个子任务 → 各自执行 → 汇总」的完整流程。

13.1.1 单 Agent 的天花板

先说清楚为什么要拆。单个 Agent 的瓶颈主要有三条，它们都不是「模型不够聪明」造成的，而是上下文与工具的混装造成的：

这三条的共同点：它们都是「把不同职责塞进一个 Agent」的代价。职责一混，指令、工具、上下文就都得混在一起，而模型对混装的容忍度有限。

多 Agent 的解法很朴素：给每种职责配一个专用 Agent。检索的 Agent 只挂检索工具、只写检索指令；计算的 Agent 只挂计算工具、只写计算指令。职责一分开，上面三条瓶颈同时缓解——这就是 supervisor/worker 模式想解决的问题。

13.1.2 Worker：带专用 system 与工具子集的 Agent

Worker 不是什么新东西，它就是第 3 章的 Agent（run 循环）套一层壳，壳里固定了两样：一个专用的 system 提示和一份专用的工具子集。

# minagent/multi/worker.py
from dataclasses import dataclass

from minagent.loop import run
from minagent.tools import ToolRegistry
from minagent.types import Model, RunResult


@dataclass
class Worker:
    name: str
    system: str
    model: Model
    registry: ToolRegistry

    def handle(self, task: str) -> RunResult:
        return run(self.model, self.registry, task, system=self.system)

就这么多。Worker 的价值全在构造时：你给它的 system 和 registry 是专属的，别的 worker 看不到。一个「计算员」只挂 add / mul，一个「检索员」只挂 lookup。handle 只是把任务丢进标准的 run 循环——分工发生在装配期，而不是运行期。

这带来两个直接好处。第一，工具隔离：计算员的注册表里没有 lookup，它不可能误调检索工具，因为那个工具压根不在它的 schema 列表里。第二，上下文隔离：计算员的消息历史只包含它自己这一轮的对话，检索员读到的长文档不会污染计算员的推理。这两点都是「提示词里加一句『请勿使用检索工具』」做不到的——约束靠的是能力边界，不是模型的自律。

13.1.3 Supervisor：派发与汇总

有了 worker，还需要一个「工头」决定谁干什么。Supervisor 的职责有两件：把目标拆成子任务并指派执行者（派发），以及把所有子结果合成最终答复（汇总）。这两件事它都用一次模型调用来完成：

# minagent/multi/supervisor.py
import json
from dataclasses import dataclass

from minagent.multi.worker import Worker
from minagent.types import Message, Model

ROUTER_SYSTEM = (
    "你是调度器。把用户目标拆成 1~3 个子任务，每个子任务指定一个执行者。"
    '只输出 JSON 数组，形如 [{"worker": "...", "task": "..."}]，不要多余文字。'
    "可用执行者：计算员（四则运算）、检索员（查常量与事实）。"
)


@dataclass
class Supervisor:
    model: Model
    workers: dict[str, Worker]

    def plan(self, goal: str) -> list[dict[str, str]]:
        resp = self.model.complete(
            [Message(role="system", content=ROUTER_SYSTEM),
             Message(role="user", content=goal)],
            [],
        )
        return json.loads(resp.message.content)

    def run(self, goal: str) -> str:
        results = []
        for item in self.plan(goal):
            worker = self.workers[item["worker"]]
            r = worker.handle(item["task"])
            results.append((item["worker"], item["task"], r.final))
        brief = "\n".join(f"- {n} 完成「{t}」-> {f}" for n, t, f in results)
        resp = self.model.complete(
            [Message(role="user",
                     content=f"目标：{goal}\n子结果：\n{brief}\n请用一句话汇总。")],
            [],
        )
        return resp.message.content

plan 是一次结构化输出（第 6 章的技巧）：让模型吐一段 JSON 数组，每项含 worker 和 task。run 先调 plan，逐条派发给对应 worker，收集每个 worker 的 final，再调一次模型把子结果合成一句话。

注意 ROUTER_SYSTEM 里那句「可用执行者：计算员、检索员」——它把可选的 worker 名单写进了路由提示。这是 supervisor 模式的关键约束：调度器的选择空间被限定在已知 worker 集合内，它不能凭空发明一个不存在的执行者。真实系统里，这份名单通常由 worker 注册表自动生成，而不是手写。

13.1.4 路由决策从哪来

plan 里的那次模型调用，就是整个 supervisor 模式的大脑。它从哪来，直接决定了系统的成本、可靠性与可调试性。实际有三种来源：

规则路由最简单，一行判断就能派活：

def rule_plan(goal: str) -> list[dict[str, str]]:
    if any(k in goal for k in ("算", "+", "*", "阶乘")):
        return [{"worker": "计算员", "task": goal}]
    if any(k in goal for k in ("查", "是什么", "多少")):
        return [{"worker": "检索员", "task": goal}]
    return [{"worker": "写手", "task": goal}]

它不花任何模型调用，行为完全可预测，但只能处理「任务类型能用关键词识别」的场景。一旦任务需要理解语义（比如「把这份财报的结论和去年对比一下」），关键词就失灵了。

模型路由就是本节的写法：让模型读目标、吐 JSON。它的优势是能理解模糊意图，代价是多一次调用、且要防它输出非法 JSON。混合路由是工程上最常用的：先用规则处理高频、明确的请求（省钱又快），规则拿不准时才回退到模型路由。第 6 章的结构化输出技巧（JSON schema、解析容错）在这里正好派上用场。

选哪种取决于请求的分布：如果 80% 的请求是少数几种固定类型，规则路由能砍掉大部分调用；如果请求五花八门，模型路由才值得。别一上来就模型路由——先问「这些请求真的需要模型来分类吗」。

13.1.5 真跑：一个 supervisor 带两个 worker

把两个 worker 装配起来真跑一遍。计算员挂 add / mul，检索员挂 lookup，各自的假模型脚本写死自己的行为：

calc = ToolRegistry()
calc.add(Tool("add", "两数相加", NUM_SCHEMA, lambda a, b: a + b))
calc.add(Tool("mul", "两数相乘", NUM_SCHEMA, lambda a, b: a * b))

facts = {"4!": "24", "5!": "120"}
research = ToolRegistry()
research.add(Tool("lookup", "查常量或阶乘", KEY_SCHEMA,
                  lambda key: facts.get(key, "未知")))

workers = {
    "计算员": Worker("计算员", "你是计算员，只做四则运算。",
                   FakeModel([[ToolCall("c1", "add", {"a": 2, "b": 3})],
                              [ToolCall("c2", "mul", {"a": 5, "b": 4})],
                              "结果是 20。"]), calc),
    "检索员": Worker("检索员", "你是检索员，只查常量与事实。",
                   FakeModel([[ToolCall("r1", "lookup", {"key": "4!"})],
                              "4 的阶乘是 24。"]), research),
}

router = FakeModel([
    '[{"worker": "计算员", "task": "计算 (2+3)*4"}, '
    '{"worker": "检索员", "task": "查 4 的阶乘"}]',
    "总结：(2+3)*4=20，4 的阶乘是 24。",
])
sup = Supervisor(router, workers)
print(sup.run("帮我算 (2+3)*4，再查 4 的阶乘，最后给个总结"))

本机实测（Python 3.14.6），把派发与回收都打了出来：

[supervisor] 目标：帮我算 (2+3)*4，再查 4 的阶乘，最后给个总结
[supervisor] 派发 计算员 <- 计算 (2+3)*4
[worker 计算员] 3 步 | 结果是 20。
[supervisor] 派发 检索员 <- 查 4 的阶乘
[worker 检索员] 2 步 | 4 的阶乘是 24。
[supervisor] 汇总：总结：(2+3)*4=20，4 的阶乘是 24。
模型调用：supervisor=2, 计算员=3, 检索员=2, 合计=7

整条链路一目了然：supervisor 先花一次调用做路由（拆成两条子任务），再依次把任务交给计算员和检索员，各自在自己的 run 循环里跑完，最后 supervisor 再花一次调用做汇总。7 次模型调用完成了「一个全能 Agent 大概要 4~5 次、且更容易出错」的事——多出来的调用，买的是职责隔离。

这里 worker 是串行执行的：supervisor 先等计算员跑完，再派检索员。它们之间没有依赖，本可以并行——这正是 13.2 消息总线要解决的问题。本节先把「分工」这件事立住，把「怎么让它们并行、怎么交换中间结果」留给下一节。

顺带看一个「上下文隔离」的证据：把两个 worker 各自的 RunResult.messages 长度打出来，会看到两条历史完全不相干：

计算员 messages: 7（system + user + 两轮工具 + 两轮结果 + 收尾）
检索员 messages: 5（system + user + 一轮工具 + 一轮结果 + 收尾）

计算员的历史里有它自己的 add / mul 结果，没有检索员的 lookup 结果；反之亦然。如果这两个任务塞进同一个 Agent，它的历史就会是两条之和（11 条），其中约一半与当前那步无关——无关历史挤占上下文，正是 13.1.1 说的「上下文污染」。分工在数据上的直接体现，就是每个 worker 的历史都短而专。

13.1.6 取舍：单 Agent 还是 supervisor-worker

多 Agent 不是免费的午餐。把 13.1.4 的收益和成本摊开看：

这张表给出一条实用判据：只有当「职责确实可以干净地分开」时，多 Agent 才划算。如果两个 worker 的工具几乎重叠、system 提示也差不多，那拆开只会增加调用次数和调试难度，收益为零。

反过来，只要满足下面任一条，就值得拆：

• 工具体系差异大（检索工具 vs 代码执行器 vs 数据库客户端）；
• system 指令互相冲突（严谨引用 vs 创意改写）；
• 子任务可以并行且互不依赖（下一节会展开）。

记住第 12 章那句提醒：能用一个大循环解决的就别上多 Agent。supervisor/worker 是「一个 Agent 真的忙不过来」时的答案，不是默认架构。

最后给一条粗估公式：supervisor/worker 的模型调用次数 ≈ 1（路由）+ Σ（每个 worker 的循环步数）+ 1（汇总）。worker 越多、每个 worker 循环越长，这个和涨得越快。在动手拆分之前，先用这条公式估一下总调用量，再问自己「这些多出来的调用换来的隔离，值不值」——十有八九，答案是「不值」，那就退回单 Agent。

13.1.7 真实模型下会怎样

本节的路由决策是脚本写死的：router 的脚本第一项直接给出了「拆成哪两条子任务、派给谁」。真实 Claude 会自己决定怎么拆——你把目标丢过去，它现场判断该派给几个 worker、每个 worker 领什么任务。

这带来三点差异：

• 拆法不唯一。同一个目标，真实模型可能拆成两条子任务，也可能拆成三条、或干脆判断「不用拆，我自己就能做」。所以 plan 的返回要做健壮性校验：worker 名不在注册表里怎么办、任务为空怎么办、返回的不是合法 JSON 怎么办。真实系统里这些都要兜底。
• 路由会错。真实模型偶尔会把「查阶乘」派给计算员（因为阶乘看起来像计算）。这类错误的代价是 worker 拿到不匹配的工具、要么报错要么给出错误结果。缓解办法是让 worker 的 system 明确说清自己的边界，并让 supervisor 在汇总时核对子结果。
• 汇总也是模型判断。真实模型汇总时可能漏掉某个子结果、或把两个子结果张冠李戴。因此 brief 里要带足上下文（谁完成了什么），让汇总这一步有据可依。

一句话：假模型证明「派发—执行—汇总」这条链路能转，真实模型负责证明它在真实任务上转得对。把链路跑通、把兜底补齐，是本节真正交付的东西。

还有一个本节没解决的问题：worker 之间不能直接说话，一切都要经 supervisor 中转。下一节会给它们修一条直接通道。

阅读导航：上一节：12.3 人类在环 · 下一节：13.2 消息总线与共享状态 。