1. 从单兵作战到团队协作多智能体系统到底在解决什么问题很多人第一次接触 Agent 开发都是从让大模型帮我干一件事开始的。写个提示词调个接口模型返回结果任务结束。这种单 Agent 模式在简单场景下确实够用但一旦任务链条变长、涉及多个专业领域、需要反复校验和迭代单 Agent 的短板就暴露得非常明显——上下文窗口塞不下、职责边界模糊、一个环节出错整条链路崩盘。我最初做 Agent 项目时也踩过这个坑。当时接了一个需求要让系统自动完成读取需求文档→拆解任务→生成代码→自测→输出报告这一整套流程。用单 Agent 硬扛的结果是提示词写到三千多字模型开始遗忘前面的指令代码生成环节和自测环节互相干扰生成的代码还没跑就先被自测 Agent否定掉了。后来拆成四个独立 Agent每个只负责一件事通过消息传递串联稳定性直接上了一个台阶。这就是多智能体系统Multi-Agent System的核心价值把复杂任务拆解成多个职责单一的智能体通过协同机制让它们像团队一样工作。每个 Agent 有自己的角色定位、工具集和记忆空间彼此之间通过消息总线或共享状态进行通信。这样做的好处不只是分工明确更重要的是每个 Agent 的上下文窗口可以专注于自己的领域推理质量显著提升。但多智能体不是银弹。我见过不少团队一上来就搞五六个 Agent结果通信开销比任务本身还大调试起来像在追一群乱跑的猫。所以这篇文章不会只讲怎么搭多智能体而是从单 Agent 的边界讲起逐步过渡到多引擎同步优化的完整方案把每一步的选型理由、踩坑经验和实操细节都摊开来说。适合谁看如果你已经写过至少一个能跑通的 Agent Demo对提示词工程、工具调用、上下文管理有基本概念那这篇内容能帮你少走至少三个月的弯路。如果你是完全的新手建议先把单 Agent 的感知-决策-执行循环跑通再回来否则多智能体的复杂度会让你怀疑人生。2. 单 Agent 的能力天花板与多智能体的拆分时机2.1 单 Agent 什么时候开始扛不住判断一个任务该不该拆成多 Agent我总结了一个简单的经验法则当你的系统提示词超过 1500 字或者单次任务需要调用超过 5 个不同工具或者任务流程中存在明显的阶段切换时就该考虑拆分了。具体来说单 Agent 的瓶颈主要体现在三个维度上下文污染。一个 Agent 既要理解用户意图又要记住历史对话还要携带工具返回结果上下文窗口很快就被塞满。更麻烦的是不同阶段的信息会互相干扰——比如代码生成阶段的历史记录会污染测试阶段的判断导致 Agent 在测试时还在想着怎么生成代码。职责冲突。让同一个 Agent 既当创作者又当审核者它在生成内容时会不自觉地放水因为审核标准是自己定的很难做到客观。这就像让同一个人既写代码又做 Code Review效果通常不如交叉审核。错误传播。单 Agent 一旦在某个环节产生幻觉后续所有步骤都会基于错误前提继续推进而且很难定位问题出在哪一步。多 Agent 架构下每个环节的输出都是显式的消息排查时可以直接看到是哪一环出了问题。2.2 拆分粒度的权衡拆太细和拆太粗都是坑拆分粒度是多智能体设计中最容易翻车的地方。我见过两种极端一种是拆得太细一个任务拆出十几个 Agent每个只做一件微不足道的小事。结果是 Agent 之间的通信成本远超任务本身而且每个 Agent 的上下文都太窄缺乏全局视野做出的决策经常互相矛盾。比如一个格式化 Agent只负责把 JSON 转成 Markdown这种工作完全没必要独立成 Agent写个工具函数就够了。另一种是拆得太粗只分了规划 Agent和执行 Agent两个角色。结果执行 Agent 又变成了一个什么都干的单 Agent只是换了个名字而已。我的经验是按决策边界拆分而不是按步骤拆分。所谓决策边界是指一个 Agent 需要独立做出判断的地方。比如需求分析和代码生成是两个不同的决策边界因为前者需要理解业务语义后者需要遵循编程规范两者的知识体系和判断标准完全不同。而生成代码和保存代码到文件就不需要拆因为后者没有决策只是执行动作。一个实用的拆分检查清单判断维度该拆不该拆是否需要独立判断需要不同领域知识做决策纯执行动作无判断上下文是否冲突两个阶段信息互相干扰信息可以共享是否需要独立工具集工具集完全不重叠工具集高度重叠错误是否需要隔离一个环节出错不应影响其他错误可以统一处理是否需要独立记忆需要长期记忆特定领域知识无需记忆或记忆可共享2.3 多智能体协同的三种基础拓扑拆完之后Agent 之间怎么连接常见的有三种拓扑结构各有适用场景。流水线式PipelineAgent 按顺序串联前一个的输出是后一个的输入。适合流程固定、阶段明确的任务比如需求→设计→开发→测试。优点是逻辑清晰、易于调试缺点是缺乏反馈回路一旦某个环节出错后面全错。中心辐射式Hub-Spoke有一个协调者 Agent负责任务分发和结果汇总其他 Agent 只和协调者通信。适合任务可以并行拆分的场景比如同时让多个 Agent 从不同角度分析同一份数据。优点是扩展性好缺点是协调者容易成为瓶颈。黑板式Blackboard所有 Agent 共享一个黑板共享状态空间每个 Agent 可以读取黑板上的信息也可以写入自己的结果。适合需要频繁信息交换的协作场景比如多 Agent 联合调试。优点是灵活性高缺点是状态管理复杂容易出现竞态条件。实际项目中这三种拓扑经常混合使用。比如我最近做的一个内容生产系统顶层是中心辐射式一个调度 Agent 分配任务每个子任务内部是流水线式研究→撰写→审核审核环节又用了黑板式多个审核 Agent 共享一份稿件各自标注问题。3. 多引擎同步优化的核心机制让多个 Agent 真正同步起来3.1 什么是多引擎为什么要同步优化这里的多引擎指的是系统中同时运行的多个 Agent 实例或多种推理引擎。在实际生产环境中一个多智能体系统往往不是所有 Agent 都用同一个模型、同一套配置。比如规划类 Agent 需要强推理能力用高参数量的模型执行类 Agent 需要快速响应用轻量模型审核类 Agent 需要高准确性可能需要多次采样投票工具调用类 Agent 需要低延迟可能用本地小模型。这些异构的引擎如果各自为政就会出现节奏不一致的问题规划 Agent 还在思考执行 Agent 已经等不及开始瞎猜审核 Agent 还没返回结果流水线已经推进到下一步。更严重的是不同引擎对同一份上下文的理解可能不一致导致协同失败。同步优化要解决的就是这个问题让多个异构引擎在时间维度、状态维度和语义维度上保持一致。3.2 时间同步用事件驱动替代轮询最朴素的同步方式是轮询——每个 Agent 定期检查依赖的上游是否完成。这种方式实现简单但延迟高、资源浪费大。我早期的一个项目就是用轮询结果系统空转率高达 40%大部分时间都在等。后来改成了事件驱动架构每个 Agent 完成自己的任务后向消息总线发布一个事件订阅了该事件的 Agent 被唤醒执行。这样既降低了延迟又节省了资源。具体实现上我用的是一个轻量级的发布-订阅模式。核心数据结构是一个事件队列每个事件包含event_type、payload、source_agent和timestamp。Agent 注册自己关心的事件类型事件到达时触发回调。class EventBus: def __init__(self): self.subscribers {} # event_type - [callback] self.event_log [] # 用于调试和回放 def subscribe(self, event_type, callback): self.subscribers.setdefault(event_type, []).append(callback) def publish(self, event_type, payload, source_agent): event { type: event_type, payload: payload, source: source_agent, timestamp: time.time() } self.event_log.append(event) for callback in self.subscribers.get(event_type, []): callback(event)注意事件驱动虽然好但要小心事件风暴——如果 Agent 之间频繁互相触发可能导致系统过载。我的做法是给每个事件类型设置一个速率限制超过阈值就降级为批量处理。3.3 状态同步共享状态与私有状态的边界多 Agent 系统中哪些状态该共享哪些该私有是个需要仔细设计的问题。我的原则是共享事实私有推理。共享状态包括任务的整体进度、已确认的中间结果、全局约束条件。这些信息所有 Agent 都需要知道而且必须保持一致。私有状态包括每个 Agent 的推理过程、临时变量、内部草稿。这些信息如果共享反而会造成上下文污染。实现上我用一个共享状态存储可以是内存字典也可以是 Redis来管理共享状态每个 Agent 有自己的私有上下文。Agent 在需要时读取共享状态在产生确定结果时写入共享状态。class SharedState: def __init__(self): self._state {} self._version {} # 用于乐观锁 def read(self, key): return self._state.get(key) def write(self, key, value, expected_versionNone): if expected_version is not None: if self._version.get(key) ! expected_version: raise ConflictError(fState {key} has been modified) self._state[key] value self._version[key] self._version.get(key, 0) 1版本号机制是为了处理并发写入冲突。当两个 Agent 同时想更新同一个状态时后写入的会失败需要重新读取最新状态再决定是否重试。这在多 Agent 并行工作时非常关键。3.4 语义同步让不同引擎说同一种语言这是最容易被忽视但最致命的一环。不同模型对同一份提示词的理解可能不同输出的格式也可能不一致。比如规划 Agent 输出的是自然语言描述的任务列表执行 Agent 期望的是结构化 JSON两者对不上系统就卡住了。解决方法是定义统一的中间表示Intermediate Representation。所有 Agent 之间的通信都通过这个中间表示进行每个 Agent 负责把自己的输出转换成中间表示也负责把接收到的中间表示转换成自己理解的格式。我通常用 JSON Schema 来定义中间表示并在每个 Agent 的提示词中明确要求输出符合该 Schema。同时在 Agent 之间加一层适配器负责格式校验和转换。TASK_SCHEMA { type: object, properties: { task_id: {type: string}, description: {type: string}, dependencies: {type: array, items: {type: string}}, expected_output: {type: string}, priority: {type: integer, minimum: 1, maximum: 5} }, required: [task_id, description, expected_output] } def validate_and_adapt(raw_output, schema): try: data json.loads(raw_output) jsonschema.validate(data, schema) return data except (json.JSONDecodeError, jsonschema.ValidationError) as e: # 触发修复流程让原 Agent 重新生成或调用修复 Agent return repair_output(raw_output, schema, str(e))提示适配器层不要做得太聪明。我见过有人让适配器自动猜测意图并补全缺失字段结果引入了难以追踪的隐性错误。适配器只做校验和格式转换不做语义推断。校验失败就退回给原 Agent 重做这样问题定位清晰。4. 从零搭建多智能体协同系统的完整实操4.1 环境准备与基础框架选型动手之前先确定技术栈。多智能体框架这两年冒出来很多我实际用过的有 AutoGen、CrewAI、LangGraph 这几个。选哪个取决于你的需求框架核心特点适合场景上手难度AutoGen对话驱动Agent 之间通过消息交互研究型项目、需要灵活对话中CrewAI角色驱动强调 Agent 的角色和任务分配流程化任务、团队协作模拟低LangGraph图驱动用状态图定义 Agent 流转复杂流程、需要精细控制高自研完全可控按需定制生产环境、有特殊需求高我的建议是原型阶段用 CrewAI 快速验证生产阶段考虑 LangGraph 或自研。CrewAI 的角色抽象很直观几行代码就能跑起来一个多 Agent 系统但它的流程控制能力有限复杂场景下会捉襟见肘。LangGraph 用状态图的方式定义流程控制力强但学习曲线陡峭。如果你决定自研核心需要实现三个模块Agent 基类、消息总线、状态管理器。Agent 基类封装模型调用、工具执行和上下文管理消息总线负责 Agent 间通信状态管理器维护共享状态。class BaseAgent: def __init__(self, name, role, model, tools, system_prompt): self.name name self.role role self.model model self.tools {t.name: t for t in tools} self.system_prompt system_prompt self.private_context [] def think(self, input_message): 核心推理循环构建提示词 - 调用模型 - 解析输出 - 执行工具 messages self._build_messages(input_message) response self.model.chat(messages) return self._parse_response(response) def _build_messages(self, input_message): messages [{role: system, content: self.system_prompt}] messages.extend(self.private_context[-10:]) # 保留最近10轮 messages.append({role: user, content: input_message}) return messages4.2 定义 Agent 角色与职责边界角色定义是多智能体系统设计的核心。一个好的角色定义应该包含四个要素身份、目标、约束、输出格式。以我做过的一个技术文档生成系统为例四个 Agent 的角色定义如下研究员 Agent身份是资深技术调研员目标是从给定资料中提取关键信息约束是只陈述事实不做推断输出格式是结构化的信息点列表。架构师 Agent身份是系统架构师目标是根据信息点设计文档结构约束是结构必须符合技术文档规范输出格式是带层级的目录树。撰写者 Agent身份是技术写作者目标是根据结构填充内容约束是语言通俗、示例充分输出格式是 Markdown 正文。审核者 Agent身份是严格的技术编辑目标是找出内容中的错误和遗漏约束是必须给出具体的修改建议输出格式是问题列表加修改建议。每个角色的提示词都严格围绕这四个要素展开避免职责重叠。我踩过的一个坑是研究员和架构师的职责边界没划清研究员在提取信息时就开始设计结构导致架构师拿到的是已经被加工过的信息反而丢失了原始细节。4.3 消息传递与任务编排的实现细节Agent 之间的消息传递有两种模式直接传递和通过总线传递。直接传递适合简单的流水线A 直接调用 B 的方法通过总线传递适合复杂的多对多通信。我推荐统一用总线传递即使当前只有两个 Agent。原因是总线模式天然支持扩展——以后加新 Agent 只需要订阅事件不需要修改现有代码。而且总线可以记录所有消息方便调试和回放。任务编排的核心是依赖管理。一个任务可能依赖多个上游任务的输出必须等所有依赖都完成才能开始。我用一个有向无环图DAG来表示任务依赖关系用拓扑排序确定执行顺序。class TaskOrchestrator: def __init__(self, event_bus): self.event_bus event_bus self.tasks {} # task_id - task_info self.completed set() def add_task(self, task_id, agent_name, dependencies, input_builder): self.tasks[task_id] { agent: agent_name, deps: dependencies, input_builder: input_builder, status: pending } def try_execute(self, task_id): task self.tasks[task_id] if task[status] ! pending: return if not all(d in self.completed for d in task[deps]): return # 依赖未满足 # 构建输入并触发 Agent input_data task[input_builder](self.completed) self.event_bus.publish(fexecute:{task[agent]}, input_data, orchestrator) task[status] running def on_task_complete(self, task_id, result): self.completed.add(task_id) self.tasks[task_id][status] done # 检查哪些任务现在可以执行了 for tid, task in self.tasks.items(): if task[status] pending: self.try_execute(tid)注意DAG 中要避免循环依赖。我在代码里加了一个启动时的环检测如果发现环就直接报错不要等到运行时才发现死锁。4.4 同步优化的三个实操技巧技巧一用心跳超时处理慢 Agent。多引擎环境下不同 Agent 的响应速度差异很大。如果一个 Agent 超过预期时间还没返回不能无限等待。我的做法是给每个 Agent 设置超时时间超时后触发降级策略——要么用备用模型重试要么跳过该步骤并标记为待人工处理。技巧二用检查点实现断点续跑。多 Agent 流程跑一半失败是常事。如果没有检查点机制只能从头再来浪费大量 token 和时间。我在每个关键节点保存状态快照失败后可以从最近的检查点恢复只重跑失败的部分。技巧三用影子模式做灰度验证。当你想替换某个 Agent 的模型或提示词时不要直接上线。先让新旧两个版本并行运行对比输出结果。如果新版本在多次测试中都优于或等于旧版本再正式切换。这个技巧帮我避免了好几次看似优化实则退化的事故。5. 多智能体协同中的典型故障与排查链路5.1 死循环Agent 之间互相踢皮球这是多智能体系统最常见的故障。表现是系统一直运行但不产出结果日志里能看到两个 Agent 反复互相调用。根本原因通常是终止条件不明确。比如撰写 Agent 把稿子交给审核 Agent审核 Agent 提出修改意见退回撰写 Agent 修改后又交给审核审核又提意见……如果审核标准过于严格或模糊这个循环永远不会结束。排查链路先看事件日志找到循环的两个或多个Agent然后检查它们之间的消息内容看审核 Agent 的意见是否在重复最后检查终止条件——是否有最多修改 N 次或连续两次意见相同则通过这样的硬性约束。修复方案是加循环计数器和收敛判断。循环计数器限制最大迭代次数超过就强制通过并标记收敛判断检测连续两次输出是否足够相似相似则视为已收敛。class LoopGuard: def __init__(self, max_iterations5, similarity_threshold0.9): self.max_iterations max_iterations self.similarity_threshold similarity_threshold self.history [] def check(self, agent_pair, output): self.history.append((agent_pair, output)) # 检查迭代次数 pair_count sum(1 for p, _ in self.history if p agent_pair) if pair_count self.max_iterations: return force_pass # 检查收敛 if len(self.history) 2: prev self.history[-2][1] if self._similarity(prev, output) self.similarity_threshold: return converged return continue5.2 状态不一致两个 Agent 看到的数据不一样这个故障比较隐蔽表现是系统偶尔产出矛盾的结果但日志看起来都正常。根因通常是共享状态更新没有加锁两个 Agent 同时读写同一份状态导致其中一个的修改被覆盖。排查方法是给共享状态加版本号每次写入时检查版本。如果发现版本冲突记录冲突日志。我通常会在开发环境开启严格模式任何版本冲突都直接抛异常这样能快速定位问题。修复方案有两种乐观锁和悲观锁。乐观锁适合冲突少的场景写入时检查版本冲突则重试悲观锁适合冲突多的场景读取时就加锁其他 Agent 必须等待。多 Agent 系统中我倾向于乐观锁因为 Agent 的思考时间较长悲观锁会导致大量等待。5.3 上下文爆炸消息越传越多token 越烧越快多 Agent 系统中消息会在 Agent 之间传递和累积。如果不加控制上下文会迅速膨胀。我见过一个项目跑到第十轮时单次请求的 token 数已经超过 10 万成本高得离谱。解决方法是上下文压缩和消息摘要。每个 Agent 在传递消息前先对消息做摘要只保留关键信息。对于历史消息用滑动窗口只保留最近 N 轮更早的消息压缩成一段摘要。def compress_context(messages, max_tokens4000): 将历史消息压缩到指定 token 数以内 if count_tokens(messages) max_tokens: return messages # 保留最近的消息压缩早期的 recent messages[-5:] early messages[:-5] summary summarize(early) # 调用模型生成摘要 return [{role: system, content: f历史摘要{summary}}] recent提示摘要本身也要消耗 token所以不要每轮都摘要。我的做法是当上下文超过阈值时才触发摘要且摘要结果缓存起来避免重复计算。5.4 工具调用冲突多个 Agent 抢同一个资源当多个 Agent 需要调用同一个外部工具比如写同一个文件、调用同一个 API时可能产生冲突。表现是数据被覆盖、API 触发限流、文件内容错乱。解决方案是资源锁和调用队列。给每个共享资源分配一个锁Agent 调用前先申请锁用完释放。对于 API 限流用一个令牌桶控制调用速率。class ResourceLock: def __init__(self): self.locks {} self.queues {} def acquire(self, resource_id, agent_name, timeout30): if resource_id not in self.locks: self.locks[resource_id] agent_name return True # 资源被占用加入等待队列 self.queues.setdefault(resource_id, []).append(agent_name) return False def release(self, resource_id, agent_name): if self.locks.get(resource_id) agent_name: del self.locks[resource_id] # 唤醒队列中的下一个 if self.queues.get(resource_id): next_agent self.queues[resource_id].pop(0) self.locks[resource_id] next_agent return next_agent return None6. 生产环境下的性能与成本控制6.1 并发处理多 Agent 如何扛住高并发多智能体系统在生产环境面临的第一个挑战就是并发。当多个用户同时提交任务时系统需要同时运行多组 Agent 实例。如果每组实例都独占资源很快就会耗尽。我的做法是Agent 池化 任务队列。预先创建一定数量的 Agent 实例放入池中任务到达时从池中取一个空闲实例执行执行完归还。这样避免了频繁创建销毁的开销也便于控制资源上限。class AgentPool: def __init__(self, agent_factory, pool_size10): self.pool queue.Queue(maxsizepool_size) for _ in range(pool_size): self.pool.put(agent_factory()) def execute(self, task): agent self.pool.get() # 阻塞直到有空闲 try: return agent.run(task) finally: self.pool.put(agent) # 归还但池化有个问题Agent 的私有上下文需要清理否则上一个任务的残留会影响下一个任务。我的做法是在归还前重置 Agent 的私有状态只保留必要的配置。对于需要长时间运行的任务池化就不合适了因为会占着实例不放。这时候用异步任务队列任务提交后立即返回后台异步执行完成后通过回调或轮询通知。6.2 Token 成本优化的五个手段多 Agent 系统的 token 消耗通常是单 Agent 的数倍因为消息在 Agent 之间传递会重复携带上下文。控制成本是生产环境必须考虑的问题。手段一模型分级。不是所有 Agent 都需要用最强的模型。规划、审核类 Agent 用强模型执行、格式化类 Agent 用轻量模型。我实测下来合理分级能降低 40% 以上的成本。手段二提示词精简。定期审查每个 Agent 的系统提示词删掉冗余的描述和示例。很多提示词是开发阶段不断加需求堆出来的里面有不少重复和过时的内容。手段三缓存复用。对于相同的输入缓存 Agent 的输出。比如多个任务都需要分析同一份需求文档第一次分析后缓存结果后续直接复用。手段四批量处理。如果多个任务可以合并尽量合并成一次请求。比如需要生成 10 个类似的代码片段可以一次性让 Agent 生成而不是调用 10 次。手段五早停机制。当系统检测到任务已经无法完成比如关键依赖失败立即终止后续所有 Agent不要继续烧 token。6.3 可观测性怎么知道系统在干什么多 Agent 系统的调试难度远高于单 Agent因为问题可能出在任何一个 Agent 或它们之间的交互上。没有完善的可观测性排查问题就像盲人摸象。我通常搭建三个层次的监控Agent 层记录每个 Agent 的输入、输出、耗时、token 消耗、工具调用情况。这些数据用于分析单个 Agent 的表现。消息层记录所有 Agent 之间的消息传递包括消息内容、发送方、接收方、时间戳。这些数据用于分析协同过程。任务层记录每个任务的完整生命周期从提交到完成或失败包括经过了哪些 Agent、每个阶段的状态、最终结果。这些数据用于分析整体流程。class Tracer: def __init__(self): self.spans [] def start_span(self, name, metadataNone): span { name: name, start: time.time(), metadata: metadata or {}, children: [] } self.spans.append(span) return span def end_span(self, span, resultNone): span[end] time.time() span[duration] span[end] - span[start] span[result] result这些数据汇总到一个可视化面板上可以直观看到任务在哪个环节卡住、哪个 Agent 耗时最长、哪条消息导致了异常。我用的是一套自研的轻量级追踪系统核心就是上面的 Tracer 类配合一个简单的 Web 界面展示。7. 多智能体系统的安全边界与风险控制7.1 Agent 权限的最小化原则多 Agent 系统中每个 Agent 能访问哪些资源、能执行哪些操作必须严格限制。我见过一个案例一个负责整理文件的 Agent 被赋予了文件系统的完全访问权限结果它在整理过程中误删了重要文件。原则是最小权限每个 Agent 只拥有完成自己任务所必需的最小权限。整理文件的 Agent 只应该有特定目录的读写权限不应该有删除权限调用 API 的 Agent 只应该有特定接口的调用权限不应该有管理权限。实现上我给每个 Agent 配置一个权限清单工具执行前先检查权限。权限清单在 Agent 初始化时确定运行过程中不可修改。class PermissionChecker: def __init__(self, permissions): self.permissions permissions # {file:read: [/data/*], api:call: [weather_api]} def check(self, action, resource): allowed self.permissions.get(action, []) for pattern in allowed: if fnmatch(resource, pattern): return True return False7.2 输出内容的过滤与审核Agent 生成的内容在输出给用户之前必须经过审核。这不仅是合规要求也是质量保证。我在系统中加了一个独立的安全审核 Agent负责检查所有输出内容。审核 Agent 的检查项包括是否包含敏感信息、是否符合格式要求、是否存在明显的逻辑错误、是否包含不当内容。审核不通过的内容会被拦截并触发重新生成或人工介入。注意审核 Agent 本身也可能出错所以不能完全依赖它。我的做法是审核 Agent 做第一道过滤关键场景再加人工复核。审核 Agent 的判定结果也要记录定期分析误判率持续优化审核规则。7.3 异常情况的降级与熔断生产环境中任何组件都可能失败。多 Agent 系统的健壮性取决于对失败的处理能力。降级策略当某个 Agent 不可用时用备用方案替代。比如强模型不可用降级到轻量模型某个工具不可用跳过该步骤或用手动方式替代。熔断机制当某个 Agent 的失败率超过阈值时暂时停止调用它避免连锁失败。熔断后定期尝试恢复成功则重新启用。class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout60): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.failures 0 self.state closed # closed, open, half-open self.last_failure_time None def call(self, func, *args, **kwargs): if self.state open: if time.time() - self.last_failure_time self.recovery_timeout: self.state half-open else: raise CircuitOpenError(Circuit is open) try: result func(*args, **kwargs) if self.state half-open: self.state closed self.failures 0 return result except Exception as e: self.failures 1 self.last_failure_time time.time() if self.failures self.failure_threshold: self.state open raise8. 我在多智能体项目中的几条实战心得做过多智能体项目的人都知道理论上的架构和实际跑起来的效果之间往往隔着无数个细节。分享几条我踩坑后总结的经验都是文档里不会写的。第一条先跑通两个 Agent 的协同再扩展到更多。很多人一上来就设计五六个 Agent 的复杂架构结果调试时根本不知道问题出在哪。我的做法是先实现两个 Agent 的最小协同闭环确保消息传递、状态同步、错误处理都跑通再逐步增加 Agent。每增加一个都要重新验证整个链路。第二条给每个 Agent 写单元测试。Agent 的行为有不确定性但基本的能力边界是可以测试的。比如给规划 Agent 输入一个明确的需求它应该输出包含特定字段的 JSON给审核 Agent 输入一段有明显错误的文本它应该能识别出来。这些测试不能保证 Agent 永远正确但能在修改提示词或换模型后快速发现退化。第三条日志要记录为什么不只是是什么。普通的日志记录 Agent 输入了什么、输出了什么。但排查问题时你更想知道 Agent 为什么做出这个决策。我的做法是让 Agent 在输出结果的同时输出一段简短的决策理由。这段理由不参与后续流程只用于调试。第四条不要追求一次设计完美。多智能体系统的设计是迭代出来的不是设计出来的。我第一个版本的多 Agent 架构和最终上线的版本相比Agent 数量、职责划分、通信方式都改了很多。接受先跑起来再优化的理念比追求完美设计更重要。第五条人工兜底永远要有。无论系统多智能总会有它处理不了的情况。在关键节点设置人工介入的入口当系统连续失败或置信度低时转交人工处理。这不是系统不完善的表现而是负责任的设计。最后说一个我最近在用的技巧给每个 Agent 加一个置信度输出。Agent 在给出结果时同时给出一个 0 到 1 的置信度分数。系统根据置信度决定后续动作——高置信度直接通过中等置信度触发交叉验证低置信度转人工。这个简单的机制让系统的整体准确率提升了不少因为很多错误在低置信度时就被拦截了。