多智能体协作与编排

把一个复杂任务交给单个 Agent，常见结局是：上下文越堆越长、注意力被无关信息稀释、在几个步骤之间反复横跳。多智能体（Multi-Agent） 的思路是把大任务拆给多个各司其职的 Agent，用明确的编排（Orchestration）把它们串起来。

但多智能体不是银弹。它用更多的 token 与更长的延迟换取分工带来的专注度。用错了，你只是把一个 Agent 的错误变成了三个 Agent 互相甩锅。本文讲清什么时候该拆、怎么拆、怎么让它们收敛。

为什么需要多智能体

单 Agent 的三重边界

• 上下文边界：所有工具 Schema、历史观察、中间结果都塞进一个上下文窗口，长任务下必然溢出或被稀释。
• 能力边界：一个系统提示很难同时把「严谨的代码审查」和「发散的创意头脑风暴」两种模式都表达好。
• 权限边界：单 Agent 若同时持有读与写、内部与外部的工具，权限面太大，一次提示注入影响全局。

协作的三种动机

分工与隔离是刚需，冗余是可选。很多团队一上来就做「辩论式」多智能体，结果成本翻三倍而质量没提升——辩论只在任务有明确对错、且单模型容易犯系统性错误时才有收益。

编排拓扑

多智能体的差异，八成体现在谁来决定下一步谁发言。常见四种拓扑：

顺序流水线（Pipeline）

Agent 依次执行，前者输出即后者输入。最简单、最可预测，适合有明确阶段的任务（抽取 → 校验 → 入库）。

extractor → validator → loader

缺点是任何一环出错都会传导，且无法回退。

层级式（Supervisor / Manager）

一个「主管」Agent 负责拆解任务、分派给下属、汇总结果；下属不直接对话，只与主管交互。这是最容易落地的多智能体形态。

          ┌────────────┐
          │ Supervisor │
          └──┬───┬───┬─┘
     ┌───────┘   │   └───────┐
 ┌───▼───┐  ┌────▼───┐  ┌────▼───┐
 │ 检索   │  │ 计算   │  │ 写作   │
 └───────┘  └────────┘  └────────┘

主管持有全局状态，下属只持有被分派的子任务——这天然实现了上下文隔离。

群聊式（Group Chat）

多个 Agent 在同一「聊天室」里轮流发言，靠发言选择器决定下一位。灵活但容易发散、成本高，需要强终止条件。

黑板式（Blackboard）

所有 Agent 读写一块共享状态（黑板），谁有可做的事谁就动手，由调度器决定触发。适合事件驱动、参与者动态增减的场景。

角色与共享状态设计

角色定义

每个 Agent 的角色提示要回答三件事：你是谁、你能用哪些工具、你的输出契约是什么。

你是「数据检索员」。
职责：根据给定问题，从内部知识库检索最相关的文档片段。
工具：仅可使用 search_kb；不得调用任何写操作。
输出契约：返回 JSON {"snippets": [{"doc_id": str, "text": str, "score": float}]}，
         最多 5 条，按 score 降序。不要输出解释性文字。

输出契约是协作的关键。若每个 Agent 都返回自由文本，下游就无从解析，整个编排退化成「一堆模型互相聊天」。

共享状态

状态该放什么、谁来改，决定了系统是否可调试。经验做法：

• 只放事实，不放过程：中间推理（Chain-of-Thought）留在各自 Agent 内部，不进共享状态。
• 结构化：用带字段的字典而非大段文本，便于程序消费。
• 单一写入者：每个字段明确由谁写，避免多个 Agent 竞争同一字段。

from dataclasses import dataclass, field

@dataclass
class SharedState:
    question: str
    snippets: list = field(default_factory=list)   # 检索员写
    draft: str = ""                                 # 写作员写
    critique: str = ""                              # 审查员写
    approved: bool = False                          # 审查员写
    steps: int = 0                                  # 调度器写

通信与协调机制

消息传递 vs 共享状态

• 消息传递：Agent 之间显式发消息，适合层级与群聊拓扑。消息可审计、可回放。
• 共享状态：Agent 读写同一块内存，适合黑板拓扑。耦合更低，但竞态与覆盖需要设计约束。

多数工程选混合：控制流用消息（谁触发谁），数据流用共享状态（结果放哪）。

冲突解决

当两个 Agent 给出矛盾结论时，需要仲裁策略：

• 投票：多个 Agent 独立作答，多数决。适合分类、判断类任务。
• 加权：按角色可信度或历史准确率加权，而非等权。
• 裁判：引入一个专门的 Judge Agent，输入双方论点做裁决。
• 升级：自动仲裁不确定时，升级给人类（Human-in-the-loop）。

消息协议设计

消息传递模式下，定义一套统一的消息结构能省去大量胶水代码：

from dataclasses import dataclass
from typing import Literal

@dataclass
class Message:
    sender: str                              # 发送者角色名
    recipient: str                           # 接收者角色名，"*" 表示广播
    kind: Literal["task", "result", "critique", "control"]
    payload: dict                            # 结构化内容
    trace_id: str
    step: int

kind 字段区分消息语义：task 是分派、result 是结果回传、critique 是审查意见、control 是终止/回退等控制指令。有了它，接收方就知道该把消息放进「待办」还是「素材」，而不用靠自然语言猜测意图。

一个经验：控制消息与数据消息分离。让「任务完成，请停止」这类控制信号走独立通道，而不是混在自然语言里让模型去理解——否则模型很可能忽略它，循环停不下来。

终止条件

多智能体最容易「聊不完」。必须有多层终止：

TERMINATORS = [
    lambda s: s.approved,                    # 任务达成
    lambda s: s.steps >= MAX_STEPS,          # 步数上限
    lambda s: s.tokens >= MAX_TOKENS,        # 预算上限
    lambda s: s.no_progress >= 3,            # 连续无进展
    lambda s: s.elapsed >= MAX_WALL,         # 墙钟上限
]

「连续无进展」是容易被忽略的一条：两个 Agent 反复交换相似内容却不推进时，应主动打断并回退到更简单的策略。

一个最小实现：Supervisor 编排

下面是一个不依赖框架的 Supervisor 循环，展示编排的核心结构：

AGENTS = {"retriever": retriever_agent,
          "calculator": calculator_agent,
          "writer": writer_agent}

def supervisor(state):
    """返回下一个要调用的 agent 名，或 None 表示结束"""
    if not state.snippets:
        return "retriever"
    if state.needs_calc and not state.calc_done:
        return "calculator"
    if state.snippets and not state.draft:
        return "writer"
    if state.draft and not state.approved:
        return "writer"          # 让写手依据 critique 修订
    return None

def run(question, max_steps=10):
    state = SharedState(question=question)
    for _ in range(max_steps):
        nxt = supervisor(state)
        if nxt is None:
            break
        state = AGENTS[nxt](state)     # 每个 agent 读状态、写状态
        state.steps += 1
    return state.draft or "未能完成任务"

这段代码把「谁下一步」「做什么」「何时停」三件事分开：supervisor 决定控制流，各 Agent 只负责处理被分派的子任务，max_steps 兜底。理解这个骨架后，再看 LangGraph、CrewAI、AutoGen 等框架，就能看清它们各自在骨架的哪一层做了封装。

各 Agent 的写法同样简单——读状态、做一件事、写状态：

def retriever_agent(state):
    hits = search_kb(state.question, top_k=5)
    state.snippets = hits
    return state

def calculator_agent(state):
    expr = extract_expression(state.snippets)
    state.calc_result = safe_eval(expr)      # 只允许白名单运算
    state.calc_done = True
    return state

def writer_agent(state):
    ctx = "\n".join(s["text"] for s in state.snippets)
    state.draft = llm.chat(
        f"依据以下资料回答问题，不要编造：\n{ctx}\n\n问题：{state.question}"
    )
    state.approved = is_complete(state.draft)
    return state

注意每个 Agent 只写自己负责的字段（snippets / calc_* / draft），这正是前面强调的「字段单一写入者」原则。一旦某个 Agent 越界改了别人的字段，调试就会变成噩梦。

上下文隔离与状态管理

多智能体相对于单 Agent 最本质的收益是上下文隔离：每个 Agent 的上下文里只有它需要的东西，不被无关信息稀释。设计时要刻意维护这条边界：

• 子任务描述而非全量历史：派发给下属的是「问题 + 必要背景」，不是整段对话历史。
• 结果摘要而非原始观察：检索员回给主管的是 Top-K 片段，不是整篇文档；抓取员回的是结构化字段，不是原始 HTML。
• 按需展开：主管需要细节时，再向下属要「展开第 3 条」，而不是一开始就把所有细节塞进来。

一个反模式是「主管把所有下属的完整输出都塞进自己的上下文」，这让主管的上下文迅速膨胀，最终和单 Agent 一样被稀释——多智能体白做了。

与工具调用及记忆的关系

多智能体建立在两样东西之上：

• 工具调用：每个 Agent 的能力边界由它持有的工具决定。工具集越小、越聚焦，角色越稳定。
• 记忆：Agent 之间以及跨会话的状态延续，需要一套记忆机制（短期会话状态 + 长期知识）。协作里最常见的需求是「让某个 Agent 记住上次的结论」，这属于记忆系统的职责。

换句话说，多智能体是工具调用 + 记忆 + 编排三者的组合产物，它不引入新的模型能力，只改变组织方式。

常见协作模式

反思（Reflection）

Agent 产出结果后，让同一个或另一个 Agent 审查并给出修改意见，再修订。对写作、代码生成提升明显。

draft → critique → revise → (可选) 再 critique → 直到 approved

关键是限制修订轮数（通常 1~2 轮），否则收益递减而成本线性增长。

辩论（Debate）

两个或多个 Agent 持不同立场论证，由裁判裁决。适合有客观正误、单模型易犯系统性偏差的任务。对开放性创作任务，辩论往往只是增加成本。

规划-执行分离（Planner-Executor）

规划者只负责把任务拆成步骤清单（不调工具），执行者按清单逐步执行。好处是规划不受执行细节污染，且步骤清单可人工审阅后再放行。

分工-汇总（Map-Reduce）

把大任务切成 N 份并行处理，再汇总。适合可并行的大规模处理（如对 100 篇文档分别抽取要点）。这一模式与 多智能体系统 里讨论的并行编排是同一思想。

成本与延迟

多智能体的代价必须算清。假设单 Agent 一次任务消耗 C 个 token、耗时 T：

延迟上，能并行的部分要并行：多个无依赖的子任务同时发起，墙钟时间接近单个。token 上，最大的浪费是重复上下文——群聊模式下每轮都要把全部历史重发，成本随轮数平方增长。对策是把长历史摘要化，或改成共享状态（只传增量）。

失败模式与排错

排错的第一原则是能重放：记录每个 Agent 的输入、输出与共享状态快照，出问题时逐步回放，定位是哪一步的分派或输出出了问题。

生产实践：观测与降级

上线后，多智能体比单 Agent 更难观测，因为一次请求横跨多个模型调用。需要做到：

• 统一 trace：给一次任务分配 trace_id，每个 Agent 的每次调用都挂上它，附带角色名、步号、耗时、token 数。
• 状态快照：每一步落一份共享状态快照（脱敏），出问题可逐步回放。
• 角色级指标：按角色统计成功率与耗时，快速定位是哪个 Agent 拖后腿。
• 成本归因：把 token 消耗按角色拆分，识别「谁在烧钱」。

降级策略同样要提前设计：

• 超时降级：某个 Agent 超时，主管改用已有信息给出「部分答案 + 说明缺什么」。
• 失败隔离：非关键 Agent（如审查员）失败不阻塞主流程，记录告警即可。
• 简化回退：多智能体整体不可用时，回退到「单 Agent + 精简提示」的保底路径。
• 限流保护：给每个角色的调用次数设上限，防止某个 Agent 异常刷调用。

这些机制的核心思想与单 Agent 一致：任何自动流程都要有预算与兜底，只是多智能体把「预算」的维度从单点扩到了角色与轮次。

评测

多智能体不能只看最终答案，还要看过程：

• 任务成功率：最终结果是否正确。
• 步数分布：是否用了过多轮次（低效或发散）。
• 单任务成本：token 与调用次数。
• 角色贡献度：去掉某个 Agent 后质量是否下降（消融实验）。
• 冲突率：Agent 间结论矛盾的比例。

消融实验尤其有价值：如果去掉「审查员」后质量没变化，说明这个角色是冗余的，应当删掉以省成本。评测集还要注意：

• 覆盖「应该拒绝」的用例（如越权请求），防止 Agent 为了完成任务而越界。
• 记录同一任务多次运行的方差，方差过大说明流程不稳定。
• 对比「多智能体」与「单 Agent + 好提示」两条基线，证明拆分确实带来收益。

小结

多智能体的价值来自分工、隔离与冗余，代价是成倍的 token 与延迟。落地时优先选层级式（Supervisor） 拓扑，因为它最容易控制与调试；把角色边界、输出契约、共享状态、终止条件这四件事设计清楚，系统才可控。协作之外，Agent 还需在跨会话中保持连续性，这依赖 Agent 记忆机制 ；多 Agent 之间的互相调用会放大提示注入等风险，务必配合 LLM 护栏 做输入输出过滤。若想进一步了解工程化的编排框架与消息协议，可阅读 多智能体编排 。