1. 多Agent协作架构到底在解决什么问题1.1 从单Agent的瓶颈说起过去一年我经手过不少大模型应用项目从最简单的问答机器人到稍微复杂一点的文档分析流水线单Agent架构能覆盖的场景其实比很多人想象的要窄。一个Agent加一套提示词再挂几个工具函数处理单一任务时确实够用但只要任务链条超过三步、涉及多个专业领域的知识交叉或者需要在执行过程中动态调整策略单Agent就开始力不从心了。最典型的表现是上下文窗口被撑爆。你让一个Agent既要做需求分析、又要写代码、还要做代码审查它得把前面所有步骤的完整上下文都带着走token消耗呈指数级增长而且随着上下文变长模型对早期信息的注意力会衰减后面生成的代码可能完全偏离最初的需求。另一个问题是角色冲突同一个模型实例很难同时扮演好“严格的代码审查者”和“富有创造力的方案设计者”这两个角色提示词里写再多“你现在是XX专家”也压不住模型底层的概率分布。多Agent协作架构就是冲着这些痛点来的。它的核心思路很朴素把一个大任务拆成若干个子任务每个子任务交给一个专门的Agent去处理Agent之间通过消息传递来协调。这跟人类团队协作的逻辑是一样的——你不会让一个人同时干产品经理、程序员和测试的活而是分给不同的人每个人专注自己擅长的部分通过会议、文档、即时消息来同步进度。1.2 协作架构的几种主流形态目前业界常见的多Agent协作架构大致可以归为三类我在不同项目里都实际用过各有各的适用场景。第一类是流水线式Pipeline。Agent按照预定义的顺序依次执行前一个Agent的输出是后一个Agent的输入。这种架构最简单、最可控适合流程固定的任务比如“需求解析→方案设计→代码生成→代码审查→测试用例生成”这种线性流程。缺点是缺乏灵活性如果中间某个环节发现问题需要回溯整个流水线就得重新跑。第二类是层级式Hierarchical。有一个协调者AgentCoordinator负责拆解任务、分配子任务、汇总结果下面挂若干个执行者AgentWorker。协调者不直接干活只做调度和决策。这种架构适合任务结构不太固定、需要动态调整的场景。我在一个科研论文辅助写作的项目里用过这种架构协调者根据论文的当前状态决定下一步是让“文献调研Agent”去补充资料还是让“数据分析Agent”去处理实验数据还是让“写作Agent”去生成段落。第三类是去中心化式Decentralized。没有明确的协调者Agent之间通过共享消息总线或者黑板Blackboard来通信每个Agent根据自己的能力决定是否响应某个消息。这种架构最灵活但也最难调试因为系统的行为是涌现出来的你很难预测最终会走到哪里。适合探索性任务比如创意生成、复杂问题求解。实际项目中我大多数时候用的是混合架构——顶层用层级式做任务分解和调度底层用流水线式保证执行效率关键节点上允许Agent之间点对点通信来做异常处理。1.3 任务调度的核心挑战多Agent系统里任务调度是最容易出问题的地方。单Agent的时候你只需要考虑“下一步做什么”多Agent的时候你要考虑“谁来做、什么时候做、做完了怎么通知别人、做错了怎么回滚”。我踩过的一个坑是任务依赖管理。早期我用简单的队列来调度Agent结果发现有些Agent在等上游输出的时候空转有些Agent在依赖还没就绪的时候就被触发了导致拿到脏数据。后来引入了DAG有向无环图来管理任务依赖每个节点是一个Agent任务边表示数据依赖关系只有入度为0的节点才能被调度执行。这个改动让系统的稳定性提升了一个档次。另一个坑是超时和重试策略。Agent调用大模型API是有延迟的有时候几秒有时候几十秒网络抖动或者模型服务限流的时候可能直接超时。如果没有合理的超时机制一个Agent卡住会导致整个系统挂起。我的做法是给每个Agent任务设置独立的超时时间超时后触发重试或者降级策略同时把失败信息写回调度器让调度器决定是重新分配任务还是跳过。还有一个容易被忽视的问题是Agent之间的消息格式。如果每个Agent都用自己的一套输入输出格式协调者就得写大量的适配代码。我的经验是尽早定义一套统一的消息协议至少包含任务ID、发送者、接收者、消息类型、负载内容、时间戳这几个字段后面扩展起来会轻松很多。2. 核心组件拆解与技术选型2.1 Agent的抽象设计在多Agent系统里每个Agent本质上是一个封装了特定能力、状态和行为的计算单元。我通常会把Agent抽象成四个核心部分角色定义Role、工具集Tools、记忆Memory和决策逻辑Policy。角色定义决定了Agent的“人设”和能力边界。比如一个“代码审查Agent”的角色定义里会明确它只负责审查代码质量、安全性和性能问题不负责修改代码。这个边界很重要没有边界的Agent会越权操作导致系统行为不可预测。工具集是Agent与外部世界交互的接口。一个Agent可以挂载多个工具函数比如搜索、计算、文件读写、API调用等。工具的设计要遵循单一职责原则一个工具只做一件事参数尽量简单。我见过有人把“搜索并总结”做成一个工具结果这个工具内部又调了一次大模型导致嵌套调用和延迟叠加调试起来非常痛苦。记忆模块负责存储Agent的历史交互信息。短期记忆通常就是当前会话的上下文长期记忆则需要外部存储支持比如向量数据库。在多Agent场景下记忆的共享和隔离是个关键设计决策——哪些信息所有Agent都能看到哪些信息只有特定Agent能访问这直接影响到系统的安全性和效率。决策逻辑是Agent的“大脑”决定它在当前状态下应该做什么。最简单的决策逻辑是基于规则的if-else复杂一点的是基于大模型的推理。我的经验是能用规则就用规则规则搞不定的再用模型推理因为模型推理的成本和不确定性都更高。2.2 通信机制的选择Agent之间的通信机制直接决定了系统的耦合度和可扩展性。我实践下来主要有三种方式直接函数调用、消息队列和共享状态。直接函数调用最简单Agent A直接调用Agent B的方法同步等待返回结果。这种方式适合Agent数量少、调用关系简单的场景缺点是耦合度高A必须知道B的存在和接口B挂了A也会挂。消息队列是更常用的方式Agent之间通过发布/订阅消息来通信发送方不需要知道接收方是谁。这种方式解耦彻底扩展性好但引入了消息中间件的运维成本而且消息的顺序和可靠性需要额外保证。我在一个多Agent内容生成的项目里用过Redis的Pub/Sub轻量够用但如果要保证消息不丢还是得上RabbitMQ或者Kafka。共享状态的方式是维护一个全局的状态存储比如Redis或者内存中的字典所有Agent都读写这个状态。这种方式适合需要频繁同步状态的场景但并发写的时候需要加锁否则会出现竞态条件。我一般只在Agent数量少、状态变更不频繁的时候用这种方式。实际项目中我倾向于组合使用核心的任务调度用消息队列Agent之间的实时状态同步用共享状态紧急的异常处理用直接调用。2.3 调度器的实现要点调度器是多Agent系统的心脏它的职责是接收任务、解析依赖、分配Agent、监控执行、处理异常。我实现过的调度器有简单的也有复杂的这里说几个关键设计点。任务队列的设计。我用过优先队列和普通队列优先队列适合有紧急任务插队的场景普通队列适合FIFO的批处理场景。队列的持久化很重要如果调度器重启后任务丢了整个系统就乱了。我一般用Redis的List或者RabbitMQ的Queue来做持久化。依赖解析。前面提到用DAG来管理依赖具体实现上可以用拓扑排序来确定执行顺序。每个任务节点记录它的前置任务列表调度器每次从入度为0的节点中选取任务执行。执行完成后更新后继节点的入度把新的入度为0的节点加入就绪队列。并发控制。多Agent系统通常需要并发执行多个任务来提高效率但并发度太高会导致资源竞争和API限流。我一般会设置一个最大并发数用信号量或者线程池来控制。同时对于有依赖关系的任务要保证它们不会并发执行。状态追踪。每个任务的状态等待、执行中、成功、失败、重试中需要实时记录方便监控和调试。我通常会把状态写到数据库或者Redis里同时提供一个查询接口方便排查问题。2.4 大模型选型与接入多Agent系统里不同Agent对模型能力的要求是不一样的。协调者Agent需要强推理能力来做任务分解和决策执行者Agent可能只需要基础的文本生成能力。全部用同一个大模型当然可以但成本和效率上不划算。我的做法是分层选型。协调者用能力最强的模型比如GPT-4或者Claude 3 Opus这个级别的执行者根据任务类型选择性价比更高的模型。比如代码生成用专门的代码模型文本摘要用轻量级的模型。如果对数据隐私有要求可以考虑本地部署的开源模型比如Qwen2.5-7B或者Llama 3系列用Ollama或者vLLM来部署。接入方式上我一般会封装一个统一的模型调用层屏蔽不同模型API的差异。这个调用层负责处理认证、重试、限流、日志记录等通用逻辑。Agent只需要调用model.generate(prompt)这样的接口不需要关心底层用的是哪个模型。这里有个细节要注意不同模型的提示词格式和参数不一样比如有些模型对system message的支持更好有些模型对temperature更敏感。统一调用层里最好能根据模型类型自动适配这些参数减少Agent开发者的心智负担。3. 完整构建一个多Agent协同任务系统3.1 场景定义与任务拆解为了把上面的理论落地我以一个实际做过的项目为例完整走一遍构建过程。这个项目的需求是给定一个技术主题自动生成一篇结构完整、内容准确的技术博客文章。这个任务看起来简单但拆开来看涉及多个子任务主题分析、大纲生成、资料检索、段落撰写、事实核查、风格润色、格式排版。如果用一个Agent从头做到尾上下文会非常长而且质量很难保证。用多Agent来做每个Agent专注一个环节质量可控得多。我的任务拆解方案是这样的协调者Agent负责接收主题拆解任务调度其他Agent汇总最终结果。大纲生成Agent根据主题生成文章大纲包括章节标题和每章要点。资料检索Agent根据大纲中的要点检索相关技术资料和案例。段落撰写Agent根据大纲和资料逐段生成文章内容。事实核查Agent检查生成内容中的技术事实是否准确。风格润色Agent统一文章风格优化表达。格式排版Agent按照Markdown格式要求排版。这个拆解不是唯一的你可以根据实际需求调整。比如如果对事实准确性要求不高可以去掉事实核查Agent如果文章长度固定可以合并撰写和润色。3.2 环境准备与依赖安装我用的技术栈是Python 3.10 LangChain Redis OpenAI API。LangChain提供了Agent和工具的基础抽象Redis用来做消息队列和状态存储。如果你不想用LangChain自己手写也不是不行但会多花不少时间在基础设施上。先建一个虚拟环境把依赖装上python -m venv multi_agent_env source multi_agent_env/bin/activate # Windows用 multi_agent_env\Scripts\activate pip install langchain langchain-openai redis python-dotenv pydantic如果你打算用本地模型把langchain-openai换成langchain-community然后装ollama或者vllm的客户端库。环境变量文件.env里配置好API密钥和Redis连接信息OPENAI_API_KEYyour_key_here REDIS_HOSTlocalhost REDIS_PORT6379 REDIS_DB0Redis我用Docker起一个最简单的实例docker run -d --name multi-agent-redis -p 6379:6379 redis:7-alpine这个配置够用了生产环境的话需要加密码、持久化和集群。3.3 Agent基类的实现为了让不同Agent的代码结构统一我先定义一个Agent基类。这个基类封装了模型调用、消息收发、状态管理的通用逻辑。import json import time import redis from abc import ABC, abstractmethod from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage class BaseAgent(ABC): def __init__(self, name, role_prompt, model_namegpt-4, redis_clientNone): self.name name self.role_prompt role_prompt self.model ChatOpenAI(modelmodel_name, temperature0.3) self.redis redis_client or redis.Redis(hostlocalhost, port6379, db0) self.memory [] def _build_messages(self, user_input): messages [SystemMessage(contentself.role_prompt)] for item in self.memory[-5:]: # 只保留最近5轮记忆 messages.append(HumanMessage(contentitem)) messages.append(HumanMessage(contentuser_input)) return messages def think(self, user_input): messages self._build_messages(user_input) response self.model.invoke(messages) self.memory.append(user_input) self.memory.append(response.content) return response.content def send_message(self, channel, message): self.redis.publish(channel, json.dumps({ sender: self.name, timestamp: time.time(), content: message })) def receive_message(self, channel, timeout30): pubsub self.redis.pubsub() pubsub.subscribe(channel) start time.time() while time.time() - start timeout: msg pubsub.get_message(timeout1) if msg and msg[type] message: return json.loads(msg[data]) return None abstractmethod def execute(self, task): pass这个基类里think方法负责调用大模型send_message和receive_message负责Redis消息收发execute是抽象方法由子类实现具体的任务逻辑。记忆模块我做了简化只保留最近5轮避免上下文过长。3.4 协调者Agent的实现协调者Agent是整个系统的入口和调度中心。它接收用户输入的主题生成任务计划然后依次调度其他Agent执行。class CoordinatorAgent(BaseAgent): def __init__(self, redis_clientNone): role_prompt 你是一个技术博客写作团队的协调者。 你的职责是 1. 分析用户给出的技术主题 2. 制定写作计划确定需要哪些环节 3. 按顺序调度各个专家Agent 4. 汇总最终结果 你不需要自己写文章只需要做计划和调度。 输出格式必须是JSON包含steps字段每个step有agent和task两个属性。 super().__init__(Coordinator, role_prompt, gpt-4, redis_client) def execute(self, topic): plan_prompt f请为以下技术主题制定写作计划{topic} plan_json self.think(plan_prompt) try: plan json.loads(plan_json) except json.JSONDecodeError: # 模型有时候会输出markdown包裹的json做个清洗 cleaned plan_json.strip().strip().strip(json).strip() plan json.loads(cleaned) results {topic: topic, sections: []} for step in plan[steps]: agent_name step[agent] task step[task] self.send_message(fagent:{agent_name}, {task: task, context: results}) response self.receive_message(fresult:{agent_name}, timeout120) if response: results[sections].append({ agent: agent_name, output: response[content] }) return results这里有个细节协调者生成计划后通过Redis的发布/订阅把任务发给对应的Agent然后等待结果。超时时间设了120秒因为有些Agent的任务确实比较耗时。3.5 执行者Agent的实现执行者Agent负责具体的子任务。以大纲生成Agent为例class OutlineAgent(BaseAgent): def __init__(self, redis_clientNone): role_prompt 你是一个技术博客大纲生成专家。 根据给定的技术主题生成一份结构清晰的文章大纲。 大纲需要包含 - 文章标题 - 3到5个一级章节 - 每个章节下2到3个二级要点 - 每个要点的简要说明 输出格式为Markdown列表。 super().__init__(OutlineAgent, role_prompt, gpt-4, redis_client) def execute(self, task): topic task[task] outline self.think(f请为以下主题生成大纲{topic}) self.send_message(result:OutlineAgent, outline) return outline其他Agent的结构类似只是角色提示词和模型选择不同。比如事实核查Agent我会用temperature更低的模型减少随机性风格润色Agent可以用temperature稍高的模型增加表达的多样性。3.6 调度器的实现调度器负责启动所有Agent监听消息管理生命周期。我用一个简单的多线程模型来实现import threading class AgentScheduler: def __init__(self): self.redis redis.Redis(hostlocalhost, port6379, db0) self.agents {} self.threads [] def register(self, agent): self.agents[agent.name] agent def _agent_worker(self, agent): pubsub self.redis.pubsub() pubsub.subscribe(fagent:{agent.name}) for message in pubsub.listen(): if message[type] message: task json.loads(message[data]) try: agent.execute(task) except Exception as e: agent.send_message(fresult:{agent.name}, fERROR: {str(e)}) def start(self): for agent in self.agents.values(): t threading.Thread(targetself._agent_worker, args(agent,), daemonTrue) t.start() self.threads.append(t) def stop(self): for t in self.threads: t.join(timeout5)这个调度器很简陋但够用。生产环境的话需要考虑Agent的动态注册和注销、任务优先级、失败重试等。3.7 完整运行流程把所有代码串起来主程序大概长这样def main(): redis_client redis.Redis(hostlocalhost, port6379, db0) scheduler AgentScheduler() scheduler.register(OutlineAgent(redis_client)) scheduler.register(ResearchAgent(redis_client)) scheduler.register(WritingAgent(redis_client)) scheduler.register(FactCheckAgent(redis_client)) scheduler.register(PolishAgent(redis_client)) scheduler.start() coordinator CoordinatorAgent(redis_client) result coordinator.execute(大模型多Agent协作架构的设计与实现) print(json.dumps(result, ensure_asciiFalse, indent2)) if __name__ __main__: main()运行后协调者会先调用大纲Agent生成大纲然后根据大纲调用资料检索Agent再调用撰写Agent逐段生成内容最后调用事实核查和润色Agent。整个过程是自动的你只需要在最后拿到结果。4. 常见问题与排查技巧实录4.1 Agent之间消息丢失怎么办这是我在早期项目里遇到最多的问题。Redis的Pub/Sub是不保证消息可靠投递的如果订阅者当时不在线消息就丢了。表现就是某个Agent一直等不到上游的消息整个流程卡住。解决方案有两个方向。一是改用可靠的消息队列比如RabbitMQ或者Redis的Stream数据结构它们支持消息持久化和确认机制。二是加一层应用层的确认和重试发送方发出消息后等待接收方的ACK超时没收到就重发。我一般用第二种因为改动小不需要引入新的中间件。具体实现上我在消息体里加了一个message_id字段接收方处理完后往一个确认频道发一条ACK消息发送方监听这个频道收到ACK就把消息标记为已送达超时没收到就重发重发超过3次就报错。4.2 大模型输出格式不稳定怎么处理多Agent系统里Agent之间的消息格式必须严格一致否则下游解析会出错。但大模型的输出天然是不稳定的你让它输出JSON它有时候会加markdown代码块标记有时候会加解释性文字有时候字段名拼错。我的处理策略是三层防护。第一层是在提示词里尽可能明确格式要求给出示例。第二层是在解析代码里做容错处理比如用正则提取JSON部分尝试多种解析方式。第三层是加一个格式校验Agent专门检查上游输出是否符合预期格式不符合就要求重新生成。实测下来第一层能解决80%的问题第二层能解决15%剩下5%需要第三层兜底。如果你用的是支持结构化输出的模型比如OpenAI的JSON mode可以直接开启这个功能能省不少事。4.3 Agent执行超时或卡死Agent卡死的原因通常有三种模型API调用超时、工具函数执行超时、死锁。模型API超时最好处理在调用层设置timeout参数超时后抛异常由调度器决定重试还是跳过。我一般设置30秒超时重试2次。工具函数超时要看具体工具比如搜索工具可能因为网络问题卡住文件读写可能因为磁盘IO卡住。我的做法是给每个工具函数也设置超时用Python的signal.alarm或者concurrent.futures来实现。死锁在多Agent系统里比较隐蔽通常是两个Agent互相等待对方的消息。比如A等B的结果B等A的结果谁都不动。预防死锁的办法是设计任务依赖时保证无环以及给每个等待操作设置超时。如果超时后还没等到就主动放弃并报错而不是无限等待。4.4 成本控制与性能优化多Agent系统调用大模型的次数远多于单Agent成本很容易失控。我做过一个统计同样一个任务单Agent方案调用模型1次多Agent方案可能调用5到10次。如果不做优化API账单会很难看。优化手段有几个。一是模型分层协调者用强模型执行者用弱模型能省不少钱。二是缓存对于相同的输入缓存模型的输出下次直接读缓存。三是并行化没有依赖关系的Agent任务可以并发执行缩短总耗时。四是精简上下文只把必要的信息传给下游Agent不要把所有历史都带上。我实测下来模型分层能省40%左右的成本缓存能省20%并行化能把总耗时缩短一半以上。这些优化加起来多Agent方案的成本可以控制在单Agent方案的2到3倍以内但质量提升是显著的。4.5 常见问题速查表问题现象可能原因排查方法解决方案Agent一直等待不执行消息丢失或订阅未生效检查Redis频道订阅状态改用可靠队列或加ACK机制下游Agent解析失败上游输出格式不符合预期打印上游原始输出加格式校验Agent或开启结构化输出系统整体卡死死锁或超时未处理查看各Agent状态和等待时间设置超时和重试检查依赖是否有环API成本过高模型调用次数过多统计各Agent调用次数和token消耗模型分层、缓存、精简上下文输出质量不稳定模型temperature过高对比不同temperature下的输出降低temperature或加后处理Agent越权操作角色边界不清晰检查Agent的工具集和提示词明确角色定义限制工具权限4.6 几个踩坑心得第一个心得是不要过早优化。我一开始就想着把系统做得特别完善加了各种重试、降级、监控结果开发周期拉得很长而且很多优化在实际运行中根本用不上。后来我改成先跑通最小可用版本再根据实际遇到的问题逐步优化效率高很多。第二个心得是日志要打全。多Agent系统的调试比单Agent难得多因为问题可能出在任何一个Agent或者它们之间的通信上。我的做法是每个Agent的输入输出都打日志消息的发送和接收也打日志日志里带上时间戳和消息ID方便追踪完整的调用链。第三个心得是给Agent起好名字。听起来是小事但当你有十几个Agent的时候名字起得清晰能省很多沟通成本。我一般用“功能Agent”的命名方式比如OutlineAgent、FactCheckAgent一看就知道是干什么的。第四个心得是定期做端到端测试。多Agent系统的行为是涌现的单元测试只能保证单个Agent没问题但Agent之间的交互可能出各种意外。我每周会跑一次完整的端到端测试用固定的输入验证输出是否符合预期及时发现回归问题。5. 协作架构的扩展与演进方向5.1 从固定流程到动态规划我上面展示的协调者Agent还是基于预定义的任务列表来调度的虽然比硬编码的流水线灵活但本质上还是“计划-执行”的模式。更高级的做法是让协调者具备动态规划能力根据执行过程中的中间结果实时调整后续任务。比如在写技术博客的场景里如果资料检索Agent发现某个主题的资料特别少协调者可以决定增加一个“资料补充Agent”去更广泛地搜索或者调整大纲把资料不足的章节合并或删除。这种动态调整能力需要协调者有更强的推理能力和更丰富的状态感知。实现上可以把协调者的决策逻辑从“一次性生成计划”改成“每执行完一个任务就重新评估状态并决定下一步”。这本质上是一个强化学习的问题但在实际工程中用大模型做few-shot推理就能达到不错的效果。5.2 Agent能力的动态发现与组合现在的系统里Agent的能力是预先定义好的协调者知道有哪些Agent可用。但在更开放的环境里Agent可能是动态加入和退出的协调者需要能够发现新Agent的能力并决定是否使用。这需要一套能力描述和注册机制。每个Agent注册时声明自己擅长什么任务、需要什么输入、产出什么输出。协调者根据任务需求匹配可用的Agent。这种机制在Agent数量多、能力重叠的场景下特别有用可以实现负载均衡和故障转移。5.3 多模态Agent的协作目前我做的项目主要是文本Agent之间的协作但多模态是明显的趋势。图像生成Agent、语音合成Agent、视频处理Agent都可以纳入协作框架。挑战在于不同模态的数据格式和传输方式差异很大消息协议需要扩展调度器需要处理更复杂的资源依赖。我的初步尝试是在消息体里加一个modality字段标识内容的类型然后为每种模态定义专门的序列化和反序列化方法。调度器层面对于需要GPU资源的Agent比如图像生成单独做一个资源池来管理。5.4 人机协作的混合模式完全自动化的多Agent系统在可预见的未来还很难达到人类专家的水平所以人机协作的混合模式是更现实的选择。在关键决策点引入人工审核比如大纲生成后让人确认事实核查发现问题后让人判断这样既能利用Agent的效率又能保证最终质量。实现上可以在调度器里加一个“人工审核”的任务类型执行到这个任务时暂停流程通过界面或者消息通知人工介入人工确认后再继续。这种模式在我做过的内容创作类项目里效果很好用户接受度也高。5.5 性能与可靠性的持续优化随着Agent数量增加和任务复杂度提升系统的性能和可靠性会面临更大挑战。我目前关注几个方向一是用异步IO替代多线程减少线程切换开销二是引入分布式调度把Agent部署到多台机器上三是做更精细的限流和熔断防止单个Agent的故障扩散到整个系统。这些优化不是一蹴而就的需要根据实际负载和故障模式逐步迭代。我的建议是先把单机版本做稳定再考虑分布式扩展不要一开始就追求大而全的架构。我在实际项目里最深的一个体会是多Agent系统的复杂度不是线性增长的而是指数增长的。每增加一个Agent可能的交互路径就多一倍调试难度也翻倍。所以控制Agent数量、简化交互关系比堆砌功能更重要。一个只有三个Agent但协作流畅的系统价值远大于一个有十个Agent但经常出错的系统。
多Agent协作架构实战：从设计到落地的完整指南
1. 多Agent协作架构到底在解决什么问题1.1 从单Agent的瓶颈说起过去一年我经手过不少大模型应用项目从最简单的问答机器人到稍微复杂一点的文档分析流水线单Agent架构能覆盖的场景其实比很多人想象的要窄。一个Agent加一套提示词再挂几个工具函数处理单一任务时确实够用但只要任务链条超过三步、涉及多个专业领域的知识交叉或者需要在执行过程中动态调整策略单Agent就开始力不从心了。最典型的表现是上下文窗口被撑爆。你让一个Agent既要做需求分析、又要写代码、还要做代码审查它得把前面所有步骤的完整上下文都带着走token消耗呈指数级增长而且随着上下文变长模型对早期信息的注意力会衰减后面生成的代码可能完全偏离最初的需求。另一个问题是角色冲突同一个模型实例很难同时扮演好“严格的代码审查者”和“富有创造力的方案设计者”这两个角色提示词里写再多“你现在是XX专家”也压不住模型底层的概率分布。多Agent协作架构就是冲着这些痛点来的。它的核心思路很朴素把一个大任务拆成若干个子任务每个子任务交给一个专门的Agent去处理Agent之间通过消息传递来协调。这跟人类团队协作的逻辑是一样的——你不会让一个人同时干产品经理、程序员和测试的活而是分给不同的人每个人专注自己擅长的部分通过会议、文档、即时消息来同步进度。1.2 协作架构的几种主流形态目前业界常见的多Agent协作架构大致可以归为三类我在不同项目里都实际用过各有各的适用场景。第一类是流水线式Pipeline。Agent按照预定义的顺序依次执行前一个Agent的输出是后一个Agent的输入。这种架构最简单、最可控适合流程固定的任务比如“需求解析→方案设计→代码生成→代码审查→测试用例生成”这种线性流程。缺点是缺乏灵活性如果中间某个环节发现问题需要回溯整个流水线就得重新跑。第二类是层级式Hierarchical。有一个协调者AgentCoordinator负责拆解任务、分配子任务、汇总结果下面挂若干个执行者AgentWorker。协调者不直接干活只做调度和决策。这种架构适合任务结构不太固定、需要动态调整的场景。我在一个科研论文辅助写作的项目里用过这种架构协调者根据论文的当前状态决定下一步是让“文献调研Agent”去补充资料还是让“数据分析Agent”去处理实验数据还是让“写作Agent”去生成段落。第三类是去中心化式Decentralized。没有明确的协调者Agent之间通过共享消息总线或者黑板Blackboard来通信每个Agent根据自己的能力决定是否响应某个消息。这种架构最灵活但也最难调试因为系统的行为是涌现出来的你很难预测最终会走到哪里。适合探索性任务比如创意生成、复杂问题求解。实际项目中我大多数时候用的是混合架构——顶层用层级式做任务分解和调度底层用流水线式保证执行效率关键节点上允许Agent之间点对点通信来做异常处理。1.3 任务调度的核心挑战多Agent系统里任务调度是最容易出问题的地方。单Agent的时候你只需要考虑“下一步做什么”多Agent的时候你要考虑“谁来做、什么时候做、做完了怎么通知别人、做错了怎么回滚”。我踩过的一个坑是任务依赖管理。早期我用简单的队列来调度Agent结果发现有些Agent在等上游输出的时候空转有些Agent在依赖还没就绪的时候就被触发了导致拿到脏数据。后来引入了DAG有向无环图来管理任务依赖每个节点是一个Agent任务边表示数据依赖关系只有入度为0的节点才能被调度执行。这个改动让系统的稳定性提升了一个档次。另一个坑是超时和重试策略。Agent调用大模型API是有延迟的有时候几秒有时候几十秒网络抖动或者模型服务限流的时候可能直接超时。如果没有合理的超时机制一个Agent卡住会导致整个系统挂起。我的做法是给每个Agent任务设置独立的超时时间超时后触发重试或者降级策略同时把失败信息写回调度器让调度器决定是重新分配任务还是跳过。还有一个容易被忽视的问题是Agent之间的消息格式。如果每个Agent都用自己的一套输入输出格式协调者就得写大量的适配代码。我的经验是尽早定义一套统一的消息协议至少包含任务ID、发送者、接收者、消息类型、负载内容、时间戳这几个字段后面扩展起来会轻松很多。2. 核心组件拆解与技术选型2.1 Agent的抽象设计在多Agent系统里每个Agent本质上是一个封装了特定能力、状态和行为的计算单元。我通常会把Agent抽象成四个核心部分角色定义Role、工具集Tools、记忆Memory和决策逻辑Policy。角色定义决定了Agent的“人设”和能力边界。比如一个“代码审查Agent”的角色定义里会明确它只负责审查代码质量、安全性和性能问题不负责修改代码。这个边界很重要没有边界的Agent会越权操作导致系统行为不可预测。工具集是Agent与外部世界交互的接口。一个Agent可以挂载多个工具函数比如搜索、计算、文件读写、API调用等。工具的设计要遵循单一职责原则一个工具只做一件事参数尽量简单。我见过有人把“搜索并总结”做成一个工具结果这个工具内部又调了一次大模型导致嵌套调用和延迟叠加调试起来非常痛苦。记忆模块负责存储Agent的历史交互信息。短期记忆通常就是当前会话的上下文长期记忆则需要外部存储支持比如向量数据库。在多Agent场景下记忆的共享和隔离是个关键设计决策——哪些信息所有Agent都能看到哪些信息只有特定Agent能访问这直接影响到系统的安全性和效率。决策逻辑是Agent的“大脑”决定它在当前状态下应该做什么。最简单的决策逻辑是基于规则的if-else复杂一点的是基于大模型的推理。我的经验是能用规则就用规则规则搞不定的再用模型推理因为模型推理的成本和不确定性都更高。2.2 通信机制的选择Agent之间的通信机制直接决定了系统的耦合度和可扩展性。我实践下来主要有三种方式直接函数调用、消息队列和共享状态。直接函数调用最简单Agent A直接调用Agent B的方法同步等待返回结果。这种方式适合Agent数量少、调用关系简单的场景缺点是耦合度高A必须知道B的存在和接口B挂了A也会挂。消息队列是更常用的方式Agent之间通过发布/订阅消息来通信发送方不需要知道接收方是谁。这种方式解耦彻底扩展性好但引入了消息中间件的运维成本而且消息的顺序和可靠性需要额外保证。我在一个多Agent内容生成的项目里用过Redis的Pub/Sub轻量够用但如果要保证消息不丢还是得上RabbitMQ或者Kafka。共享状态的方式是维护一个全局的状态存储比如Redis或者内存中的字典所有Agent都读写这个状态。这种方式适合需要频繁同步状态的场景但并发写的时候需要加锁否则会出现竞态条件。我一般只在Agent数量少、状态变更不频繁的时候用这种方式。实际项目中我倾向于组合使用核心的任务调度用消息队列Agent之间的实时状态同步用共享状态紧急的异常处理用直接调用。2.3 调度器的实现要点调度器是多Agent系统的心脏它的职责是接收任务、解析依赖、分配Agent、监控执行、处理异常。我实现过的调度器有简单的也有复杂的这里说几个关键设计点。任务队列的设计。我用过优先队列和普通队列优先队列适合有紧急任务插队的场景普通队列适合FIFO的批处理场景。队列的持久化很重要如果调度器重启后任务丢了整个系统就乱了。我一般用Redis的List或者RabbitMQ的Queue来做持久化。依赖解析。前面提到用DAG来管理依赖具体实现上可以用拓扑排序来确定执行顺序。每个任务节点记录它的前置任务列表调度器每次从入度为0的节点中选取任务执行。执行完成后更新后继节点的入度把新的入度为0的节点加入就绪队列。并发控制。多Agent系统通常需要并发执行多个任务来提高效率但并发度太高会导致资源竞争和API限流。我一般会设置一个最大并发数用信号量或者线程池来控制。同时对于有依赖关系的任务要保证它们不会并发执行。状态追踪。每个任务的状态等待、执行中、成功、失败、重试中需要实时记录方便监控和调试。我通常会把状态写到数据库或者Redis里同时提供一个查询接口方便排查问题。2.4 大模型选型与接入多Agent系统里不同Agent对模型能力的要求是不一样的。协调者Agent需要强推理能力来做任务分解和决策执行者Agent可能只需要基础的文本生成能力。全部用同一个大模型当然可以但成本和效率上不划算。我的做法是分层选型。协调者用能力最强的模型比如GPT-4或者Claude 3 Opus这个级别的执行者根据任务类型选择性价比更高的模型。比如代码生成用专门的代码模型文本摘要用轻量级的模型。如果对数据隐私有要求可以考虑本地部署的开源模型比如Qwen2.5-7B或者Llama 3系列用Ollama或者vLLM来部署。接入方式上我一般会封装一个统一的模型调用层屏蔽不同模型API的差异。这个调用层负责处理认证、重试、限流、日志记录等通用逻辑。Agent只需要调用model.generate(prompt)这样的接口不需要关心底层用的是哪个模型。这里有个细节要注意不同模型的提示词格式和参数不一样比如有些模型对system message的支持更好有些模型对temperature更敏感。统一调用层里最好能根据模型类型自动适配这些参数减少Agent开发者的心智负担。3. 完整构建一个多Agent协同任务系统3.1 场景定义与任务拆解为了把上面的理论落地我以一个实际做过的项目为例完整走一遍构建过程。这个项目的需求是给定一个技术主题自动生成一篇结构完整、内容准确的技术博客文章。这个任务看起来简单但拆开来看涉及多个子任务主题分析、大纲生成、资料检索、段落撰写、事实核查、风格润色、格式排版。如果用一个Agent从头做到尾上下文会非常长而且质量很难保证。用多Agent来做每个Agent专注一个环节质量可控得多。我的任务拆解方案是这样的协调者Agent负责接收主题拆解任务调度其他Agent汇总最终结果。大纲生成Agent根据主题生成文章大纲包括章节标题和每章要点。资料检索Agent根据大纲中的要点检索相关技术资料和案例。段落撰写Agent根据大纲和资料逐段生成文章内容。事实核查Agent检查生成内容中的技术事实是否准确。风格润色Agent统一文章风格优化表达。格式排版Agent按照Markdown格式要求排版。这个拆解不是唯一的你可以根据实际需求调整。比如如果对事实准确性要求不高可以去掉事实核查Agent如果文章长度固定可以合并撰写和润色。3.2 环境准备与依赖安装我用的技术栈是Python 3.10 LangChain Redis OpenAI API。LangChain提供了Agent和工具的基础抽象Redis用来做消息队列和状态存储。如果你不想用LangChain自己手写也不是不行但会多花不少时间在基础设施上。先建一个虚拟环境把依赖装上python -m venv multi_agent_env source multi_agent_env/bin/activate # Windows用 multi_agent_env\Scripts\activate pip install langchain langchain-openai redis python-dotenv pydantic如果你打算用本地模型把langchain-openai换成langchain-community然后装ollama或者vllm的客户端库。环境变量文件.env里配置好API密钥和Redis连接信息OPENAI_API_KEYyour_key_here REDIS_HOSTlocalhost REDIS_PORT6379 REDIS_DB0Redis我用Docker起一个最简单的实例docker run -d --name multi-agent-redis -p 6379:6379 redis:7-alpine这个配置够用了生产环境的话需要加密码、持久化和集群。3.3 Agent基类的实现为了让不同Agent的代码结构统一我先定义一个Agent基类。这个基类封装了模型调用、消息收发、状态管理的通用逻辑。import json import time import redis from abc import ABC, abstractmethod from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage class BaseAgent(ABC): def __init__(self, name, role_prompt, model_namegpt-4, redis_clientNone): self.name name self.role_prompt role_prompt self.model ChatOpenAI(modelmodel_name, temperature0.3) self.redis redis_client or redis.Redis(hostlocalhost, port6379, db0) self.memory [] def _build_messages(self, user_input): messages [SystemMessage(contentself.role_prompt)] for item in self.memory[-5:]: # 只保留最近5轮记忆 messages.append(HumanMessage(contentitem)) messages.append(HumanMessage(contentuser_input)) return messages def think(self, user_input): messages self._build_messages(user_input) response self.model.invoke(messages) self.memory.append(user_input) self.memory.append(response.content) return response.content def send_message(self, channel, message): self.redis.publish(channel, json.dumps({ sender: self.name, timestamp: time.time(), content: message })) def receive_message(self, channel, timeout30): pubsub self.redis.pubsub() pubsub.subscribe(channel) start time.time() while time.time() - start timeout: msg pubsub.get_message(timeout1) if msg and msg[type] message: return json.loads(msg[data]) return None abstractmethod def execute(self, task): pass这个基类里think方法负责调用大模型send_message和receive_message负责Redis消息收发execute是抽象方法由子类实现具体的任务逻辑。记忆模块我做了简化只保留最近5轮避免上下文过长。3.4 协调者Agent的实现协调者Agent是整个系统的入口和调度中心。它接收用户输入的主题生成任务计划然后依次调度其他Agent执行。class CoordinatorAgent(BaseAgent): def __init__(self, redis_clientNone): role_prompt 你是一个技术博客写作团队的协调者。 你的职责是 1. 分析用户给出的技术主题 2. 制定写作计划确定需要哪些环节 3. 按顺序调度各个专家Agent 4. 汇总最终结果 你不需要自己写文章只需要做计划和调度。 输出格式必须是JSON包含steps字段每个step有agent和task两个属性。 super().__init__(Coordinator, role_prompt, gpt-4, redis_client) def execute(self, topic): plan_prompt f请为以下技术主题制定写作计划{topic} plan_json self.think(plan_prompt) try: plan json.loads(plan_json) except json.JSONDecodeError: # 模型有时候会输出markdown包裹的json做个清洗 cleaned plan_json.strip().strip().strip(json).strip() plan json.loads(cleaned) results {topic: topic, sections: []} for step in plan[steps]: agent_name step[agent] task step[task] self.send_message(fagent:{agent_name}, {task: task, context: results}) response self.receive_message(fresult:{agent_name}, timeout120) if response: results[sections].append({ agent: agent_name, output: response[content] }) return results这里有个细节协调者生成计划后通过Redis的发布/订阅把任务发给对应的Agent然后等待结果。超时时间设了120秒因为有些Agent的任务确实比较耗时。3.5 执行者Agent的实现执行者Agent负责具体的子任务。以大纲生成Agent为例class OutlineAgent(BaseAgent): def __init__(self, redis_clientNone): role_prompt 你是一个技术博客大纲生成专家。 根据给定的技术主题生成一份结构清晰的文章大纲。 大纲需要包含 - 文章标题 - 3到5个一级章节 - 每个章节下2到3个二级要点 - 每个要点的简要说明 输出格式为Markdown列表。 super().__init__(OutlineAgent, role_prompt, gpt-4, redis_client) def execute(self, task): topic task[task] outline self.think(f请为以下主题生成大纲{topic}) self.send_message(result:OutlineAgent, outline) return outline其他Agent的结构类似只是角色提示词和模型选择不同。比如事实核查Agent我会用temperature更低的模型减少随机性风格润色Agent可以用temperature稍高的模型增加表达的多样性。3.6 调度器的实现调度器负责启动所有Agent监听消息管理生命周期。我用一个简单的多线程模型来实现import threading class AgentScheduler: def __init__(self): self.redis redis.Redis(hostlocalhost, port6379, db0) self.agents {} self.threads [] def register(self, agent): self.agents[agent.name] agent def _agent_worker(self, agent): pubsub self.redis.pubsub() pubsub.subscribe(fagent:{agent.name}) for message in pubsub.listen(): if message[type] message: task json.loads(message[data]) try: agent.execute(task) except Exception as e: agent.send_message(fresult:{agent.name}, fERROR: {str(e)}) def start(self): for agent in self.agents.values(): t threading.Thread(targetself._agent_worker, args(agent,), daemonTrue) t.start() self.threads.append(t) def stop(self): for t in self.threads: t.join(timeout5)这个调度器很简陋但够用。生产环境的话需要考虑Agent的动态注册和注销、任务优先级、失败重试等。3.7 完整运行流程把所有代码串起来主程序大概长这样def main(): redis_client redis.Redis(hostlocalhost, port6379, db0) scheduler AgentScheduler() scheduler.register(OutlineAgent(redis_client)) scheduler.register(ResearchAgent(redis_client)) scheduler.register(WritingAgent(redis_client)) scheduler.register(FactCheckAgent(redis_client)) scheduler.register(PolishAgent(redis_client)) scheduler.start() coordinator CoordinatorAgent(redis_client) result coordinator.execute(大模型多Agent协作架构的设计与实现) print(json.dumps(result, ensure_asciiFalse, indent2)) if __name__ __main__: main()运行后协调者会先调用大纲Agent生成大纲然后根据大纲调用资料检索Agent再调用撰写Agent逐段生成内容最后调用事实核查和润色Agent。整个过程是自动的你只需要在最后拿到结果。4. 常见问题与排查技巧实录4.1 Agent之间消息丢失怎么办这是我在早期项目里遇到最多的问题。Redis的Pub/Sub是不保证消息可靠投递的如果订阅者当时不在线消息就丢了。表现就是某个Agent一直等不到上游的消息整个流程卡住。解决方案有两个方向。一是改用可靠的消息队列比如RabbitMQ或者Redis的Stream数据结构它们支持消息持久化和确认机制。二是加一层应用层的确认和重试发送方发出消息后等待接收方的ACK超时没收到就重发。我一般用第二种因为改动小不需要引入新的中间件。具体实现上我在消息体里加了一个message_id字段接收方处理完后往一个确认频道发一条ACK消息发送方监听这个频道收到ACK就把消息标记为已送达超时没收到就重发重发超过3次就报错。4.2 大模型输出格式不稳定怎么处理多Agent系统里Agent之间的消息格式必须严格一致否则下游解析会出错。但大模型的输出天然是不稳定的你让它输出JSON它有时候会加markdown代码块标记有时候会加解释性文字有时候字段名拼错。我的处理策略是三层防护。第一层是在提示词里尽可能明确格式要求给出示例。第二层是在解析代码里做容错处理比如用正则提取JSON部分尝试多种解析方式。第三层是加一个格式校验Agent专门检查上游输出是否符合预期格式不符合就要求重新生成。实测下来第一层能解决80%的问题第二层能解决15%剩下5%需要第三层兜底。如果你用的是支持结构化输出的模型比如OpenAI的JSON mode可以直接开启这个功能能省不少事。4.3 Agent执行超时或卡死Agent卡死的原因通常有三种模型API调用超时、工具函数执行超时、死锁。模型API超时最好处理在调用层设置timeout参数超时后抛异常由调度器决定重试还是跳过。我一般设置30秒超时重试2次。工具函数超时要看具体工具比如搜索工具可能因为网络问题卡住文件读写可能因为磁盘IO卡住。我的做法是给每个工具函数也设置超时用Python的signal.alarm或者concurrent.futures来实现。死锁在多Agent系统里比较隐蔽通常是两个Agent互相等待对方的消息。比如A等B的结果B等A的结果谁都不动。预防死锁的办法是设计任务依赖时保证无环以及给每个等待操作设置超时。如果超时后还没等到就主动放弃并报错而不是无限等待。4.4 成本控制与性能优化多Agent系统调用大模型的次数远多于单Agent成本很容易失控。我做过一个统计同样一个任务单Agent方案调用模型1次多Agent方案可能调用5到10次。如果不做优化API账单会很难看。优化手段有几个。一是模型分层协调者用强模型执行者用弱模型能省不少钱。二是缓存对于相同的输入缓存模型的输出下次直接读缓存。三是并行化没有依赖关系的Agent任务可以并发执行缩短总耗时。四是精简上下文只把必要的信息传给下游Agent不要把所有历史都带上。我实测下来模型分层能省40%左右的成本缓存能省20%并行化能把总耗时缩短一半以上。这些优化加起来多Agent方案的成本可以控制在单Agent方案的2到3倍以内但质量提升是显著的。4.5 常见问题速查表问题现象可能原因排查方法解决方案Agent一直等待不执行消息丢失或订阅未生效检查Redis频道订阅状态改用可靠队列或加ACK机制下游Agent解析失败上游输出格式不符合预期打印上游原始输出加格式校验Agent或开启结构化输出系统整体卡死死锁或超时未处理查看各Agent状态和等待时间设置超时和重试检查依赖是否有环API成本过高模型调用次数过多统计各Agent调用次数和token消耗模型分层、缓存、精简上下文输出质量不稳定模型temperature过高对比不同temperature下的输出降低temperature或加后处理Agent越权操作角色边界不清晰检查Agent的工具集和提示词明确角色定义限制工具权限4.6 几个踩坑心得第一个心得是不要过早优化。我一开始就想着把系统做得特别完善加了各种重试、降级、监控结果开发周期拉得很长而且很多优化在实际运行中根本用不上。后来我改成先跑通最小可用版本再根据实际遇到的问题逐步优化效率高很多。第二个心得是日志要打全。多Agent系统的调试比单Agent难得多因为问题可能出在任何一个Agent或者它们之间的通信上。我的做法是每个Agent的输入输出都打日志消息的发送和接收也打日志日志里带上时间戳和消息ID方便追踪完整的调用链。第三个心得是给Agent起好名字。听起来是小事但当你有十几个Agent的时候名字起得清晰能省很多沟通成本。我一般用“功能Agent”的命名方式比如OutlineAgent、FactCheckAgent一看就知道是干什么的。第四个心得是定期做端到端测试。多Agent系统的行为是涌现的单元测试只能保证单个Agent没问题但Agent之间的交互可能出各种意外。我每周会跑一次完整的端到端测试用固定的输入验证输出是否符合预期及时发现回归问题。5. 协作架构的扩展与演进方向5.1 从固定流程到动态规划我上面展示的协调者Agent还是基于预定义的任务列表来调度的虽然比硬编码的流水线灵活但本质上还是“计划-执行”的模式。更高级的做法是让协调者具备动态规划能力根据执行过程中的中间结果实时调整后续任务。比如在写技术博客的场景里如果资料检索Agent发现某个主题的资料特别少协调者可以决定增加一个“资料补充Agent”去更广泛地搜索或者调整大纲把资料不足的章节合并或删除。这种动态调整能力需要协调者有更强的推理能力和更丰富的状态感知。实现上可以把协调者的决策逻辑从“一次性生成计划”改成“每执行完一个任务就重新评估状态并决定下一步”。这本质上是一个强化学习的问题但在实际工程中用大模型做few-shot推理就能达到不错的效果。5.2 Agent能力的动态发现与组合现在的系统里Agent的能力是预先定义好的协调者知道有哪些Agent可用。但在更开放的环境里Agent可能是动态加入和退出的协调者需要能够发现新Agent的能力并决定是否使用。这需要一套能力描述和注册机制。每个Agent注册时声明自己擅长什么任务、需要什么输入、产出什么输出。协调者根据任务需求匹配可用的Agent。这种机制在Agent数量多、能力重叠的场景下特别有用可以实现负载均衡和故障转移。5.3 多模态Agent的协作目前我做的项目主要是文本Agent之间的协作但多模态是明显的趋势。图像生成Agent、语音合成Agent、视频处理Agent都可以纳入协作框架。挑战在于不同模态的数据格式和传输方式差异很大消息协议需要扩展调度器需要处理更复杂的资源依赖。我的初步尝试是在消息体里加一个modality字段标识内容的类型然后为每种模态定义专门的序列化和反序列化方法。调度器层面对于需要GPU资源的Agent比如图像生成单独做一个资源池来管理。5.4 人机协作的混合模式完全自动化的多Agent系统在可预见的未来还很难达到人类专家的水平所以人机协作的混合模式是更现实的选择。在关键决策点引入人工审核比如大纲生成后让人确认事实核查发现问题后让人判断这样既能利用Agent的效率又能保证最终质量。实现上可以在调度器里加一个“人工审核”的任务类型执行到这个任务时暂停流程通过界面或者消息通知人工介入人工确认后再继续。这种模式在我做过的内容创作类项目里效果很好用户接受度也高。5.5 性能与可靠性的持续优化随着Agent数量增加和任务复杂度提升系统的性能和可靠性会面临更大挑战。我目前关注几个方向一是用异步IO替代多线程减少线程切换开销二是引入分布式调度把Agent部署到多台机器上三是做更精细的限流和熔断防止单个Agent的故障扩散到整个系统。这些优化不是一蹴而就的需要根据实际负载和故障模式逐步迭代。我的建议是先把单机版本做稳定再考虑分布式扩展不要一开始就追求大而全的架构。我在实际项目里最深的一个体会是多Agent系统的复杂度不是线性增长的而是指数增长的。每增加一个Agent可能的交互路径就多一倍调试难度也翻倍。所以控制Agent数量、简化交互关系比堆砌功能更重要。一个只有三个Agent但协作流畅的系统价值远大于一个有十个Agent但经常出错的系统。