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 以上版本,依赖核心库包括:
openai>=1.35.0 pydantic>=2.6.0 redis>=5.0.0 fastapi>=0.110.0 uvicorn>=0.29.0 python-dotenv>=1.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_factory=datetime.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. 多智能体协作系统的扩展方向
这个案例只是把最基础的协作流程跑通了,但多智能体系统的可能性远不止于此。我在做完这个案例后,马上在思考几个扩展方向。
一个是引入反思与自我修正机制。现在每个智能体完成任务后直接返回结果,如果它能对自身输出做一次快速反思,比如检查是否遗漏关键要素、是否有明显逻辑漏洞,再自我修正一轮,整体质量会再上一个台阶。这个机制不需要额外增加新智能体,只要在执行流程里加一个“反思-修正”循环即可。
另一个是引入人类反馈回路。对于高风险任务,比如医疗建议或金融建议,系统不能全自动跑完就对外输出,应该在关键节点设置人工审批环节。协调者生成任务树时,可以把某些子任务标记为“需人工确认”,调度模块遇到这类任务就暂停并等待人工审核结果。这个改动在架构上只需增加一个审核状态,核心代码不需要大改。
再有一个方向是智能体动态扩展。目前系统中智能体的数量和角色是写死的,如果希望系统能根据任务类型自动创建新智能体,可以在协调者之上再加一个“智能体工厂”模块。协调者发现现有智能体能力不足时,触发工厂模块根据任务需求动态生成新的智能体实例,并把对应工具和提示词装配好。这就是多智能体系统从“固定团队”走向“动态组织”的关键一步。
最后说说选型层面的建议。如果你准备在自己的项目里引入多智能体协作系统,先别急着搭框架,而是拿一个真实业务场景跑通最小闭环。用最简单的方式,比如两三个智能体配合,跑通后再逐渐加模块。多智能体系统的复杂度是累积的,每一步都要有明确收益才值得引入。没有明确任务边界、没有稳定工具依赖的场景,用单智能体反而更省心。
我在实现这个案例的过程中最深的体会是,多智能体系统真正考验的不是某个单一模型的聪明程度,而是系统架构对不确定性的容纳能力。模型会出错、工具会超时、任务会互相依赖,这套系统的价值不在于消灭这些不确定性,而在于把出错的影响限制在局部,并通过重试、校验、隔离等机制保证整体流程的可靠推进。把这一层想明白了,后面遇到任何具体实现问题,其实都有清晰的解决路径。