《AI Agent 开发入门》13.3 协作模式对照

13.3 协作模式对照

13.1 拆了角色，13.2 修了通道，但「多个 Agent 具体怎么组织」还没定。同样是三个 worker，可以串成一条流水线，也可以让它们互相辩论，还可以让它们各投一票。三种组织方式用的 worker 一模一样，产出的成本与质量却差很多。这一节把三种拓扑放在同一个任务上真跑，用数字说话。

本节把 minagent 推进到 minagent/multi/topology.py：用同一批假模型 worker 实现顺序、辩论、投票三种协作模式，量化对比步数、模型调用、总耗时与一致性，并给出「什么任务适合哪种拓扑」的决策表。

13.3.1 三种拓扑

先给三种拓扑一个精确的刻画。设任务需要若干次「worker 产出文本」，三种拓扑的差别全在信息如何流动：

三者的核心差异在于：pipeline 里信息是单向累积的（越到后面看到的越多），debate 里信息是双向对抗的（一方能看到并反驳另一方），voting 里信息是互相隔离的（投票者彼此看不见，保证独立性）。

这三种结构对应三种不同的目的：pipeline 要的是逐级加工，debate 要的是暴露分歧，voting 要的是用独立性换稳健。下一小节用同一批 worker 把它们都实现出来。

13.3.2 同一批 worker，三种组织方式

先做一个统一的「带延迟的假模型」，它模拟真实调用的耗时，让总耗时可测：

# minagent/multi/topology.py
import time
from collections import Counter
from dataclasses import dataclass

from minagent.types import Message

CALL_MS = 0.03  # 模拟一次模型调用 30ms


class LatencyModel:
    def __init__(self, inner, delay: float = CALL_MS) -> None:
        self.inner = inner
        self.delay = delay

    def complete(self, messages: list[Message], tools: list[dict]):
        time.sleep(self.delay)
        return self.inner.complete(messages, tools)


def say(model: LatencyModel, text: str) -> str:
    return model.complete([Message(role="user", content=text)], []).message.content

三种拓扑共用同一个任务：判断 17 是不是质数。三种实现各自把 worker 组织起来：

def pipeline() -> tuple[str, int, int]:
    stages = [LatencyModel(FakeModel(["17 的因数只有 1 和 17。"])),
              LatencyModel(FakeModel(["试除 2、3、4 均不整除，故为质数。"])),
              LatencyModel(FakeModel(["结论：17 是质数。"]))]
    data, steps = "判断 17 是否为质数", 0
    for m in stages:                      # 上一环输出喂给下一环
        data = say(m, data)
        steps += 1
    return data, steps, steps


def debate() -> tuple[str, int, int]:
    pro = LatencyModel(FakeModel(["17 只能被 1 和自身整除，是质数。"]))
    con = LatencyModel(FakeModel(["17 是奇数，奇数里很多是合数，所以不是质数。"]))
    rebut = LatencyModel(FakeModel(["奇数不等于合数，9 才是合数；17 无其他因数。"]))
    judge = LatencyModel(FakeModel(["综合双方论证：反方论据不成立，判定 17 是质数。"]))
    a = say(pro, "判断 17 是否为质数，给出你的立场。")
    b = say(con, f"对方说：{a}。你的立场？")        # 反方看到正方发言
    c = say(rebut, f"反方说：{b}。请反驳。")        # 正方看到反方发言
    verdict = say(judge, f"正方：{a}\n反方：{b}\n反驳：{c}\n请裁决。")
    return verdict, 4, 4


def voting() -> tuple[str, int, int]:
    voters = [LatencyModel(FakeModel(["是，17 是质数。"])),
              LatencyModel(FakeModel(["是，17 是质数。"])),
              LatencyModel(FakeModel(["不是，17 是合数。"]))]  # 噪声票
    votes = [say(m, "17 是质数吗？只回答「是」或「不是」。") for m in voters]
    tally = Counter("是" if "不是" not in v else "不是" for v in votes)
    return f"结论：17 {'是' if tally.most_common(1)[0][0] == '是' else '不是'}质数", len(votes), len(votes)

三段代码长度相近，但信息流的差别一眼可见：pipeline 里 data 被反复覆盖，每个 worker 只看到上一个的输出；debate 里 con 的输入里嵌着 a（正方发言），rebut 的输入里嵌着 b（反方发言）——这一层「看到对方」是辩论区别于投票的关键；voting 里三个投票者拿到的输入完全一样，互相不可见。

注意 debate 的 worker 数量（4 个：正、反、反驳、裁决）和 voting（3 个投票者）与 pipeline（3 个阶段）并不相等——拓扑决定了需要几个 worker，而不是反过来。

13.3.3 真跑：量化对比

把三种拓扑各跑一遍，记录步数、模型调用、总耗时和一致性。一致性定义为「独立产出的判断中，与最终裁决一致的比例」：pipeline 只有一个最终判断，记 1.00；debate 有两个开场立场（一正一反），记 0.50；voting 三票里两票一致，记 0.67。

本机实测（Python 3.14.6）：

拓扑            步数    模型调用      总耗时(ms)     一致性
顺序 pipeline   3     3         108.9       1.00
辩论 debate     4     4         135.2       0.50
投票 voting     3     3         123.5       0.67

（总耗时随机器负载略有浮动，但相对大小稳定：debate 因为多一次调用，最慢。）

三种拓扑的完整中间输出也打了出来，可以看到每条链路的实际流转：

[pipeline] 分析员 -> 17 的因数只有 1 和 17。
[pipeline] 计算员 -> 试除 2、3、4 均不整除，故为质数。
[pipeline] 写手 -> 结论：17 是质数。
[debate] 正方 -> 17 只能被 1 和自身整除，是质数。
[debate] 反方 -> 17 是奇数，奇数里很多是合数，所以不是质数。
[debate] 正方反驳 -> 奇数不等于合数，9 才是合数；17 无其他因数。
[debate] 裁决 -> 综合双方论证：反方论据不成立，判定 17 是质数。
[voting] 投票者1 -> 是，17 是质数。
[voting] 投票者2 -> 是，17 是质数。
[voting] 投票者3 -> 不是，17 是合数。

13.3.4 指标是怎么采集的

那张表不是手写的，是跑出来的。三个指标各有采集方式，装在一个统一的 Outcome 里：

from dataclasses import dataclass


@dataclass
class Outcome:
    topology: str
    final: str
    steps: int
    calls: int
    elapsed: float
    agreement: float


AGREEMENT = {"顺序 pipeline": 1.0, "辩论 debate": 0.5, "投票 voting": 2 / 3}


def measure(name: str, fn) -> Outcome:
    t0 = time.perf_counter()
    final, steps, calls = fn()
    return Outcome(name, final, steps, calls,
                   time.perf_counter() - t0, AGREEMENT[name])


outcomes = [measure("顺序 pipeline", pipeline),
            measure("辩论 debate", debate),
            measure("投票 voting", voting)]

步数和模型调用在每种拓扑内部累加——本例里两者恰好相等（每个 worker 恰好一次调用），但一般不相等：一个 worker 内部可能跑多轮 run 循环。总耗时由 time.perf_counter() 前后相减得到，配合 LatencyModel 注入的 30ms 模拟延迟，让数字有意义。一致性这里是手工标注的：pipeline 只有一个判断记 1.0，debate 两个开场立场记 0.5，voting 三票两中记 2/3。

要诚实地说清一个局限：假模型测不出「质量」。本节的一致性只是「判断之间同不同」，不等于「判断对不对」——真实场景里必须用第 16 章的评测集去量正确性。本节先用确定性指标把成本结构（调用数、耗时）和分歧结构（一致性）测清楚，这两样是拓扑选型最直接的依据。

换句话说，这张表里没有「答案正确率」这一列。把成本、分歧、正确率三样凑齐，才是完整的选型依据——前两样本节给了，第三样留给第 16 章。

13.3.5 怎么读这张表

数字背后有三条规律，逐条看：

第一，模型调用次数直接决定成本与延迟。 debate 用 4 次调用、最慢；pipeline 和 voting 各 3 次。真实系统里每次调用都要花钱花时间，所以「多一轮辩论」是有明确代价的——多出的那一次调用换来的是「让反方有机会发言」。

第二，一致性不是越高越好，要看任务要什么。 pipeline 一致性 1.00，但那是因为它只有一个声音，没有任何独立校验；voting 只有 0.67，说明确实存在分歧——而这个分歧正是「值得警惕」的信号。高一致性可能是「只有一个声音」，低一致性可能是「真的众说纷纭」，两者含义完全不同，不能简单比较。

第三，debate 的价值不在「一致」，而在「暴露」。它的开场立场一致性只有 0.50，看似最差，但这恰恰是它的用途：把隐藏的分歧摆到台面上。本例里反方提出的是个伪论据（「奇数很多是合数」），辩论让它显形，裁决才有的放矢。如果一开始就让一个 Agent 直接下结论，这个错误的推理过程根本不会被看见。

一个容易踩的坑：这三张数字不能横向比「谁更好」。它们解决的是不同问题——pipeline 求「把复杂任务分步做对」，debate 求「把分歧挖出来」，voting 求「在不确定时用多数压住噪声」。选错拓扑，再多的 worker 也没用。

13.3.6 决策表：什么任务适合哪种拓扑

把 13.3.1 的结构差异落到具体任务上：

这张表有一条压倒性的使用原则，也是第 12、13 章反复强调的：能用顺序解决的，就不要上辩论和投票。pipeline 是最便宜、最可控、最易调试的拓扑；debate 和 voting 都要多花调用、多担风险（辩论可能被谬误带偏，投票可能被多数谬误压制）。只有在「顺序解决不了」时才升级——要么需要暴露分歧（debate），要么需要抗随机错误（voting）。

如果对着表还是拿不准，按顺序问三个问题：

1. 任务有唯一正确答案吗？ 有 → 事实题，直接排除 debate。
2. 单次调用可能随机出错吗？ 可能 → 上 voting 抗噪。
3. 任务能拆成有先后依赖的步骤吗？ 能 → pipeline 最省。

三个问题走完，多数场景会落到 pipeline 或 voting；只有「答案本身就是多元观点」时才轮到 debate。先用最便宜的拓扑，只有当它确实解决不了时才升级——这条升级路径比一开始就上重拓扑靠谱得多。

13.3.7 三个反模式

选拓扑时最容易踩的三个坑，值得单独拎出来：

反模式一：为了「更聪明」无脑堆 worker。 直觉是「三个臭皮匠顶个诸葛亮」，于是给每个任务都配上投票或辩论。但 13.3.3 的表已经说明：每多一个 worker 就多一次调用、多一份延迟，而质量未必提升。worker 数量应该由「任务的分歧程度」决定，而不是由「想要多少聪明」决定。

反模式二：用投票解决「大家都没知识」的问题。 如果三个投票者用的是同一个模型、同一份上下文、都没有某条关键信息，它们会同错——三票全错，多数票只会让错误更自信。投票能压住的是随机噪声，压不住相关的系统性错误。要解决后者，得靠外部检索（第 10 章）或人工，而不是加票数。

反模式三：用辩论解决有唯一答案的事实题。 「17 是不是质数」这种有确定答案的问题，辩论纯属浪费时间——正确的一方不需要靠嘴赢。辩论的价值在观点类、权衡类任务（方案选型、文案取舍），那里「没有标准答案、只有不同立场」。事实题走 pipeline 或 voting，观点题才走 debate。

这三个反模式的共同教训：拓扑是手段，不是目的。先问清「这个任务的核心难点是什么」，再挑能对症的拓扑；挑不出来，说明根本不需要多 Agent。

还有一个更隐蔽的坑：把多 Agent 当成「并行加速」的万能钥匙。多 Agent 默认是串行协作，真加速要靠 asyncio 并发，而并发又会引入 13.2 讲的竞态与顺序问题。想快，先把并发搞对，而不是先堆 worker。

13.3.8 本章小结

第 13 章到这里收尾，三节各管一段：13.1 用 supervisor/worker 划清职责（谁干什么），13.2 用消息总线与共享状态搭好通道（怎么传数据），13.3 用三种拓扑给出组织方式（怎么排布）。三者合起来，就是卷一应对「一个 Agent 不够用」时的完整工具箱。

但要记住贯穿全章的克制：多 Agent 是重武器。它带来能力隔离、上下文隔离和并行可能，也带来成倍的调用、延迟和调试难度。用之前先问一句「单 Agent 真的不够吗」——这个判断，比会写 supervisor 更重要。下一章转入可靠性：当循环跑得越来越复杂、调用越来越多，出错与超时就不再是偶发，而是常态，需要一套分类与重试机制来兜底。

13.3.9 真实模型下会怎样

本节的 worker 全是脚本写死的，每个 worker 说什么、说几轮、最终判什么，都由 FakeModel 的脚本决定。真实 Claude 的差别在于这三个拓扑的核心变量都由模型现场生成：

• pipeline 的「加工质量」不可控。假模型每一步的文本是写死的；真实模型可能在第 2 步就把结论丢了，导致第 3 步基于空信息收尾。链越长，误差累积越严重——所以真实 pipeline 通常要在每步加校验或早停。
• debate 的「对抗」会失控。假模型的反方只会说一句脚本台词；真实模型可能越辩越跑题，甚至互相附和（同质化）。控制它的办法是限制轮数（本例是 1 轮反驳）并给裁决者明确的判据。
• voting 的「独立性」会打折。假模型三个投票者的输入完全一样；真实模型若共享了同一段上下文（比如都看过同一份可能有错的资料），就会同错同对，多数票反而放大错误。独立性要靠输入多样化来保障，而不是靠「多跑几次」。
• 一致性可以真算。本例的一致性是我们手工标注的；真实系统里可以跑 N 次、统计最终答案的分布，用真实的分歧率来选择拓扑——这正是第 16 章对比实验的做法。

一句话：假模型把三种拓扑的骨架和成本结构演示清楚，真实模型负责让每一环都带上不确定性。拓扑选得对不对，最终要靠真跑一遍、量一遍才知道。

阅读导航：上一节：13.2 消息总线与共享状态 · 下一节：14.1 分类错误与重试 。