Agent 工程体系 · 第 63/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

多 Agent 路由：分类、能力匹配、动态选择、回退和评测

多 Agent 系统的核心问题，不是“如何同时运行多个模型”，而是：面对一个请求，系统如何决定由谁处理、以什么上下文处理、失败后交给谁、什么时候停止，以及如何证明这个决定是有效的。

在单 Agent 系统中，模型通常直接面对用户请求和工具集合。多 Agent 系统则增加了一个决策层：路由器需要在多个候选 Agent 之间进行选择。这个选择可能基于固定规则，也可能由模型动态完成；可能只决定第一次处理者，也可能在执行过程中不断转移控制权。

Anthropic 将这类系统区分为两种基本形态：Workflow 通过预先编排的代码路径控制模型和工具，Agent 则允许模型动态决定后续步骤与工具调用。Routing 属于一种典型 Workflow：先对输入分类，再将请求送入专门的后续流程。OpenAI 的 Agents 文档则将多 Agent 编排进一步分为 Agent 作为工具调用，以及 handoff 两种控制权转移方式。(anthropic.com)

一、先定义问题：路由到底在决定什么

设一次用户请求为：

x=(u,c,p,m)x = (u, c, p, m)

其中：

  • uu：用户输入，例如“我的订单已经签收，但我没有收到”；
  • cc：已有上下文，例如会话历史、用户身份、订单号；
  • pp：业务策略，例如是否允许退款、是否需要人工审批；
  • mm：运行时元数据，例如时间、地区、当前负载、Agent 健康状态。

系统中有候选 Agent 集合：

A={a1,a2,…,an}\mathcal{A} = \{a_1, a_2, \ldots, a_n\}

每个 Agent 不只是一个模型名称，而应被看作一个带有契约的执行单元：

ai=(Ii,Oi,Ti,Ri,Si,Ci)a_i = (I_i, O_i, T_i, R_i, S_i, C_i)

其中：

  • IiI_i：可接受的输入类型；
  • OiO_i：输出类型；
  • TiT_i：可使用的工具；
  • RiR_i：权限和风险边界；
  • SiS_i：状态与上下文要求；
  • CiC_i：能力描述，例如“处理退款”“查询物流”“解释技术错误”。

路由器要计算的是：

π(x,A,σ)→(a,d,γ)\pi(x, \mathcal{A}, \sigma) \rightarrow (a, d, \gamma)

其中：

  • aa：选中的 Agent；
  • dd：决策原因和分类结果；
  • γ\gamma：路由执行状态，例如重试次数、已尝试 Agent、剩余预算。

因此，“路由到哪个 Agent”只是最终结果的一部分。一个可生产化的路由决策至少还应说明：

  1. 请求属于什么任务类别；
  2. 为什么候选 Agent 满足能力要求；
  3. 传递了哪些上下文；
  4. 是否发生过回退；
  5. 当前 Agent 是否拥有最终回复权；
  6. 失败后是否允许再次尝试。

如果这些信息没有被显式记录，系统出现错误时通常只能看到“模型答错了”，而无法判断是分类错、能力声明错、上下文丢失、权限不足，还是回退策略造成了二次损害。

二、分类不是能力匹配：两个经常被混淆的决策

1. 分类回答“这是什么任务”

分类器输出一个类别：

y=f(x)y = f(x)

例如：

refund
shipping
technical_support
general

分类的对象是请求，不是 Agent。分类结果可以帮助系统缩小候选范围，但不能直接推出最终处理者。

例如：

“我想退掉昨天买的路由器，但页面显示已签收，我还没有收到货。”

这句话同时包含：

  • 订单状态问题；
  • 物流异常；
  • 退款意图；
  • 可能需要人工介入。

如果分类器被迫选择唯一标签，可能输出 refund，但真正的第一处理者应当是物流 Agent，因为退款资格依赖于确认包裹是否丢失。

因此更稳妥的分类输出不是单一字符串，而是结构化任务描述：

{
  "primary_intent": "delivery_exception",
  "secondary_intents": ["refund_request"],
  "entities": {
    "order_id": null,
    "product_type": "router"
  },
  "risk": "medium",
  "requires_external_state": true,
  "confidence": 0.86
}

这里的 primary_intent 描述主要任务，secondary_intents 保留伴随目标，entities 提供后续能力匹配需要的参数，requires_external_state 表示不能仅凭语言生成完成。

2. 能力匹配回答“谁能完成它”

能力匹配是一个约束问题：

Eligible(ai,x)=InputCompatible(ai,x)∧CapabilitySufficient(ai,x)∧PermissionAllowed(ai,x)∧Healthy(ai)\text{Eligible}(a_i, x) = \text{InputCompatible}(a_i,x) \land \text{CapabilitySufficient}(a_i,x) \land \text{PermissionAllowed}(a_i,x) \land \text{Healthy}(a_i)

只有满足硬约束的 Agent 才能进入候选集合：

E(x)={ai∈A∣Eligible(ai,x)}\mathcal{E}(x) = \{a_i \in \mathcal{A} \mid \text{Eligible}(a_i,x)\}

例如，退款 Agent 可能具备“读取订单”和“创建退款申请”的能力，但如果当前用户没有通过身份认证，则：

CapabilitySufficient = true
PermissionAllowed = false

它不能被选择。把“有能力”误认为“有权限”，是多 Agent 系统中较危险的设计错误。

分类和能力匹配的因果关系应当是：

原始请求
  -> 任务解析
  -> 生成结构化需求
  -> 过滤不合格 Agent
  -> 对合格 Agent 排序
  -> 选择或拒绝

而不是：

原始请求
  -> LLM 猜一个 Agent 名称
  -> 直接调用

后者将类别判断、能力判断、权限判断和运行状态判断全部压缩进了一次不可审计的生成。

三、从固定分类到动态选择

1. 静态路由：规则优先，路径可预测

静态路由使用确定性规则：

def route_by_rule(intent: dict) -> str:
    if intent["risk"] == "high":
        return "human_review"

    if intent["primary_intent"] == "delivery_exception":
        return "shipping_agent"

    if intent["primary_intent"] == "refund_request":
        return "refund_agent"

    if intent["primary_intent"] == "technical_support":
        return "technical_agent"

    return "general_agent"

它的优点是：

  • 决策可解释；
  • 延迟低；
  • 便于测试；
  • 适合高风险边界；
  • 不依赖路由模型的随机性。

缺点是规则覆盖范围有限。当类别变多、任务交叉、输入表达变化时，规则会快速膨胀。

静态路由最适合处理“类别边界稳定、后续流程明确”的任务。Anthropic 将 routing 描述为先分类、再将请求交给专门后续任务，并指出它适用于存在清晰类别且分类可以被可靠完成的场景。(anthropic.com)

2. 动态路由：先描述需求，再匹配能力

动态路由不直接要求模型从 Agent 名称中选择一个，而是让模型生成任务需求：

{
  "goal": "判断包裹是否属于物流异常，并给出下一步处理",
  "required_capabilities": [
    "query_delivery_status",
    "inspect_delivery_exception"
  ],
  "optional_capabilities": [
    "prepare_refund_case"
  ],
  "input_requirements": [
    "order_id"
  ],
  "risk_level": "medium",
  "output_format": "case_resolution"
}

然后由程序执行能力匹配：

def eligible_agents(requirement, agents):
    required = set(requirement["required_capabilities"])
    result = []

    for agent in agents:
        capabilities = set(agent["capabilities"])
        if not required.issubset(capabilities):
            continue
        if requirement["risk_level"] not in agent["risk_levels"]:
            continue
        if not agent["healthy"]:
            continue
        result.append(agent)

    return result

这一步很重要：模型可以负责理解请求，但不应独自决定权限和健康状态。权限、健康、租户隔离、配额和工具可用性应由程序或策略引擎验证。

3. 动态评分：在合格候选中排序

设候选 Agent aia_i 的评分为：

S(ai∣x)=wqQi+wcCi+wlLi+wrRi−wkKiS(a_i \mid x) = w_q Q_i + w_c C_i + w_l L_i + w_r R_i - w_k K_i

其中：

  • QiQ_i：历史质量，例如在相同任务集上的成功率；
  • CiC_i：能力匹配程度；
  • LiL_i：当前延迟或负载得分；
  • RiR_i：可靠性，例如最近错误率、超时率；
  • KiK_i：成本；
  • w∗w_*：业务权重。

注意：评分只能在硬约束过滤之后进行。不能因为某个 Agent 便宜、快速，就让它处理不具备权限或能力的任务。

一个简单的选择器如下：

from dataclasses import dataclass

@dataclass
class Agent:
    name: str
    capabilities: set[str]
    quality: float
    latency_score: float
    reliability: float
    cost: float
    healthy: bool
    risk_levels: set[str]

def choose_agent(requirement, agents):
    required = set(requirement["required_capabilities"])
    eligible = []

    for agent in agents:
        if not agent.healthy:
            continue
        if not required.issubset(agent.capabilities):
            continue
        if requirement["risk_level"] not in agent.risk_levels:
            continue

        score = (
            0.40 * agent.quality
            + 0.20 * agent.latency_score
            + 0.30 * agent.reliability
            - 0.10 * agent.cost
        )
        eligible.append((score, agent))

    if not eligible:
        return None, "no_eligible_agent"

    eligible.sort(key=lambda item: item[0], reverse=True)
    return eligible[0][1], "selected"

输入是结构化需求和 Agent 注册表，输出是 Agent 以及决策状态。实际系统中，quality 不应由单次模型自评产生，而应来自离线评测、线上成功结果和人工复核。

四、能力描述必须是可验证的契约

“擅长金融问题”不是一个可执行能力声明；“可以查询账户余额，但不能转账”才接近可验证契约。

一个 Agent 的能力描述至少需要包含：

{
  "id": "refund_agent",
  "version": "2026-08-12",
  "capabilities": [
    {
      "name": "prepare_refund_case",
      "description": "根据订单、支付和售后政策生成退款申请草稿",
      "input_schema": "RefundCaseInput",
      "output_schema": "RefundCaseDraft",
      "side_effect": "none"
    },
    {
      "name": "submit_refund",
      "description": "提交退款申请",
      "input_schema": "RefundSubmitInput",
      "output_schema": "RefundSubmission",
      "side_effect": "external_write",
      "requires_approval": true
    }
  ],
  "limits": {
    "max_context_tokens": 12000,
    "supported_languages": ["zh-CN", "en-US"]
  }
}

这里需要区分三种能力：

  1. 信息能力：查询、检索、计算；
  2. 决策能力：判断资格、分类、生成建议；
  3. 行动能力：写入系统、发起退款、修改工单。

行动能力必须额外声明副作用。一个 Agent 能够“生成退款建议”，并不代表它可以“执行退款”。如果路由器只看到能力名称而看不到副作用等级，就可能把只读任务错误地路由到具有写权限的 Agent。

在跨系统协作中，A2A 的 Agent Card 可以承担类似的能力发现职责，声明 Agent 身份、端点、能力、技能和认证要求；Task 表示有状态的工作单元，Message 表示交互内容，Artifact 表示任务产生的结果。不同实现和协议版本的字段、传输方式可能变化，因此生产系统应以所采用版本的规范和实现为准，而不能把某个示例 JSON 当作永久兼容接口。(a2aproject.github.io)

五、Supervisor、Router 和 Handoff 的边界

1. Router：做选择，不拥有全部执行控制

Router 通常完成：

解析请求
  -> 选择 Agent
  -> 创建执行上下文
  -> 交给 Agent

如果 Agent 执行期间不能再次请求其他 Agent，那么 Router 只是入口分发器。

2. Supervisor：持续拥有控制权

Supervisor 是一个持续运行的协调者。它可以：

  • 拆分任务；
  • 并行调用多个 Agent；
  • 汇总结果；
  • 检查中间产物；
  • 决定是否重试；
  • 决定是否切换 Agent；
  • 最终生成回复。

其状态可以抽象为：

st+1=F(st,et)s_{t+1} = F(s_t, e_t)

其中：

  • sts_t：第 tt 步的工作状态；
  • ete_t：当前 Agent、工具或人工审核产生的事件；
  • FF：协调策略。

Supervisor 适合处理：

“先查询物流，再判断是否符合退款政策，最后生成客服回复。”

这类任务的关键不是一次选择，而是多个阶段之间的依赖关系。

3. Handoff：转移回复权或控制权

Handoff 不只是“调用另一个模型”。它表示当前 Agent 将后续处理责任交给另一个 Agent。

例如：

triage_agent
  --handoff--> shipping_agent

转移后需要明确：

  • 新 Agent 是否看到完整历史；
  • 是否只收到摘要；
  • 原 Agent 是否还能继续执行；
  • 谁向用户发送最终回复；
  • 新 Agent 完成后是否返回原 Agent；
  • handoff 是否允许再次转回。

OpenAI 的多 Agent 编排文档将 handoff 作为一种控制权转移模式，并强调多 Agent 设计需要明确专业 Agent 的职责和交接关系。(developers.openai.com)

可以用以下模型区分两种方式：

Supervisor-as-tool:
  Supervisor -> 调用 Agent -> 收到结果 -> Supervisor 继续决策

Handoff:
  Agent A -> 转移控制权 -> Agent B 成为当前处理者

前者的最终回复权仍在 Supervisor；后者的最终回复权通常随控制权转移。

六、上下文传递：不是“把聊天记录全部复制过去”

上下文至少可以分成四层：

用户上下文：身份、会话、偏好
任务上下文：目标、约束、已完成步骤
证据上下文：工具结果、文档、数据库记录
控制上下文：预算、权限、超时、已访问 Agent

向下游 Agent 传递完整原始历史，容易造成三个问题：

  1. 上下文过长，增加成本和延迟；
  2. 无关内容干扰当前 Agent；
  3. 让下游 Agent 误以为历史文本中的指令仍然有效。

更稳妥的做法是传递结构化交接包：

{
  "task_id": "task-7821",
  "goal": "判断订单是否存在物流异常",
  "user_visible_summary": "用户称订单已显示签收但本人未收到",
  "facts": [
    {
      "source": "order_service",
      "order_id": "O1001",
      "status": "delivered",
      "timestamp": "2026-08-31T10:20:00+08:00"
    }
  ],
  "unresolved": [
    "需要查询签收凭证",
    "需要判断是否满足异常件政策"
  ],
  "constraints": {
    "may_issue_refund": false,
    "must_preserve_language": "zh-CN"
  }
}

facts 应当标记来源和时间，避免把模型猜测当成事实。unresolved 让下游 Agent 知道尚未完成的工作。constraints 则防止权限在 Agent 之间被“继承”得过多。

上下文压缩不是简单摘要。摘要如果删除了订单号、时间、权限限制等关键字段，可能使下游 Agent无法继续工作。因此应把上下文分为：

  • 必须保留的结构化事实；
  • 可重新检索的引用；
  • 可压缩的自然语言对话；
  • 不能传递的敏感数据。

七、动态选择中的“不确定”：路由器必须允许拒绝选择

设路由器对候选 Agent 的置信度为 pip_i。即使最大值较高，也不代表选择可靠。至少应检查两个条件：

max⁡ipi≥τ\max_i p_i \ge \tau

以及：

p(1)−p(2)≥δp_{(1)} - p_{(2)} \ge \delta

其中：

  • p(1)p_{(1)}：最高置信度；
  • p(2)p_{(2)}：第二高置信度；
  • τ\tau：最低置信度阈值；
  • δ\delta：候选间最小间隔。

例如：

refund_agent:   0.46
shipping_agent: 0.43
general_agent:   0.11

最大概率并不低，但差距只有 0.03。此时直接选择退款 Agent，可能将一个物流异常问题错误地变成退款流程。更合理的结果是：

route = clarification_or_supervisor
reason = ambiguous_between_shipping_and_refund

对于高风险任务，路由器还应使用保守规则：

def require_safe_route(requirement, confidence, margin):
    if requirement["risk_level"] == "high":
        return confidence >= 0.95 and margin >= 0.15

    return confidence >= 0.75 and margin >= 0.08

阈值不是普遍常数，而应通过验证集和线上代价函数确定。漏路由到专门 Agent 的代价、误路由到错误 Agent 的代价、请求澄清的代价，通常并不相同。

八、回退：不是失败后随便换一个 Agent

1. 回退的三种原因

回退至少有三类：

能力回退

原 Agent 发现任务超出能力边界：

technical_agent -> supervisor

这种回退通常是业务上的“无法处理”，不是系统故障。

运行时回退

原 Agent 因为超时、限流、服务不可用而失败：

refund_agent timeout -> refund_agent retry
                         -> refund_readonly_agent
                         -> human_review

质量回退

Agent 返回了格式正确但证据不足、事实矛盾或验证失败的结果：

shipping_agent -> evaluator
                -> alternative_shipping_agent

三者不能混在一起。能力不足时继续重试没有意义；服务超时时换同一个 Agent 可能有效；质量失败时则必须保留原结果和验证证据。

2. 回退状态机

一个最小状态机如下：

stateDiagram-v2
    [*] --> Classified
    Classified --> Selected: 存在合格候选
    Classified --> Clarification: 分类不确定
    Classified --> Rejected: 无权限或无能力

    Selected --> Running
    Running --> Completed: 通过验证
    Running --> Retry: 短暂性错误
    Running --> Fallback: 能力不足或质量失败
    Running --> HumanReview: 高风险阻塞

    Retry --> Running
    Fallback --> Running: 选择未尝试候选
    Fallback --> HumanReview: 无安全候选

    Completed --> [*]
    Clarification --> [*]
    Rejected --> [*]
    HumanReview --> [*]

其中最关键的状态不是 Running，而是 Retry 和 Fallback 的边界。

一次调用可以记录为：

{
  "attempt": 2,
  "agent": "shipping_agent",
  "failure_class": "quality_failure",
  "evidence": {
    "missing": ["delivery_proof"],
    "contradiction": false
  },
  "fallback_candidates": ["shipping_agent_v2", "human_review"]
}

3. 回退必须满足单调性

一个安全的回退策略应满足：

Riskt+1≤Riskt\text{Risk}_{t+1} \leq \text{Risk}_{t}

也就是说，回退后不能自动获得更多权限或执行更危险的动作。

例如：

可接受：
refund_write_agent 失败
  -> refund_readonly_agent
  -> human_review

不可接受：
refund_write_agent 失败
  -> unrestricted_admin_agent

除非用户重新授权或人工审批，否则系统不能因为失败而扩大权限范围。

4. 重试必须幂等

如果 Agent 已经成功调用外部支付系统，但响应在网络中丢失，系统无法直接判断“未执行”还是“已执行”。再次调用可能造成重复退款。

因此，具有副作用的任务应携带幂等键：

idempotency_key = tenant_id + task_id + action_name

并在外部服务或本地事务表中记录：

task_id      action             status       external_id
task-7821    submit_refund      succeeded    refund-991

回退逻辑应先查询动作状态，再决定是否重试。模型层面的“我还没有执行”不能作为事实依据，外部系统返回的状态才是依据。Agent 在执行过程中应持续从工具结果、代码执行结果等环境反馈中获得事实，并设置最大迭代次数或其他停止条件。(anthropic.com)

九、如何防止 Handoff 死循环

最典型的死循环是：

triage_agent -> shipping_agent
shipping_agent -> triage_agent
triage_agent -> shipping_agent

循环可能由三种原因产生：

  1. 两个 Agent 的能力边界重叠；
  2. 每个 Agent 都把“不确定”解释成“交给对方”；
  3. handoff 没有携带历史路径和剩余预算。

每次交接都应维护路由轨迹：

{
  "visited_agents": ["triage_agent", "shipping_agent"],
  "handoff_count": 1,
  "max_handoffs": 3,
  "last_reason": "missing_delivery_proof"
}

最小防护条件包括：

handoff_count<H\text{handoff\_count} < H

anext∉visited_agentsa_{\text{next}} \notin \text{visited\_agents}

或者允许回到已访问 Agent，但必须满足状态发生了有效变化：

Δevidence≠0\Delta \text{evidence} \neq 0

例如，物流 Agent 查询到新的签收凭证后，回到 Supervisor 重新判断是合理的；如果没有任何新证据，只是再次转回，则应终止。

def can_handoff(state, next_agent, new_evidence):
    if state["handoff_count"] >= state["max_handoffs"]:
        return False

    if next_agent in state["visited_agents"] and not new_evidence:
        return False

    return True

不要只依赖模型提示词说“不要循环”。循环是系统状态问题，应由运行时强制限制。

十、Agent2Agent：路由不只发生在进程内

进程内多 Agent 通常共享代码、状态和调用协议；跨组织或跨技术栈的 Agent 协作则需要网络级互操作协议。

A2A 的价值在于把“发现 Agent、提交任务、传输消息、获取产物、跟踪状态”从某个框架的内部调用，提升为可协商的外部契约。其核心对象可以这样理解：

Agent Card
  描述“我是谁、我在哪里、我能做什么、如何认证”

Message
  描述“一轮交互中发送了什么内容”

Task
  描述“这项工作当前处于什么状态”

Artifact
  描述“这项工作产生了什么可消费结果”

一个路由器发现远程 Agent 时，不应仅根据名称选择，而应检查：

Agent Card
  -> 能力和技能
  -> 输入/输出模态
  -> endpoint
  -> 认证要求
  -> 协议版本
  -> 是否支持流式或长任务

调用远程 Agent 后，路由器还需要把网络协议状态映射为本地状态：

submitted  -> queued
working    -> running
input-required -> waiting_for_input
completed  -> completed
failed     -> failed
canceled   -> canceled

Task 与本地运行实例的关系应通过稳定的 task_id 和 context_id 维护，而不能只依赖一次 HTTP 请求的连接生命周期。长任务可能需要轮询、流式事件或推送通知；消息和最终产物也不应混为一体。A2A 规范将 Task、Message 和 Artifact 分别用于任务状态、交互内容和任务结果，并定义了消息发送、任务查询、取消及流式更新等交互形式。(a2aproject.github.io)

一个常见误解是：

A2A 能力声明等于能力真实性保证。

实际上，Agent Card 是声明，不是测试报告。远程 Agent 可能版本已变、服务降级、部分工具不可用，或者声明的能力只在特定租户和权限下有效。因此能力发现后仍需要：

  1. 认证和授权；
  2. 输入 schema 校验；
  3. 健康检查；
  4. 小规模能力探测；
  5. 结果验证；
  6. 超时和取消机制。

十一、一个可运行的端到端路由示例

下面的示例不依赖具体 Agent 框架，使用 Python 标准库模拟分类、能力过滤、动态评分、执行、质量验证和回退。它的意义是展示生命周期，而不是提供生产级模型调用代码。

from dataclasses import dataclass
from typing import Callable


@dataclass
class Request:
    text: str
    risk_level: str = "medium"


@dataclass
class Requirement:
    intent: str
    required_capabilities: set[str]
    confidence: float


@dataclass
class Result:
    agent: str
    answer: str
    evidence: dict
    valid: bool


@dataclass
class Agent:
    name: str
    capabilities: set[str]
    quality: float
    reliability: float
    cost: float
    healthy: bool
    run: Callable[[Request], Result]


def classify(request: Request) -> Requirement:
    text = request.text

    if "签收" in text and ("没收到" in text or "未收到" in text):
        return Requirement(
            intent="delivery_exception",
            required_capabilities={
                "query_delivery_status",
                "inspect_delivery_exception",
            },
            confidence=0.91,
        )

    if "退款" in text:
        return Requirement(
            intent="refund_request",
            required_capabilities={"prepare_refund_case"},
            confidence=0.82,
        )

    return Requirement(
        intent="general",
        required_capabilities={"general_qa"},
        confidence=0.55,
    )


def run_shipping(request: Request) -> Result:
    # 模拟外部物流查询
    return Result(
        agent="shipping_agent",
        answer="订单显示签收，但尚未取得有效签收凭证，建议转人工核验。",
        evidence={
            "delivery_status": "delivered",
            "delivery_proof": None,
        },
        valid=True,
    )


def run_refund(request: Request) -> Result:
    return Result(
        agent="refund_agent",
        answer="已生成退款申请草稿，尚未提交退款。",
        evidence={"refund_submitted": False},
        valid=True,
    )


def run_general(request: Request) -> Result:
    return Result(
        agent="general_agent",
        answer="当前问题需要更多订单信息。",
        evidence={},
        valid=False,
    )


def select_agents(requirement: Requirement, agents: list[Agent]):
    selected = []

    for agent in agents:
        if not agent.healthy:
            continue
        if not requirement.required_capabilities.issubset(agent.capabilities):
            continue

        score = (
            0.45 * agent.quality
            + 0.40 * agent.reliability
            - 0.15 * agent.cost
        )
        selected.append((score, agent))

    selected.sort(key=lambda item: item[0], reverse=True)
    return [agent for _, agent in selected]


def route(request: Request, agents: list[Agent]) -> Result:
    requirement = classify(request)

    if requirement.confidence < 0.70:
        return Result(
            agent="supervisor",
            answer="请提供订单号，或说明您希望查询物流还是申请退款。",
            evidence={"reason": "classification_uncertain"},
            valid=True,
        )

    candidates = select_agents(requirement, agents)

    if not candidates:
        return Result(
            agent="human_review",
            answer="当前没有可用的专门处理 Agent，已转人工处理。",
            evidence={"reason": "no_eligible_agent"},
            valid=True,
        )

    attempted = set()

    for agent in candidates:
        if agent.name in attempted:
            continue

        attempted.add(agent.name)

        try:
            result = agent.run(request)
        except TimeoutError:
            continue
        except Exception as exc:
            print(f"{agent.name} failed: {exc}")
            continue

        if result.valid:
            return result

    return Result(
        agent="human_review",
        answer="自动处理未得到可验证结果，已转人工处理。",
        evidence={"reason": "all_candidates_failed"},
        valid=True,
    )


if __name__ == "__main__":
    agents = [
        Agent(
            name="shipping_agent",
            capabilities={
                "query_delivery_status",
                "inspect_delivery_exception",
            },
            quality=0.92,
            reliability=0.95,
            cost=0.20,
            healthy=True,
            run=run_shipping,
        ),
        Agent(
            name="refund_agent",
            capabilities={"prepare_refund_case"},
            quality=0.90,
            reliability=0.94,
            cost=0.25,
            healthy=True,
            run=run_refund,
        ),
        Agent(
            name="general_agent",
            capabilities={"general_qa"},
            quality=0.60,
            reliability=0.98,
            cost=0.05,
            healthy=True,
            run=run_general,
        ),
    ]

    request = Request("订单显示已经签收，但我没有收到货")
    result = route(request, agents)

    print(result.agent)
    print(result.answer)
    print(result.evidence)

预期输出类似：

shipping_agent
订单显示签收，但尚未取得有效签收凭证，建议转人工核验。
{'delivery_status': 'delivered', 'delivery_proof': None}

这个例子中有几个重要的因果关系：

  1. classify() 先把自然语言转换成任务需求；
  2. select_agents() 只根据能力和健康状态过滤；
  3. 评分只在候选 Agent 之间进行；
  4. shipping_agent 的结果必须包含证据；
  5. valid 表示结果通过了最小质量检查；
  6. 所有候选失败后，系统进入人工回退，而不是继续无界重试。

真实系统中，classify() 可以由模型完成，但必须使用结构化输出并在程序侧校验枚举值、字段完整性和权限约束。run() 则应封装工具调用、超时、取消、幂等键和外部状态查询。

十二、评测：不能只看“路由准确率”

多 Agent 路由至少有四个层次的评测对象。

1. 分类质量

给定标注数据集：

D={(xj,yj)}j=1ND = \{(x_j, y_j)\}_{j=1}^{N}

可以计算：

Accuracy=#正确分类N\text{Accuracy} = \frac{\#\text{正确分类}}{N}

但多标签、层级标签和拒答场景不能只看 Accuracy。还应关注：

  • 每个类别的 Precision；
  • 每个类别的 Recall；
  • 混淆矩阵；
  • 高风险任务的漏检率；
  • clarification 的过度使用率；
  • 置信度校准误差。

例如，物流异常被分到退款的代价可能远高于一般咨询被分到技术支持，因此需要代价加权：

RoutingCost=∑j=1NC(yj,y^j)\text{RoutingCost} = \sum_{j=1}^{N} C(y_j,\hat{y}_j)

其中 CC 是业务定义的错误代价矩阵。

2. 能力匹配质量

分类正确不代表 Agent 合适。应单独评测：

任务要求 -> 候选能力集合 -> 实际可完成性

可定义：

CapabilityRecall=被选 Agent 能完成的任务数需要专门能力的任务数\text{CapabilityRecall} = \frac{\text{被选 Agent 能完成的任务数}} {\text{需要专门能力的任务数}}

还要测量越权率：

PrivilegeViolationRate=发生权限不匹配的执行次数总执行次数\text{PrivilegeViolationRate} = \frac{\text{发生权限不匹配的执行次数}} {\text{总执行次数}}

这个指标应尽可能接近零，因为越权不是普通质量错误。

3. 执行质量

执行质量关注最终任务是否完成，而不是 Agent 名称是否选对：

TaskSuccess=GoalSatisfied∧EvidenceValid∧PolicyCompliant\text{TaskSuccess} = \text{GoalSatisfied} \land \text{EvidenceValid} \land \text{PolicyCompliant}

例如，用户询问“包裹是否丢失”，Agent 即使正确调用了物流查询工具，如果最终没有返回可信状态或后续动作，仍不能算任务成功。

4. 路由系统代价

完整目标通常是多目标优化：

J=α⋅SuccessRate−β⋅Latency−γ⋅Cost−η⋅RiskJ = \alpha \cdot \text{SuccessRate} -\beta \cdot \text{Latency} -\gamma \cdot \text{Cost} -\eta \cdot \text{Risk}

其中：

  • SuccessRate：任务成功率；
  • Latency：端到端延迟；
  • Cost：模型、工具和人工处理成本；
  • Risk：错误执行、越权或不可逆副作用风险。

不应只优化路由命中率。例如，一个永远把请求交给最强模型的路由器，可能分类准确，但成本和延迟不可接受；一个永远选择最快 Agent 的路由器，可能在复杂任务上失败率很高。

OpenAI 将 Agent workflow 的评测作为独立能力，强调应对工作流进行系统化评估，而不是只观察个别对话结果。Anthropic 也建议先用全面评测验证简单方案，再在确有收益时增加多步骤和 Agent 复杂度。(developers.openai.com)

十三、评测集如何覆盖真实路由边界

一个有效的路由评测集不能只包含“典型问题”，还应包括边界和故障样本：

1. 明确单一意图：
   “查询订单 O1001 的物流状态”

2. 多意图：
   “包裹没收到，我可以直接退款吗？”

3. 缺少实体：
   “帮我查一下订单”

4. 领域混合：
   “路由器连不上，是物流问题还是设备问题？”

5. 权限冲突：
   已认证用户请求执行未授权操作

6. Agent 不可用：
   首选 Agent 超时或返回 503

7. 结果无证据：
   Agent 给出结论，但工具没有返回依据

8. 循环风险：
   Agent A 和 Agent B 互相 handoff

9. 外部副作用：
   网络超时发生在扣款或退款之后

10. 提示注入：
   文档或工具结果要求改变路由策略

每条样本应记录：

{
  "input": "...",
  "expected_intent": "delivery_exception",
  "allowed_agents": ["shipping_agent", "supervisor"],
  "forbidden_agents": ["refund_write_agent"],
  "must_ask_clarification": false,
  "side_effect_allowed": false,
  "expected_terminal_state": "completed_or_human_review"
}

这使评测从“模型回答是否像人”变成了可验证的状态和约束检查。

十四、常见失败模式与诊断路径

失败一：分类器把用户最后一句话当成全部目标

表现：

用户：包裹没收到，可以退款吗？
分类：refund

诊断：

  • 检查是否丢失主任务和前置条件；
  • 比较单标签分类与多字段任务解析；
  • 查看是否把“可以退款吗”错误当成“立即退款”。

修复：

  • 使用主意图、次意图和前置依赖；
  • 将“询问资格”和“执行退款”区分；
  • 对存在外部事实依赖的请求先走查询 Agent。

失败二：Agent 注册表描述过于宽泛

表现：

technical_agent: handles all technical issues

结果是任何包含“不能用”的请求都被路由到技术 Agent，即使问题实际属于账号、物流或权限。

诊断：

  • 对能力名称做反向测试；
  • 为每个能力添加正例和反例；
  • 检查 Agent 能力是否对应具体工具和输入 schema。

失败三：回退只按异常类型，不检查已发生副作用

表现：

submit_refund timeout -> retry submit_refund

诊断：

  • 查询外部系统的动作状态；
  • 对照幂等键和事务日志；
  • 检查网络超时发生在请求发送前还是发送后。

修复：

unknown outcome
  -> query action status
  -> succeeded: adopt result
  -> not_found: retry with same idempotency key
  -> ambiguous: human review

失败四：Handoff 只传自然语言，不传状态

表现：

“请继续处理这个问题。”

下游 Agent 不知道已做过哪些查询、哪些权限被禁止，也无法判断是否需要重复调用工具。

诊断：

  • 检查交接包是否有 task_id；
  • 检查是否包含事实来源和未解决事项；
  • 检查是否记录已访问 Agent 和 handoff 原因。

失败五：只评测首选 Agent，不评测失败路径

很多系统在健康环境中表现良好，但首选 Agent 不可用时直接返回错误，或进入无限重试。

评测必须显式注入：

timeout
rate_limit
invalid_output
tool_failure
permission_denied
partial_success
unknown_side_effect

并验证每种故障是否进入预期终态。

十五、生产取舍：什么时候不应该使用多 Agent

如果一个任务：

  • 只有一个稳定意图；
  • 只需要少量工具；
  • 不需要独立权限边界；
  • 不需要不同上下文策略；
  • 单 Agent 已经满足质量和延迟要求；

那么增加路由层通常只会增加：

  • 一次或多次额外模型调用；
  • 上下文转换；
  • 调试复杂度；
  • 回退路径；
  • 评测矩阵；
  • 状态一致性问题。

Anthropic 明确建议从最简单的方案开始，只有当额外复杂度能带来可测量收益时才引入 Agentic 系统。复杂度可能换来更强的任务表现，但同时会增加延迟、成本和错误累积风险。(anthropic.com)

多 Agent 真正有价值的条件通常是：

不同任务确实需要不同工具或知识边界
且
不同 Agent 的能力差异能被验证
且
系统能够处理上下文、权限、回退和终止
且
离线与线上评测证明收益大于额外成本

最后需要把路由系统看成一个受约束的决策系统，而不是一个“让模型选择专家”的提示词技巧。完整的路由闭环应当是：

请求解析
  -> 分类与结构化任务
  -> 能力和权限过滤
  -> 动态选择
  -> 执行与证据采集
  -> 结果验证
  -> 完成、重试、回退或人工介入
  -> 记录轨迹并评测

其中，分类决定“请求像什么”，能力匹配决定“谁有资格处理”，动态选择决定“此刻选谁”，回退决定“失败后如何保持安全”，评测则决定“这套决策是否真的比简单方案更好”。只有这五个部分同时成立，多 Agent 路由才是可靠执行体系的一部分，而不是多个模型之间的随机转发。

系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。