多智能体协作的几种常见模式

多智能体协作的几种常见模式

一个 Agent 不够用时，自然会想到“多找几个 Agent 分工”。但多智能体并不是数量越多越好——它带来分工优势的同时，也带来了通信成本、调试难度和 token 开销。本文梳理几种常见协作模式及其适用场景，给出主管—工人、评审循环的 Python 示例和 Java 版的消息结构，并总结常见坑、检查清单与 FAQ。

一、背景：单个 Agent 的天花板在哪里

单 Agent 加工具，已经能解决很大一类问题。但随着任务变复杂，常见瓶颈会逐渐出现：

• 上下文被挤满：几十个工具的说明、长长的历史、大量检索结果挤在一个窗口里，模型注意力被稀释；

• 角色冲突：同一个提示词里既要求“大胆创作”又要求“严格挑错”，两种要求互相拉扯；

• 工具过多导致选择不准：一个 Agent 挂 40 个工具，选错的概率明显上升；

• 缺少制衡：自己写、自己审，很难发现自己的盲点。

多智能体的基本思路，是把一个“大而全”的 Agent 拆成多个“小而专”的 Agent：每个 Agent 拥有更短的提示词、更少的工具、更干净的上下文，再通过某种协作机制把它们组合起来。

需要强调的是，多智能体不是“更高级”的架构，而是一种以复杂度换取隔离与分工的取舍。只有当分工带来的收益大于通信和协调的成本时，它才值得。

二、什么时候该用多智能体

• 任务可以拆成 相对独立 的子任务（调研、编码、审查）；

• 不同子任务需要 不同的提示词、工具或模型；

• 单个 Agent 的上下文已经被工具说明和历史塞满；

• 需要“互相检查”来提升质量，例如写作者 + 审稿人。

反过来，如果一个 Agent 加几个工具就能完成，就不要急着拆分。

三、四种常见模式

1. 主管—工人（Supervisor / Worker）

一个“主管”Agent 负责理解任务、分派子任务、汇总结果；多个“工人”各司其职。结构清晰，是最常用的模式。

              ┌────────────┐
   用户 ────▶ │  Supervisor │ ◀──── 汇总结果
              └─────┬──────┘
          ┌─────────┼─────────┐
          ▼         ▼         ▼
     [调研 Agent] [编码 Agent] [测试 Agent]

2. 流水线（Pipeline）

Agent 按固定顺序处理，前一个的输出是后一个的输入，例如：需求分析 → 编码 → 审查 → 文档。可预测、易调试，但缺乏灵活性。

3. 评审/辩论（Critic / Debate）

一个 Agent 生成，另一个 Agent 挑错，循环几轮后输出。适合写作、代码审查、方案评估。需要限制轮数，避免互相“客套”或无限争论。

4. 对等协作（Peer-to-Peer / 黑板模式）

多个 Agent 通过共享的“黑板”（共享状态）读写信息，没有固定的中心。灵活但最难控制，生产环境要谨慎使用。

四、模式对比

再从成本与风险角度补充一张表：

五、主管—工人：一个可运行的 Python 示例

下面的示例演示主管如何分派任务、收集结构化结果，并在最后汇总。llm、llm_json、make_agent 需要接入真实模型，控制流本身可直接使用。

from dataclasses import dataclass
import uuid, time

@dataclass
class TaskItem:
    worker: str          # 指派给哪个工人
    instruction: str     # 给工人的具体指令
    depends_on: list[int]  # 依赖的任务序号（从 0 开始）

WORKERS = {
    "research": make_agent("你是调研员，只负责查资料并给出要点，不写代码", tools=[search]),
    "coder":    make_agent("你是程序员，只负责写代码并说明用法", tools=[run_python]),
    "reviewer": make_agent("你是审查员，只负责挑问题并给出修改建议，不直接修改", tools=[]),
}

def supervisor(task: str, max_items: int = 6, budget_tokens: int = 50_000) -> str:
    trace_id = uuid.uuid4().hex[:8]          # 整个任务共用一个 trace id，便于排查
    plan = llm_json(
        "把任务拆成不超过 %d 个子任务，每个子任务指派给 %s 之一，"
        "输出 JSON：{\"items\":[{\"worker\":..,\"instruction\":..,\"depends_on\":[..]}]}\n任务：%s"
        % (max_items, list(WORKERS), task)
    )
    items = [TaskItem(**i) for i in plan["items"][:max_items]]
    outputs: dict[int, str] = {}
    used = 0

    for idx, item in enumerate(items):
        if item.worker not in WORKERS:                      # 主管可能“幻想”出不存在的角色
            outputs[idx] = f"无效的工人：{item.worker}"
            continue
        deps = {d: outputs.get(d, "") for d in item.depends_on}   # 只传必要的上游结果
        start = time.time()
        out = WORKERS[item.worker].run(item.instruction, context=deps)
        used += out.tokens_used
        print(f"[{trace_id}] #{idx} {item.worker} 用时 {time.time()-start:.1f}s")
        outputs[idx] = out.text
        if used > budget_tokens:                             # 总预算护栏
            outputs[idx] += "\n（已达预算上限，提前结束）"
            break

    return llm(f"根据以下子任务结果给出最终答复，未完成的部分请明确说明：{outputs}")

这段代码里有几个值得借鉴的工程细节：

• 主管的输出要校验：工人名称、任务数量都可能不合法；

• 传递依赖结果而不是全部历史：每个工人只看到自己需要的信息；

• 统一的 trace id 与预算：出了问题可追溯，也不会无限烧钱。

六、评审循环：让一个 Agent 挑另一个的错

评审/辩论模式的关键是限制轮数并设置通过标准。下面是一个“写作者 + 审稿人”的循环：

def write_with_review(topic: str, max_rounds: int = 3) -> str:
    draft = writer.run(f"请就主题写一份初稿：{topic}")
    for round_no in range(1, max_rounds + 1):
        review = reviewer.run(
            "请逐条指出草稿的问题，输出 JSON："
            '{"pass": bool, "issues": [{"level": "high|mid|low", "desc": str}]}\n'
            f"草稿：{draft}"
        )
        data = parse_json(review)
        high = [i for i in data["issues"] if i["level"] == "high"]
        if data["pass"] or not high:          # 通过标准：没有高优先级问题
            return draft
        draft = writer.run(
            f"请根据以下问题修改草稿，只修改有问题的部分，不要重写全文：{high}\n草稿：{draft}"
        )
    return draft + "\n\n（注意：达到最大评审轮数，仍可能存在未解决问题）"

注意两点：一是把“通过标准”写成程序可判断的条件（没有 high 级问题），而不是让两个 Agent 自行决定何时结束；二是让审稿人只挑问题、不动手修改，职责分离才能避免两个 Agent 互相覆盖对方的成果。

七、Java 视角：用统一的消息结构连接 Agent

Agent 之间最好传结构化消息。在 Java 里可以用 record 定义协议，并在每条消息上带 trace id、发送者和类型，便于日志与审计：

import java.time.Instant;
import java.util.Map;

/** Agent 之间传递的统一消息 */
public record AgentMessage(
        String traceId,         // 同一任务的所有消息共用
        String from,            // 发送者 Agent 名称
        String to,              // 接收者 Agent 名称
        MessageType type,       // 消息类型
        String content,         // 正文（自然语言或 JSON 字符串）
        Map<String, Object> meta,   // 扩展信息：token 用量、耗时等
        Instant createdAt) {

    public enum MessageType { TASK, RESULT, REVIEW, ERROR }

    public static AgentMessage task(String traceId, String from, String to, String content) {
        return new AgentMessage(traceId, from, to, MessageType.TASK, content,
                Map.of(), Instant.now());
    }
}

/** 所有 Agent 实现这个接口，主管只依赖接口 */
public interface Agent {
    String name();
    AgentMessage handle(AgentMessage input) throws Exception;
}

/** 一个简单的主管：按顺序把任务交给工人，遇到异常时记录并继续 */
public class Supervisor {
    private final Map<String, Agent> workers;

    public Supervisor(Map<String, Agent> workers) {
        this.workers = workers;
    }

    public AgentMessage dispatch(AgentMessage task) {
        Agent worker = workers.get(task.to());
        if (worker == null) {
            return new AgentMessage(task.traceId(), "supervisor", task.from(),
                    AgentMessage.MessageType.ERROR, "未知工人：" + task.to(),
                    Map.of(), Instant.now());
        }
        try {
            return worker.handle(task);
        } catch (Exception e) {
            return new AgentMessage(task.traceId(), worker.name(), "supervisor",
                    AgentMessage.MessageType.ERROR, "工人执行失败：" + e.getMessage(),
                    Map.of(), Instant.now());
        }
    }
}

统一的消息类型让你可以很自然地在日志中过滤出所有 ERROR，也方便之后接入消息队列，把同步调用换成异步并行。

八、一个场景：自动化代码评审系统

假设团队想做一个“提交代码后自动给出评审意见”的系统，可以这样设计：

1. 主管收到一次 PR 的变更，判断涉及哪些类型的文件（业务代码、SQL、配置）；

2. 安全审查员：只看是否有硬编码密钥、SQL 拼接、权限检查缺失，拥有读取变更文件的只读工具；

3. 性能审查员：只看循环内查询、大对象拷贝等，同样只读；

4. 风格审查员：检查命名和注释规范；

5. 主管汇总三份意见，去重并按严重程度排序，输出一份评审报告。

为什么不用单个 Agent？因为三种审查的关注点不同，提示词写在一起容易互相干扰；拆开后每个 Agent 的提示词短、任务聚焦，也可以针对不同类型使用不同大小的模型。同时代价也很明确：模型调用次数增加约 3 到 4 倍，需要设置预算，并且对于很小的变更（例如只改了一行注释）应直接跳过，由主管判断是否真的需要动用三个审查员。

九、如何选模式：一个决策流程

面对一个具体需求，可以按下面的顺序自问，而不是凭感觉挑一个“看起来高级”的模式：

1. 步骤顺序是否固定？ 如果每次都是“分析 → 生成 → 检查”，直接用流水线，最便宜也最好调试；

2. 子任务是否需要运行时动态决定？ 如果需要根据输入决定调用哪些专家，用主管—工人；

3. 质量是否是首要目标，且有明确的判断标准？ 如果是，在关键产出后加一个评审循环；

4. 是否真的需要没有中心的自由协作？ 绝大多数业务场景并不需要，把它留给研究型或探索型任务，并配上严格的预算与轮数限制。

还可以用“拆分成本”做一次自检：拆分后每个 Agent 的提示词是否明显更短？工具数量是否明显减少？如果拆完之后每个 Agent 仍然背着一大堆工具和冗长提示词，那说明拆分并没有解决真正的瓶颈，只是增加了通信开销。

另外要提醒的是，模式之间并不互斥。一个常见的组合是：外层用主管—工人负责分派，某个关键工人内部再使用评审循环，例如“代码工人”写完代码后，由“审查工人”挑一轮问题再交付。组合时仍然要保证每一层都有轮数与预算上限。

十、工程上的注意事项

• 明确边界：每个 Agent 的职责和可用工具都要写清楚，避免越权和重复工作。

• 约定消息格式：Agent 之间最好传结构化数据（JSON），而不是一大段自然语言。

• 控制成本：每多一个 Agent，往往意味着多一轮模型调用。设置总步数和总 token 上限。

• 可观测：给每次任务一个 trace id，记录是哪个 Agent 在什么时候做了什么，否则出了问题无从排查。

• 防止错误放大：上游的错误会被下游当作事实。关键节点加入校验。

原则：先用单 Agent 跑通，再在确实遇到瓶颈时拆分。 多智能体是解决问题的手段，而不是目的。

十一、常见坑与修复

十二、多智能体落地检查清单

• ☐ 先验证过单 Agent 方案确实达不到要求，并记录了瓶颈。

• ☐ 每个 Agent 有清晰的角色、独立的提示词和最小工具集。

• ☐ 选定了协作模式，并能说出选择理由。

• ☐ 消息格式结构化，带 trace id、发送者和类型。

• ☐ 有总步数、总 token、总耗时三类上限。

• ☐ 评审类循环有轮数上限和程序化的通过条件。

• ☐ 关键结论有校验或依据，不因上游错误而全盘皆错。

• ☐ 日志可回放，能看出每个 Agent 的输入输出与耗时。

十三、常见问题（FAQ）

Q1：多智能体一定比单 Agent 效果好吗？不一定。分工能降低单个 Agent 的负担，但也引入了通信损耗与错误放大。是否更好要用评测集对比，同一批任务分别跑单 Agent 和多 Agent 方案，比较正确率、成本和延迟。

Q2：不同 Agent 可以使用不同的模型吗？可以，而且是常见的做法：让较强的模型担任主管或审稿人，较小的模型处理简单、重复的子任务。但要留意不同模型对工具调用格式和提示词的敏感度不同，切换后需要重新评测。

Q3：Agent 之间应该共享记忆吗？默认建议隔离：每个 Agent 只拿到完成任务所需的信息。确实需要共享时，使用明确的共享状态（如黑板），并规定谁可以写、谁只读，避免互相覆盖。

Q4：“辩论”模式真的有用吗？在某些需要多角度审视的任务上可能有帮助，但它也可能导致互相迎合或无意义的循环。建议先从“生成 + 评审”的简单结构开始，限制轮数，并通过评测确认确实带来了改善再加大投入。

十四、小结与下一步

多智能体的核心在于“分工 + 通信 + 控制”。选对模式比堆数量更重要，主管—工人模式是大多数场景的稳妥起点。

建议的实践顺序：

1. 用单 Agent 把任务跑通，记录它卡在哪里（上下文？工具太多？缺少制衡？）；

2. 针对瓶颈选择最小的拆分方案，例如仅增加一个“审稿人”；

3. 给每次任务加上 trace id 与预算护栏；

4. 用同一批测试任务对比拆分前后的效果与成本，再决定是否继续扩展。