Agent 开发实战（七）：多 Agent 协作——规划者 / 执行者 / 审查者

前几篇都是"单 Agent 一杆到底"。当任务变复杂（"对比近三月三类数据、交叉验证、出一份带结论的报告"），单 Agent 容易在中途迷失、漏步骤、自说自话。本篇引入多 Agent 协作：一个负责拆解规划，一个负责调工具执行，一个负责审查校验。

1. 单 Agent 的瓶颈

上下文被稀释：长任务里，早期目标和约束会被后来的工具结果淹没。
缺乏自检：模型一边查数一边写结论，容易"顺手编"。
难并行：多个独立子任务挤在一次循环里，串行且易错。

多 Agent 不是银弹，但在"复杂、多步、要对结果负责"的场景里，分工确实更稳。

2. 三个角色

角色	职责	是否调工具
Planner（规划者）	把用户问题拆成有序子任务，明确每步要什么	否
Worker（执行者）	执行单个子任务，可调用工具、查数据	是
Reviewer（审查者）	核对结论是否有数据支撑、有无遗漏、有无编造	否（只读）

3. Planner：输出结构化任务

让 Planner 返回 JSON 任务列表，便于程序驱动：

PLANNER_PROMPT = """你是任务规划者。把用户问题拆成有序子任务。
只输出 JSON：{"tasks":[{"id":1,"goal":"...","tool":"query_table|query_sheet|search_documents|vcf_summary|...","args":{}}]}
不要执行，只规划。"""

def plan(question: str) -> list[dict]:
    resp = client.chat.completions.create(
        model=os.getenv("LLM_MODEL"),
        messages=[{"role":"system","content":PLANNER_PROMPT},
                  {"role":"user","content":question}],
        response_format={"type":"json_object"})
    return json.loads(resp.choices[0].message.content)["tasks"]

让 Planner 显式声明每个子任务要用的工具，等于逼它把"要什么数据"想清楚，后续 Worker 执行也更可控。

4. Worker：执行单个子任务

Worker 复用前面的工具执行器（第（三）篇），但每次只跑一个子任务，上下文更干净：

def work(task: dict) -> str:
    messages = [
        {"role":"system","content":"你是执行者，用工具完成给定子任务，给出该步骤的发现（基于真实数据）。"},
        {"role":"user","content": f"目标：{task['goal']}\n请调用工具：{task['tool']}，参数：{task.get('args')}"},
    ]
    while True:
        resp = client.chat.completions.create(
            model=os.getenv("LLM_MODEL"), messages=messages,
            tools=registry.openai_schemas(), tool_choice="auto", stream=False)
        msg = resp.choices[0].message
        if not msg.tool_calls:
            return msg.content
        messages.append(msg)
        for call in msg.tool_calls:
            res = execute_one(call)
            messages.append({"role":"tool","tool_call_id":call.id,
                             "content": compress_tool_result(res)})

每个 Worker 的输出是一段"基于真实数据的发现"，作为证据留待汇总。

5. Reviewer：核对与补漏

REVIEWER_PROMPT = """你是审查者。给你原始问题和各步骤发现，请判断：
1. 结论是否有数据支撑（禁止无依据断言）；
2. 是否遗漏关键子任务；
3. 是否存在内部矛盾。
输出 JSON：{"pass":true/false,"issues":[...],"verdict":"..."}"""

def review(question, findings: list[str]) -> dict:
    resp = client.chat.completions.create(
        model=os.getenv("LLM_MODEL"),
        messages=[{"role":"system","content":REVIEWER_PROMPT},
                  {"role":"user","content": f"问题：{question}\n发现：\n" + "\n".join(findings)}],
        response_format={"type":"json_object"})
    return json.loads(resp.choices[0].message.content)

若 pass=false，可把 issues 回喂给 Planner 补任务，形成"规划→执行→审查→补规划"的闭环（限定最大轮次，避免死循环）。

6. 编排主流程

def run_multiagent(question: str) -> str:
    tasks = plan(question)
    findings = []
    for t in tasks:
        findings.append(work(t))
    report = review(question, findings)
    if not report["pass"]:
        # 简单处理：把问题附上审查意见，交给一个汇总 Agent 出最终答复
        findings.append("审查意见：" + "; ".join(report["issues"]))
    return summarize(question, findings)

def summarize(question, findings):
    resp = client.chat.completions.create(
        model=os.getenv("LLM_MODEL"),
        messages=[{"role":"system","content":"根据发现写一份简洁、有结论、带数据引用的报告。"},
                  {"role":"user","content": f"问题：{question}\n发现：\n" + "\n".join(findings)}])
    return resp.choices[0].message.content

7. 什么时候该用多 Agent

✅ 任务步骤多、要对结果负责（报告、审计、交叉验证）
✅ 子任务彼此独立、可并行
❌ 简单问答（"今天注册多少"）——单 Agent 足够，多 Agent 只增加延迟和成本

8. 成本与边界

多 Agent 每轮都烧 token，务必设上限（最大任务数、最大审查轮次）。
Planner/Reviewer 用较小模型（如 gpt-4o-mini），Worker 才上大模型，省成本。
所有"事实"仍来自工具结果，Reviewer 只看证据链，不接触数据库。

9. 小结

多 Agent 的本质是把"思考"和"执行"分开，再加一道质检：

Planner 负责想清楚要什么；
Worker 负责真实取数；
Reviewer 负责别瞎编、别漏步。

它不神秘，就是用程序把"规划—执行—审查"这段人类工作流显式化。复杂任务里，这比让一个模型硬扛要可靠得多。

别为了"多 Agent"而多 Agent。能单 Agent 解决的，单 Agent 永远更便宜更快。