1. 多智能体协作系统到底在解决什么问题先把这个项目的核心价值说透。多智能体协作系统英文里叫 Multi-Agent Collaboration System简单理解就是让多个具备独立思考能力的 AI 智能体Agent组成一个团队各自负责不同的子任务通过消息传递、任务分配、结果汇总等方式协同完成一个单靠单个智能体难以搞定的复杂目标。我最初接触到这类系统时第一反应是“这不就是把几个大模型 API 串起来吗”实际深入做下来才发现完全不是这么回事。单个大模型确实能完成不少事但它的瓶颈非常明显上下文窗口有限、单一角色定位难以覆盖复杂流程、长链路任务容易在中间步骤丢失信息。多智能体协作系统的本质不是“多个模型轮流跑”而是“多个有明确分工的智能体按照一套协作协议并行或串行地推进任务”每个智能体有自己的记忆、工具、行为边界和输出规范。回到这个项目标题——“第 28 章 案例四 多智能体协作系统”这显然是某本技术书籍或培训教程中的一个完整实战案例。从章节编号来看前面已经有 27 章铺垫到这里开始进入一个综合性的多智能体案例。这种案例通常不会只讲理论而是要把前面讲过的提示词工程、工具调用、记忆机制、任务编排等能力整合到一起最终搭建出一个能跑的、解决真实场景问题的协作系统。这个案例适合谁来参考有两类人。一类是已经熟悉大模型 API 调用、想更进一步掌握智能体架构的开发者另一类是正在设计 AI 产品的技术负责人想搞清楚多智能体系统到底比单智能体强在哪里、哪些场景值得用、哪些场景用了反而更糟。后面我会把设计思路、核心原理、实操搭建、踩坑记录全部拆开讲手把手把这个案例复现出来。2. 整体设计与方案选型为什么非要用多智能体2.1 先想清楚单智能体真的不够用吗在做多智能体系统之前必须先回答一个问题这个任务单智能体做不了吗我在实际项目里见过不少为了炫技而强行套多智能体架构的情况结果推理延迟翻倍、token 成本飙升、系统稳定性下降最终又退回单智能体方案。所以判断标准应该落在任务本身的复杂度上。以这个案例中可能涉及的典型场景来说比如“自动生成一份行业分析报告”单智能体的处理方式是一次请求把资料塞进上下文让模型一口气输出报告。问题来了——如果资料超过上下文窗口数据会被截断如果报告结构复杂模型容易在长文生成中遗忘早期结论如果你希望先调研、再分析、最后排版单智能体很难在同一个会话里保持三种角色状态的切换清晰。多智能体的处理方式则是拆解调研智能体负责收集和筛选资料分析智能体基于调研结果做数据解读和趋势判断写作智能体把分析结论转化为结构化报告审校智能体再对报告做事实核查和格式修正。每个智能体只做自己擅长的一步输入输出都是明确的结构化数据这样既控制了上下文长度又让每个环节的专业度更高。我在这个案例里最终确认的方案就是这种“流水线 分角色”的混合架构不是所有智能体同时并行而是按流程分段推进在关键节点设置汇合点由协调者统一调度。2.2 三种主流协作模式的取舍多智能体的协作方式大致有三种选型时直接决定系统的复杂度和灵活性。第一种是中心化编排模式一个中央调度智能体掌握全部任务清单按顺序或按依赖关系把子任务分发给各个工作智能体再回收结果。这种模式最直观也最容易调试因为所有流程都围绕调度者展开问题定位时只要看调度者的决策日志就行。缺点是调度者容易成为瓶颈如果子任务过多调度者的上下文和决策压力会非常大。第二种是去中心化自组织模式各智能体之间直接通信通过协商或投票决定任务归属。这种模式适合开放性问题比如多个智能体共同 brainstorm 一个方案但缺点是收敛性差智能体之间可能出现互相等待、重复劳动甚至冲突工程上极难控制。第三种是分层混合模式也是我在这类实战案例中最推荐的一种。顶层有一个管理者智能体下面按业务域拆成若干子团队每个子团队内部有一个小队长小队长再管理具体执行智能体。这种模式既保留了中心化编排的可控性又通过层级分解减轻了单一调度者的压力扩展性也更好。这个案例我选的就是分层混合模式因为案例的目标是演示一个完整的、可运行的协作系统而不是演示一个研究原型。分层结构能让读者清晰地看到任务是怎么逐级拆解的也方便后续替换或扩充子团队。2.3 智能体之间的通信协议设计这是多智能体系统里最容易被忽略但又最关键的部分。很多新手会让智能体直接输出自然语言给下一个智能体比如调研智能体输出“我觉得这些资料挺有用的你看看吧”下一个智能体根本没法稳定解析。必须设计一套明确的通信协议让每个智能体的输出都是结构化、可校验的。我在案例中定义了一个统一的消息格式核心字段包括任务 ID、发送者、接收者、消息类型如任务分配、结果返回、错误报告、状态更新、数据载荷、时间戳。数据载荷部分尽量使用 JSON字段固定不允许智能体自由发挥。比如调研智能体返回的数据载荷必须包含“sources”数组每个 source 对象里必须有“title”“url”“summary”三个字段缺一不可。这套协议的作用相当于给智能体之间建立了一套“契约”。智能体不需要理解对方的长篇大论只需要按照契约解析 JSON、提取字段、执行下一步。这也意味着每个智能体的提示词里必须写清楚“你输出的内容必须符合以下 JSON Schema”而且要给出正反两个示例让模型学会输出格式正确的数据。2.4 工具注册与环境隔离多智能体系统里每个智能体通常需要调用外部工具比如搜索、计算、读写文件、调用数据库等。工具不能直接写死在代码里而是要通过“工具注册表”统一管理。每个工具声明自己的名称、描述、输入参数格式和输出格式智能体通过名字调用系统在运行时做参数校验和权限校验。我还特别强调环境隔离。多个智能体如果共享同一组环境变量或同一个临时目录很容易出现互相覆盖文件、API 密钥混乱等问题。这个案例里我给每个智能体分配了独立的临时工作目录所有读写操作限制在自己目录内只有在协调者明确指令下才允许跨目录访问。这样既能防止数据污染又能在出错时快速定位是哪个智能体的操作导致的。3. 核心细节与实操要点从零搭起这套系统3.1 系统整体模块划分动手写代码之前先把模块切清楚。这个案例的系统架构我拆成了六个核心模块入口路由模块、智能体注册中心、任务调度模块、消息总线、工具服务模块、日志与监控模块。入口路由模块负责接收外部请求把用户的需求转化成标准化的任务描述然后交给任务调度模块。智能体注册中心维护所有可用智能体的清单包含每个智能体的能力描述、当前状态、关联的模型配置。任务调度模块根据任务的复杂度和智能体的忙闲状态决定任务分配给谁、是否需要拆解子任务、子任务之间的依赖顺序。消息总线承担所有智能体之间的消息传递可以在进程内用内存队列实现也可以引入 Redis Stream取决于部署规模。工具服务模块把搜索、文件读写、数据计算等能力封装成可被智能体调用的工具接口。日志与监控模块记录每个智能体的输入输出、耗时、token 消耗和错误信息debug 时全靠它。3.2 环境准备与依赖清单这个案例的代码我建议使用 Python 3.10 以上版本依赖核心库包括openai1.35.0 pydantic2.6.0 redis5.0.0 fastapi0.110.0 uvicorn0.29.0 python-dotenv1.0.0需要说明的是这里的 openai 库指的是兼容 OpenAI API 协议的客户端你完全可以通过设置 base_url 来对接任意兼容的大模型服务。我在实际案例中用的是硅基流动的 API因为国内访问更稳定而且支持多种开源模型切换。这里不涉及任何网络代理问题纯粹是 API 对接。模型选择上我建议协调者使用更强更稳的模型比如推理能力靠前的旗舰模型负责任务拆解和决策执行智能体可以使用性价比更高的模型因为它们的任务相对固定输出结构明确不需要太强的临场推理能力。这样组合下来整体成本能比全用旗舰模型降低一半以上。3.3 智能体基类与个性化扩展为了让代码不重复我抽象了一个 Agent 基类统一处理消息接收、工具调用、上下文管理、结果输出。基类的核心方法如下class BaseAgent: def __init__(self, name: str, role_prompt: str, model: str, tools: list[dict]): self.name name self.role_prompt role_prompt self.model model self.tools tools self.memory [] async def handle_message(self, message: dict) - dict: # 1. 将消息转为对应的上下文 # 2. 调用 LLM 并携带工具定义 # 3. 解析模型输出为结构化结果 # 4. 返回符合协议的消息 pass def add_to_memory(self, content: str): # 每个智能体维护自己的短期记忆 pass具体的业务智能体只需要继承 BaseAgent然后传入对应的角色提示词和工具列表。比如调研智能体的角色提示词会强调“你是一个专业的资料调研员你的任务是根据给定主题搜索高质量信息源并输出结构化摘要列表”同时挂上搜索工具。这里有一个实操细节不要把所有提示词都写在一个超长字符串里建议拆成“系统角色描述”“任务指令模板”“输出格式规范”“负面约束”四段拼接时动态插入具体任务内容。这样每个智能体的提示词结构清晰后续修改也方便。3.4 任务调度与状态机设计任务调度是整个系统的中枢我直接用状态机来管理每个任务的生命周期。任务状态包括pending、running、waiting、completed、failed。pending 表示任务已创建但未开始running 表示有智能体正在执行waiting 表示子任务已完成但需要等待其他依赖任务completed 和 failed 一目了然。协调者负责生成任务树。比如初始任务“撰写市场分析报告”协调者会拆出三个子任务“收集行业数据”“分析竞品动态”“输出报告初稿”。这三个子任务之间有依赖关系前两个可以并行第三个必须等前两个完成。调度模块用一张依赖表记录这些关系每个子任务完成时调度模块检查它的下游任务是否所有上游都已完成若是则把下游任务从 waiting 置为 running。状态机的价值在于系统可以随时接受外部查询“当前任务进展到哪一步了”也可以在中途失败时精准重跑失败的分支而不是整个流程推倒重来。这个案例里我通过一个简单的任务表加轮询方式实现数据量不大时完全够用。3.5 记忆与上下文管理技巧多智能体系统里每个智能体都有自己的上下文窗口但单个智能体不需要也没有必要记住全部信息只需要记住自己任务范围内的信息。我在基类里实现了“滚动窗口记忆”每个智能体维护一个按时间排序的消息列表当总 token 数超过阈值时把最旧的交互压缩成一段摘要只保留关键结论丢弃详细过程。这个机制很关键。比如调研智能体在与搜索工具交互过程中可能会产生大量中间结果它不需要把全部搜索结果原文保留到最终报告阶段只需要保留每条信息的标题、来源和摘要。因此调研智能体在完成每次搜索后会把原始文本交给一个压缩步骤用一次轻量的 LLM 调用生成结构化摘要再存入记忆。这样既保留了信息又控制了 token 成本。另一个技巧是共享记忆与私有记忆分离。协调者有一个全局记忆记录任务目标和当前进度各执行智能体只有私有记忆。执行智能体完成任务后只把结构化结果提交给协调者而不是把私有记忆全部同步过去。避免信息过载的同时也保护了各智能体的上下文纯度。4. 实操过程与核心环节实现4.1 第一步定义通信协议与数据模型我先把通信协议用 Pydantic 定义成强类型模型这样在解析智能体输出时能立刻校验格式错误而不需要等下游智能体用的时候才发现问题。核心模型如下from pydantic import BaseModel, Field from typing import Optional, Any from datetime import datetime class AgentMessage(BaseModel): task_id: str sender: str receiver: str msg_type: str Field(pattern^(task_assign|task_result|task_error|status_update)$) payload: dict {} timestamp: datetime Field(default_factorydatetime.utcnow)每个智能体在返回结果时由外层代码负责构造 AgentMessage。模型本身不直接输出这个 JSON 结构因为让模型以纯文本形式精确输出 JSON 再解析稳定性远不如让模型输出字段值、由代码组装 JSON。经验是模型负责“思考”和“提炼内容”代码负责“包装格式”两者分工成功率会高很多。比如调研智能体模型只输出一个包含若干条信息的 Markdown 列表外层代码用正则和字符串解析把每条信息拆成 title、url、summary 三个字段再组装成 AgentMessage。这个过程比让模型直接输出 JSON 再二次 JSON.parse 要稳得多。原因是模型生成大段文本时对 JSON 格式的遵守度会随着内容长度增加而下降但 Markdown 列表格式的错误率相对低。4.2 第二步实现协调者智能体协调者是整个系统里唯一可以直接对外交互的智能体它接收初始目标拆解子任务分发给执行智能体然后汇总结果。它的角色提示词需要强调“你是项目经理不负责具体执行只负责拆解、调度和最终整合”。协调者的核心逻辑是一个循环只要任务树里还有未完成的状态它就检查下一个需要触发的子任务调用对应的执行智能体获取返回结果并更新任务树。由于主流程是串行的协调者的实现不复杂但我在里面加了重试机制——执行智能体返回失败时协调者会先记录错误然后把该子任务重新打包附加错误信息发给同一个智能体重试一次。如果重试仍失败才标记任务失败。这里的细节是重试时的上下文处理需要把前一次错误结果和出错原因一起传给智能体让它知道之前错在哪里。否则直接重跑大概率还是同样的错误。协调者会在重试消息中加入“你上一次的输出不符合要求具体错误是缺少 sources 字段。请重新生成并确保输出包含完整字段。”这样一个修正指示。4.3 第三步实现调研智能体调研智能体挂了一个搜索工具我封装了一个通过 API 获取搜索结果的函数。考虑到稳定性我没有直接用普通网页爬虫而是使用了一个搜索 API 服务输入关键词返回前若干条网页结果每条结果包含标题、链接、摘要。这样省去了抓取网页再清洗的麻烦。调研智能体的执行流程大概是接收任务指令 → 从指令中提取核心搜索词 → 调用搜索工具获取结果 → 对每条结果做简单去重和相关性打分 → 从高相关性的结果里提取正文关键段落 → 形成结构化摘要列表 → 返回结果。这里有一个值得注意的点搜索词提取不要直接拿整个任务描述去搜而是先让模型提炼出 3-5 个检索词再搜索。比如任务描述是“分析 2024 年新能源车市场格局”直接搜索会返回大量无关新闻但提炼成“新能源车 市场规模 2024”“新能源 车企 销量排名 2024”这样的精确检索词结果质量会高很多。相关性打分我也会交给模型来做但不会让模型输出分数而是让模型对每条结果打标签“高相关/中相关/低相关”只保留高相关和中相关的结果。这些过滤逻辑放在代码里不占模型上下文。4.4 第四步实现分析与写作智能体分析智能体接收调研智能体输出的结构化摘要然后结合自己的领域知识做趋势判断。它的输出不要求太长的篇幅关键是逻辑链完整从哪些数据能得出哪些结论。我要求它在输出中必须带有“依据”字段引用调研结果里的 source 标识这样后续审校智能体能做事实核对。写作智能体负责把分析结论转化为一篇完整文章。它的输入是分析结果的结构化列表输出是一篇带标题、小标题、段落、结论的文章草稿。为了让文章质量更高我给写作智能体增加了一段“受众感知”指示先让模型用一句话描述目标读者是谁再按这个定位调整文风。这个步骤看似多余但实测对最终文章的可读性提升非常明显。这里解释一下为什么要拆成分析和写作两步而不是让一个智能体同时做。分析与写作是两种完全不同的认知任务分析需要批判性思维和归纳推理写作需要表达能力和结构组织。混在一起时模型容易偏向某一方面分析深度不够或者文章结构混乱。拆开后每个智能体的任务边界清晰提示词可以完全围绕单一目标设计效果自然更好。4.5 第五步实现审校智能体与流程闭环审校智能体是我认为最容易被人忽略但价值极高的环节。它在写作智能体完成后介入检查三件事事实性错误、逻辑一致性、格式规范性。事实性检查方面审校智能体把文章中的关键数字和结论与调研摘要做比对如果有出入就标记出来。逻辑一致性方面检查文章前后结论是否矛盾比如前文说“市场增长放缓”后文又写“高速增长”这种问题肉眼不容易发现但模型能捕捉到。格式规范方面检查是否缺少必要段落或标题层级是否混乱。审校通过后协调者收到最终版本整个流程结束。如果审校发现问题协调者会把审校意见连同原文章一起发回写作智能体要求它根据意见修改。这个循环允许最多两轮超过两轮仍不合格则人工介入。实测下来大多数情况下一轮修改就能通过只有资料本身相互矛盾的场景需要第二轮。5. 常见问题与排查技巧实录5.1 问题一智能体输出 JSON 格式不稳定这是我在多智能体系统里遇到的最高频问题。模型在短输出时 JSON 格式基本没问题但只要输出内容变长很容易出现遗漏逗号、字符串未转义、输出前后夹杂解释文字等情况。解决思路我刚才提过不要依赖模型直接输出完整 JSON而是让模型输出结构化 Markdown 或纯文本列表由代码负责 JSON 组装。如果确实需要模型输出 JSON必须在提示词里给出一个最小示例并明确要求“只输出 JSON不要输出任何其他文字”。同时在外层代码加一个兜底解析函数尝试 json.loads 前先剥离代码块标记和首尾多余文字。实测这样能把解析成功率从 70% 提升到 95% 以上。5.2 问题二多个智能体并行时资源竞争如果多个调研智能体同时发起搜索请求很容易触发第三方搜索 API 的限流。我的处理方式是给工具调用加信号量限制并发数同时在每个工具函数里做指数退避重试——第一次失败等 1 秒第二次等 2 秒第三次等 4 秒。这样既保证了并行度又不会把上游服务打挂。资源竞争的另一个表现是 API Key 并发配额不足。解决方法是准备多个 API Key按智能体维度分配不同的 Key并在注册中心里记录每个 Key 的用量。当然如果用的是统一网关在网关层做限流和配额管理更规范案例里为了简单就直接在代码层处理了。5.3 问题三任务陷入循环或长时间不结束最典型的场景是审校智能体反复不通过写作智能体反复修改形成死循环。我的对策是给每个任务设置最大重试轮数达到上限后强制结束返回“需要人工介入”的结果。还有一种场景是任务依赖关系配置错误导致某个子任务永远在 waiting 状态。排查方法是看依赖表里是否有环启动调度时先做一次拓扑排序检测检测到环就直接报错。日志是定位这类问题的唯一有效手段。我在每个智能体的入口和出口各打一条结构化日志包含 task_id、agent_name、status、耗时、token 数。出问题时先查任务状态机流转记录找到停在哪个状态再看对应智能体的输入输出日志基本能定位到原因。5.4 问题四模型上下文污染当执行智能体重试次数多了以后它的记忆里积累了太多历史信息这些信息可能包含前几次错误尝试的细节干扰最终输出。我在记忆机制里做了处理每次重试时不把上一次的完整输出加入记忆只加入错误摘要和修正指令这样智能体不会被大量无关内容干扰。另外不同子任务之间如果共用一个智能体实例子任务的记忆可能残留到下一个任务。解决方法是每完成一个任务就清理该智能体的短期记忆只保留一些与角色相关的系统提示词。这一步很多人会忽略直到发现第二个任务的结果里出现了第一个任务的内容才意识到问题。5.5 实操心得如何做性能与成本优化整个系统跑通后性能优化空间主要集中在两个方向减少多余的 LLM 调用和降低单次调用的 token 量。减少多余调用靠的是合理的任务合并。比如多个子任务都需要调用同一份数据不要让每个智能体各查一遍而是在协调者层面做一次数据获取然后通过消息总线广播给需要的智能体。降低 token 量方面核心手段是把长文本压缩成摘要。我在调研、分析、写作这个链条里都插入了摘要步骤摘要长度控制在原文的 10%-20%。别小看这一步实际算下来一个完整流程能节省 40% 以上的 token 消耗代价只是多两次轻量模型调用成本几乎可以忽略。成本控制还有一条经验优先用便宜模型做“体力活”用贵模型做“脑力活”。调研、摘要、格式转换这些任务我用的是轻量级模型任务拆解、最终审校这种需要深度理解的环节我才用旗舰模型。整体成本实测能比全旗舰模型方案降 50% 以上而输出质量几乎没有差别。6. 多智能体协作系统的扩展方向这个案例只是把最基础的协作流程跑通了但多智能体系统的可能性远不止于此。我在做完这个案例后马上在思考几个扩展方向。一个是引入反思与自我修正机制。现在每个智能体完成任务后直接返回结果如果它能对自身输出做一次快速反思比如检查是否遗漏关键要素、是否有明显逻辑漏洞再自我修正一轮整体质量会再上一个台阶。这个机制不需要额外增加新智能体只要在执行流程里加一个“反思-修正”循环即可。另一个是引入人类反馈回路。对于高风险任务比如医疗建议或金融建议系统不能全自动跑完就对外输出应该在关键节点设置人工审批环节。协调者生成任务树时可以把某些子任务标记为“需人工确认”调度模块遇到这类任务就暂停并等待人工审核结果。这个改动在架构上只需增加一个审核状态核心代码不需要大改。再有一个方向是智能体动态扩展。目前系统中智能体的数量和角色是写死的如果希望系统能根据任务类型自动创建新智能体可以在协调者之上再加一个“智能体工厂”模块。协调者发现现有智能体能力不足时触发工厂模块根据任务需求动态生成新的智能体实例并把对应工具和提示词装配好。这就是多智能体系统从“固定团队”走向“动态组织”的关键一步。最后说说选型层面的建议。如果你准备在自己的项目里引入多智能体协作系统先别急着搭框架而是拿一个真实业务场景跑通最小闭环。用最简单的方式比如两三个智能体配合跑通后再逐渐加模块。多智能体系统的复杂度是累积的每一步都要有明确收益才值得引入。没有明确任务边界、没有稳定工具依赖的场景用单智能体反而更省心。我在实现这个案例的过程中最深的体会是多智能体系统真正考验的不是某个单一模型的聪明程度而是系统架构对不确定性的容纳能力。模型会出错、工具会超时、任务会互相依赖这套系统的价值不在于消灭这些不确定性而在于把出错的影响限制在局部并通过重试、校验、隔离等机制保证整体流程的可靠推进。把这一层想明白了后面遇到任何具体实现问题其实都有清晰的解决路径。