AI 工程基础体系 · 第 100/100 篇。内容覆盖机器学习、深度学习与生成式 AI;模型、数据、评测、权限和成本会作为同一生产系统处理。

多 Agent 协作模式:角色、路由、共享状态、冲突和评测

多 Agent 系统不是“让多个模型同时回答一个问题”这么简单。它是一个由多个具有不同职责的决策组件组成的生产系统:组件接收任务、读取上下文、调用工具、修改状态、把结果交给其他组件,并在权限、延迟、成本和失败约束下完成目标。

这里的 Agent 可以是:

  • 机器学习模型或深度学习模型;
  • 负责规划、检索、执行、审查的生成式 AI;
  • 传统规则引擎、SQL 查询器、搜索服务或人工审批节点。

因此,多 Agent 的核心问题不是模型数量,而是如何划分职责、如何选择下一步、如何共享事实、如何处理不一致,以及如何证明系统确实变好了


一、先建立系统模型:Agent、任务、状态和环境

1. Agent 不只是一个模型调用

一个 Agent 至少包含以下部分:

Ai=(Pi,Mi,Ti,Si,πi)A_i=(P_i,M_i,T_i,S_i,\pi_i)

其中:

  • PiP_i:角色提示、策略或规则;
  • MiM_i:模型,例如分类器、LLM、视觉模型或搜索排序模型;
  • TiT_i:可调用工具集合;
  • SiS_i:它能读取和写入的状态范围;
  • πi\pi_i:决策策略,即在当前上下文下选择下一动作的函数。

动作可以是:

a{生成文本,调用工具,转交 Agent,修改状态,请求人工,结束}a \in \{\text{生成文本},\text{调用工具},\text{转交 Agent},\text{修改状态},\text{请求人工},\text{结束}\}

因此,“研究 Agent”和“写作 Agent”的差异不应只体现在提示词上。研究 Agent 通常拥有检索和引用验证工具,写作 Agent 拥有文档草稿写入权限;如果二者都能任意修改最终文档,角色划分就只是名义上的。

2. 多 Agent 系统是一个受约束的状态转移系统

令系统状态为 StS_t,包括:

  • 用户请求;
  • 当前任务阶段;
  • 中间结论;
  • 来源和证据;
  • 工具调用结果;
  • 权限上下文;
  • token、时间和金额预算;
  • 审批记录;
  • 版本号或事件序列号。

在时刻 tt,系统选择 Agent ii 和动作 ata_t

(it,at)=π(St)(i_t,a_t)=\pi(S_t)

执行动作后得到:

St+1=F(St,it,at,ot)S_{t+1}=F(S_t,i_t,a_t,o_t)

其中 oto_t 是模型或工具的观测结果。

一个合格的协作系统必须使状态变化可解释:能够回答“谁在什么时间、基于什么输入、通过什么工具、修改了哪一部分状态”。如果只能看到最终文本,就无法区分模型推理错误、路由错误、工具错误和并发覆盖错误。

3. Agent、工具和协议是不同层次

这三个概念经常混淆:

  • Agent 决定下一步做什么;
  • 工具 执行一个相对明确的操作,例如查数据库、发起退款、搜索文档;
  • 协议 规定 Agent 如何发现、调用和交换工具或上下文。

Model Context Protocol(MCP)主要解决模型应用与外部工具、资源、提示模板之间的标准化连接。其典型结构包括 Host、Client 和 Server:应用侧 Host 管理连接,Client 与某个 MCP Server 通信,Server 暴露工具、资源或提示能力。

MCP 并不会自动完成:

  • 多 Agent 的任务拆分;
  • 哪个 Agent 应该被路由;
  • 多个 Agent 如何达成共识;
  • 业务状态如何提交;
  • 冲突如何解决;
  • 评测指标如何定义。

所以可以用 MCP 连接工具,但仍然需要单独设计 Agent 编排、状态存储和冲突控制。


二、角色设计:按决策边界拆分,而不是按名词拆分

1. 角色的基本条件

一个角色划分是有意义的,至少应满足以下条件之一:

  1. 目标不同:例如生成候选答案与验证候选答案;
  2. 可见信息不同:例如风控 Agent 不能读取完整个人资料;
  3. 工具权限不同:例如查询 Agent 只能读库,执行 Agent 才能写库;
  4. 错误代价不同:高风险动作需要独立审批;
  5. 评测方法不同:分类器用准确率或 F1,文档审查器用事实一致性和引用覆盖率;
  6. 生命周期不同:长期运行的监控 Agent 与一次性规划 Agent 的状态管理不同。

反之,仅仅把一个模型复制三次并分别命名为“分析 Agent”“推理 Agent”“专家 Agent”,通常不会产生稳定的协作收益。

2. 常见角色及其边界

路由器

路由器不负责解决完整任务,而是选择工作流或下一位 Agent:

r(x)=argmaxkP(kx)r(x)=\arg\max_{k} P(k\mid x)

它的输入通常包括用户请求、租户、风险等级、历史上下文和预算;输出应是结构化路由结果,而不是任意自然语言。

例如:

{
  "route": "refund_review",
  "risk": "high",
  "required_agents": ["policy_checker", "transaction_reader", "human_approver"]
}

路由器的常见失败是“自信地选错流程”。因此高风险领域不能只依赖模型分类概率,还应加入规则门槛:

route(x)={human_review,if risk(x)τargmaxkP(kx),otherwise\text{route}(x)= \begin{cases} \text{human\_review}, & \text{if } risk(x)\geq \tau \\ \arg\max_k P(k\mid x), & \text{otherwise} \end{cases}

规划器

规划器把目标拆成有依赖关系的子任务。若任务图为 G=(V,E)G=(V,E),边 uvu\rightarrow v 表示 vv 必须等待 uu 完成。

例如:

读取交易记录 ─┐
              ├─> 判断政策适用性 ─> 生成处理建议 ─> 审批
读取账户状态 ──┘

规划器不应直接拥有所有执行权限。否则规划错误会直接转化为业务副作用。

专家或执行器

执行器完成单一领域操作,例如检索、代码运行、SQL 查询、特征计算、图像分析或文档生成。它的输入输出应尽量结构化:

{
  "customer_id": "C001",
  "eligible": true,
  "evidence": ["policy:v3:section_4", "transaction:T88"],
  "confidence": 0.93
}

结构化输出可以被后续 Agent 验证,而不是让后续 Agent 从长文本中重新猜测事实。

审查器

审查器的目标不是“再写一遍答案”,而是检测违反约束的地方:

  • 是否引用不存在的来源;
  • 是否把推测写成事实;
  • 是否违反业务规则;
  • 是否越权访问或执行;
  • 是否遗漏必需字段;
  • 是否满足格式和安全要求。

审查器与生成器最好使用不同的提示、不同的工具权限,必要时使用不同模型或不同算法,以减少同源错误。

仲裁器或审批器

当多个结果不一致时,仲裁器根据明确规则选择结果、要求补证据或升级人工。仲裁器不应简单采用“多数票”,因为三个 Agent 可能共享同一个错误来源。

3. 角色边界必须包含权限边界

权限应绑定到 Agent 身份和任务上下文,而不是只写在提示词里。

例如:

Agent 可读 可写 高风险动作
检索 Agent 文档索引 检索缓存
规划 Agent 用户请求、摘要 计划草稿
执行 Agent 已批准参数 业务草稿 受限
审批 Agent 证据、策略、执行结果 审批记录 需要人工或强策略

提示词中的“不要退款”不是权限控制。真正的控制应由服务端在工具调用前检查:

allow(agent,action,resource,context){0,1}\text{allow}(agent, action, resource, context)\in\{0,1\}

工具服务拒绝未授权调用,即使模型生成了合法格式的调用参数,也不能执行。


三、路由:决定谁做、何时做,以及何时停止

1. 路由有三种不同含义

顺序路由

一个 Agent 的输出是下一个 Agent 的输入:

请求 → 规划 → 检索 → 生成 → 审查

适合依赖关系明确、上下文需要逐步收敛的任务。

并行路由

多个 Agent 同时处理相互独立的子任务:

请求 ─┬─> 法规检索
      ├─> 交易查询
      └─> 风险分类

适合降低延迟,但要求共享写入可隔离或可合并。

条件路由

根据结果选择下一条路径:

风险低 → 自动执行
风险高 → 审批
证据不足 → 继续检索
结果冲突 → 仲裁

条件不能只依据自然语言中的“看起来完成了”,应依据结构化状态和显式终止条件。

2. 路由器的输入不能只有用户问题

一个实际路由函数通常依赖:

r=f(q,h,p,b,ρ,δ)r=f(q, h, p, b, \rho, \delta)

其中:

  • qq:当前请求;
  • hh:历史上下文;
  • pp:权限和租户信息;
  • bb:剩余预算;
  • ρ\rho:风险等级;
  • δ\delta:当前任务状态和依赖完成情况。

同一个问题,在不同租户、不同权限或不同预算下可能必须走不同路径。

3. 路由的优化目标不只是准确率

可以把一次工作流的效用写为:

U=QλcCλlLλrRU=Q-\lambda_c C-\lambda_l L-\lambda_r R

其中:

  • QQ:结果质量;
  • CC:模型、工具和基础设施成本;
  • LL:延迟;
  • RR:风险,例如越权、错误执行或隐私泄露;
  • λc,λl,λr\lambda_c,\lambda_l,\lambda_r:业务对各项代价的权重。

例如,退款、药品建议和代码部署应提高 λr\lambda_r,而普通摘要可以更重视成本和延迟。

路由器还必须有停止条件:

stop(St)=valid(St)evidence_sufficient(St)budget_available(St)\text{stop}(S_t)= \text{valid}(S_t)\land \text{evidence\_sufficient}(S_t)\land \text{budget\_available}(S_t)

若只使用“继续思考直到满意”,系统会出现循环调用、成本失控和重复检索。

4. 路由失败的诊断方法

记录至少以下字段:

{
  "trace_id": "tr-17",
  "task_id": "task-4",
  "agent": "router",
  "input_hash": "…",
  "route": "policy_checker",
  "route_reason": {
    "risk": "high",
    "missing_evidence": true
  },
  "policy_version": "router-v5",
  "budget_remaining": 0.42,
  "next_state_version": 8
}

诊断时先区分:

  1. 路由分类错;
  2. 路由正确但 Agent 执行错;
  3. Agent 正确但工具返回错;
  4. 结果正确但状态合并错;
  5. 所有局部结果正确,但整体目标定义错。

如果没有 trace、输入快照和状态版本,这些问题通常会被混为“模型不稳定”。


四、共享状态:共享事实,不共享任意上下文

1. 共享状态的三种形式

消息传递

Agent 通过消息传递结果:

A → B

优点是边界清楚、容易审计;缺点是上下文可能重复传输,且长链路容易丢失信息。

共享黑板

所有 Agent 读写同一个任务状态:

State = {
  facts: ...,
  hypotheses: ...,
  evidence: ...,
  decisions: ...,
  status: ...
}

优点是协作方便;缺点是写入冲突、越权和隐式依赖更严重。

事件日志

每次变化追加为不可变事件:

EvidenceAdded(...)
PlanCreated(...)
ReviewFailed(...)
ApprovalGranted(...)

当前状态由事件重放得到:

Sn=F(F(F(S0,e1),e2),en)S_n=F(F(\cdots F(S_0,e_1),e_2)\cdots,e_n)

事件日志适合审计、回放和故障恢复,但需要处理事件版本、重放兼容性和数据保留。

2. 状态应分层

一个可靠的任务状态至少分为:

  1. 事实:来自数据库、文档或工具的观察结果;
  2. 假设:Agent 当前推断,可能被推翻;
  3. 证据:事实对应的来源、时间和版本;
  4. 决策:经过规则或审批后的结论;
  5. 副作用记录:已经发送的邮件、退款、部署等;
  6. 控制状态:任务阶段、重试次数、预算、锁和版本。

不能把事实和假设放进同一个字符串字段。例如:

{
  "fact": {
    "transaction_status": "settled",
    "source": "payments_db",
    "observed_at": "2025-01-10T12:00:00Z"
  },
  "hypothesis": {
    "refund_reason": "duplicate_charge",
    "confidence": 0.71
  }
}

如果“可能重复扣款”被写入 transaction_status,后续执行器可能把推测误当成数据库事实。

3. 共享上下文不等于共享全部历史

将所有对话、工具返回和内部推理都复制给每个 Agent 会导致:

  • token 成本增长;
  • 无关信息干扰;
  • 隐私和权限泄露;
  • 上下文窗口溢出;
  • 一个 Agent 的错误假设污染全局。

更安全的方式是为每个 Agent 构造最小上下文:

Ci=view(S,permissioni,taski)C_i = \text{view}(S, permission_i, task_i)

其中 view 只返回该角色完成任务所需的信息。状态存储层负责过滤,而不是依赖模型自觉忽略字段。


五、并发和冲突:多个正确结果也可能合并成错误状态

1. 冲突的形式

假设两个 Agent 同时读取版本 v=3v=3 的状态:

初始:limit = 100,version = 3
Agent A:读取 limit=100,建议改为 80
Agent B:读取 limit=100,建议改为 60

若系统采用“最后写入覆盖”,最终可能是 60 或 80,另一个决策会静默丢失。这不是模型冲突,而是并发控制错误。

常见冲突包括:

  • 写写冲突:两个 Agent 修改同一字段;
  • 读写冲突:一个 Agent 基于过期事实做决策;
  • 语义冲突:两个字段各自合法,但组合后违反约束;
  • 权限冲突:一个 Agent 生成了另一个角色才允许提交的动作;
  • 证据冲突:不同来源对同一事实给出不同值。

2. 乐观并发控制

对可变状态使用版本号:

UPDATE task_state
SET payload = :new_payload,
    version = version + 1
WHERE task_id = :task_id
  AND version = :expected_version;

返回行数为:

  • 1:提交成功;
  • 0:版本已变化,提交失败,必须重新读取并处理冲突。

关键点是:重试不能直接重复副作用。如果一次提交已经触发外部支付,再因为网络超时而重试,可能重复扣款。需要幂等键:

idempotency_key = task_id + action_type + logical_step

外部系统必须以该键去重,或者先写入待执行动作,再由可靠执行器异步提交。

3. 冲突解决不是统一采用“最新值”

不同状态适合不同策略:

  • 日志和证据:追加,不覆盖;
  • 计数器:使用原子增量;
  • 集合:可使用并集合并,但要处理删除语义;
  • 文档草稿:按段落或字段合并;
  • 金额、权限、审批结论:冲突时拒绝自动合并;
  • 低风险偏好:可以使用最后写入,但必须记录来源。

若两个 Agent 对同一事实给出不同值,应保留:

{
  "claims": [
    {"value": "eligible", "source": "policy_agent", "evidence": ["p1"]},
    {"value": "ineligible", "source": "policy_agent", "evidence": ["p2"]}
  ],
  "resolution": "pending_review"
}

直接选置信度最高的结果并不总是正确,因为置信度通常是模型自报分数,未必经过校准。

4. 语义不变量比字段锁更重要

假设系统有约束:

approved_amounteligible_amount\text{approved\_amount}\leq \text{eligible\_amount}

即使两个字段分别由不同 Agent 写入,也必须在提交事务中一起验证。字段级锁只能避免同时写入,不能保证跨字段不变量。

状态提交应类似:

读取当前状态
→ 校验版本
→ 校验权限
→ 校验业务不变量
→ 写入状态和事件
→ 提交外部副作用或创建待执行命令

六、一个可运行的最小协作示例

下面的程序只使用 Python 标准库,模拟一个“规划—审查—执行”的系统。它不调用真实模型,但完整展示了角色、路由、共享状态、版本控制和评测接口。运行环境为 Python 3.10 及以上。

from dataclasses import dataclass, field
from typing import Dict, List, Tuple


@dataclass
class State:
    request: str
    facts: Dict[str, str] = field(default_factory=dict)
    plan: List[str] = field(default_factory=list)
    review: str = "pending"
    status: str = "new"
    version: int = 0
    events: List[str] = field(default_factory=list)


class Conflict(Exception):
    pass


class Store:
    def __init__(self, state: State):
        self.state = state

    def read(self) -> State:
        # 生产系统应返回不可变快照,而不是直接暴露内部对象。
        import copy
        return copy.deepcopy(self.state)

    def commit(self, snapshot: State, changes: Dict[str, object], event: str):
        if snapshot.version != self.state.version:
            raise Conflict(
                f"stale version: expected={snapshot.version}, "
                f"actual={self.state.version}"
            )

        for key, value in changes.items():
            setattr(self.state, key, value)

        self.state.version += 1
        self.state.events.append(event)


def planner(snapshot: State) -> Dict[str, object]:
    if "account_status" not in snapshot.facts:
        plan = ["read_account", "check_policy", "prepare_answer"]
    else:
        plan = ["check_policy", "prepare_answer"]
    return {"plan": plan, "status": "planned"}


def reviewer(snapshot: State) -> Dict[str, object]:
    required = {"account_status", "policy_result"}
    missing = required - snapshot.facts.keys()
    if missing:
        return {"review": "reject: missing " + ",".join(sorted(missing))}
    return {"review": "approve"}


def execute(snapshot: State) -> Dict[str, object]:
    if snapshot.review != "approve":
        return {"status": "stopped"}
    # 这里仅产生无副作用的回答;真实写操作应经过权限和幂等控制。
    return {"status": "completed"}


def evaluate(state: State) -> Dict[str, float]:
    return {
        "completed": float(state.status == "completed"),
        "review_passed": float(state.review == "approve"),
        "event_count": float(len(state.events)),
    }


def main():
    store = Store(State(
        request="判断账户是否可以继续使用服务",
        facts={
            "account_status": "active",
            "policy_result": "allowed",
        },
    ))

    # 路由:状态为 new 时选择规划器。
    snap = store.read()
    if snap.status == "new":
        result = planner(snap)
        store.commit(snap, result, "PlannerCreatedPlan")

    # 路由:计划完成后选择审查器。
    snap = store.read()
    if snap.status == "planned":
        result = reviewer(snap)
        store.commit(snap, result, "ReviewerFinished")

    # 路由:只有审查通过才执行。
    snap = store.read()
    if snap.review == "approve":
        result = execute(snap)
        store.commit(snap, result, "ExecutorFinished")

    final = store.read()
    print("status:", final.status)
    print("plan:", final.plan)
    print("review:", final.review)
    print("version:", final.version)
    print("events:", final.events)
    print("metrics:", evaluate(final))


if __name__ == "__main__":
    main()

预期输出类似:

status: completed
plan: ['check_policy', 'prepare_answer']
review: approve
version: 3
events: ['PlannerCreatedPlan', 'ReviewerFinished', 'ExecutorFinished']
metrics: {'completed': 1.0, 'review_passed': 1.0, 'event_count': 3.0}

每次 read() 得到一个带版本号的快照。commit() 只有在快照版本仍然是当前版本时才写入,因此可以检测过期写入。示例中的 events 让状态变化可追踪;真实系统通常会把事件写入持久化日志,并在事务中同时更新状态。

下面模拟冲突:

store = Store(State(request="demo"))

a = store.read()
b = store.read()

store.commit(a, {"status": "planned"}, "A")
try:
    store.commit(b, {"status": "completed"}, "B")
except Conflict as exc:
    print("conflict detected:", exc)

输出:

conflict detected: stale version: expected=0 actual=1

这只检测到冲突,尚未解决冲突。生产实现需要重新读取当前状态,判断第二个更新是否仍然适用;如果更新是金额、权限或审批结论,通常应转人工或专门仲裁,而不是盲目重试。


七、通信和协议:何时用消息,何时用工具协议

1. Agent 间通信的最小契约

Agent 通信消息至少应包含:

{
  "task_id": "t-100",
  "trace_id": "tr-9",
  "sender": "retriever",
  "recipient": "writer",
  "schema_version": 2,
  "payload": {
    "claims": [],
    "sources": []
  },
  "deadline": "2025-01-10T12:01:00Z",
  "idempotency_key": "t-100-retrieval-1"
}

消息契约需要说明:

  • 字段含义和类型;
  • 哪些字段是事实,哪些是推断;
  • 是否允许缺失;
  • 失败如何表示;
  • 是否可重试;
  • 结果是否具有副作用;
  • 版本如何兼容。

自然语言消息适合传递复杂解释,但不适合直接驱动高风险动作。高风险动作应转换为结构化命令,并由服务端重新校验。

2. MCP 在协作系统中的位置

可以让一个 Agent 通过 MCP 发现或调用检索、文件、数据库等能力。此时要区分:

  • MCP Server 暴露的是工具、资源或提示能力;
  • Agent 编排器决定调用哪个 Server;
  • 业务服务决定调用是否有权限;
  • 状态层决定结果如何落库;
  • 评测系统决定调用是否改善结果。

工具返回的数据也不能自动视为可信事实。应记录来源、时间、权限主体和工具版本。对于写操作,工具服务还应执行参数校验、授权、幂等和审计。

3. OpenAI Agents 类框架的职责边界

OpenAI 的 Agents 相关指南和 SDK 主要围绕 Agent、工具、handoff、guardrail 和 tracing 等编排概念展开。使用这类框架时,仍应自己明确:

  • handoff 是控制权转移还是仅传递建议;
  • 工具错误是否重试;
  • guardrail 是输入检查、输出检查还是业务审批;
  • tracing 是否包含足够的状态版本和工具结果;
  • Agent 是否能直接执行副作用。

框架可以降低编排代码量,但不会替代业务状态机、权限系统、数据库事务和领域评测。具体 API 以所使用版本的官方文档为准,不能把实验性接口当作稳定协议。


八、失败路径:超时、重复、部分成功和失去上下文

1. 超时不等于未执行

客户端调用超时可能有两种情况:

  1. 请求确实没有到达服务;
  2. 服务已执行,但响应在返回途中丢失。

因此,重试前必须查询幂等键或执行记录:

发送 action(idempotency_key)
→ 超时
→ 查询 action 状态
→ 若已完成,读取结果
→ 若未创建,再安全重试

不能根据一次网络超时直接判断“动作失败”。

2. 部分成功必须显式建模

例如三个并行 Agent 中两个成功、一个失败,状态不应只有 failed。可以表示为:

{
  "subtasks": {
    "search_policy": "succeeded",
    "read_transaction": "succeeded",
    "risk_score": "timeout"
  },
  "overall_status": "waiting_for_retry"
}

这样路由器可以只重试失败子任务,而不是重新执行全部流程。

3. 循环和重复工作

常见循环是:

审查失败 → 生成器重写 → 审查失败 → 重新检索 → 生成器重写

必须限制:

  • 最大轮数;
  • 每类错误的重试次数;
  • 总 token 和金额预算;
  • 单任务最长时间;
  • 相同输入和相同工具参数的重复次数。

若第 kk 轮结果没有改善:

QkQk1<ϵQ_{k}-Q_{k-1}<\epsilon

就应停止自动循环,转人工或返回带缺陷说明的结果。


九、冲突的完整算例:两个 Agent 都有合理依据

假设退款政策规定:

  • 交易在 24 小时内重复扣款,可以退款;
  • 账户存在欺诈风险时,必须冻结自动退款;
  • 支付系统记录交易已结算;
  • 风控系统给出“高风险”。

退款 Agent 依据第一条政策输出:

eligible = true

风控 Agent 依据第二条输出:

auto_refund = false

这不是简单的真假冲突。两个结论分别作用于不同变量:

eligible=true\text{eligible}=true

表示业务上符合退款条件;而:

auto_refund=false\text{auto\_refund}=false

表示不能自动执行。

正确的仲裁结果应是:

{
  "eligible": true,
  "auto_refund": false,
  "next_action": "human_review",
  "evidence": ["policy:refund-24h", "risk:high"]
}

错误做法包括:

  • 看到“eligible=true”就自动退款;
  • 看到“high risk”就把退款资格改成 false;
  • 按 Agent 数量投票;
  • 让最终生成 Agent 自由解释并选择。

仲裁必须依据领域变量和不变量,而不是语言表面的冲突。


十、评测:评测协作系统,而不只是评测最终文本

1. 评测对象至少分四层

单 Agent 能力

测试每个角色是否完成自己的局部任务:

  • 分类准确率、精确率、召回率、F1;
  • 检索的 Recall@k、MRR、引用命中率;
  • 工具参数正确率;
  • 结构化输出通过率;
  • 代码测试通过率;
  • 事实核验准确率。

对于生成模型,不能只用 BLEU 或 ROUGE 判断事实正确性;摘要词面相似并不代表证据充分。

编排能力

评测:

  • 路由正确率;
  • 不必要的 Agent 调用率;
  • 子任务依赖是否满足;
  • 循环率;
  • 部分失败后的恢复率;
  • 并行化带来的延迟变化。

系统级结果

评测:

  • 任务成功率;
  • 关键业务约束违反率;
  • 人工升级率;
  • 副作用误执行率;
  • 隐私和越权事件;
  • P50、P95 延迟;
  • 每任务 token、工具调用和基础设施成本。

协作增益

将多 Agent 系统与强单 Agent 基线比较:

ΔQ=QmultiQsingle\Delta Q=Q_{\text{multi}}-Q_{\text{single}}

同时计算单位成本收益:

ROIQ=ΔQCmultiCsingleROI_Q=\frac{\Delta Q}{C_{\text{multi}}-C_{\text{single}}}

如果质量只提高 1%,成本提高 5 倍,或者延迟使业务不可用,就不能仅凭“准确率上升”宣布成功。

2. 评测必须包含对照组和故障集

至少建立:

  1. 单 Agent 基线;
  2. 多 Agent 无共享状态基线;
  3. 多 Agent 有共享状态版本;
  4. 多 Agent 加审查或仲裁版本。

故障集应覆盖:

  • 缺失字段;
  • 相互矛盾的来源;
  • 工具超时;
  • 工具返回恶意内容;
  • 重复消息;
  • 过期状态提交;
  • 低置信度路由;
  • 超预算;
  • 越权工具调用;
  • 长上下文截断。

只有正常样本上的平均分,无法说明系统具备生产鲁棒性。

3. 评测结果要能归因

一次任务应保存:

输入快照
→ 路由决策
→ 每个 Agent 的输入视图
→ 工具调用及返回
→ 状态版本变化
→ 冲突与重试
→ 最终输出
→ 人工或自动评分

同时要固定模型版本、提示版本、工具版本和数据集版本。否则回归测试失败时无法知道变化来自模型、提示、路由还是外部数据。

4. 人工评分与模型评分的边界

模型作为评审器可以提高规模,但容易与生成器共享偏差。高风险指标应加入:

  • 规则校验;
  • 数据库事实核对;
  • 专家抽检;
  • 双盲人工评分;
  • 不同模型交叉评审。

评分器也需要校准。例如把“引用格式正确”和“引用真正支持结论”分成两个指标,避免格式合格掩盖证据无关。


十一、机器学习、深度学习和生成式 AI 的统一处理方式

多 Agent 不是生成式 AI 专属模式。

一个传统机器学习流水线可以包含:

数据质量 Agent → 特征验证 Agent → 模型评测 Agent → 发布审批 Agent

深度学习系统可以包含:

数据采样 Agent → 训练监控 Agent → 漂移检测 Agent → 回滚 Agent

生成式 AI 应用可以包含:

路由 Agent → 检索 Agent → 生成 Agent → 事实审查 Agent

三者共同需要管理:

  • 数据版本;
  • 模型版本;
  • 评测结果;
  • 权限;
  • 成本;
  • 可回滚状态;
  • 失败和审计记录。

例如,模型发布不是“训练完成就部署”,而是一个受约束状态机:

训练完成
→ 离线评测通过
→ 安全与偏差检查通过
→ 灰度发布
→ 在线指标满足阈值
→ 扩大流量

任何一个 Agent 都不能绕过审批状态直接把模型推向全量生产。


十二、常见误解和真实边界

误解一:Agent 越多,效果越好

Agent 数量增加会引入通信、上下文、延迟和协调成本。若各 Agent 使用相同模型、相同数据和相同提示,它们可能产生高度相关的错误。协作只有在角色具有互补信息、工具或判断标准时才有价值。

误解二:让一个监督 Agent 负责所有审查即可

监督 Agent 也可能复用生成 Agent 的错误假设。对关键事实,应尽量使用独立来源或确定性检查;对支付、权限、部署等动作,应由服务端策略阻断,而不是依赖另一个模型“看出来”。

误解三:共享完整上下文能减少信息损失

完整上下文会放大无关信息、隐私暴露和错误锚定。正确目标是共享足够的事实和证据,并保留来源和不确定性,而不是共享所有内部过程。

误解四:MCP 本身就是多 Agent 协议

MCP 解决的是工具和上下文能力的标准化连接。多 Agent 的角色、路由、状态、冲突和评测仍需由应用层设计。

误解五:重试就是可靠性

没有幂等键、版本检查和副作用记录的重试,会把偶发故障变成重复执行。可靠性来自可恢复的状态机,而不是无限增加重试次数。


十三、一个可落地的设计顺序

应先写清楚业务不变量和失败代价,再决定是否拆分 Agent。随后为每个角色定义:

  1. 输入视图;
  2. 输出 schema;
  3. 可用工具;
  4. 权限边界;
  5. 成功和失败状态;
  6. 可重试性;
  7. 评测指标。

然后把工作流表示成显式状态图,而不是让 Agent 自由决定所有流程:

stateDiagram-v2
    [*] --> Received
    Received --> Routed
    Routed --> Executing
    Executing --> Reviewing
    Reviewing --> Completed: pass
    Reviewing --> Rework: recoverable failure
    Reviewing --> HumanReview: high risk or conflict
    Rework --> Executing
    Executing --> Retry: transient failure
    Retry --> Executing
    HumanReview --> Completed: approved
    HumanReview --> Rejected: denied
    Completed --> [*]
    Rejected --> [*]

关键路径是:路由只决定进入哪个受控阶段;执行阶段产生事实或候选结果;审查阶段验证约束;高风险或无法自动解决的冲突进入人工;所有状态变化都带版本、事件和审计信息。

最终,多 Agent 协作的评价标准不是“有多少个 Agent”,而是:

系统价值=目标质量错误风险延迟成本运行成本\text{系统价值} = \text{目标质量} - \text{错误风险} - \text{延迟成本} - \text{运行成本}

只有当角色边界减少了错误、路由利用了互补能力、共享状态保持了事实来源、冲突处理阻止了错误副作用,并且评测能够证明这些改进时,多 Agent 才构成一种工程能力,而不是把一次模型调用拆成更多调用。


系列导航与关联阅读

官方资料

本文依据研究论文、标准组织与主流框架官方文档重新梳理;正文、示例与工程清单由 WR BLOG 编写。