多 Agent 协作

目录

上一篇说明怎样把任务变成可恢复的工作流。本篇沿用订单异常调查:当支付排查和履约排查都需要各自检索资料、选择工具、追踪线索时,可以把它们交给独立 Agent,再由协调者汇聚证据。

多 Agent 的价值来自任务与上下文的合理分工。 单个 Agent 也能并行调用工具;只有需要独立决策循环、上下文或执行边界时,拆分 Agent 才提供额外价值。本文讨论如何选择协作关系、交接结果和恢复失败,以及怎样判断增加的协调工作是否值得。

1. 先找出需要拆分的工作

订单调查可能包含三个不同问题:订单记录是否一致、支付是否完成、履约事件是否缺失。查询三个接口可以由一个工作流并行完成。如果每条线索都需要进一步决定查什么、何时停止和怎样解释矛盾,再考虑为它们建立独立 Agent。

观察到的问题 拆分可能带来的收益 先确认什么
不同方向的资料难以放入同一上下文 各自调查,返回压缩后的证据 汇总是否丢失关键条件,原始出处能否追溯
工具、知识与验收要求差异明显 为各角色提供专门配置 专门配置是否在对应任务集上确有改善
子任务可独立推进 并行探索不同线索 普通并行工具调用是否已经足够
需要独立复核 分离产出与检查职责 检查者是否拥有不同证据与可执行的验收条件

角色名称不会自动产生专业能力;“分析师”和“审查者”使用相同材料,也可能重复同一个错误。拆分还会带来摘要损失、任务遗漏和状态同步问题,收益需要用同一组任务比较。

Anthropic 在其 2025 年研究系统工程记录中介绍了协调者与独立子 Agent 的分工,以及并行探索和结果压缩的作用。这能支持“独立探索适合某些研究任务”的设计判断,不能替代订单调查场景的评测。工程记录

2. 协作关系由信息依赖决定

四种常见模式可以承担不同工作,不必组合齐全:

模式 订单调查中的用法 交接重点
Supervisor–Worker 协调者分派支付与履约调查,再综合结论 子任务边界、证据格式、返回时限
Peer-to-Peer 两个调查者交换冲突证据并复核 哪个结论被质疑、依据是什么、何时停止
Pipeline 证据整理 → 原因分析 → 报告检查 前一阶段的输出满足后一阶段的输入契约
Dynamic Routing 根据问题类型选择支付或履约调查者 路由目标有效,模糊请求可以澄清

中央协调:让并行调查真正分工

协调者应给出具体问题,而不是让两个 Worker 都“检查订单异常”。例如,支付 Worker 核对交易记录、回调和状态变化;履约 Worker 追踪出库与配送事件。二者返回带来源与时间的发现,协调者检查遗漏与矛盾。

协调者分派支付与履约调查,各 Worker 独立检索并返回证据,最后统一检查与交付

下面是 Python 3.11+ 的可执行调度骨架:输入是已通过授权与预算检查的只读 Worker 工厂,每个工厂返回一个协程。它演示失败隔离,不包含模型、持久化和全局预算实现。

import asyncio

async def collect_workers(jobs, *, timeout):
    if timeout <= 0:
        raise ValueError("timeout must be positive")

    async def run_one(factory):
        try:
            async with asyncio.timeout(timeout):
                return {"status": "ok", "data": await factory()}
        except TimeoutError:
            return {"status": "timeout", "data": None}
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            return {"status": "failed", "error": type(exc).__name__}

    async with asyncio.TaskGroup() as group:
        tasks = {
            name: group.create_task(run_one(factory))
            for name, factory in jobs.items()
        }
    return {
        name: (
            {"status": "cancelled", "data": None}
            if task.cancelled() else task.result()
        )
        for name, task in tasks.items()
    }

普通异常被转成单个 Worker 的结果,其他 Worker 可以继续。外部取消仍向上传播,不能包装成正常交付;主动取消的子任务单独标记。这里的超时依赖协作式取消,阻塞事件循环或吞掉取消信号的代码不受硬时限保证。TaskGroup 的异常与取消行为参见 Python 文档。

调用方必须检查每项状态。支付证据是判断是否扣款的必要输入时,支付 Worker 失败就应交付“支付状态待确认”,不能让汇总模型补出成功结果。大批任务还需要并发上限、排队与总预算,不能一次创建无限量 Worker。

对等协作:交换证据,不以相似度判定正确

支付 Agent 认为已付款,履约 Agent 却看到未支付订单。有效交流应附上交易 ID、记录版本和发生时间,而不是反复陈述立场。

每轮可以让双方基于同一份上一轮记录独立回应,也可以顺序回应;这两种时序会产生不同的信息条件,需要明确选择。停止条件可以是矛盾已经由新证据解释、没有新增证据或轮次预算耗尽。

文本余弦相似度只能反映表示上的接近,不能证明结论一致,更不能证明正确。投票结果也应保留为多数意见;业务状态最终仍由可核验的记录支撑。

流水线:降级结果也必须满足契约

“证据整理 → 原因分析 → 报告检查”存在明确依赖。每个阶段保存输入版本、输出和下一阶段位置,恢复时继续未完成的部分。

若分析阶段失败,不能把原始证据直接冒充分析结果交给检查者。可用的降级结果应明确标记“原因未确认”,并满足下游可接受的结构。保存降级结果时,同样需要更新阶段位置,避免恢复后重复处理。

执行外部动作与保存检查点之间可能发生崩溃。检查点只能说明记录到了哪里,不能单独证明动作没有执行;涉及写入仍需稳定业务请求 ID、幂等处理和结果查询。

动态路由:分派任务不等于授予权限

路由可以使用明确规则,也可以使用模型分类。服务端应校验目标是否在允许集合内,并为无法判定的请求提供澄清入口。模型自报的 confidence 不能直接当作准确率,阈值应由验证集校准。

Agent 被选中之后仍须经过权限、资源范围和预算检查。把请求路由到退款专家,不等于允许它执行退款。

3. 交接的是有出处的结果,不是无边界聊天记录

任务消息至少应能关联到运行、子任务、计划版本和上游消息。以下是说明性数据结构,不是某个框架的协议:

{
  "run_id": "investigation-001",
  "task_id": "payment-check",
  "plan_version": 2,
  "in_reply_to": "assignment-008",
  "status": "completed",
  "findings": [
    {
      "claim": "交易记录状态为已支付",
      "source_ref": "payment-record-007",
      "observed_version": 4
    }
  ],
  "unresolved": ["订单状态尚未同步的原因"]
}

这里的 completed 表示调查子任务已交付,不表示整笔订单的问题已经解决。“事实记录”“对原因的解释”和“尚待验证的假设”应分别表达,汇总时保留来源与限制。

通信方式 减少了什么耦合 仍需管理什么
共享黑板 Worker 不必直接寻址其他 Worker 数据结构、写入权、版本及汇聚条件
显式消息 每次交接有发送方、接收方与关联 ID 重复、乱序、超时及接收方契约
事件总线 发布者不必知道所有订阅者 事件版本、消费状态、幂等和链路追踪

它们都不能消除协作契约。小规模单进程调查可以先用受控结果集合,跨进程和恢复需求出现后再选择持久存储或消息设施。

4. 状态分区减少冲突,版本检查阻止旧结果覆盖新结果

支付和履约 Worker 分别写自己的结果区,可以减少直接覆盖。但重试、旧计划和并发实例仍可能写入同一分区;多个分区之间也可能需要一致的汇总版本。

服务端应从认证后的执行身份确定可写分区,以运行 ID、计划版本、任务 ID 和尝试 ID 关联结果。不要信任消息自填的 agent_id,也不要用 Python assert 做授权检查:优化模式可以移除 assert。

推荐的写入顺序是:

  1. 校验当前执行身份、任务归属和资源范围。
  2. 校验计划版本与尝试状态,拒绝迟到的旧版本结果。
  3. 通过事务或原子条件更新保存结果;重复交付返回已有结果。
  4. 汇总者读取同一计划版本的必要结果,保留缺失和冲突状态。

权限校验回答“谁能写”;条件更新回答“这份结果现在是否还有效”。分区、授权和并发控制解决的是不同问题。

协调者主备切换时,也需要租约或 fencing 机制防止旧实例继续提交。复制一个 Supervisor 进程并不能自动完成一致的故障接管。

5. 失败恢复先看动作状态,再决定是否重试

已知状态 处理方式
只读查询暂时失败 在时限与重试预算内重试
结果结构不合格 返回明确错误,修订输入或改用满足同一契约的 Worker
可选证据缺失 标记缺失后继续,缩小结论范围
必要证据缺失 暂停相关判断或升级人工处理
写入结果未知 先按业务请求 ID 查询,不能盲目重新执行
已成功写入但后续失败 依据业务规则补偿或继续,不能假定回滚能抹除所有影响

对等协作还有等待问题。A 等 B,而 B 又等 A,可能形成停滞;但“等待图有环”只有在各方必须等待环内条件、且没有其他解除途径时才足以判断死锁。如果任一外部回复即可唤醒任务,单看环会误判。

维护等待对象、截止时间、唤醒条件与负责人,比只打印一个环更有用。超时后应撤销等待或重新分派任务,并同步状态,不能只杀掉一个协程就宣称业务已恢复。

6. 可靠性要按实际成功条件计算

如果三个阶段必须全部成功,且不考虑恢复,链式法则给出:

P(全部成功) = P(A) × P(B 成功 | A 成功) × P(C 成功 | A、B 成功)

只有在成功事件独立时,才能直接相乘三个边际成功率。若三个条件成功率都为 0.8,结果为 0.512。它描述的是这条指定链路,不是任意多 Agent 系统的成功率。

在“每个阶段条件成功率相同、全部成功才算完成”的分配假设下,端到端 95% 对应每阶段约 0.95 ** (1 / 3) = 98.30%。实际系统可以给不同阶段分配不同目标,并通过验证、恢复或改变任务分解改善结果。

不存在通用的“LLM 单步 99% 工程极限”,也不能根据单 Agent 的成功率直接推算拆分后的成功率:任务已经变化,错误相关性和恢复路径也变了。比较时应记录端到端完成率、事实错误、越权动作、未解决状态及人工介入,而不是只统计 Worker 是否返回文字。

7. 成本与延迟由调用轨迹决定

模型费用按每次调用的计费类别累加:输入、输出、缓存或其他适用项目分别计量,再加工具与基础设施费用。每轮重复发送的系统提示和历史消息也可能计入输入,不能只计算一次;缓存是否命中与如何收费取决于实际服务。

下面只演示调用数量,不代表实测成本或质量:

假设路径 模型调用数 未计入的变化
单 Agent 直接给出答案 1 检索、工具调用后的续写与修订
1 次分派 + 3 个各调用一次模型的 Worker + 1 次汇总 5 Worker 内部循环、重试、复核

这个差异不能直接换算成 5 倍费用,更不能推出固定质量提升。各次上下文长度、输出长度和模型单价都可能不同。

独立 Worker 并行时,该批耗时接近最慢 Worker 的耗时,再加分派、汇聚、排队与检查;共享限流和依赖会改变这一关系。便宜模型适合哪些角色也应实测:错误分派引发的返工可能超过节省的调用费。

评估至少保留同一批任务、验收标准、模型版本、工具条件、预算限制和重复运行结果。报告成本与延迟的分布,以及单位成功任务成本,才知道收益来自分工、额外算力还是放宽了预算。

8. 调试要还原交接与状态变化

Trace 应关联运行、任务、父子 Span、模型与工具版本、计划版本、输入输出引用、耗时、用量和终止原因。多源汇聚不一定能由单棵调用树完整表示,可以用关联链接表达交接。

诊断记录需要足够的上下文,但不能默认无限期保存全部敏感输入。按用途脱敏、控制访问和保留期;详细原文放在受控存储中,Trace 保存引用。

回放也要区分两件事:重放已记录的模型和工具响应,可以检查确定性的调度与状态迁移;重新调用模型则是重新运行,固定 seed 或 temperature=0 不能保证完整系统复现。外部数据、模型服务和并发时序都可能变化,写操作必须使用隔离环境或记录响应替身。

全连接的 N 个 Agent 可能形成 O(N²) 条潜在通信关系,但这不是“调试难度必然平方增长”的定律。限制交接关系、清晰记录版本和减少共享写入,都能让问题更容易定位。

多 Agent 是否值得引入,最终取决于分工能否改善目标任务:独立探索是否带回新证据,汇总是否保留关键条件,错误能否定位和恢复,额外成本是否换来了可测的收益。下一篇讨论框架与协议,继续把这些职责落实到实现与互操作边界上。

评论