多 Agent 编排模式

哓番茄大约 5 分钟

多 Agent 编排模式

单 agent + 十个工具，超过一定复杂度后会「手忙脚乱」——工具选错、注意力被稀释。解法是拆成多个各司其职的 agent，用明确的拓扑结构编排。

本篇手写实现各模式（原生 SDK），每节末尾标注 LangGraph 里的对应物；框架用法详解见 LangChain/LangGraph 笔记。子代理的运行时机制见 Harness 子代理篇。

一、为什么拆：三个信号

信号 1：系统提示写到了 3000 字，一半在说「不要用 X 工具做 Y」
        → 角色混杂，拆成各管一摊的多个 agent

信号 2：任务天然分阶段（调研→写→审），且每阶段要不同工具/模型
        → 流水线，每阶段独立优化

信号 3：主 agent 上下文总被中间结果挤爆
        → 把「过程大、结论小」的子任务扔给子 agent

先记住反直觉的一条：多 agent 不是为了更聪明，是为了更可控——每个 agent 职责单一、提示词短、工具少，各自都更稳定。

二、模式 1：Router（路由器）

最简单的多 agent：一个分类器决定走哪条专家线。

from pydantic import BaseModel
from typing import Literal

class Route(BaseModel):
    target: Literal["tech_support", "billing", "chitchat"]
    reason: str

async def route(user_msg: str) -> str:
    """廉价小模型做路由判断"""
    resp = await call_llm(
        system="把用户消息路由到合适的处理方。",
        messages=[{"role": "user", "content": user_msg}],
        schema=Route,
        model="claude-haiku-4-5",           # 路由用小模型，省钱快速
    )
    return resp.parsed.target

HANDLERS = {
    "tech_support": tech_support_agent,     # 各自独立的单 agent
    "billing": billing_agent,               # 各有自己的工具与提示词
    "chitchat": chitchat_agent,
}

answer = await HANDLERS[await route("发票怎么开")](user_msg)

适用：意图可枚举、各分支独立
优点：结构最简单，每个分支可独立测试与迭代
LangGraph 对应：条件边（conditional edges）

三、模式 2：Pipeline（流水线）

固定阶段串联，每阶段一个 agent，输出是下一段输入。

async def content_pipeline(topic: str) -> str:
    outline = await agent("大纲师", f"为主题「{topic}」拟三级大纲", tools=[])

    draft = await agent("写手",
        f"按此大纲写正文：\n{outline}",
        tools=[search_tool])                   # 写手可以联网查资料

    review = await agent("审稿人",
        f"审这篇稿子，只输出修改清单：\n{draft}",
        tools=[])                              # 审稿人无工具，纯判断

    return await agent("编辑",
        f"按修改清单定稿：\n稿件：{draft}\n修改清单：{review}", tools=[])

适用：流程确定、阶段异构（不同提示词/工具/模型）
优点：每阶段可缓存、可重跑、可单独换模型
要点：阶段间只传「结论」不传全部过程，接口即提示词
LangGraph 对应：顺序边（add_edge 串节点）

四、模式 3：Supervisor（主管）

主管 agent 动态派活给工人，工人交结果，主管决定继续派还是结束。这是「模型自主编排」的起点。

WORKERS = {
    "researcher": ("调研员，搜集资料", [search_tool, read_tool]),
    "writer":     ("写手，产出文案", []),
    "critic":     ("批评家，找问题提改进点", []),
}

async def supervisor(task: str, max_iters: int = 15) -> str:
    context = f"任务：{task}\n\n[工作日志]\n"

    for _ in range(max_iters):
        # 主管本身也是一个 LLM 调用：决定下一步派谁（或结束）
        decision = await call_llm(
            system=f"""你是团队主管。根据任务与工作日志，决定下一步：
1. 派哪个工人做什么（给出 worker 名与具体指令）
2. 或输出 DONE: <最终答案> 结束任务。
可用工人：{ {k: v[0] for k, v in WORKERS.items()} }""",
            messages=[{"role": "user", "content": context}],
            schema=NextAction,   # {action: "delegate"|"done", worker, instruction, final}
        )
        act = decision.parsed

        if act.action == "done":
            return act.final

        role_desc, tools = WORKERS[act.worker]
        result = await run_agent(                    # 复用 02 篇的 agent loop
            system=f"你是{role_desc}", task=act.instruction, tools=tools)
        context += f"- 派 {act.worker}：{act.instruction}\n  结果：{result[:2000]}\n"

    return "（主管达到最大迭代数）"

适用：步骤无法预知、需要按中间结果调度
优点：灵活，能处理开放任务
代价：不可预测（多跑几轮很正常）、token 成本高、必须配迭代上限
LangGraph 对应：supervisor 模式（官方示例库有同款）

五、模式 4：Debate（辩论/对抗验证）

同一问题多个独立视角，互相批判后裁决。适合判断类任务而非执行类。

async def debate(question: str, rounds: int = 2) -> str:
        # 正方 / 反方各自独立上下文，避免互相「感染」
    pro = await run_agent("你是方案倡导者，论证该方案可行", question, [])
    con = await run_agent("你是风险审查者，全力找出该方案的问题", question, [])

    for _ in range(rounds):
        con  = await run_agent("风险审查者，回应并强化你的质疑",
                               f"原问题：{question}\n对方最新观点：{pro}\n你此前：{con}", [])
        pro  = await run_agent("倡导者，逐条回应质疑",
                               f"原问题：{question}\n对方质疑：{con}\n你此前：{pro}", [])

    return await run_agent("你是中立裁判，汇总双方观点给出结论与理由",
                           f"问题：{question}\n正方：{pro}\n反方：{con}", [])

适用：方案评审、事实核查、有争议的决策
原理：单 agent 自查会「顺着说服自己」；独立上下文才投得出反对票
成本：token ≈ 单 agent 的 5-8 倍，用在刀刃上

六、模式选型速查

模式	结构	可预测性	成本	适用
Router	一分多	高	低	意图分类分发
Pipeline	线性	高	中	固定阶段流程
Supervisor	星形动态	中	高	开放任务调度
Debate	环形	中	高	判断/评审
Fan-out 并行	扇形	高	中（wall-clock 低）	独立子任务批量处理

决策顺序：能 Router 不 Pipeline，能 Pipeline 不 Supervisor——自主性阶梯在多 agent 内部同样成立。

七、工程要点

1. agent 间通信 = 提示词传结论
   传结构化摘要，不传对方完整上下文（那是子代理模式存在的意义）

2. 每层都要上限
   主管迭代数、单 agent 轮数、总 token——层层设闸，防连锁失控

3. 模型分级
   路由/机械步骤用小模型，判断/创作核心步骤才用旗舰模型

4. 可观测
   多 agent 排查 = 看链路：哪个 agent 哪一步开始跑偏的
   → 07 篇的 tracing 在多 agent 场景不是可选，是刚需

5. 先单后多
   动手写多 agent 前，先确认单 agent + 好工具真的到瓶颈了

本篇小结

拆多 agent 是为了可控（职责单一）而非更聪明
四大模式：Router 分发、Pipeline 串联、Supervisor 动态调度、Debate 对抗
选型从简单到复杂；每层设上限；agent 间只传结论
手写完这些模式，再看 LangGraph 就是「把这些拓扑声明化」

下一篇：生产化工程——把 agent 从 demo 变成服务。