多智能体协作系统实战：从概念拆解到工程落地

去年我在做自动化流程改造时碰到一个特别现实的问题单独一个 AI 模型再聪明也只能按提示词死板地执行单线程任务。真正要做的内容调研、报告生成、代码审查这类活儿往往需要好几道工序来回配合。后来我把目光落在一个叫 agency-agents 的项目设计思路上——它不是某个特定产品的名字而是一套把多个 AI 智能体组合成“数字化小团队”的工程模式。简单说就是让每个智能体扮演一个业务角色比如情报收集、方案起草、质量检查然后通过统一的消息通道协作最终输出一份完整结果。我现在写这篇文章就是想把这段实践完整沉淀下来从概念拆解、架构选型到一个能跑起来的最小系统再到真实场景里的踩坑与排查。无论你是刚接触智能体开发的新手还是已经写过几个 Agent Demo、想往工程化方向走的开发者这份笔记应该都能给你一条可以照抄的路线。1. 什么是 agency-agents先搞清楚概念边界1.1 名字拆开看Agency 和 Agents 分别意味着什么第一次看到 agency-agents 这个名字大多数人会愣一下。Agency 在英文里有“代理机构”“中介”的意思Agents 又跟中文“代理”撞车两个词叠在一起特别容易绕晕。我自己的理解是这里的 Agency重点不是“代理”的接口含义而是“组织、机构、能动性”的混合体。它强调的是一群 Agent 出于共同目标聚在一起各自承担职责像一个微型组织那样运转。Agents 则指那些具备自主决策能力的智能体它们内部通常有一个大语言模型作为推理内核负责理解目标、拆解动作、判断要不要调用工具、生成最终内容。所以 agency-agents 合在一起最精准的翻译应该是“机构化智能体群体”或者“多智能体协作组织”。它回答的问题是如果一个人形单影只那我们就拉一队人用工程手段让它们像公司部门一样配合。1.2 它到底解决了什么问题单智能体和多智能体之间差的不是数量而是能力边界。第一个问题是任务延续性差。单次对话里模型生成完一段内容之后就不会主动往下推进了。它没有办法“过一会儿再回来检查结果”更不可能在发现素材不足时自行去补一轮搜索。多智能体结构可以做到规划者派活、执行者干活、检查者验收不合格就退回重做直到通过。第二个问题是复杂度超过单提示词的承载能力。你把十步流程写进一个提示词里模型大概率会在中间某一步开始丢三落四。但如果你把这些步骤分配给三个不同角色每个角色只关心自己那一段出错的概率就小得多。这跟软件工程里“模块职责单一”是一个道理。第三个问题是结果无人把关。单一模型直接生成报告经常出现数据过时、引用错误、逻辑跳跃。多智能体结构里专门安排一个质检角色等于给流程加了一道人工评审的替代品不依赖用户自己去发现所有问题。你甚至可以把它想象成开一家小型内容公司一个人既采访又写作还校对效率低质量差但如果分成记者、编辑、校对三个人每个人都聚焦在自己的专业环节上整体质量会立刻上一个台阶。1.3 它适合用在哪些场景里我现在判断一个需求要不要上多智能体就看三个特征是否有明确分工、是否有反复迭代、是否有独立验收动作。比较典型的有四类场景内容调研与报告生成先收集资料再整理结构然后撰写全文最后核对事实。客户工单处理先理解用户诉求再查知识库答复前做风险判断必要时转人工。代码修复流程先读异常栈再定位代码位置给出修改建议然后执行测试验证。数据分析流水线先写清洗脚本再跑统计分析生成图表说明最后输出业务结论。这些场景有一个共同点只靠一次模型调用远远不够必须靠多条消息链路把任务层层推进最终得到一个经过校验的产出物。2. 设计一个能跑起来的 agency-agents 框架核心模块选型2.1 先决定编排方式中心化、去中心化还是混合模式开始写代码之前我建议你先定拓扑结构。这不是理论洁癖而是直接影响后续消息怎么转发、出问题怎么排查。中心化编排是最容易上手的。系统里有一个主控 Agent它负责任务分配、状态跟踪、结果回收其他 Agent 收到指令后执行跑完就把结果交回主控。优点是流程透明好调试缺点是主控会成为瓶颈所有消息都要经过它中转。去中心化编排则是每个 Agent 都直接给别的 Agent 发消息靠消息总线或者共享黑板来协作。它的扩展性好适合多角色自由讨论的场景但调试难度高很容易出现消息环回和状态不一致。混合模式一般是这样常态下主控负责调度但遇到特定分支时允许子 Agent 之间临时直接交互比如研究员和数据分析师私聊获取某个指标然后把信息带回主流程。我个人的建议是第一个版本别去做去中心化先老老实实用中心化等你对协作模式有把握了再放开其他路线。2.2 把角色定义成“部门”而不是“工种”在设计角色时我的一个经验是不要只考虑“这个 Agent 做什么动作”要考虑“它代表哪个部门来做事”。规划者岗位的性质本质上是项目经理。它不负责具体的信息检索也不负责最终撰稿只负责把一个大目标拆成可执行小任务并判断哪些任务可以并行。执行者更像是业务骨干。例如内容策划、代码实现、数据分析、资料整理。这类 Agent 是实际产生产出的部分需要至少掌握一个外部工具比如搜索接口、代码执行环境或者数据库查询。检查者负责质量审核。它的提示词应该明确列出验收标准告诉它什么情况可以放行什么情况需要退回修改。它的反馈不能只说“不行”必须指出具体问题并给出修正建议。汇总者类似于编辑。它要把各执行者交付的内容拼成最终成品处理格式、去重、润色形成一个完整一致的输出。把角色设计成部门还有一个额外的好处升级时你只需要替换某个部门的能力不需要推翻整个流程。比如想换执行策略改一个 Agent 内部提示词就行其他部分完全不受影响。2.3 消息协议让智能体之间说同一种语言多智能体最忌讳的事情就是各说各话。一个 Agent 回复纯文本另一个回复 JSON主控还得临时解析各种稀奇古怪的格式。所以第一件事就是定义一套统一的消息协议。我常用的消息结构长这样{ id: msg_001, type: task, from: planner, to: researcher, status: pending, payload: { task: 收集过去三个月智能体开发框架的更新动态, deadline: 2025-03-01, context_ids: [req_01] } }字段一定要固定。id 是唯一标识用于追踪type 决定消息类别是普通任务、审核反馈还是完成通知from 和 to 决定路由方向status 表示消息当前状态payload 里放实际内容。用 Python 定义的话可以写成 dataclassfrom dataclasses import dataclass, field from typing import Any, Dict dataclass class Message: msg_id: str msg_type: str sender: str receiver: str status: str pending payload: Dict[str, Any] field(default_factorydict)统一协议之后主控只管像邮局一样把消息从 A 路由到 B不需要理解消息内部的具体业务含义。所有 Agent 也都用同一套对象处理入站和出站消息逻辑变得非常干净。2.4 工具和权限边界不能把所有钥匙都交给智能体设计多智能体系统时最容易忽略但又最容易出事的地方就是工具权限。每个执行型 Agent 都会绑定若干工具例如搜索引擎接口、内部文档库、数据库连接器、代码执行沙箱。但你要清楚模型本身不知道哪些操作是安全的它只是根据上下文推断“该用什么工具”。我的建议是三条铁律只给最小必要权限。查询类 Agent 只给只读接口写操作必须单独分配给特定角色不能人人都有写库能力。高风险操作必须加确认环节。比如删除数据库记录、发送邮件、修改生产配置这类操作应该返回一个待确认状态由主控或者人工审批后再执行。工具结果要做校验。工具返回的内容不能直接当成事实需要由检查者做二次确认。这套权限设计看着像限制了智能体的自由度实际恰恰相反。边界清晰之后智能体才能真正发挥能力因为我们可以放心地把更复杂的任务交给它。3. 手把手实现一个最小可用的 agency-agents 系统3.1 环境和项目结构这一节我们直接动手。我假设你本地已经有了 Python 3.10 以上的环境并且可以在命令行里安装依赖。除了标准库之外我们还需要一个能调用大模型接口的客户端你可以根据自己实际情况选型。项目目录我建议这样安排agency_agents_demo/ ├── main.py ├── agents/ │ ├── base.py │ ├── planner.py │ └── researcher.py ├── core/ │ ├── message.py │ └── scheduler.py └── tools/ └── search_stub.py把代码拆成 agents、core、tools 三个目录长期维护起来会很舒服。agents 目录放角色逻辑core 目录放调度和消息tools 目录放外部工具接口。3.2 先写消息对象和引擎基底core/message.py 里就放前面提到的 Message 类。然后我们再补一个 MessageQueue用来做简单流转from collections import deque class MessageQueue: def __init__(self): self.queue deque() self.waiting_map {} def publish(self, msg: Message): self.queue.append(msg) def consume(self) - Message | None: if self.queue: return self.queue.popleft() return None def register_reply(self, msg: Message): self.waiting_map[msg.msg_id] msg注意到这里我还留了一个 waiting_map用途是记录某个请求有没有返回结果方便排查丢失消息。3.3 定义智能体基类agents/base.py 里的职责很简单接收消息、调用模型、执行工具、返回消息。from abc import ABC, abstractmethod from typing import Dict from core.message import Message class BaseAgent(ABC): def __init__(self, name: str, system_prompt: str): self.name name self.system_prompt system_prompt def build_prompt(self, task: str) - str: return f{self.system_prompt}\n\n任务{task} abstractmethod def handle(self, msg: Message, context: Dict) - Message: pass这里有一点需要解释我没有把模型调用直接写死在基类里。不同角色可能希望用不同参数比如规划者用温度低一点的配置保证稳定创作者用适度随机的配置保证多样性。所以模型调用放到子类里实现更合适。3.4 实现规划者智能体和执行者智能体规划者的核心工作是把一个用户需求拆成多个子任务然后分配出去。在这个最小示例里为了让事情可复现我先不依赖真正的大模型调用而是用一个简化的规则实现from agents.base import BaseAgent from core.message import Message class PlannerAgent(BaseAgent): def handle(self, msg: Message, context: Dict) - Message: user_request msg.payload.get(task, ) subtasks [ {type: research, task: f收集关于「{user_request}」的基础资料}, {type: draft, task: f基于资料撰写初稿}, ] return Message( msg_idplan_1, msg_typetask_batch, senderself.name, receiverscheduler, statusok, payload{subtasks: subtasks} )执行者我们写一个 Researcher负责搜索资料。这里工具调用我用的是一个 search_stub只返回固定文本方便演示class ResearcherAgent(BaseAgent): def handle(self, msg: Message, context: Dict) - Message: task msg.payload.get(task, ) result search_stub.query(task) return Message( msg_idres_1, msg_typeresult, senderself.name, receiverplanner, statusok, payload{research: result} )实际项目中research 部分应该换成真实的搜索 API。但你注意这个骨架只需要改工具实现不需要改 Agent 的协作逻辑这就体现出分层的好处。3.5 调度器把消息在智能体之间转运调度器是这个小系统的“邮局”。它从队列里取消息根据消息里的 receiver 字段找到对应 Agent把消息交给它处理然后把返回结果再入队。一个简化版本是这样的class Scheduler: def __init__(self, agents: Dict[str, BaseAgent]): self.agents agents self.queue MessageQueue() def run(self, root_msg: Message, max_rounds: int 10) - str: self.queue.publish(root_msg) final_result for _ in range(max_rounds): msg self.queue.consume() if msg is None: break agent self.agents.get(msg.receiver) if agent is None: continue response agent.handle(msg, context{}) if response.receiver scheduler: # 如果是需要继续分配的子任务就再入队 for sub in response.payload.get(subtasks, []): self.queue.publish(Message( msg_idsub[task], msg_typetask, senderplanner, receiverresearcher, payloadsub )) else: final_result response.payload return final_result注意我这里用的是循环上限 max_rounds这个非常关键。真实系统里防死循环要靠这个上限兜底否则一旦 Agent 之间来回转发程序就永远不会停下来。3.6 跑起来看效果组合主程序from agents.planner import PlannerAgent from agents.researcher import ResearcherAgent from core.scheduler import Scheduler agents { planner: PlannerAgent(nameplanner, system_prompt你是一个项目规划者。), researcher: ResearcherAgent(nameresearcher, system_prompt你是一个资料研究员。), } scheduler Scheduler(agents) root Message( msg_idroot_1, msg_typeuser_request, senderuser, receiverplanner, payload{task: 新一代智能体框架的市场动态} ) result scheduler.run(root, max_rounds10) print(result)运行后你会看到任务从 user 到 planner被拆成两个子任务然后每个子任务又转给 researcher最后由 researcher 返回搜索结果。虽然这个结果目前还很简陋但完整链路的雏形已经出来了后续所有高级功能都是在这个骨架上长出来的。4. 真实场景用三个智能体完成一次内容调研4.1 把需求拆成可执行任务理论说多了容易飘我们把前面这套骨架压到一个真实用例里做一个“智能体框架选型”的调研报告目标读者是技术负责人。原始需求拆出来是这样收集智能体框架的社区活跃度包括 GitHub Star 增长、Discord 讨论量。对比各框架的编排能力是否支持多智能体协作。输出一份选型建议包含优缺点和适用场景。这三个点互相之间有依赖但又有明显边界。社区数据是一个方向能力对比是另一个方向最后选型建议要结合前面两个结果。正好对应执行者、分析者、汇总者三种角色。4.2 角色和工具映射表我给出一个工具与角色的映射你照着搭即可角色需要完成的职责需要的工具数量配置规划者拆解任务、设置优先级、调派资源无外部工具只靠模型推理1调研员采集社区数据、整理原始事实搜索接口、GitHub API、社区归档2分析员对比框架能力、识别优劣文档库检索、表格处理1审查员核对数据、检查逻辑、退回修改文件读取、元数据校验1汇总者合并结果、润色语言、输出终稿文档模板、格式化工具1注意我在这里给调研员配了 2 个实例。并不是所有 Agent 都只能有一个有些角色因为工作量大是可以多实例并行的。这也是多智能体系统比单模型有优势的地方。4.3 任务流转过程真实的流转过程比前面 demo 要丰富很多。我用文字描述一遍完整链路第一步用户把需求提交给规划者。规划者判断这个问题可以拆成三条并行分支于是发出三条 task 消息分别给两个调研员和一个分析员。第二步调研员收到消息后各自去查数据。一个查社区增长一个查框架文档然后分别把结果发回规划者。规划者收集齐后再发给分析员做能力对比。第三步分析员把对比结果整理成带结论的结构化数据发给审查员。审查员检查发现某个框架的数据缺失于是回了一条 feedback 消息给分析员要求补数据。第四步分析员重新调用工具补齐再次发给审查员这次通过。审查员把通过结果发给汇总者。第五步汇总者把调研结果、分析结论、审查意见合并形成最终报告。这个流程里的每个步骤都留下了消息记录所以哪怕中途出错我们也能从日志里看到卡在哪个环节。4.4 结果组装与验收多智能体系统的最后一步一定不能省略。汇总者输出初稿之后要再过一道程序化检查比如每个论证是否都能追溯到原始调研数据。是否包含至少一个“不推荐某框架”的明确理由。字数是否在目标范围内。是否带有可执行的下一步建议。我建议把这类验收判断也交给一个专门做质检的 Agent或者写成一个规则检查器。不要把验收动作交给汇总者自己否则等于让运动员自己做裁判系统很容易陷入“我写得很好”的自我感觉良好里。5. 常见问题与排查技巧实录5.1 智能体陷入死循环消息来回转这是我见过最多的问题。两个智能体你一句我一句消息永远停不下来。原因多数是A 给 B 发了一条消息B 不知道该不该把结果回传给 A于是又回了一条“请确认”A 又把它当成新任务继续处理。排查方法有三个重点。第一设置最大轮数像 demo 里那样用一个 max_rounds 参数兜底。第二在消息结构里带上“链路追踪号”从根消息继承下来这样能快速圈定一条循环链路。第三在 Agent 提示词里明确写清楚“除非收到明确的修正指令否则不要主动发起新一轮任务”。这句话看着不起眼实际非常有用。另一个技巧是给消息做哈希去重。每个 Agent 在处理消息之前可以把消息内容摘要比对一遍如果发现最近已经处理过相似内容就直接忽略。5.2 上下文越来越长成本失控多智能体系统跑的时间越长每个 Agent 积累的上下文越大。一个大语言模型每次调用都要重新处理全部上下文费用随 Token 数线性上涨最后可能比人工做还贵。解决思路不是“清理上下文”而是“各人只管各家事”。每个 Agent 只携带自己完成当前任务所需的最小上下文比如调研员只需要它的查询词和最新结果不需要知道整个项目的来龙去脉。规划者需要全局视图但规划者上下文也不会太长因为规划者只关注任务列表不需要读懂每份调研原文。我习惯给每个任务块单独建立“工作记忆”文件任务结束后自动归档Agent 启动时不加载历史只在必要时通过检索拿到相关片段。5.3 工具调用失败任务卡住不动工具调用崩溃几乎是必然发生的。搜索引擎接口超时、API 密钥过期、返回格式不符合预期任何一个出问题都会让 Agent 停止推进。处理办法分三层。第一层是重试发现失败时让 Agent 稍等片刻再调用一次。第二层是降级如果接口超时可以暂时用缓存数据或更简单的模拟数据顶上并在报告里标注“此处数据来自备用源”。第三层是转人工连续失败超过指定次数后把消息状态标记为 escalation由人工介入处理。这里的核心经验是工具调用失败不能只记录日志必须让系统有能力改变后续行为。如果一条路走不通Agent 要能换一条路走完任务而不是单纯崩溃退出。5.4 并发写同一份文件结果互相覆盖我在一次实践里遇到两个调研员同时写同一个结果文件后写的把先写的覆盖了整份报告丢了三分之一。排查的时候发现消息队列正常任务也没有死循环就是并发写文件没做控制。解决办法几种一是给写操作加锁同一时刻只允许一个 Agent 写文件二是用版本号控制每个 Agent 写之前带上自己的版本号提交时发现版本陈旧就要重新拉取合并三是尽量把所有写操作都收敛到汇总者一个角色上其他执行者只返回消息不直接落盘。第三种方式最简单推荐你先用这个。收敛写权限本质上跟前面说权限设计是一致的。5.5 效果评估一个报告的好坏靠什么判定多智能体系统最容易被质疑的一点是结果到底靠谱吗我的做法是建立两层评估。客观层看结构化指标任务完成率、消息丢失率、平均轮数、工具调用成功率。这些指标都该有日志埋点用一次跑批任务就能统计出来。主观层看内容质量每个报告都要人来做简易盲评可以从信息准确性、逻辑连贯性、可操作性三个维度打分。初期效果不稳定很正常关键是分数要能反馈回提示词和流程设计里。我自己的经验是先跑五到十次“黄金测试集”也就是固定几个已知答案的问题用同一套配置跑十遍看结果波动多大。如果波动大到不可接受就说明角色分工或提示词还要继续收敛。6. 从 demo 到可用系统边界与扩展建议6.1 安全与合规底线要先划清楚demo 阶段怎么折腾都行但真要把 agency-agents 用到生产环境第一条就是安全边界。数据脱敏必须提前做。用户输入里可能带手机号、邮箱、内部系统路径这些信息一旦被 Agent 塞进无关工具就会造成泄露。我通常在入口处加一个脱敏模块把所有可识别信息替换成占位符等最终生成结果后再还原。权限最小化不能只在文档里写。实际配置工具时每个 Agent 的 API Key 都要单独创建不能共用一个全局超级权限账号。比如调研员只需要只读接口给它写权限等于埋雷。人工审批环节不能省。涉及对外发布、删除数据、修改生产配置的操作至少要有一个“人工确认”的步骤卡在中间。哪怕这个步骤只是点一下按钮都能在很大程度上挡住误操作。6.2 成本和质量控制多智能体系统看起来炫酷但如果不控制成本改造成本会很惊人。我做过一次估算一个三智能体的调研任务如果每个 Agent 都调用一次大模型合计 Token 数是单模型直接生成的 4 倍左右。也就是说你必须换掉“什么都交给大模型”的思路。省钱的办法有几条简单任务不要上多智能体一个正则匹配能解决的问题别绕弯。能用规则地方不要调用模型比如格式转换、字段校验、去重识别。模型选型分级规划者用更便宜的模型执行者用能力更强的模型生成类任务用最高规格模型。对结果做缓存相同或相似任务不再重复跑全套流程。质量控制上核心是建立“结果版本”思维。每个任务都要有追踪 ID每个结果都要标明由哪个 Agent 产出、基于什么材料。这样一发现问题可以快速回滚到特定版本。6.3 几个值得继续深挖的方向agency-agents 这套模式落地到一定程度后你会发现可扩展的空间还很大。一个方向是意图路由。用户输入进来先由一个轻量分类模型判断该走哪条智能体链路而不是所有需求都进同一套流程。比如“帮我看一下报错日志”和“帮我梳理数据库表结构”走完全不同的协作链条。另一个方向是人工反馈闭环。加入人类在环机制在关键节点上允许人工修改 Agent 的产出然后把修改结果作为反馈数据回灌到提示词优化里。这也是一种低成本的数据飞轮。还有一个方向是可观测性平台。多智能体的消息流转其实很像分布式系统日志、链路追踪、指标监控缺一不可。如果你把这个系统当正式服务来运维建议早点接上结构化的监控面板不然跑半个月之后你根本不知道是哪条消息出问题。我在实际操作中最大的体会是agency-agents 的成功不在于智能体数量多而在于边界清晰。每个智能体知道自己能干什么、不能干什么消息该怎么传、工具该用到什么程度这些设计清晰了哪怕底层模型能力一般整体效果也不会差。反过来如果你把所有复杂度都堆在一起再强的模型也救不了混乱的协作流程。后面我会继续把这个最小框架扩展成一份可以复用在不同业务场景的基础设施如果你也在折腾类似的东西建议从今天这个三智能体骨架开始试跑通一次闭环之后你对整体的理解立刻会上一个台阶。