智能软件工程 · 案例实战：多智能体协作 Planner–Executor–Critic 三角色闭环

本篇核心论点（原则驱动）：多智能体不是"免费的智力增强"，而是一笔可量化的工程开销。三角色闭环用"多次调用 + 重试轮数"换成功率，只有当成功率提升的收益 > 成本与延迟的代价时才划算。判定权交给对照实验，不交给直觉。

阅读路径：§1 先立单 Agent 对照基线，§2–§3 讲角色分工与共享状态契约（自学），§4 是协调失败模式（对接 pr09 失败编码），§5–§6 是核心交付：对照实验口径与成本放大公式（动手实验），§7 给可跑的三节点闭环代码，§8–§10 是度量、诚实结论与接口。

1. 单 Agent 为什么不够 / 什么时候够：先立对照基线

任何"要不要上多智能体"的讨论，都必须先回答一个前置问题：单 Agent（一个模型 + 一轮 ReAct 式"思考—调工具—观察"循环，见 pr03 五件套）到底在哪些任务上失败？没有这条基线，"多角色更好"就是无法证伪的空话。

单 Agent 在下列情形确实够用，此时引入多智能体只是过度设计：

• 短程、单目标：步骤 ⩽ 5 且无需中途改变计划（查一个数、改一处配置）。
• 无自我校验需求：输出正确性由外部确定性测试即可判定（pr05），不必让模型自审。
• 上下文能装下：全部相关信息在一窗内，无需跨"子 Agent"分而治之。

相反，当任务呈现以下特征时，单 Agent 的结构性缺陷才暴露出来：

对照原则：本篇所有"多智能体收益"都相对 §1 这条基线衡量。同一个任务集，先测单 Agent 的成功率 SR_single、每任务 tokens、端到端延迟，再测三角色闭环——三个数字一起看，才允许下结论。

2. 三角色分工：Planner 分解、Executor 调工具、Critic 审校回退

三角色闭环把一次任务运行拆成三种职责，用共享状态（blackboard / scratchpad）串联，并用 Critic 的门控决定"通过交付"还是"退回重规划"：

• Planner（规划）：读任务与当前状态，产出/更新子目标序列与下一步意图；不亲自调工具。
• Executor（执行）：拿当前子目标去调用工具 / 生成中间产物，把结果写回状态；不做全局决策。
• Critic（评审）：对照验收标准审校产物，输出 pass 交付 revise 补做 replan 回规划 三类裁决之一；只裁不做。

关键设计取舍：Critic 必须"只裁不做"。一旦 Critic 既评审又改写，角色边界模糊，就会退化成"一个 Agent 自言自语"，白白多付两次调用成本却拿不到独立视角的收益。分工的价值来自视角独立性，不是来自"人多"。

3. 共享状态与通信契约：避免"传话游戏"式信息衰减

多智能体真正的工程难点不是"提示词怎么写三个角色"，而是状态如何在角色间传递而不失真。三种主流通信模式各有代价：

"传话游戏"信息衰减：若每轮把全部对话原文再喂给下一个角色，早期约束（预算上限、禁止项）会被反复摘要、逐字稀释，几轮后 Critic 审校的已不是原始需求。工程对策：把硬约束放进契约的固定字段（如 acceptance、constraints）而非塞进自然语言历史，使其不随轮次衰减。

一个最小 handoff 契约应当包含（这也是 §7 代码里 state 字典的字段）：

{
  "task":        "为公开 API 生成并验证 SDK",
  "constraints": ["不得写入真实密钥", "预算 <= 20 万 token"],
  "acceptance":  ["编译通过", "单元测试全绿", "无硬编码密钥"],
  "subgoals":    ["生成骨架", "实现方法", "补测试"],
  "plan_step":   2,
  "observations":["骨架已生成", "方法实现有 1 处 TODO"],
  "critique":    {"verdict": "revise", "reason": "TODO 未清"},
  "retry_round": 1
}

与 pr03/pr05 的衔接：单 Agent 的"scratchpad"在这里升级成"带写权限约定的共享黑板 + 结构化裁决"。而 Critic 的"验收"应尽量复用 pr05 的确定性测试（编译/单测），把"模型主观打分"降到最低——这样多智能体的增益才可归因、可复现。

4. 失败模式（对接 pr09 失败编码）

多智能体在单 Agent 的失败之上，额外引入一类协调失败（coordination failure）。下表给出可编码、可统计的失败清单，编码沿用 pr09 风格，便于跨案例聚合：

CF3 与 CF4 是一对相反病灶：Critic 太"松"就空转（永不重试、质量不升），太"严"或缺乏单调判据就震荡（重试不停、成本失控）。健康信号是 retry_round 呈单调收敛分布：多数任务 0–1 轮通过，少数多轮，且封顶后强制交付并标失败。

5. 单 vs 多 Agent 对照实验（本篇核心交付）

这是本案例要交给读者的可运行练习：在同一任务集上，用同一底座模型、同一工具集，仅改变"控制结构"（单 Agent 循环 vs 三角色闭环），测三件事并对齐统计口径（对接 pr08 度量、pr14 对照实证）。

5.1 实验设计：因子、任务集与重复

效度声明：必须随机化运行顺序、固定预热，避免上下文缓存等混杂虚增某一结构的"优势"。成功率判定优先用确定性验收（编译/单测/字段核对），把 LLM 主观评审的占比降到最低——这是内部效度的关键（对接 pr08 / pr14）。

5.2 "收益 vs 成本"权衡曲线口径

把两个结构在同一任务集上的结果画成散点：x 轴 = 每任务平均 total_tokens（或成本），y 轴 = 成功率 SR。多智能体"值得"的判据是：它相对单 Agent 带来的 SR 增益，落在成本可接受的前沿上。经验性的结构性趋势（非实测真值，见文末诚实标注）：

• easy 档：单 Agent 已高 SR，多智能体 SR 增益 ≈ 0 甚至为负（协调开销引入新失败 CF1–CF2），成本却翻数倍 → 过度设计。
• hard 档（长程 / 需自我校验）：单 Agent SR 明显偏低，闭环的 retry 把"一次做对"换成"反复做到对"，SR 增益可覆盖成本 → 值得。
• 成本随重试轮数近似线性放大：Critic 判一次 revise/replan 就多一整轮 Planner+Executor 调用（见 §6 公式）。

6. 用公式表达成本放大

设单 Agent 一次任务平均调用 k₁ 步、每步平均 token 为 T。三角色闭环一轮至少包含 Planner + Executor + Critic 三次调用；若 Critic 判 revise/replan 就多一整轮。设重试轮数 R（含首轮），则期望总 token 成本近似：

更一般地，多智能体成本是"各角色调用之和 + 重试轮数放大项"：

于是"值不值"的判据可以写成一条不等式：设单 Agent 成功率为 S、闭环提升到 S'（S' > S），一个成功任务的工程价值为 V，单位 token 成本为 c。多智能体当且仅当下式成立才划算：

其中 κ 为"单轮闭环相对单 Agent 一期的调用倍数"（三角色通常 κ ≈ 3 起）。直白结论：成功率提升 S'−S 必须大到能覆盖"约 3 倍 × (1+期望重试轮数)"的成本膨胀，否则多智能体就是净亏。

注明示意：以上为把"分工 = 更聪明"祛魅成可计算权衡的简化模型，假设各角色调用近似等价、重试相互独立，用以说明成本随角色数与重试轮数乘性放大这一结构性趋势。真实系数应以 §5 的实测 traces.csv 拟合，而非套用本式取真值。

7. 实现路径：最小三节点闭环（代码块）

下面给一段不依赖具体框架、LangGraph 风格的最小手写实现：节点 = 角色，state（dict）在节点间传递，Critic 决定 continue / replan / pass；同时采集每轮 agent_role / retry_round / total_tokens / latency 写入 traces.csv。traces schema 沿用 pr03，新增上述列（对接 pr03 单 Agent 五件套基础）。

7.1 三角色闭环 + 每轮 trace 采集

import time, json, csv, random
from llm import call_llm           # 你的底座封装：返回 (text, usage)
from tools import TOOLS           # 单 Agent 与多 Agent 共用同一工具集

MAX_RETRY = 4                      # CF4 震荡重试的封顶（成本熔断）
TRACE_COLS = ["task_id","structure","agent_role","retry_round",
              "verdict","total_tokens","latency_s","success"]

def run_planner(state):             # 读:task/observations  写:subgoals/plan_step
    plan = call_llm(
        "你是 Planner，只分解子目标不调工具。硬约束必须逐条保留："
        + json.dumps(state["constraints"], ensure_ascii=False)
        + "\n已有观察：" + str(state["observations"]),
        system="planner")
    state["subgoals"] = plan["parsed"]["subgoals"]
    return state

def run_executor(state):            # 读:subgoals[plan_step]  写:observations
    goal = state["subgoals"][min(state["plan_step"], len(state["subgoals"]) - 1)]
    obs  = call_llm(f"执行子目标并调用工具：{goal}", tools=TOOLS, system="executor")
    state["observations"].append(obs["text"])
    return state

def run_critic(state):              # 只裁不做：优先用确定性验收
    checks = run_acceptance(state)  # 编译/单测/字段核对（复用 pr05）
    if all(checks.values):
        state["critique"]["verdict"] = "pass"
    elif state["retry_round"] >= MAX_RETRY:
        state["critique"]["verdict"] = "pass"          # 封顶强制交付并标失败
    else:
        state["critique"]["verdict"] = "replan"
        state["critique"]["reason"] = [k for k,v in checks.items if not v]
    return state

def closed_loop(task):
    state = {"task": task["desc"], "constraints": task["constraints"],
             "acceptance": task["acceptance"], "observations": [],
             "subgoals": [], "plan_step": 0, "retry_round": 0,
             "critique": {"verdict": "replan", "reason": []}}
    for r in range(MAX_RETRY + 1):
        state["retry_round"] = r
        for role, fn in (("planner", run_planner),
                         ("executor", run_executor),
                         ("critic", run_critic)):
            t0 = time.time
            state = fn(state)
            emit_trace(task["id"], "multi", role, r,
                       state["critique"]["verdict"], t0)   # 记 tokens/latency
        if state["critique"]["verdict"] == "pass":
            break
    state["success"] = bool(run_acceptance(state).get("单元测试全绿", False))
    return state

7.2 最小消息总线与 trace 落盘

from collections import deque

class Bus:                          # 结构化 handoff 的最小实现
    def __init__(self): self.q = deque
    def send(self, frm, to, kind, payload):
        self.q.append({"from": frm, "to": to, "kind": kind,
                       "payload": payload, "seq": len(self.q)})
    def recv(self, to):
        out = [m for m in self.q if m["to"] == to]
        for m in out: self.q.remove(m)
        return out                  # 只投递给指定接收方，杜绝脏读/串台

_rows = []
def emit_trace(task_id, structure, role, rnd, verdict, t0):
    _rows.append({
        "task_id": task_id, "structure": structure, "agent_role": role,
        "retry_round": rnd, "verdict": verdict,
        "total_tokens": last_usage["total_tokens"],   # 由 call_llm 侧记录
        "latency_s": round(time.time - t0, 3),
        "success": "",                                 # 运行后回填
    })

def dump(path="traces.csv"):
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=TRACE_COLS)
        w.writeheader; w.writerows(_rows)
    print(f"已写出 {path}（{len(_rows)} 行）—— 下一步按 structure 分组聚合")

# 对照运行：同一任务集先跑 single_react（pr03 基线），再跑 closed_loop（multi）
# tasks = load_tasks("tasks.jsonl"); random.shuffle(tasks)   # 随机化，抗混杂
# for t in tasks: single_agent(t); closed_loop(t)             # 每任务各重复 5 次

诚实提示：LangGraph / AutoGen / CrewAI 等框架的实际 API 以官方文档为准；上面是"框架无关"的等价骨架，用 state dict + 三函数直观展示"节点 = 角色、Critic 决定 continue/replan"的闭环，便于读者看清成本从哪里来。把 emit_trace 接进任一真实框架同样适用——关键是每轮都落 agent_role/retry_round/total_tokens/latency。

8. 度量与看板（对接 pr08）

有了 traces.csv，聚合口径固定如下——每格同时报 SR、p50/p95 延迟、成本、平均重试轮数，四者不可只看其一：

看板告警阈值（建议）：若 retry_round 均值接近 MAX_RETRY 却 SR 不涨 → 大概率 CF4 震荡，检查 Critic 是否有单调判据；若恒为 0 且 SR 与单 Agent 无差 → CF3 空转，Critic 形同虚设。

9. 诚实结论：何时更好、何时是过度设计

一句话判定：把"要不要多智能体"从审美问题变成测量问题。默认答案是"先上单 Agent 基线（pr03），只在 hard 档、有确定性验收锚点、且对照实验证明 SR 增益覆盖成本时"，才升级为 Planner–Executor–Critic 闭环。分工不天然等于更好。

10. 接口：与系列其它案例的连接

一页纸总结：三角色闭环 = Planner 分解 / Executor 调工具 / Critic 只裁不做，靠共享状态与结构化 handoff 串联，用 retry 把"一次做对"换成"反复做到对"；代价是 token 与 p95 延迟约按 3×(1+E[R]) 乘性放大。因此"是否值得"必须由与单 Agent 严格对照的成功率/成本/延迟三方权衡判定——分工不天然等于更好。

诚实标注：①文中 LangGraph / AutoGen / CrewAI 等框架的实际 API 与用法以各自官方文档为准，本篇代码为框架无关等价骨架；②§5–§6 的曲线与系数为结构性趋势与示意，非某型号实测真值，实际结论须以自采 traces.csv 拟合为准。