1. 多智能体协作架构到底在解决什么问题第一次接触 Loop Engineering 这个概念是在一个需要同时调度十几个自动化任务的项目里。当时的状态很原始每个任务单独写脚本任务之间靠文件轮询传递状态一个环节卡住整条链路就停摆。后来团队里有人提出能不能让这些任务像一支球队一样有前锋、有中场、有后卫各自知道什么时候该跑位、什么时候该传球。这就是我理解多智能体协作架构的起点——它不是让一个万能程序干所有事而是把复杂目标拆成若干角色让角色之间通过约定好的协议协同完成。Loop Engineering 这个词拆开看就是“循环”加“工程化”。循环指的是智能体之间不断迭代、反馈、修正的过程工程化指的是这套东西不能停留在演示阶段得能稳定跑在生产环境里。多智能体协作架构的核心价值在于当任务复杂度超过单个智能体的处理上限时通过分工和协作把问题降维。比如一个电商客服场景单个智能体既要理解用户意图、又要查订单、又要处理退换货政策、还要安抚情绪很容易顾此失彼。拆成意图识别智能体、订单查询智能体、政策匹配智能体、话术生成智能体之后每个角色的职责边界清晰出错也容易定位。这套架构适合谁来参考如果你正在做自动化流程编排、智能客服、内容生产流水线、数据分析管道或者任何需要多个步骤串联且步骤之间有依赖关系的系统多智能体协作架构都值得认真研究。它不要求你一开始就上很重的框架哪怕先用两个智能体做最简单的“生产者-消费者”模型也能体会到协作带来的结构性优势。我见过太多项目一开始追求大而全的架构结果连最基本的消息传递都没跑通最后不了了之。所以我的建议是从最小闭环开始先让两个智能体跑起来再逐步扩展。2. 核心架构设计与角色拆解2.1 为什么选择多智能体而不是单体巨兽单体智能体的思路是“一个大脑搞定所有事”这在任务边界清晰、步骤固定的场景下没问题。但一旦任务涉及多轮交互、动态决策、外部工具调用单体架构就会变得极其臃肿。我做过一个对比实验同一个文档处理任务单体智能体需要在一个提示词里塞进格式解析、内容摘要、关键词提取、分类归档四个功能提示词长度超过三千字模型经常顾此失彼摘要质量忽高忽低。拆成四个智能体之后每个智能体的提示词控制在五百字以内职责单一输出稳定性明显提升。多智能体架构的另一个优势是可替换性。单体智能体如果要换一个摘要模型整个提示词都得重写。多智能体架构下摘要智能体是一个独立模块换模型只需要改这个模块的配置其他部分不受影响。这种模块化带来的维护便利在项目迭代半年之后会体现得淋漓尽致。当然多智能体不是没有代价。最大的代价是通信开销和状态一致性。智能体之间传递消息需要序列化和反序列化如果消息格式设计得不好光解析消息就能吃掉大量时间。状态一致性更麻烦智能体 A 认为任务已经完成智能体 B 还在等 A 的输出这种死锁在多智能体系统里非常常见。所以架构设计的核心不是“怎么拆”而是“拆完之后怎么保证消息不丢、状态不乱”。2.2 角色划分的三种常见模式在实际项目中我总结出三种比较实用的角色划分模式。第一种是流水线模式智能体按顺序排列前一个的输出是后一个的输入。这种模式最简单适合步骤固定的场景比如“抓取数据→清洗数据→分析数据→生成报告”。流水线模式的缺点是容错性差中间任何一个环节出错整条线就断了。解决办法是在每个环节加校验智能体校验不通过就回退到上一个环节重做。第二种是主管-工人模式有一个主管智能体负责拆解任务和分配工作多个工人智能体负责执行具体任务。主管智能体不干具体活只做调度和结果汇总。这种模式适合任务可以并行拆分的场景比如“同时分析十个不同来源的数据最后汇总成一份报告”。主管-工人模式的关键在于主管的拆解能力如果拆得不好工人之间会出现重复劳动或者遗漏。第三种是辩论模式多个智能体对同一个问题给出不同答案然后通过投票或者辩论达成共识。这种模式适合需要高可靠性的决策场景比如风险评估、内容审核。辩论模式的成本最高因为同一个问题要被处理多次但换来的是更高的准确率。我一般只在关键决策节点使用辩论模式日常任务还是用流水线或主管-工人模式。2.3 通信协议与消息格式设计智能体之间怎么说话这是架构设计里最容易被忽视但最影响稳定性的部分。我见过用自然语言直接通信的方案智能体 A 输出一段话智能体 B 去解析这段话。这种方案在演示阶段很酷但在生产环境里就是灾难自然语言有歧义解析规则稍微不严谨就会出错。我的做法是强制使用结构化消息格式通常是 JSON字段包括消息 ID、发送者、接收者、消息类型、负载内容、时间戳、状态标记。消息类型需要提前定义好比如“任务请求”“任务结果”“错误报告”“心跳检测”。每种消息类型的负载结构也要固定不能这次传字符串、下次传数组。我一般会写一个消息 schema 文件所有智能体在发送和接收消息时都按照 schema 校验校验不通过直接拒绝避免脏数据在系统里流转。还有一个细节是消息的幂等性。智能体可能因为网络抖动或者超时重试而收到重复消息如果消息处理不是幂等的就会产生重复操作。解决办法是在消息里带一个唯一 ID智能体处理之前先查一下这个 ID 是否已经处理过处理过就直接返回缓存结果。这个机制在分布式系统里很常见但在多智能体架构里经常被忽略导致一些莫名其妙的重复执行问题。3. 核心细节解析与实操要点3.1 智能体生命周期管理每个智能体从启动到销毁经历的状态包括初始化、就绪、忙碌、等待、错误、终止。很多项目只关注“忙碌”状态忽略了其他状态的管理结果智能体卡在错误状态里没人管整个系统慢慢僵死。我的做法是给每个智能体加一个状态机状态之间的转换有明确的触发条件。比如从“忙碌”到“等待”是因为需要外部输入从“忙碌”到“错误”是因为执行超时或者抛出异常。状态机的好处是可视化。你可以画一张状态转换图一眼看出哪些状态之间缺少转换路径。我遇到过一个 bug智能体执行完任务后直接进入“终止”状态但主管智能体还在等它的结果导致主管一直阻塞。后来在状态机里加了“完成”状态智能体执行完后先进入“完成”等主管确认收到结果后再进入“终止”问题就解决了。心跳检测也是生命周期管理的一部分。每个智能体定期向注册中心发送心跳超过一定时间没收到心跳就认为该智能体已经下线主管智能体需要重新分配它的任务。心跳间隔的设置需要权衡太短会增加通信负担太长会导致故障发现延迟。我一般设置心跳间隔为 5 秒超时阈值为 15 秒也就是连续三次心跳丢失才判定下线。3.2 任务拆解与依赖管理任务拆解是多智能体协作的第一步。拆解的原则是每个子任务有明确的输入和输出子任务之间尽量解耦但依赖关系必须显式声明。我通常用有向无环图来表示任务依赖节点是子任务边是依赖关系。有向无环图的好处是可以做拓扑排序确定哪些任务可以并行执行哪些必须串行。拆解粒度是个经验活。拆得太粗智能体负担重容易出错拆得太细通信开销大调度复杂。我的经验是一个子任务的执行时间在 10 秒到 2 分钟之间比较合适。如果某个子任务经常超过 2 分钟考虑进一步拆分如果某个子任务经常在 1 秒内完成考虑和相邻子任务合并。依赖管理还有一个坑是循环依赖。智能体 A 等 B 的输出B 等 A 的输出两个都卡住。有向无环图在理论上可以避免循环依赖但在动态任务拆解的场景下循环依赖可能意外产生。我的做法是在任务提交时做一次环检测发现环就拒绝执行并报警。环检测用深度优先搜索就能实现代码量不大但能避免很多死锁问题。3.3 上下文传递与状态共享智能体之间需要共享上下文比如用户的历史对话、任务的中间结果、全局配置。上下文传递有两种方式一种是随消息传递每个消息里带上必要的上下文片段另一种是集中存储智能体从共享存储里读取上下文。两种方式各有优劣随消息传递简单直接但消息会变得很大集中存储消息轻量但需要额外的存储服务和一致性保证。我一般根据上下文的大小和访问频率来选择。小片段、高频访问的上下文随消息传递比如当前任务的 ID、用户的身份标识。大块、低频访问的上下文集中存储比如完整的对话历史、知识库文档。集中存储我通常用键值数据库键是任务 ID 或者会话 ID值是序列化后的上下文对象。状态共享的一个常见问题是并发写冲突。两个智能体同时更新同一个上下文对象后写的覆盖先写的。解决办法是加版本号或者乐观锁智能体读取上下文时记下版本号写入时检查版本号是否变化变化了就重新读取再写入。这个机制在数据库里很成熟但在多智能体架构里需要自己实现因为智能体之间的状态共享往往不走数据库。4. 实操过程与核心环节实现4.1 环境准备与基础框架搭建开始搭建之前先确定技术栈。我用的比较多的是 Python因为生态丰富异步编程支持好。基础框架包括消息队列、注册中心、日志系统。消息队列我用 Redis 的发布订阅功能轻量且够用。注册中心可以用 Redis 的键过期机制实现智能体启动时写入一个带过期时间的键定期刷新过期了就认为下线。日志系统用标准的 logging 模块每个智能体写自己的日志文件方便排查问题。目录结构建议这样组织agents 目录放各个智能体的实现每个智能体一个文件core 目录放消息协议、状态机、任务调度这些公共模块config 目录放配置文件logs 目录放日志。每个智能体继承一个基类基类里实现消息收发、心跳、状态管理这些通用逻辑子类只需要实现具体的任务处理函数。# core/agent_base.py import json import time import threading import redis class BaseAgent: def __init__(self, agent_id, redis_client): self.agent_id agent_id self.redis redis_client self.state init self.heartbeat_interval 5 self.running True def start(self): self.state ready threading.Thread(targetself._heartbeat_loop, daemonTrue).start() self._message_loop() def _heartbeat_loop(self): while self.running: self.redis.setex(fagent:{self.agent_id}:heartbeat, 15, alive) time.sleep(self.heartbeat_interval) def _message_loop(self): pubsub self.redis.pubsub() pubsub.subscribe(fagent:{self.agent_id}) for message in pubsub.listen(): if message[type] message: self._handle_message(json.loads(message[data])) def _handle_message(self, message): msg_type message.get(type) if msg_type task: self.state busy result self.process_task(message[payload]) self.send_message(message[sender], result, result) self.state ready def process_task(self, payload): raise NotImplementedError def send_message(self, receiver, msg_type, payload): message { id: f{self.agent_id}-{time.time()}, sender: self.agent_id, receiver: receiver, type: msg_type, payload: payload, timestamp: time.time() } self.redis.publish(fagent:{receiver}, json.dumps(message))这个基类提供了心跳、消息循环、状态管理的基本骨架。子类只需要实现 process_task 方法处理具体的任务逻辑。消息格式统一用 JSON包含消息 ID、发送者、接收者、类型、负载、时间戳。消息 ID 用于幂等性检查时间戳用于超时判断。4.2 任务调度器的实现细节任务调度器是整个系统的中枢负责接收外部请求、拆解任务、分配给智能体、收集结果。我实现调度器的时候踩过几个坑这里详细说一下。第一个坑是任务拆解后的依赖管理。一开始我用简单的列表存储子任务按顺序执行结果发现有些子任务其实可以并行白白浪费了时间。后来改成有向无环图用拓扑排序确定执行顺序并行度明显提升。第二个坑是超时处理。某个智能体执行任务超时了调度器如果一直等整个任务就卡住了。我的做法是给每个子任务设置超时时间超时后调度器重新分配任务给其他智能体或者标记任务失败并通知上游。超时时间的设置需要根据任务的历史执行时间来确定我一般设置为历史平均执行时间的三倍。第三个坑是结果收集。多个智能体并行执行结果返回的顺序是不确定的。调度器需要根据任务 ID 把结果对应到正确的子任务上。我一开始用列表存储结果按返回顺序对应结果经常错位。后来改成字典键是任务 ID值是结果问题就解决了。# core/scheduler.py import json import time import redis from collections import defaultdict class Scheduler: def __init__(self, redis_client): self.redis redis_client self.tasks {} self.results defaultdict(dict) def submit_task(self, task_id, dag): self.tasks[task_id] { dag: dag, status: running, created_at: time.time() } self._execute_dag(task_id) def _execute_dag(self, task_id): dag self.tasks[task_id][dag] completed set() while len(completed) len(dag[nodes]): ready_nodes [ node for node in dag[nodes] if node[id] not in completed and all(dep in completed for dep in node[deps]) ] for node in ready_nodes: self._dispatch_node(task_id, node) time.sleep(0.1) completed self._check_completed(task_id, dag) def _dispatch_node(self, task_id, node): message { id: f{task_id}-{node[id]}, sender: scheduler, receiver: node[agent], type: task, payload: node[payload], timestamp: time.time() } self.redis.publish(fagent:{node[agent]}, json.dumps(message)) def _check_completed(self, task_id, dag): completed set() for node in dag[nodes]: result_key fresult:{task_id}:{node[id]} if self.redis.exists(result_key): completed.add(node[id]) return completed这个调度器实现了基本的有向无环图执行逻辑。submit_task 接收任务 ID 和有向无环图定义_execute_dag 循环找出所有依赖已满足的节点分发给对应的智能体。_check_completed 检查 Redis 里是否有结果键有就认为该节点已完成。实际生产中还需要加超时检测、失败重试、结果校验等逻辑但核心思路就是这样。4.3 智能体之间的消息传递实战消息传递的可靠性直接决定系统的稳定性。我用 Redis 发布订阅做消息传递但 Redis 发布订阅有个问题如果订阅者不在线消息就丢了。对于关键任务消息这个丢失是不可接受的。解决办法是加一个确认机制发送者发送消息后等待接收者的确认消息超时没收到就重发。重发次数超过阈值就标记任务失败。确认机制会增加通信开销所以不是所有消息都需要确认。我的做法是分级任务分配消息需要确认心跳消息不需要确认结果返回消息需要确认。确认消息本身不需要再确认否则就无限递归了。还有一个细节是消息的顺序性。Redis 发布订阅不保证消息顺序智能体 A 先发消息 M1 再发 M2智能体 B 可能先收到 M2 再收到 M1。对于有顺序要求的场景需要在消息里带序列号接收者按序列号排序后再处理。我一般给每个发送者维护一个递增的序列号接收者缓存乱序消息等缺失的序列号到达后再按顺序处理。# core/reliable_messaging.py import json import time import threading import redis class ReliableMessenger: def __init__(self, agent_id, redis_client): self.agent_id agent_id self.redis redis_client self.pending {} self.seq 0 self.lock threading.Lock() def send_with_ack(self, receiver, msg_type, payload, timeout10, max_retries3): with self.lock: self.seq 1 msg_id f{self.agent_id}-{self.seq} message { id: msg_id, sender: self.agent_id, receiver: receiver, type: msg_type, payload: payload, timestamp: time.time() } for attempt in range(max_retries): self.redis.publish(fagent:{receiver}, json.dumps(message)) self.pending[msg_id] time.time() time.sleep(timeout) if msg_id not in self.pending: return True return False def handle_ack(self, msg_id): with self.lock: self.pending.pop(msg_id, None)这个可靠消息模块实现了带确认的消息发送。send_with_ack 发送消息后等待确认超时重发重发次数超过阈值返回失败。handle_ack 在收到确认消息时调用从待确认列表中移除消息 ID。实际使用中还需要一个后台线程定期清理超时的待确认消息避免内存泄漏。5. 常见问题与排查技巧实录5.1 智能体卡死与死锁排查智能体卡死是多智能体系统里最常见的问题。表现是某个智能体一直处于“忙碌”状态不返回结果也不响应心跳。排查思路是先看日志确认卡在哪个步骤再看消息队列确认是否有未处理的消息最后看外部依赖确认是否在等数据库或者 API 响应。我遇到过一次典型的死锁智能体 A 等智能体 B 的结果智能体 B 等智能体 A 的结果。两个都卡在“等待”状态心跳正常但任务永远完不成。排查的时候看消息日志发现 A 和 B 互相发了请求消息但都没有发结果消息。解决办法是在调度器里加环检测发现循环依赖就拒绝执行并报警。还有一种卡死是资源耗尽。智能体开了太多线程或者连接系统资源不够新任务无法执行。排查方法是看系统监控CPU、内存、文件描述符、网络连接数。我一般给每个智能体设置资源上限比如最多开 10 个线程、最多 100 个连接超过就拒绝新任务。5.2 消息丢失与重复处理消息丢失的表现是任务一直处于“进行中”但没有任何进展。排查方法是看消息队列的监控指标确认消息是否被发送和接收。如果发送了但没接收可能是订阅者不在线或者订阅关系断了。如果接收了但没处理可能是处理逻辑抛异常了。重复处理的表现是同一个任务被执行了多次产生了重复的结果。排查方法是看消息 ID确认是否有重复的消息 ID 被处理。解决办法是在处理逻辑里加幂等性检查处理之前先查一下消息 ID 是否已经处理过。我踩过的一个坑是 Redis 发布订阅的重连问题。网络抖动导致订阅连接断开Redis 客户端自动重连但重连期间的消息丢了。后来改成用 Redis 的 Stream 数据结构Stream 支持消费者组和消息确认消息不会因为消费者离线而丢失。Stream 的代价是比发布订阅重一些但对于关键任务消息这个代价是值得的。5.3 性能瓶颈定位与优化多智能体系统的性能瓶颈通常出现在三个地方消息传递、任务调度、智能体执行。定位方法是加监控指标统计每个环节的耗时。消息传递耗时包括序列化、网络传输、反序列化任务调度耗时包括依赖检查、节点分发、结果收集智能体执行耗时包括提示词构建、模型调用、结果解析。优化手段根据瓶颈位置不同而不同。消息传递慢就换更高效的序列化格式比如用 MessagePack 代替 JSON或者用二进制协议代替文本协议。任务调度慢就优化依赖检查算法用缓存减少重复计算。智能体执行慢就优化提示词减少不必要的上下文或者换更快的模型。我做过一次优化把消息序列化从 JSON 换成 MessagePack消息体积减少了 40%序列化耗时减少了 60%。但 MessagePack 的可读性不如 JSON调试的时候不方便。所以我的建议是开发和调试阶段用 JSON生产环境如果性能压力大再换 MessagePack。问题类型典型表现排查手段解决方案智能体卡死状态一直忙碌心跳正常查日志、查消息队列、查外部依赖加环检测、设资源上限消息丢失任务无进展消息队列无记录查订阅关系、查网络连接改用 Stream、加确认机制重复处理同一任务多次执行查消息 ID、查处理日志加幂等性检查性能瓶颈整体耗时增加加监控指标、分段计时优化序列化、优化调度算法5.4 智能体输出质量不稳定的应对智能体的输出质量受提示词、模型版本、输入数据的影响。我遇到过同一个智能体在不同时间对相同输入给出不同输出的情况排查后发现是模型版本更新了。解决办法是固定模型版本不要用 latest 标签用具体的版本号。如果必须用最新版本那就加输出校验校验不通过就重试或者降级。提示词的影响更大。提示词里一个词的改动可能导致输出风格完全变化。我的做法是提示词版本化管理每次修改都记录版本号和修改原因方便回滚。同时加输出格式校验比如要求输出 JSON就用 JSON schema 校验不通过就重试。输入数据的影响也不容忽视。如果输入数据里有噪声或者格式不一致智能体的输出也会不稳定。解决办法是在输入智能体之前加一个预处理步骤清洗数据、统一格式。这个预处理步骤可以是一个独立的智能体专门负责数据清洗。6. 工程化最佳实践与扩展思路6.1 可观测性建设日志、指标、追踪多智能体系统的调试难度比单体系统高一个数量级因为问题可能出现在任何一个智能体、任何一次消息传递、任何一个状态转换。没有可观测性排查问题就像盲人摸象。我的做法是三管齐下日志、指标、追踪。日志要结构化每条日志包含时间戳、智能体 ID、任务 ID、消息 ID、日志级别、日志内容。结构化日志的好处是可以被日志系统自动解析和索引排查问题时可以按任务 ID 或者消息 ID 过滤。我一般用 JSON 格式写日志每行一个 JSON 对象。指标要覆盖关键路径包括消息发送数、消息接收数、任务成功数、任务失败数、平均执行时间、超时次数。指标用 Prometheus 格式暴露配合 Grafana 做可视化。我一般给每个智能体暴露一个 /metrics 接口Prometheus 定期抓取。追踪要贯穿整个任务生命周期。一个任务从提交到完成经过多个智能体每个智能体处理时都记录一个 span所有 span 组成一个 trace。追踪用 OpenTelemetry 实现trace ID 在消息里传递智能体处理消息时从消息里提取 trace ID创建子 span。这样排查问题时可以看到整个任务的调用链路快速定位瓶颈。6.2 灰度发布与版本兼容多智能体系统的升级比单体系统麻烦因为智能体之间有依赖关系升级一个智能体可能影响其他智能体。我的做法是灰度发布新版本智能体先上线一个实例把少量流量导过去观察一段时间没问题再逐步扩大流量。灰度发布的关键是版本兼容新旧版本的智能体要能互相通信。版本兼容的实现方式是在消息里带版本号接收者根据版本号决定如何处理消息。如果新版本的消息格式变了接收者要能同时处理新旧两种格式。我一般用适配器模式新版本的消息先经过适配器转换成旧版本格式再交给旧版本的智能体处理。这样新旧版本可以共存一段时间等所有智能体都升级完成后再移除适配器。6.3 从单机到分布式的扩展路径单机多智能体系统受限于单台机器的资源智能体数量多了就跑不动。扩展路径是先把消息队列换成独立的服务再把智能体部署到多台机器上。消息队列换成独立服务后智能体之间通过网络通信不再依赖本机的 Redis。智能体部署到多台机器后需要服务发现机制智能体启动时注册自己的地址其他智能体通过服务发现找到它。分布式扩展的挑战是网络分区和时钟漂移。网络分区导致部分智能体无法通信系统需要决定是继续服务还是停止服务。我一般选择继续服务但降级功能比如只处理关键任务非关键任务排队等待。时钟漂移导致不同机器上的时间不一致消息的时间戳不可靠。解决办法是用逻辑时钟代替物理时钟逻辑时钟只保证因果关系不保证绝对时间。6.4 安全边界与权限控制多智能体系统里智能体之间的信任关系需要明确。不是所有智能体都能访问所有资源需要加权限控制。我的做法是给每个智能体分配一个角色角色决定它能访问哪些资源、能调用哪些工具。智能体发送消息时带上自己的角色令牌接收者校验令牌没有权限就拒绝。权限控制还要考虑最小权限原则。智能体只拥有完成其任务所需的最小权限不多给。比如查询订单的智能体只能读订单数据不能修改订单数据。修改订单的智能体才能写订单数据。这样即使某个智能体被攻破影响范围也有限。还有一个安全边界是输入输出过滤。智能体的输入可能包含恶意内容比如提示词注入攻击。输出可能包含敏感信息比如用户隐私数据。我的做法是在输入智能体之前加一个过滤智能体检测并清除恶意内容在输出智能体之后加一个审查智能体检测并脱敏敏感信息。这两个智能体不参与业务逻辑只做安全过滤。6.5 成本控制与资源调度多智能体系统的成本主要来自模型调用和基础设施。模型调用成本跟调用次数和每次调用的 token 数有关。降低调用次数的方法是合并任务把多个小任务合并成一个大任务一次调用完成。降低每次调用的 token 数的方法是精简提示词去掉不必要的上下文和示例。基础设施成本跟智能体数量和运行时长有关。降低智能体数量的方法是合并角色把功能相近的智能体合并成一个。降低运行时长的方法是按需启动任务来了才启动智能体任务完成就销毁。按需启动的代价是启动延迟适合对延迟不敏感的场景。我一般会做一个成本监控面板实时显示模型调用次数、token 消耗、基础设施费用。设置预算告警超过预算就报警避免意外的高额账单。成本优化是一个持续的过程需要定期审查调用日志找出可以优化的地方。6.6 实际项目中的取舍经验做了几个多智能体项目之后我最大的体会是架构复杂度要和任务复杂度匹配。任务简单就用单体智能体任务复杂才用多智能体。不要为了用多智能体而用多智能体那是本末倒置。我见过一个项目任务只是简单的文本分类却搭了一套多智能体架构结果调试成本比单体方案高了十倍效果还差不多。另一个体会是先跑通再优化。一开始不要追求完美的架构先让最基本的流程跑起来哪怕消息传递用的是最简单的轮询哪怕状态管理用的是全局变量。跑通之后再逐步替换成更可靠的方案。我见过太多项目卡在架构设计阶段讨论了几周还没写一行代码最后不了了之。最后一个体会是文档和测试不能省。多智能体系统的行为比单体系统复杂得多没有文档和测试过两个月自己都看不懂。文档要写清楚每个智能体的职责、消息格式、状态转换、依赖关系。测试要覆盖正常流程、异常流程、边界条件。我一般要求每个智能体至少有单元测试整个系统有集成测试关键路径有端到端测试。这些经验都是在实际项目中踩坑踩出来的希望能给正在做或者准备做多智能体协作架构的朋友一些参考。架构设计没有银弹适合自己的才是最好的。