LLMNotes

第 17 章:多 Agent 协作

zjc 于 2026-01-17 发布

这是《LLM 应用开发 零基础实战指南》的独立章节版。本章从概念、实操和生产排查三个视角展开,代码块保留了原书可直接运行的版本。 多 Agent 不是把模型多调几次就叫协作,而是把复杂任务拆给不同角色,通过明确的通信协议、共享状态和决策边界完成工作。它适合大型任务,也带来更高成本和更难的排查。

17.1 何时需要多 Agent

场景 是否适合 原因
客服问题路由 可用小模型或规则 单步即可
文档问答 通常不需要 RAG 更直接
代码仓库修改 适合 需要检索、编码、测试、评审分工
市场研究报告 适合 多来源收集与综合
复杂工单处理 可选 流程可受控编排
简单文本改写 不需要 成本高

引入多 Agent 前先问:

  1. 单 Agent 是否已经足够;
  2. 上下文是否会互相污染;
  3. 是否需要不同工具权限;
  4. 是否需要并行工作;
  5. 失败时如何归因;
  6. 成本预算是否明确。

17.2 常见拓扑

Supervisor
  |-- Researcher
  |-- Analyst
  |-- Writer
  +-- Reviewer
拓扑 说明
Supervisor 中央协调,易控制
Pipeline 串式流转,简单
Debate 相互评审,适合结论校验
Blackboard 共享状态,各角色读写
Peer-to-peer 自由通信,难治理

生产系统推荐 Supervisor 或 Pipeline,便于权限、预算和审计。

17.3 消息协议

{
  "message_id": "m-001",
  "run_id": "run-100",
  "from": "researcher",
  "to": "analyst",
  "type": "artifact",
  "content": "已收集 12 篇官方文档链接",
  "artifacts": ["research-notes.md"],
  "status": "done",
  "budget_used": 4200
}

消息类型:

类型 含义
task 分配任务
question 请求澄清
artifact 提交产物
review 评审意见
error 失败
decision 决策或确认

消息要持久化,支持回放和恢复。

17.4 共享状态

from dataclasses import dataclass, field

@dataclass
class SharedState:
    run_id: str
    goal: str
    constraints: list[str]
    artifacts: dict[str, str] = field(default_factory=dict)
    decisions: list[dict] = field(default_factory=list)
    agent_status: dict[str, str] = field(default_factory=dict)
    revision: int = 0

状态管理规则:

  1. 产物写对象存储,状态存引用;
  2. 每次更新带版本;
  3. 并发写入要加锁或队列;
  4. 保留决策原因;
  5. 支持从最近一致状态恢复。

17.5 角色设计

示例代码任务:

Agent 职责 工具
Planner 拆任务 无写权限
Explorer 查代码 只读仓库
Implementer 修改代码 补丁工具
Tester 运行测试 沙箱命令
Reviewer 找风险 只读代码和 diff
Coordinator 合并结果 发布审批

角色隔离的价值是权限最小化:检索 Agent 不应拥有写权限,测试 Agent 不应访问生产网络。

17.6 协作流程

Goal
  -> Planner 生成任务
     -> Researcher 并行收集
        -> Analyst 汇总
           -> Writer 输出草稿
              -> Reviewer 检查
                 -> Coordinator 接受 / 退回 / 终止

退回规则:

  1. 限制最大修订轮数;
  2. 明确退回原因;
  3. 只退回相关角色;
  4. 保留每版产物;
  5. 连续失败转人工;
  6. 不允许 Agent 互相降低安全要求。

17.7 并发与一致性

并行收集可以降低耗时:

import asyncio

async def run_agents(agents, task):
    results = await asyncio.gather(
        *[agent.run(task) for agent in agents],
        return_exceptions=True,
    )
    return results

注意:

  1. 数据库写入需要幂等;
  2. 文件冲突需要锁;
  3. 外部 API 限流;
  4. 一个 Agent 失败不一定要终止全部;
  5. 汇总前检查产物完整性。

17.8 成本控制

多 Agent 成本容易放大:

总成本 = 每个 Agent token + 共享上下文复制 + 重试 + 评审轮次

控制手段:

  1. 角色使用合适模型;
  2. 传递摘要而非完整历史;
  3. 工具产物外置;
  4. 限制轮次和人数;
  5. 缓存公共研究结果;
  6. 简单任务降级单 Agent;
  7. 设置单任务金额上限。

17.9 调试与观测

Trace 示例:

run-100
  |-- planner message
  |-- researcher A tool calls
  |-- researcher B tool calls
  |-- analyst synthesis
  |-- reviewer comments
  +-- final artifact

关键指标:

指标 说明
task success rate 最终成功率
revision count 平均修订次数
agent failure rate 角色失败率
duplicated work rate 重复工作
total cost 总成本
wall time 总耗时
human escalation 人工升级

17.10 反模式

反模式 后果
所有 Agent 共享所有工具 权限过大
自由互发消息 死循环和漂移
每步复制全量上下文 成本高
无最终评审 质量不可控
无消息持久化 无法排查
角色职责重叠 互相推责
用多 Agent 代替业务流程 状态不可预测

本章小结

多 Agent 通过角色分工、受限通信和共享状态处理复杂任务。生产推荐 Supervisor 或 Pipeline 拓扑,严格隔离工具权限,控制轮次和成本,并保存完整消息链用于回放。先用简单架构解决问题,再按需求引入多 Agent。

思考题

  1. 什么时候多 Agent 优于单 Agent?
  2. Supervisor 拓扑为什么更容易治理?
  3. 多 Agent 消息为什么要持久化?
  4. 如何避免多 Agent 成本放大?
  5. 多 Agent 系统的最终质量由谁保证?