1. 从单兵作战到团队协同为什么我们需要重新思考智能体架构做过智能体项目的人大概都有这种体会一开始写个单Agent的Demo感觉无所不能接上大模型、挂几个工具函数就能跑通问答、检索、简单任务执行。但一旦业务场景复杂起来比如需要同时处理需求分析、代码生成、测试验证、文档撰写这几件事单Agent就开始捉襟见肘了。你要么把提示词写得无比冗长让一个Agent扮演多个角色结果它哪个角色都演不好要么手动在多个Agent之间来回切换效率低不说上下文还经常丢失。这就是Agency Agents多智能体协作系统要解决的核心问题。简单说它是一套让多个各有所长的智能体像真实开发团队一样分工协作的架构方案。每个Agent有自己的专长领域、工具集和决策逻辑它们通过一套约定好的通信机制互相传递信息、协商任务、校验结果最终完成单个Agent无法胜任的复杂工作流。这套东西适合谁参考如果你正在做以下任何一件事这篇文章都值得你花时间读完正在设计多Agent系统的架构方案手头有多个单Agent但不知道怎么让它们协同想了解Agent之间任务分配和结果校验的工程实现或者单纯对多智能体协作的落地细节感兴趣。我会从架构设计思路讲到具体实现再到踩过的坑和排查技巧尽量把每个关键决策背后的逻辑说清楚。2. 整体架构设计多智能体协作的骨架怎么搭2.1 为什么选择“角色专业化消息总线”的组合方案多智能体系统的架构方案有很多种常见的有层级式一个主Agent调度多个子Agent、黑板模式所有Agent共享一个状态空间、市场式Agent通过竞价获取任务等。我在实际项目中试过几种最后发现角色专业化消息总线的组合在开发团队场景下最稳。角色专业化的意思是每个Agent只负责一个明确的职能域。比如需求分析Agent就专注理解用户意图、拆解任务、输出结构化需求文档代码生成Agent只负责根据需求写代码测试Agent只负责验证代码正确性。这样做的好处是每个Agent的提示词可以写得非常精准工具集也可以按需配置不会出现“什么都懂一点但什么都不精”的情况。消息总线则是Agent之间通信的基础设施。你可以把它理解成一个异步的消息队列每个Agent往总线里发消息也可以从总线订阅自己关心的消息类型。为什么不用直接函数调用因为直接调用会让Agent之间产生强耦合A要调B就得知道B的接口B改了接口A也得跟着改。消息总线把这种耦合解开了Agent只需要关心“我发出什么消息”和“我消费什么消息”不需要知道对方是谁。注意消息总线的引入会带来一个副作用——调试变难了。因为消息是异步流动的出问题时你很难一眼看出是哪个环节卡住了。所以我在实现时加了一个消息日志模块所有经过总线的消息都落盘记录排查问题时直接看日志时间线。2.2 核心模块拆解与职责边界整个系统我拆成了五个核心模块每个模块的职责边界必须划清楚否则后期维护会非常痛苦。Agent注册中心负责管理所有Agent的生命周期。每个Agent启动时向注册中心报到声明自己的角色、能力标签、订阅的消息类型。注册中心维护一张全局的Agent能力表当有新任务需要分配时调度器可以查这张表找到合适的Agent。消息总线是系统的神经中枢。我选用了基于发布-订阅模式的内存队列实现支持消息持久化和重放。消息格式统一为JSON结构包含消息ID、发送者、接收者可以是具体Agent或广播、消息类型、负载内容和时间戳。任务调度器负责把用户请求拆解成子任务然后根据Agent能力表把子任务分配给对应的Agent。调度器不关心任务具体怎么执行它只负责“把合适的事交给合适的人”。共享状态存储用来保存整个协作过程中的中间状态。比如需求分析Agent输出的需求文档、代码生成Agent产出的代码、测试Agent的测试报告都存在这里。每个Agent都可以读取自己需要的状态但写入时只能写自己负责的那部分避免状态冲突。监控与日志模块负责记录整个系统的运行状态包括每个Agent的活跃度、任务完成率、消息吞吐量等指标。这个模块在开发阶段可能觉得可有可无但一旦系统跑起来没有它你就是在盲人摸象。2.3 通信协议的设计取舍Agent之间的通信协议设计是整个架构里最需要仔细斟酌的部分。我试过几种方案这里说说各自的优劣。第一种是自由文本通信Agent之间直接发自然语言消息。优点是实现简单Agent不需要学习额外的协议格式。缺点是解析困难接收方需要理解自然语言的意图容易产生歧义。我早期用这个方案时经常出现代码生成Agent误解了需求分析Agent的意思写出来的东西完全跑偏。第二种是结构化消息所有通信都遵循预定义的JSON Schema。优点是解析确定性强不会产生歧义。缺点是灵活性差新增一种消息类型就要改Schema。后来我采用了折中方案核心通信走结构化消息但允许在负载字段里嵌入自然语言描述作为补充说明。第三种是混合模式这也是我最终采用的方案。消息头是结构化的包含类型、优先级、路由信息等元数据消息体可以是结构化数据也可以是自然语言文本由接收方根据消息类型决定怎么解析。这样既保证了路由和调度的确定性又保留了Agent之间传递复杂语义的灵活性。3. 核心细节解析每个Agent内部是怎么运转的3.1 Agent的认知循环感知、规划、执行、反思每个Agent内部都跑着一个认知循环我把它拆成四个阶段感知、规划、执行、反思。这个循环借鉴了认知科学里的一些概念但在工程实现上做了简化。感知阶段Agent从消息总线和共享状态里获取输入。比如代码生成Agent感知到一条“需求文档已更新”的消息就会去共享状态里拉取最新的需求文档。感知阶段的关键是过滤——Agent不需要处理所有消息只处理与自己角色相关的。规划阶段Agent根据感知到的输入制定行动计划。比如测试Agent拿到代码后会规划先跑单元测试再跑集成测试最后生成测试报告。规划阶段通常需要调用大模型来做推理这也是整个循环里最耗时的部分。执行阶段Agent按照规划调用工具或生成内容。比如代码生成Agent调用代码执行工具来验证生成的代码能否跑通测试Agent调用测试框架来执行测试用例。反思阶段Agent检查执行结果是否符合预期如果不符合就回到规划阶段重新规划。这个反思机制是保证输出质量的关键。我试过去掉反思阶段结果Agent经常输出半成品就往下传导致后面环节全部返工。实操心得反思阶段的提示词设计很关键。不要简单问“结果对不对”而要给出具体的检查清单。比如代码生成Agent的反思提示词里会明确列出“语法是否正确、边界条件是否处理、是否有硬编码”等检查项这样Agent的自我校验才有针对性。3.2 工具集配置给每个Agent配什么武器每个Agent的工具集配置直接决定了它的能力边界。我的原则是“最小必要”——只给Agent配它完成本职工作必需的工具多一个都不给。为什么因为工具越多Agent的选择困难症越严重而且容易越权操作。需求分析Agent的工具集最简单只需要一个文档读写工具和一个用户意图澄清工具。它不需要执行代码也不需要访问外部API。代码生成Agent的工具集包括代码文件读写、代码执行沙箱、依赖管理工具。代码执行沙箱是必须的因为生成的代码必须先跑一遍验证语法和基本逻辑不能直接扔给测试Agent。测试Agent的工具集包括测试框架调用、代码覆盖率分析、日志查询。测试Agent需要能独立运行测试用例并收集结果不能依赖代码生成Agent帮它跑。文档撰写Agent的工具集包括文档模板管理、Markdown渲染、图表生成。这个Agent相对独立主要消费其他Agent的输出然后整理成可读的文档。3.3 上下文管理让Agent记住该记住的多智能体系统里上下文管理是个容易被忽视但极其重要的环节。每个Agent的上下文窗口是有限的不可能把所有历史消息都塞进去。我的做法是分层管理上下文。会话级上下文保存当前任务相关的所有消息和状态这是Agent工作时主要参考的。会话级上下文有大小限制超过阈值就触发摘要压缩把早期消息压缩成摘要保留。角色级上下文保存Agent在长期运行中积累的经验和知识。比如代码生成Agent会记住“这个项目里用户偏好用TypeScript而不是JavaScript”、“这个代码库的命名规范是驼峰式”等偏好信息。角色级上下文跨会话持久化让Agent越用越顺手。全局上下文保存整个系统的配置和约束比如“所有输出必须符合安全规范”、“代码不能引入GPL协议的依赖”等。全局上下文对所有Agent可见确保协作过程中不违反系统级约束。4. 实操过程从零搭建一个多智能体协作系统4.1 环境准备与基础依赖安装动手之前先把环境理清楚。我用的技术栈是Python 3.10作为主语言消息总线用Redis的Pub/Sub加Stream做持久化共享状态存储用SQLite开发阶段或PostgreSQL生产阶段Agent之间的大模型调用统一走一个内部封装的LLM客户端。先装基础依赖pip install redis sqlalchemy pydantic httpx pip install openai # 或其他LLM客户端库Redis的安装就不展开了网上教程很多。装好后启动Redis服务确认Pub/Sub功能正常redis-cli ping # 应该返回 PONG数据库初始化用SQLAlchemy自动建表我定义了三张核心表agent_registryAgent注册信息、message_log消息日志、shared_state共享状态。建表语句在项目启动时自动执行不需要手动操作。4.2 Agent基类的实现与角色扩展所有Agent都继承一个基类基类里实现了认知循环的骨架、消息收发、状态读写这些通用逻辑。子类只需要实现三个方法perceive感知、plan规划、execute执行。class BaseAgent: def __init__(self, agent_id, role, capabilities, subscriptions): self.agent_id agent_id self.role role self.capabilities capabilities self.subscriptions subscriptions self.message_bus MessageBus() self.state_store StateStore() def run(self): while True: messages self.message_bus.consume(self.subscriptions) context self.perceive(messages) plan self.plan(context) result self.execute(plan) self.reflect(result) def perceive(self, messages): raise NotImplementedError def plan(self, context): raise NotImplementedError def execute(self, plan): raise NotImplementedError def reflect(self, result): # 默认实现检查结果是否为空非空则认为成功 return result is not None以代码生成Agent为例它的perceive方法会从消息总线里拉取“需求文档更新”类型的消息然后从共享状态里读取最新的需求文档。plan方法调用LLM生成代码实现方案。execute方法调用代码执行沙箱验证代码。reflect方法检查代码是否通过语法检查和基本测试。注意基类里的run方法是一个无限循环实际部署时需要用信号量或事件来控制启停。我一开始忘了加停止机制结果调试时Agent停不下来只能杀进程。4.3 消息总线的搭建与消息格式定义消息总线我封装了一个MessageBus类底层用Redis的Stream做持久化Pub/Sub做实时通知。消息格式定义如下class Message(BaseModel): message_id: str sender: str receiver: str # 可以是具体agent_id或broadcast msg_type: str # 如requirement_update, code_ready, test_result priority: int # 0-9, 数字越大优先级越高 payload: dict # 具体内容 timestamp: float发送消息时MessageBus会把消息写入Redis Stream同时通过Pub/Sub频道通知订阅者。消费消息时Agent根据自己的订阅列表过滤消息类型只处理自己关心的。这里有个细节消息的优先级怎么用我的做法是调度器在分配任务时根据任务紧急程度设置优先级Agent消费消息时按优先级排序处理。比如“代码有严重bug需要修复”的消息优先级设为9“文档格式调整”设为3。这样紧急任务不会被低优先级任务阻塞。4.4 任务调度器的分配逻辑与实现任务调度器是整个系统的入口。用户提交一个请求后调度器先调用LLM做任务拆解把大任务拆成子任务列表然后根据每个子任务的能力需求去Agent注册中心匹配对应的Agent。匹配逻辑我用了简单的标签匹配每个子任务声明自己需要的能力标签如“需求分析”、“代码生成”调度器在注册中心里找拥有这些标签的Agent。如果有多个Agent匹配就选当前负载最低的那个。def dispatch(self, task): subtasks self.decompose(task) for subtask in subtasks: candidates self.registry.find_by_capability(subtask.required_capability) if not candidates: raise NoAvailableAgentError(subtask.required_capability) selected min(candidates, keylambda a: a.current_load) self.message_bus.send(Message( senderscheduler, receiverselected.agent_id, msg_typetask_assigned, prioritysubtask.priority, payload{subtask: subtask.dict()} )) selected.current_load 1任务拆解的提示词我调了很多版。早期版本拆得太粗一个子任务里混了好几种能力需求导致匹配不到合适的Agent。后来改成“每个子任务只对应一种能力标签”拆解粒度就合适了。4.5 共享状态存储的读写与冲突处理共享状态存储用了一张宽表每个状态项有唯一的key、value、版本号和最后修改者。Agent读取状态时拿到的是最新版本写入时采用乐观锁——如果写入时版本号已经变了说明有其他Agent改过这个状态当前写入就失败Agent需要重新读取再写入。def write_state(self, key, value, expected_version): current self.db.query(State).filter_by(keykey).first() if current.version ! expected_version: raise VersionConflictError(key, expected_version, current.version) current.value value current.version 1 current.last_modified_by self.agent_id self.db.commit()乐观锁的好处是避免了锁竞争多个Agent可以同时读同一个状态。坏处是写入冲突时需要重试所以Agent的写入逻辑要设计成幂等的重试不会产生副作用。5. 常见问题与排查技巧实录5.1 Agent之间消息丢失或重复消费怎么办这是消息总线最常遇到的问题。消息丢失通常是因为Agent消费消息后没有正确确认导致消息在Redis Stream里被标记为已消费但实际上没处理完。我的解决办法是引入手动确认机制Agent处理完消息后才发送ACK如果处理过程中崩溃消息会在超时后重新投递。重复消费则是因为ACK超时导致消息被重新投递但Agent其实已经处理过了。解决办法是给每条消息加唯一IDAgent在处理前先检查这个消息ID是否已经处理过处理过的直接跳过。这个去重逻辑我放在基类的perceive方法里统一实现子类不用操心。实操心得去重缓存不要用内存字典因为Agent重启后缓存就丢了。我用Redis的Set来存已处理的消息ID设置24小时过期既保证了去重效果又不会无限增长。5.2 任务分配不均导致某些Agent过载早期版本里调度器只看Agent的当前负载结果所有任务都往负载最低的Agent身上堆导致那个Agent很快也过载了。后来我改进了调度算法引入加权轮询每个Agent有一个权重值权重根据历史处理速度和成功率动态调整调度器按权重比例分配任务。具体实现是维护一个滑动窗口记录每个Agent最近100个任务的平均处理时间和成功率。权重 成功率 / 平均处理时间。调度器按权重做加权随机选择这样处理快且成功率高的Agent会分到更多任务但不会无限过载。5.3 Agent输出质量不稳定怎么破Agent输出质量波动是多智能体系统最头疼的问题。同一个需求代码生成Agent有时候写得很好有时候写得一塌糊涂。我总结了几个原因和对应的解决办法。原因一是提示词不够具体。解决办法是把提示词模板化把“写一个函数”改成“写一个函数要求输入参数类型为X返回类型为Y处理边界条件Z异常情况抛出W异常”。提示词越具体输出越稳定。原因二是缺少示例。我在提示词里加了几个few-shot示例展示期望的输入输出格式。加了示例后输出格式的稳定性明显提升。原因三是反思阶段太弱。前面说过反思阶段的检查清单要具体。我后来给每个Agent的反思阶段都配了一套自动化检查工具比如代码生成Agent的反思阶段会自动跑lint和单元测试不通过就打回重做。5.4 系统整体吞吐量上不去怎么优化系统跑起来后如果发现吞吐量低先看瓶颈在哪。我的排查顺序是先看消息总线的吞吐量再看Agent的处理速度最后看共享状态的读写性能。消息总线方面Redis Stream的吞吐量通常不是瓶颈但如果消息体太大比如把整个代码文件塞进消息里吞吐量会明显下降。解决办法是消息体里只放引用比如状态key实际内容放共享状态存储Agent需要时自己去拉。Agent处理速度方面瓶颈通常在LLM调用。我的优化手段是并行化——一个Agent可以同时处理多个任务只要这些任务之间没有依赖关系。实现上就是把Agent的run方法改成多线程或异步IO每个任务在一个独立的协程里跑。共享状态方面如果读写频繁SQLite会成为瓶颈。切换到PostgreSQL后性能提升明显。另外给常用的查询字段加索引也能显著降低读取延迟。5.5 常见问题速查表问题现象可能原因排查方法解决办法消息丢失ACK机制未启用检查消息日志中是否有未ACK的消息启用手动ACK设置合理超时重复消费ACK超时导致重投查看消息日志中同一ID出现多次加消息去重缓存Agent过载调度算法只看当前负载查看各Agent的任务队列长度改用加权轮询调度输出质量波动提示词不够具体对比高质量和低质量输出的提示词差异模板化提示词加few-shot示例吞吐量低LLM调用串行监控各环节耗时并行化Agent任务处理状态冲突并发写入同一key查看VersionConflictError日志乐观锁重试机制6. 协作流程的编排与动态调整6.1 静态编排与动态编排的取舍多智能体协作的流程编排有两种思路静态编排和动态编排。静态编排是提前把Agent之间的协作流程画好比如“需求分析→代码生成→测试→文档”每个环节的输出传给下一个环节。动态编排是让调度器根据任务特点实时决定走什么流程。我一开始用的是静态编排因为实现简单、可控性强。但很快发现静态编排太死板——有些任务不需要写文档有些任务需要反复迭代代码和测试静态流程根本覆盖不了。后来改成了动态编排为主、静态编排为辅的混合模式。具体做法是调度器维护一个流程模板库每个模板定义了一组可能的Agent协作路径。调度器根据任务类型选择最匹配的模板然后在模板允许的范围内动态调整具体路径。比如“新功能开发”模板允许代码生成和测试之间反复迭代但“文档更新”模板就直接从需求分析跳到文档撰写。6.2 迭代循环的控制与退出条件代码生成和测试之间经常需要反复迭代生成代码→测试→发现bug→修复→再测试。这个循环如果不加控制可能会无限进行下去。我设置了三个退出条件满足任意一个就跳出循环。第一个条件是测试全部通过。这是最理想的退出方式。第二个条件是迭代次数达到上限我设的是5次。超过5次还没通过说明问题可能不在代码本身而在需求理解或架构设计上需要人工介入。第三个条件是连续两次迭代的测试通过率没有提升说明Agent陷入了局部最优继续迭代也是浪费时间。注意迭代次数上限不要设得太高否则会浪费大量LLM调用。我试过设10次结果一个简单bug修了8次还没修好白白烧了一堆token。5次是个比较平衡的值。6.3 Agent动态增删与热插拔系统运行过程中可能需要新增Agent或下线某些Agent。比如业务扩展后需要增加一个“性能优化Agent”或者某个Agent频繁出错需要临时下线检修。这就要求系统支持Agent的热插拔。实现上Agent注册中心监听一个特殊的控制频道当有Agent上线或下线时注册中心更新能力表并广播通知。调度器收到通知后刷新本地的Agent列表后续任务分配就基于新的列表进行。已经在处理中的任务不受影响继续由原Agent完成。热插拔的关键是状态迁移。如果一个Agent下线时还有未完成的任务这些任务需要重新分配给其他Agent。我的做法是Agent下线前先把未完成的任务状态写回共享存储注册中心检测到Agent下线后把这些任务重新入队由调度器重新分配。7. 性能调优与规模化扩展的实战经验7.1 LLM调用成本的优化策略多智能体系统跑起来后LLM调用成本会迅速攀升。每个Agent的每次规划、执行、反思都要调LLM任务一多账单就吓人。我总结了几个降本手段。第一个是缓存。相同的输入不要重复调LLM把输入做哈希后查缓存命中就直接返回。这个手段对反思阶段特别有效因为很多反思的输入是相似的。第二个是模型分级。不是所有环节都需要用最强模型。规划阶段可以用强模型保证推理质量执行阶段可以用轻量模型降低成本反思阶段用规则引擎代替LLM做初步筛选只有规则引擎拿不准的才调LLM。第三个是批处理。把多个独立的LLM调用合并成一个批次请求减少网络往返开销。这个手段对并行处理多个子任务时特别有效。7.2 水平扩展从单机到分布式单机跑多智能体系统受限于CPU和内存Agent数量一多就卡。水平扩展的思路是把Agent分布到多台机器上消息总线和共享状态存储也做成分布式的。消息总线从单机Redis换成Redis Cluster共享状态存储从SQLite换成PostgreSQL集群。Agent本身是无状态的可以随意在多台机器上部署多个实例。调度器需要感知所有实例的存在这个通过注册中心实现——每个Agent实例启动时向注册中心注册自己的地址调度器从注册中心获取全量实例列表。分布式部署后网络延迟成为新的瓶颈。我的优化手段是把频繁通信的Agent部署在同一台机器或同一个机房减少跨网络调用。另外消息体尽量精简大内容走共享存储不走消息总线。7.3 监控指标体系的建立系统规模化之后没有监控就是睁眼瞎。我建立了一套三层监控指标体系。系统层监控消息总线的吞吐量、延迟、积压消息数共享状态存储的读写QPS和响应时间以及各机器的CPU、内存、网络使用率。Agent层监控每个Agent的任务处理量、平均处理时间、成功率、LLM调用次数和token消耗。这些指标按Agent角色聚合方便对比同类Agent的表现。业务层监控端到端的任务完成率、平均完成时间、人工介入率。这些指标直接反映系统的业务价值。监控数据用Prometheus采集Grafana做可视化。告警规则设了三条消息积压超过阈值告警、Agent成功率低于阈值告警、任务平均完成时间超过阈值告警。8. 踩坑记录与个人经验总结8.1 那些让我熬夜排查的坑第一个坑是消息顺序问题。消息总线是异步的不保证消息按发送顺序到达。结果代码生成Agent有时候先收到“测试失败”的消息后收到“代码已生成”的消息导致状态混乱。解决办法是在消息里加序列号Agent按序列号排序后再处理。第二个坑是Agent死锁。两个Agent互相等待对方的消息结果都卡住了。比如代码生成Agent等测试Agent的反馈测试Agent等代码生成Agent的代码。解决办法是设置超时机制等待超过一定时间就触发超时处理逻辑避免无限等待。第三个坑是状态不一致。多个Agent同时读写共享状态导致状态出现中间态。比如需求文档更新到一半代码生成Agent就读到了不完整的需求。解决办法是引入事务机制状态更新要么全部成功要么全部回滚。8.2 关于Agent提示词设计的几点体会提示词设计是多智能体系统里最需要反复打磨的部分。我的体会是提示词要像给新员工写工作手册一样写。你不能只说“帮我写代码”你要说“你是代码生成Agent你的职责是根据需求文档生成可运行的代码。代码必须符合以下规范使用TypeScript、遵循项目现有的命名约定、每个函数必须有JSDoc注释、异常情况必须处理。生成代码后你必须调用代码执行沙箱验证代码能否跑通如果跑不通就修复后重新验证。”提示词里还要明确Agent的边界。比如代码生成Agent的提示词里要写“你不负责写测试用例测试用例由测试Agent负责。你只需要保证代码本身能跑通”。边界清晰了Agent之间才不会互相越权。8.3 什么场景适合上多智能体什么场景不适合多智能体系统不是银弹。我踩过的一个坑就是在一个简单场景里硬上了多智能体结果复杂度上去了效果反而没比单Agent好。适合上多智能体的场景有几个特征任务可以明确拆分成多个子任务每个子任务需要不同的专业能力子任务之间有依赖关系需要传递中间结果对输出质量要求高需要多轮校验和迭代。不适合的场景也很明显任务本身很简单单Agent就能搞定任务对实时性要求极高多Agent之间的通信开销无法接受任务需要全局一致性多Agent的分布式决策反而容易产生冲突。8.4 后续可以继续深挖的方向这套系统跑通之后我还在继续折腾几个方向。一个是Agent的自主学习能力让Agent从历史任务中总结经验自动优化自己的提示词和工具配置。另一个是跨系统的Agent协作让不同项目的Agent能够互相调用形成更大的协作网络。还有一个是Agent行为的可解释性让每个决策都能追溯到具体的推理过程方便调试和审计。这些方向都还在探索阶段有进展了再另开一篇细说。如果你也在做多智能体相关的东西欢迎交流踩坑经验。
