1. 从agency-agents这个标题说起一个多智能体协作框架的完整拆解第一次看到agency-agents这个命名我的直觉是这是一个把代理和智能体两个概念揉在一起的项目。拆开看agency在软件语境里通常指代理机构或代理能力而agents则是智能体的复数形式。合在一起它指向的是一个多智能体协作系统——让多个具备独立决策能力的智能体像一家代理公司里的不同角色一样各司其职、互相配合共同完成一个复杂任务。这个方向最近一年在开发者圈子里讨论度非常高。原因很直接单个大模型调用已经很难满足真实业务场景的需求了。你让一个模型既做需求分析、又写代码、又做测试、又写文档它会在某个环节掉链子。但如果把任务拆开让不同的智能体分别负责不同环节每个智能体只专注自己那一块整体输出的质量会明显提升。agency-agents 要解决的正是这个分工与协作的问题。这篇文章适合谁看如果你正在做 AI 应用开发尤其是涉及任务编排、自动化流程、多角色协作的场景那这篇内容会对你有直接帮助。如果你只是听说过多智能体这个词但没动手做过我也会从最基础的设计思路讲起把每个环节的取舍逻辑说清楚。全文基于我对这类系统的实际开发经验来写涉及具体实现的部分会给出可参考的代码结构和参数配置。2. 整体架构设计为什么这样拆分智能体2.1 核心设计思路角色化分工而非功能化分工很多人在设计多智能体系统时第一反应是按功能来拆分——一个负责搜索、一个负责计算、一个负责生成。这种拆法在简单场景下能用但一旦任务变复杂就会出现问题功能边界模糊智能体之间互相推诿或者某个智能体不知道该在什么时候介入。agency-agents 的思路更接近真实公司的组织方式——按角色拆分。每个智能体有自己的职位描述包括职责范围、输入输出规范、以及和其他角色的协作关系。比如在一个内容生产流程里会有策划者负责拆解需求、执行者负责具体产出、审核者负责质量把关。这种设计的好处是每个智能体的行为逻辑更接近人类的协作模式调试和扩展都更直观。我实测下来角色化分工在任务链路超过三步之后优势会非常明显。因为每个角色的输入输出都是明确的你可以单独替换某个角色而不影响整体流程。比如把审核者从一个简单规则引擎换成一个更强的模型其他部分完全不用动。2.2 通信机制选型消息队列 vs 直接调用智能体之间怎么传递信息这是架构设计里第一个要拍板的事。常见方案有两种一种是直接函数调用A 智能体处理完直接把结果传给 B另一种是引入消息队列所有智能体通过队列异步通信。直接调用的优点是简单、延迟低、调试方便。缺点是耦合度高一旦某个环节卡住整条链路都会阻塞。消息队列的优点是解耦彻底、支持异步、方便做重试和监控。缺点是引入了额外的基础设施调试链路变长。agency-agents 在这块的选择是混合模式核心链路用直接调用保证响应速度非关键路径比如日志记录、结果归档走消息队列。这个取舍的逻辑是——用户等待的环节不能有额外延迟但后台处理可以慢慢来。具体实现上我用的是一个轻量的内存队列加一个持久化队列的组合内存队列处理实时消息持久化队列做兜底。注意如果你的任务链路里有超过 5 个智能体强烈建议至少把日志和监控走异步。否则一旦某个智能体出问题你很难定位是哪一步出的错。2.3 状态管理共享内存还是各自维护多智能体系统里状态管理是最容易出 bug 的地方。每个智能体都有自己的上下文但有些信息需要全局共享——比如当前任务的进度、已经产出的中间结果、用户的原始需求。agency-agents 采用的是分层状态管理每个智能体维护自己的私有状态比如当前的思考过程、临时变量同时有一个全局的共享状态存储比如任务 ID、已完成步骤、关键产出。私有状态用对象内部属性存共享状态用一个独立的 Store 来管理。这样做的好处是智能体之间的状态隔离清晰不会出现 A 改了变量导致 B 行为异常的情况。同时共享状态可以持久化方便做断点续跑和故障恢复。我在实际项目里踩过一个坑早期把所有状态都放在一个全局对象里结果两个智能体同时读写同一个字段出现了竞态条件排查了大半天。后来改成分层管理这类问题就再也没出现过。3. 核心智能体的实现细节与实操要点3.1 策划者智能体任务拆解的逻辑与边界策划者Planner是整个系统的入口它的职责是把用户的模糊需求拆解成可执行的具体步骤。这个环节的质量直接决定了后续所有智能体的工作效率。实现上策划者的核心是一个任务分解提示词模板。模板里需要明确几件事任务的最终目标是什么、有哪些约束条件、可用的执行角色有哪些、每个步骤的预期输出格式是什么。我常用的模板结构是这样的PLANNER_PROMPT 你是一个任务策划专家。请将以下需求拆解为可执行的步骤序列。 需求{user_request} 可用角色 - 执行者负责具体内容产出 - 审核者负责质量检查 - 归档者负责结果整理 输出格式要求 1. 每个步骤必须指定负责角色 2. 每个步骤必须有明确的输入和输出 3. 步骤之间如果有依赖关系必须标注 请以 JSON 格式输出步骤列表。 这里有个关键细节步骤的粒度控制。拆得太粗执行者不知道从哪下手拆得太细智能体之间的通信开销会急剧增加。我的经验是每个步骤的工作量控制在一个智能体一次调用能完成的范围内。如果某个步骤需要多次调用就应该继续拆。另一个容易忽略的点是失败处理。策划者不仅要拆步骤还要为每个步骤定义什么算成功、什么算失败、失败了怎么办。我通常会在步骤定义里加一个fallback字段指定失败后的备选方案。比如如果搜索不到相关信息则基于已有知识生成。3.2 执行者智能体工具调用与结果校验执行者是真正干活的角色它的核心能力是工具调用。一个执行者通常需要访问外部工具——比如搜索引擎、数据库、代码执行环境、文件系统。agency-agents 里每个执行者都绑定了一组工具工具的定义遵循统一的接口规范。工具调用的实现要点有三个第一参数校验必须在调用前完成。我见过太多因为参数格式不对导致工具调用失败的案例。解决方案是在工具定义里加上参数 schema调用前先做一次校验。比如搜索工具的参数必须是字符串且长度不超过 200 字符不满足就直接返回错误而不是把错误参数传给工具。第二结果解析要容错。外部工具返回的数据格式可能不稳定今天返回 JSON明天返回纯文本。执行者需要有一套容错解析逻辑先尝试按预期格式解析失败则降级到通用解析。第三调用结果要记录。每次工具调用的输入、输出、耗时、是否成功都要记录下来。这些数据在排查问题时非常有用。我通常会把调用记录写到一个独立的日志文件里格式是 JSON Lines方便后续分析。def execute_with_tool(tool_name, params, context): # 参数校验 if not validate_params(tool_name, params): return {status: error, reason: invalid_params} # 调用工具 start_time time.time() try: result tools[tool_name].call(params) status success except Exception as e: result str(e) status error # 记录日志 log_call({ tool: tool_name, params: params, result: result, status: status, duration: time.time() - start_time }) return {status: status, data: result}实操心得工具调用的超时设置非常关键。我一般把默认超时设为 30 秒对于搜索类工具设为 15 秒对于代码执行类工具设为 60 秒。超时后不要直接失败而是返回一个超时状态让上层决定是重试还是降级。3.3 审核者智能体质量标准的量化与执行审核者的职责是检查执行者的产出是否符合要求。这个环节最容易做成走过场——审核者随便看一眼就说通过了。要避免这个问题关键是把质量标准量化。我通常从三个维度定义质量标准完整性是否覆盖了所有要求点、准确性是否有事实错误或逻辑矛盾、格式规范性是否符合输出格式要求。每个维度给出一个 0 到 1 的评分加权求和得到总分。总分低于阈值比如 0.7则判定为不通过需要执行者重新处理。审核者的提示词里要明确列出检查清单。比如对于一篇文档检查清单可能包括是否包含所有章节、是否有错别字、数据引用是否准确、格式是否统一。清单越具体审核结果越可靠。REVIEWER_PROMPT 请对以下内容进行质量审核。 内容{content} 审核维度 1. 完整性0-1分是否覆盖了所有要求点 2. 准确性0-1分是否有事实错误或逻辑矛盾 3. 格式规范性0-1分是否符合输出格式要求 请逐项打分并给出理由最后计算加权总分。 权重完整性 0.4准确性 0.4格式规范性 0.2。 输出格式 { completeness: {score: 0.9, reason: ...}, accuracy: {score: 0.8, reason: ...}, format: {score: 1.0, reason: ...}, total: 0.88, passed: true } 这里有个经验审核者不要和执行者用同一个模型。如果条件允许用不同厂商或不同版本的模型来做审核能发现更多问题。因为同一个模型往往有相同的盲区自己审自己很难发现错误。4. 完整实操流程从零搭建一个可运行的多智能体系统4.1 环境准备与依赖安装先把基础环境搭起来。我用的技术栈是 Python 3.10 以上核心依赖包括一个 HTTP 客户端用于调用模型 API、一个异步框架用于并发处理、一个配置管理库。具体安装命令如下pip install httpx asyncio aiofiles pydantic python-dotenv配置管理用.env文件来存 API 密钥和模型参数不要硬编码在代码里。一个典型的配置结构是这样的MODEL_API_BASEhttps://api.example.com/v1 MODEL_API_KEYyour_key_here PLANNER_MODELmodel-large EXECUTOR_MODELmodel-medium REVIEWER_MODELmodel-large MAX_RETRIES3 TIMEOUT_SECONDS30注意模型选择上策划者和审核者建议用能力更强的模型执行者可以用中等模型。因为策划和审核对逻辑推理要求高执行对生成速度要求高。这个搭配在成本和效果之间比较平衡。4.2 智能体基类的设计与实现所有智能体共享一个基类把公共逻辑抽出来。基类负责的事情包括模型调用、重试逻辑、日志记录、状态管理。子类只需要实现自己的核心处理方法。class BaseAgent: def __init__(self, name, model, system_prompt): self.name name self.model model self.system_prompt system_prompt self.private_state {} async def call_model(self, messages, retries3): for attempt in range(retries): try: response await model_client.chat( modelself.model, messagesmessages, timeoutTIMEOUT_SECONDS ) return response except Exception as e: if attempt retries - 1: raise await asyncio.sleep(2 ** attempt) async def process(self, input_data, shared_state): raise NotImplementedError这个基类里重试逻辑用的是指数退避——第一次失败等 1 秒第二次等 2 秒第三次等 4 秒。这样能有效应对临时的网络抖动或服务限流。4.3 编排器的实现任务调度与异常处理编排器Orchestrator是整个系统的中枢负责按顺序调用各个智能体、传递数据、处理异常。它的核心逻辑是一个循环从策划者拿到步骤列表然后逐步执行每个步骤每步执行完更新共享状态。class Orchestrator: def __init__(self, planner, executor, reviewer): self.planner planner self.executor executor self.reviewer reviewer self.shared_state {} async def run(self, user_request): # 第一步策划 plan await self.planner.process(user_request, self.shared_state) self.shared_state[plan] plan # 第二步逐步执行 for step in plan[steps]: result await self.executor.process(step, self.shared_state) # 第三步审核 review await self.reviewer.process(result, self.shared_state) if not review[passed]: # 审核不通过重试 result await self.executor.process( step, self.shared_state, feedbackreview[reason] ) self.shared_state[fstep_{step[id]}_result] result return self.shared_state异常处理上我设置了三个级别的容错单步重试最多 3 次、步骤降级用备选方案替代、整体中断返回已完成的部分结果。大部分情况下单步重试就能解决问题。如果某个步骤连续失败就触发降级逻辑。4.4 参数调优与性能优化系统跑起来之后性能调优是下一步。我实测下来影响性能的主要有三个参数并发数、超时时间、重试次数。并发数方面如果步骤之间没有依赖关系可以并发执行。我一般把并发数设为 3 到 5再高的话模型 API 容易限流。超时时间前面说过按工具类型分别设置。重试次数设为 3 次比较合适再多的话整体延迟会明显增加。还有一个容易被忽略的优化点缓存。如果某个步骤的输入和之前某次执行的输入相同可以直接复用之前的结果。我在执行者里加了一个简单的缓存层用输入内容的哈希值作为 key命中率大概在 20% 左右对整体速度提升很明显。参数推荐值调整依据并发数3-5API 限流阈值超时时间15-60秒工具类型重试次数3延迟与成功率平衡缓存开关开启重复输入场景日志级别INFO排查问题需要5. 常见问题与排查技巧实录5.1 智能体卡死或无限循环的排查这是最常见的问题。表现是某个智能体反复调用同一个工具或者两个智能体互相等待。根本原因通常是终止条件不明确。排查思路先看日志确认是哪个智能体在循环。然后检查它的提示词里有没有明确的终止条件。比如执行者的提示词里应该写如果连续两次调用工具都没有获得新信息则停止并返回当前结果。审核者的提示词里应该写最多审核两轮两轮后无论结果如何都返回。我踩过的一个坑是策划者拆出的步骤里有一个等待审核通过但审核者又依赖执行者的输出形成了循环依赖。后来在编排器里加了一个依赖检查发现循环依赖就直接报错而不是让它跑下去。5.2 输出格式不稳定的处理方案模型输出格式不稳定是另一个高频问题。你要求它输出 JSON它有时候输出 JSON有时候输出带 markdown 代码块的 JSON有时候还加一段解释文字。解决方案是多层解析第一层尝试直接解析 JSON第二层用正则提取代码块里的 JSON第三层用正则提取花括号内容第四层如果都失败就把原始输出传给一个格式修复提示词让模型重新格式化。def parse_json_safely(text): # 第一层直接解析 try: return json.loads(text) except: pass # 第二层提取代码块 match re.search(r(?:json)?\s*(.*?), text, re.DOTALL) if match: try: return json.loads(match.group(1)) except: pass # 第三层提取花括号 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group(0)) except: pass return None实操心得与其在解析上做各种容错不如在提示词里就把格式要求写死。我通常会在提示词末尾加一句只输出 JSON不要有任何其他文字。这句话能减少 80% 的格式问题。5.3 常见问题速查表问题现象可能原因排查方法解决方案智能体卡死终止条件缺失查看调用日志补充终止条件输出格式错误提示词不明确检查原始输出加强格式约束审核总是通过标准太宽松检查评分分布量化审核标准执行速度慢串行执行分析依赖关系改为并发执行结果不一致状态污染检查共享状态隔离私有状态API 频繁超时并发过高查看限流日志降低并发数5.4 几个我踩过的坑和对应的解法第一个坑共享状态被意外修改。早期我把共享状态直接传给每个智能体结果某个智能体在处理时不小心改了共享状态里的字段导致后续智能体拿到错误数据。解法是传给智能体的是共享状态的深拷贝智能体只能通过专门的接口来更新共享状态。第二个坑日志太多导致磁盘写满。调试阶段我把所有调用都记了详细日志跑了一天发现磁盘满了。解法是日志分级INFO 级别只记关键信息DEBUG 级别才记详细内容并且设置日志轮转最多保留 7 天。第三个坑模型版本升级导致行为变化。有一次模型服务商升级了模型版本同样的提示词输出风格变了导致审核通过率骤降。解法是在配置里锁定模型版本号不要用latest这种浮动标签。6. 扩展方向与个人经验分享这套框架跑通之后可以扩展的方向不少。我目前尝试过的有接入更多类型的工具比如数据库查询、文件操作、增加智能体之间的协商机制多个执行者投票决定结果、以及引入长期记忆把历史执行结果存起来供后续参考。协商机制这块值得多说一句。当任务比较复杂时让多个执行者分别给出方案然后由一个决策者来选择最优方案效果比单个执行者直接产出要好。代价是成本翻倍所以只在对质量要求极高的场景下用。长期记忆的实现思路是每次执行完成后把任务类型 输入特征 最终结果存到一个向量数据库里。下次遇到类似任务时先检索历史结果作为参考。这个做法在重复性任务上效果很好能显著减少模型调用次数。最后分享一个小技巧给每个智能体起一个有意义的名字。不要用 agent1、agent2 这种而是用 planner、executor、reviewer 这种。名字本身就能提醒你这个智能体的职责边界在调试和日志分析时也能快速定位问题。这个习惯看起来不起眼但在系统变复杂之后能省下大量排查时间。