多 Agent 协作的五个设计原则：从踩坑到落地

1
2
3

1. 范围偏差：A 说"优化这个服务"，B 理解为"优化延迟"，A 其实想"优化内存"
2. 粒度偏差：A 说"简单分析一下"，B 写了 2000 字深度报告
3. 隐含假设偏差：A 说"部署到生产"，A 假设 B 知道要先跑测试，B 直接部署了

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

Layer 1: 信息查询（RPC）
  → Agent A 问 Agent B 一个问题，B 返回答案
  → 本质：同步函数调用
  → 示例：A 问 B "这个函数在哪个文件"

Layer 2: 任务委托（异步）
  → Agent A 把一个完整任务交给 B，B 独立完成后返回结果
  → 本质：异步任务队列
  → 示例：A 让 B "分析这个服务的性能瓶颈"

Layer 3: 方案对弈（迭代）
  → Agent A 和 B 就一个问题来回讨论，互相挑战对方的方案
  → 本质：多轮协商
  → 示例：A 和 B 讨论"这个系统应该用什么架构"

Layer 4: 共创涌现（真协作）
  → 多个 Agent 共享上下文，持续演化一个方案，产出超越任何单个 Agent 的结果
  → 本质：集体智能
  → 示例：多个 Agent 共同设计一个复杂系统，各自贡献不同视角

1
2
3
4
5
6
7
8

用户请求："帮我重构这个函数"

过度拆分后的调用链：
User → Planner → CodeReader → Analyzer → Refactorer → CodeWriter → Tester → Reporter
                    ↕              ↕            ↕           ↕
              (语义分区风险)  (语义分区风险) (语义分区风险) (语义分区风险)

每多一跳，语义偏差累积一次。7 跳之后，最终输出可能已经偏离原始意图。

1
2
3
4
5
6
7
8

实际生成过程：
  模型倾向 → 结论方向已定 → 生成支撑该结论的推理步骤

看起来像：
  前提 → 推理步骤 1 → 推理步骤 2 → 结论

本质区别：
  推理链是结论的"事后合理化"，不是结论的"推导来源"

1
2
3
4
5

可信度 = f(锚点密度)

其中：
  锚点 = 可被独立验证的事实声明
  密度 = 锚点数量 / 推理步骤总数

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

低密度（不可信）：
  "这个服务延迟高是因为 DB 慢查询导致的"
  → 纯推理，无锚点，密度 = 0

中密度（部分可信）：
  "这个服务 P99 延迟 = 500ms [L2: 监控数据]，
   其中 DB 查询占 400ms [L2: trace 数据]，
   所以瓶颈在 DB"
  → 2 个锚点 / 3 步推理，密度 ≈ 0.67

高密度（可信）：
  "这个服务 P99 延迟 = 500ms [L2: 监控数据]，
   DB 查询占 400ms [L2: trace 数据]，
   该查询缺少索引 [L1: EXPLAIN 输出]，
   加索引后延迟降至 50ms [L1: 测试环境验证]"
  → 4 个锚点 / 5 步推理，密度 = 0.8

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

message:
  content: "服务延迟瓶颈在 DB 层"
  assertions:
    - claim: "P99 延迟 500ms"
      confidence: verified
      evidence: "grafana 监控面板 2024-01-15 14:00-15:00"
      verify_command: "curl 'http://grafana/api/query?...' "

    - claim: "DB 查询占 80% 耗时"
      confidence: verified
      evidence: "trace_id=abc123 的 span 分析"
      verify_command: "python mlog_log.py trace abc123 live.service.name"

    - claim: "根因是缺少联合索引"
      confidence: inferred
      reasoning: "查询条件涉及 user_id + created_at，当前只有单列索引"
      verify_command: "EXPLAIN SELECT ... FROM table WHERE user_id=? AND created_at>?"

    - claim: "加索引后预计延迟降至 50ms"
      confidence: uncertain
      reasoning: "基于类似场景的经验估算，未实际验证"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

                    ┌─────────────────────────────┐
                    │     Agent 输出（可能有错）     │
                    └──────────────┬──────────────┘
                                   ▼
                 ┌─────────────────────────────────┐
  Layer 1:       │  格式检查：输出结构是否完整？      │  ← 成本最低
                 └──────────────┬──────────────────┘
                                ▼
                 ┌─────────────────────────────────┐
  Layer 2:       │  逻辑检查：前后是否自洽？          │  ← 低成本
                 └──────────────┬──────────────────┘
                                ▼
                 ┌─────────────────────────────────┐
  Layer 3:       │  锚点验证：verified 声明是否属实？ │  ← 中等成本
                 └──────────────┬──────────────────┘
                                ▼
                 ┌─────────────────────────────────┐
  Layer 4:       │  交叉验证：独立 Agent 是否同意？   │  ← 较高成本
                 └──────────────┬──────────────────┘
                                ▼
                 ┌─────────────────────────────────┐
  Layer 5:       │  人类审查：关键决策人工确认        │  ← 最高成本
                 └─────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# 阶段 1：任务下发
message:
  type: task_request
  from: alice
  to: bob
  content: "分析 user-service 的性能瓶颈并给出优化建议"
  complexity_hint: medium

# 阶段 2：意图确认（接收方返回理解）
message:
  type: intent_confirmation
  from: bob
  to: alice
  understanding:
    scope: "user-service 的 API 延迟问题"
    approach: "查看监控 → 定位慢接口 → 分析 trace → 给出优化建议"
    output_format: "根因分析 + 优化建议列表（按优先级排序）"
    assumptions:
      - "重点关注 P99 延迟，不是吞吐量"
      - "优化建议限于代码层面，不涉及基础设施变更"
    questions:
      - "需要关注特定时间段吗？还是看整体趋势？"

# 阶段 3：确认/修正
message:
  type: intent_ack
  from: alice
  to: bob
  status: confirmed_with_correction
  corrections:
    - "也要看吞吐量，最近有用户反馈 QPS 上不去"

# 阶段 4：执行确认
message:
  type: execution_start
  from: bob
  to: alice
  content: "收到，开始分析。预计输出：延迟+吞吐量双维度分析，覆盖最近 7 天。"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

class Assertion:
    claim: str           # 声明内容
    confidence: str      # verified | inferred | uncertain
    evidence: str        # 证据来源（verified 必填）
    verify_cmd: str      # 验证命令（可选）
    reasoning: str       # 推理过程（inferred/uncertain 必填）

class AgentOutput:
    summary: str
    assertions: List[Assertion]

    @property
    def trust_score(self) -> float:
        """锚点密度 = verified 数量 / 总声明数量"""
        total = len(self.assertions)
        verified = sum(1 for a in self.assertions if a.confidence == "verified")
        return verified / total if total > 0 else 0

# 接收方处理逻辑
def handle_output(output: AgentOutput, risk_level: str):
    if risk_level == "low":
        return accept(output)
    elif risk_level == "medium":
        for a in output.assertions:
            if a.confidence == "inferred" and a.verify_cmd:
                validate(a)
        return accept_with_caveats(output)
    elif risk_level == "high":
        for a in output.assertions:
            if a.confidence != "verified":
                require_validation(a)
        return accept_after_validation(output)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def assess_complexity(task: str) -> str:
    signals = {
        "scope_ambiguity": check_ambiguity(task),
        "step_count": estimate_steps(task),
        "domain_count": count_domains(task),
        "reversibility": assess_reversibility(task),
        "blast_radius": assess_impact(task),
    }

    if signals["blast_radius"] == "high" or not signals["reversibility"]:
        return "high"
    if signals["step_count"] > 5 or signals["domain_count"] > 2:
        return "medium"
    if signals["scope_ambiguity"] == "low" and signals["step_count"] <= 3:
        return "low"
    return "medium"

# 复杂度 → 协作模式映射
COMPLEXITY_TO_MODE = {
    "low":    {"handshake": False, "annotation": "minimal", "verification": "L1-2"},
    "medium": {"handshake": True,  "annotation": "full",    "verification": "L1-3"},
    "high":   {"handshake": True,  "annotation": "full",    "verification": "L1-4"},
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

Week 1-2: 复杂度评估（方案 3）
  → 先有分类能力，为后续方案提供触发条件

Week 3-4: 语义握手（方案 1）
  → 对 medium/high 任务启用
  → 观察：握手后的返工率 vs 不握手的返工率

Week 5-8: 三级标注（方案 2）
  → 对所有输出启用标注
  → 观察：trust_score 与实际准确率的相关性

持续迭代：
  → 收集"语义偏差"案例，优化复杂度评估规则
  → 收集"锚点验证失败"案例，优化标注策略
  → 数据足够后，考虑引入对抗性验证