最近在做一个多智能体协作系统项目代号就叫 agency-agents。这个系统本质上是一个把多个AI智能体组合起来、让它们像一个小团队一样分工协作的框架。它解决的问题很直接单个智能体在处理复杂任务时经常顾此失彼——又要理解用户意图又要调用工具还要管理上下文最后输出质量很难保证。而通过“多个智能体各管一段、互相配合”的方式可以把复杂任务拆碎让每个智能体只做自己最擅长的事再由一个调度层统一协调。这篇文章我会把这个系统的设计思路、核心代码结构、实测数据和踩过的坑完整记录下来给正在做多智能体方案的开发者一个可参考的样本。1. 项目整体设计与方案选型1.1 多智能体系统要解决的三个核心矛盾构建多智能体系统之前得先想清楚它到底解决什么问题。如果只是做一个带工具调用的单智能体应用那没必要引入多智能体的复杂度。我在实际项目中遇到的核心矛盾有三个。第一个是上下文长度有限。单个智能体的上下文窗口再大也扛不住长时间任务积累的对话历史、中间结果和工具返回数据。比如让一个智能体同时处理数据清洗、特征工程、模型调参三个环节它在第三个环节时早就忘了第一步的具体细节输出质量断崖式下降。第二个是职责混杂导致指令跟随变差。同一个智能体既当数据分析师又当代码审查员还要负责用户沟通它的系统提示词会变得极其臃肿。我在测试中发现超过两到三层职责叠加后智能体经常在错误的时间调用错误的工具。它开始“自由发挥”明明该做数据统计它却跑去生成图表。第三个是失败定位困难。单智能体应用一旦输出错误你很难判断是提示词问题、模型问题还是工具调用的问题。多智能体把任务切成独立模块后每个环节都可以单独验证出错时可以快速定位到具体某个智能体。正是这三个矛盾让我决定采用多智能体协作架构。agency-agents 的设计目标不是做一个“更聪明的单体大脑”而是做一个“能协调多个专业大脑的中枢系统”。1.2 为什么选择四层架构而不是完全去中心化在架构选型阶段我对比了三种主流方案完全去中心化的对等网络架构、中心化编排架构、以及分层协作架构。完全去中心化听起来很酷每个智能体都能主动发起对话、请求其他智能体帮忙但实际跑起来问题很多——消息风暴难以控制循环调用经常发生调试时根本看不清消息在智能体之间怎么流转。我最终选择了四层架构接入层、编排调度层、智能体执行层、基础设施层。这个结构有点像一个公司接入层是前台接收用户请求编排调度层是项目经理负责拆解任务、分派给不同执行人智能体执行层是干活的具体员工基础设施层是共享的办公资源比如数据库、消息队列、工具API。选择这个结构的关键考量是“可观测性优先”。每一层职责清晰消息流向固定日志可以按层追踪。即使某个环节出了问题也能快速判断是编排层分派错误还是执行层某个智能体能力不足。牺牲了一点灵活性换来了可控性在多智能体系统的早期阶段这是更稳妥的选择。1.3 技术栈选型与核心依赖技术栈的选择直接决定了开发效率和系统稳定性。我在这套系统里核心依赖如下模型层接入多个大语言模型基础任务用轻量模型复杂推理用强模型。通信层异步消息队列负责智能体之间的消息传递和任务状态同步。工具层统一函数调用协议每个工具注册成标准化接口。存储层向量数据库存长期记忆关系型数据库存任务元数据。可观测层结构化日志加分布式追踪记录每个智能体的输入输出。选型时有个重要的取舍通信层最开始我只用内存队列后来因为任务量上升、消息积压才换成持久化的消息队列组件。如果你的任务是纯同步、量级也不大内存队列完全够用但一旦涉及长时间运行的任务持久化队列几乎是必须的否则进程一重启所有状态全部丢失。2. 核心模块拆解与实现细节2.1 智能体定义用统一的Agent协议封装所有能力agency-agents 里的每个智能体都遵循同一个抽象协议。我定义了一个基础类所有具体智能体都继承自这个类。协议的关键字段包括agent_id、角色描述、可用工具列表、目标函数、以及上下文缓冲区。class BaseAgent: def __init__(self, agent_id: str, role_desc: str, tools: list, model_config: dict): self.agent_id agent_id self.role_desc role_desc self.tools {tool.name: tool for tool in tools} self.model_config model_config self.context_buffer [] # 保存当前任务相关的上下文 async def run(self, task_input: dict) - dict: 执行当前任务返回结构化结果 raise NotImplementedError async def call_tool(self, tool_name: str, params: dict) - dict: 调用注册过的工具 if tool_name not in self.tools: return {status: error, message: ftool {tool_name} not found} return await self.tools[tool_name].execute(**params)这里有个特别容易被忽略的细节role_desc 不只是给智能体看的提示词它同时也是调度器做任务分派时的匹配依据。我在实际项目里用向量化方式把 task_input 和所有智能体的 role_desc 做相似度匹配实现初步路由。这个设计的好处是不需要硬编码分派规则新加一个智能体时只要注册它的角色描述调度器自动就能识别它适合处理什么任务。2.2 编排调度层任务分解与路由策略编排层是整个系统最核心的部分。它接收用户请求后执行三个步骤解析任务目标、分解子任务、分派给对应智能体。任务分解我采用的是“意图拆分加依赖分析”的方式。先用一个轻量模型快速判断任务的整体类型再基于预设的分解模板生成子任务列表。每个子任务都包含任务ID、依赖的任务ID列表、目标智能体类型、输入参数、预期输出格式。class Orchestrator: def __init__(self, agents: dict, decomposer, queue): self.agents agents self.decomposer decomposer self.queue queue async def process(self, user_request: str) - str: # 1. 任务分解 plan await self.decomposer.decompose(user_request) # 2. 按依赖关系构建DAG并调度 task_id_map {} for sub_task in plan.sub_tasks: if sub_task.dependencies: await self.queue.wait_for_dependencies(sub_task.dependencies) target_agent self.route(sub_task) result await self.agents[target_agent].run(sub_task.input) task_id_map[sub_task.task_id] result # 3. 汇总结果 return self.aggregate(task_id_map, plan.output_template)路由策略我用了两种模式组合基于角色描述的语义匹配作为主路由基于关键词规则的硬匹配作为兜底。比如包含“计算”或“统计”的任务优先路由到数据分析智能体包含“代码”或“函数”的路由到代码生成智能体。语义匹配处理未知场景硬匹配保证高频场景不跑偏。2.3 工具注册与权限控制工具是多智能体系统的能力底座工具层设计得不好整个系统就会变成“一个会聊天但什么都干不了的空壳”。我在系统里实现了统一的工具注册协议每个工具都包含名称、描述、参数Schema、执行函数、权限级别五个要素。register_tool( namedatabase_query, description执行只读SQL查询适合数据提取和统计, params_schema{ type: object, properties: { query: {type: string, description: 完整的SELECT语句} } }, permission_levelread_only ) async def database_query(query: str): # 连接只读副本不允许任何写操作 async with get_readonly_connection() as conn: result await conn.fetch(query) return {rows: result}权限控制是这里不可忽视的部分。多智能体环境下因为任务被拆分某个智能体可能只看到任务片段这反而降低了误操作风险。但如果权限设置不当也容易造成更严重的问题——一个原本只该做数据读取的智能体在生成代码时可能试图执行系统命令。我的做法是给每个工具打上权限标签并在智能体执行层加上“只能调用自身权限范围内工具”的硬校验而不是完全依赖模型自觉。2.4 记忆与上下文管理方案多智能体系统的记忆管理比单智能体复杂得多因为每个智能体都有自己的上下文缓冲同时还要共享一部分全局记忆。我把记忆分成三层第一层是各智能体的私有记忆。比如数据分析智能体在处理当前数据集时特征分布、缺失值情况这些信息只保存在它自己的上下文缓冲里。第二层是任务级共享记忆同一任务下的多个智能体可以访问用于传递中间结果。第三层是全局长期记忆存在向量数据库里记录历史任务的经验、用户的偏好、常见问题解决方案。实现上我用了“双写策略”智能体执行完一个子任务后把结果同时写入任务级存储和长期记忆库。写长期记忆库前会做一次摘要压缩避免存储爆炸。这个方案看起来简单但实际调整了好几次。最开始所有中间结果都被完整存入长期记忆库没过多久向量数据库的规模就失控了检索相关性也急剧下降。后来改成只存摘要加关键指标情况才好转。3. 实操过程与核心环节实现3.1 从零构建一个可运行的多智能体Demo如果你也想搭建自己的多智能体协作系统我建议从一个非常窄的场景入手。我当初选的第一个场景是“自动生成数据分析报告”。任务链路是用户提供一个CSV文件路径和几个分析问题系统自动完成数据加载、统计计算、图表生成、报告撰写四个环节最终输出一份Markdown格式的分析报告。这个场景非常适合用来练手因为四个环节边界清晰每个环节都可以分配一个独立智能体数据加载智能体、统计计算智能体、图表生成智能体、报告撰写智能体。它们之间的依赖关系也很明确——数据加载完才能统计统计完才能画图和写报告。实现步骤是这样的定义四个智能体的角色描述和工具列表。实现编排器的任务分解逻辑把用户请求拆成四个子任务。为每个智能体编写具体的提示词模板和执行流程。实现消息队列处理智能体间的数据传递。搭建仪表盘观察各智能体的执行状态和延迟。第一步到第三步用了一天时间第四步到第五步花了两天。大部分时间都耗在调试智能体之间的数据格式不一致上——统计智能体输出的JSON结构图表智能体期望的输入结构两者之间需要做一层字段映射。这个在单智能体系统里完全不是问题但在多智能体系统里就变成了日常摩擦。3.2 数据加载智能体的实现细节数据加载智能体是整个任务链的起点。它接收用户提供的文件路径检查文件存在性和格式读取数据后做基础清洗然后输出一个标准化数据结构。class DataLoaderAgent(BaseAgent): def __init__(self, config): tools [ FileReaderTool(), DataProfilerTool(), DataCleanerTool() ] super().__init__( agent_iddata_loader, role_desc负责读取数据文件检查数据质量执行基础清洗操作, toolstools, model_configconfig ) async def run(self, task_input: dict) - dict: file_path task_input[file_path] # 读取文件 raw_data await self.call_tool(file_reader, {path: file_path}) # 数据画像输出列名、类型、缺失值分布 profile await self.call_tool(data_profiler, {data: raw_data}) # 基础清洗剔除全空行、统一日期格式 cleaned await self.call_tool(data_cleaner, { data: raw_data, profile: profile }) return { data_reference: cleaned[storage_id], column_types: profile[column_types], missing_counts: profile[missing_counts], row_count: cleaned[row_count] }一个比较隐蔽的坑是不要把原始数据直接作为返回值传给下一个智能体。数据量一大消息队列就会吃不住。我在实现中使用共享存储ID数据清洗后存到对象存储返回值里只带一个引用ID下一个智能体通过这个ID去拉取数据。3.3 编排器的DAG执行引擎编排器内部我用了一个有向无环图来管理子任务依赖。最开始的版本是线性的一个子任务执行完接着执行下一个但碰到某些场景会卡死比如报告撰写智能体需要同时等待统计结果和图表结果如果线性执行图表子任务必须等统计子任务完全结束才开始两者之间白白浪费了一次串行等待。改成DAG执行引擎后可以并行处理没有依赖关系的子任务。DAG引擎的核心逻辑class DAGExecutor: def __init__(self, tasks: dict): self.tasks tasks # task_id - TaskDefinition self.dependencies {tid: set(deps) for tid, deps in tasks.items()} self.dependents defaultdict(set) for tid, deps in self.dependencies.items(): for dep in deps: self.dependents[dep].add(tid) async def execute(self, execute_func): ready [tid for tid, deps in self.dependencies.items() if not deps] result_store {} while ready: current ready.pop() results await asyncio.gather(*[ execute_func(self.tasks[tid], result_store) for tid in current ]) for tid, res in zip(current, results): result_store[tid] res # 清理已完成依赖解锁下一批任务 new_ready set() for tid in self.dependencies: self.dependencies[tid] - set(current) if not self.dependencies[tid]: new_ready.add(tid) ready list(new_ready) return result_store并行执行带来一个额外的好处就是整体任务耗时显著下降。在四个子任务的场景里原来的串行耗时是每个子任务耗时之和现在最短路径计算后图表生成和统计计算可以同时跑总耗时约等于“数据加载统计计算报告撰写”的链路耗时。3.4 报告撰写智能体汇总与格式化报告撰写智能体是任务链的最后一环它负责把前面的所有结果转成自然语言报告。这个智能体的系统提示词里最重要的约束是禁止编造数据。所有结论必须引用上游智能体返回的JSON字段直接取数、转述、归纳。我在提示词里专门加了一段强约束你是一个专业的报告撰写智能体。你的任务是把结构化分析结果转化为清晰易懂的Markdown报告。 可用的数据字段包括data_summary、statistical_results、chart_reference。 每条结论必须直接基于上述字段中的实际值不得补充外部知识或自行推断。 如果字段中有异常值或缺失值请在报告中明确指出不要掩盖。 输出格式要求包含标题、数据概览、统计结论、附录图表引用四个部分。实测下来加了这个约束之后报告幻觉率从原先的30%以上降到了5%以内。多智能体系统里每个智能体的输出质量直接影响下游智能体的输入所以对输出幻觉的容忍度更低必须在提示词层面就做严格限制。3.5 实测数据与效果对比我用一组模拟的销售数据做了完整测试数据包含3万行、12个字段分析目标是月度销售趋势和区域分布。实测数据如下总耗时约18秒其中数据加载及清洗用了4.2秒统计计算用了7.8秒图表生成和报告撰写各用了1.5秒和4.5秒。相比单智能体直接处理同样任务总耗时节省了约25%主要节省在统计和图表环节的并行执行上。更关键的是输出质量。单智能体在同一个任务中报告里出现了两个数据错误——它把季度汇总数值直接当成了月度值还漏掉了一个区域的缺失值说明。多智能体场景下统计计算智能体严格返回JSON数值报告撰写智能体只能照实转述数据错误率明显降低。4. 常见问题与排查技巧4.1 消息风暴与递归调用问题多智能体系统里最危险的问题之一就是消息风暴。A智能体请求B帮忙B处理不了转给CC又觉得应该让A来处理于是消息在智能体之间无休止循环消耗大量token还拖垮系统。我在最初测试时踩过这个坑。两个智能体因为角色描述存在重叠调度器把同一个任务反复路由给它们双方都在等待对方给出“最终结果”消息积压了上千条。解决办法有两个层面。第一是路由层加“最大跳数”限制任何任务最多经过三个智能体就会终止并交给兜底智能体。第二是执行层加“任务指纹去重”如果检测到同一个任务指纹在短时间内反复出现直接拒绝执行并报告循环。class LoopProtector: def __init__(self, max_hops3, max_repeats2): self.max_hops max_hops self.task_fingerprints defaultdict(list) def can_execute(self, task_fingerprint: str, current_hops: int) - bool: if current_hops self.max_hops: return False timestamps self.task_fingerprints[task_fingerprint] # 清除一分钟前的记录 recent [t for t in timestamps if time.time() - t 60] self.task_fingerprints[task_fingerprint] recent if len(recent) self.max_repeats: return False self.task_fingerprints[task_fingerprint].append(time.time()) return True4.2 上下文污染与数据错位多智能体系统中不同智能体共享任务级存储时很容易产生上下文污染。比如数据分析智能体在任务中更新了数据集字段报告撰写智能体如果读取的是初始数据引用就会产生数据错位。我遇到过一个典型案例统计计算智能体在执行过程中进行了异常值剔除但报告撰写智能体引用的还是数据加载阶段的统计描述两份数据不一致导致报告出现矛盾结论。排查思路是给每个中间结果加上版本号下游智能体读取时必须指定版本。同时在日志里记录每个智能体的数据输入来源出问题时可以快速追溯是哪一步产生了数据变更。4.3 并发限制与超时处理多智能体系统天然具有并发特性但并发也带来新问题。多个智能体同时调用同一个外部API时可能触发接口限流某个智能体响应缓慢时会阻塞整个任务链路。我处理的策略包括为每个外部API维护独立的并发信号量限制同时请求数量为每个子任务设置超时时间超时后调度器会标记该子任务失败继续执行后续可完成的子任务最后在汇总阶段标注哪些环节失败了。超时时间不能设置太短因为大模型推理本身速度较慢。根据我的测试普通分析类子任务超时设置在30到60秒比较合理涉及长时间工具调用的可以放宽到120秒。4.4 调试多智能体系统的心得调试多智能体系统最大的困难是“看不到全局”。单智能体的输出错误很容易重现多智能体的错误可能依赖当时的历史上下文、并发时序、中间结果状态复现难度高。我的调试方法有三个。第一是给每次任务生成唯一的trace_id所有日志都带上这个ID方便串联完整执行链路。第二是记录每个智能体输入输出的JSON快照存到独立的日志文件不会因为控制台日志刷屏而丢失关键信息。第三是提供一个“单步执行模式”可以手动指定某个子任务单独运行任意一个智能体传入构造好的测试数据快速验证它在隔离环境下的行为。单步执行模式非常关键。我在排查一个图表生成智能体的问题时一开始完全摸不着头脑——它在完整链路里经常生成错误图表单独跑却总是正常。后来发现是上游统计智能体返回的JSON里某个字段在极端情况下会缺失图表智能体没有做边界处理。用单步模式配合构造的异常输入几分钟就定位到了问题。4.5 性能瓶颈与扩展建议多智能体系统跑久了性能瓶颈通常出现在三个地方。第一个是LLM API的调用延迟和并发配额这个受限于服务商只能通过缓存和请求合并来缓解。第二个是消息队列的积压如果子任务执行时间很长队列会持续堆积需要增加消费者实例。第三个是向量数据库的检索延迟长期记忆不断膨胀后检索相关性和速度都会下降需要定期压缩和索引优化。我建议在系统上线初期就做好压测核心目标是弄清楚单个任务平均会派生出多少个子任务、每条消息的平均体积、智能体之间的平均往返次数。这三个指标直接决定了系统能支撑的最大并发任务量。5. 落地经验与后续扩展5.1 从Demo到生产的几个关键改造如果只是跑通Demo内存队列加进程内状态就够了。但要落地到生产环境我做了几个改造。第一是把内存队列替换成持久化消息队列保证应用重启时任务状态不丢。第二是引入配置中心把模型参数、工具开关、路由规则全部外部化不再硬编码。第三是增加监控告警当某个智能体的错误率超过阈值或者任务平均耗时异常升高时能及时收到通知。这些改造听起来都是老生常谈但在多智能体系统里的意义更为重大。因为系统天然是一门“分布式系统”任何单一节点的不可靠都会被多级链路放大。消息队列持久化保证某个执行节点崩溃后下游智能体能够重新拉取任务配置外部化让新智能体上线、旧智能体下线不用重启整个服务。5.2 智能体能力的复用与沉淀随着系统跑得更久我越来越觉得多智能体系统的核心资产不是代码而是已经调试好的智能体能力库和工具库。一个调试好的数据分析智能体换一套提示词可以服务完全不同的业务场景一个工具如果写得好在多个智能体间复用收益是指数级的。所以我在代码之外维护了两类文档。一类是智能体卡片记录每个智能体的适用场景、输入输出格式、已知限制和调试历史。另一类是工具规范文档记录每个工具的调用方式、参数边界、权限级别和常见错误。这两类文档对新加入的开发者非常友好也避免了自己半年后回来看代码时的认知断层。5.3 下一步规划自适应调度与能力学习目前的编排层还依赖预先设定的任务分解模板和角色描述匹配规则扩展性有明显上限。遇到完全没见过的任务类型时拆分质量会下降调度器可能把任务分派给不合适的智能体。下一步准备尝试两个方向。一个是基于历史路由结果动态调整路由权重让调度器在反复试错中学会哪些智能体更适合处理哪类任务。另一个是引入“能力自评”机制每个智能体在执行完任务后对自身结果做一个置信度评估调度器汇总这些置信度来动态调配更优质的智能体资源。这两个方向都在实验阶段效果还不敢打包票但整体思路是明确的多智能体系统的上限拼的不是单个模型多强而是调度层有多智能。5.4 写在最后的几点感受做完agency-agents这个项目我最大的感受是多智能体系统的开发模式跟单智能体完全不同。单智能体是“写提示词调模型”多智能体是“设计协作机制管理复杂度”。前者的思维重心在文本后者的思维重心在架构。过程中踩过消息风暴的坑也吃过上下文污染的亏但看到最终多个智能体像一个小团队一样协同完成任务时那种成就感还是相当真实的。如果你正准备上手多智能体系统我的建议是不要一开始就搭复杂的框架先把一个两到三个智能体的极简场景跑通再逐步增加智能体数量。每一步都保持系统可观测、可调试你的复杂度控制能力就会跟上架构膨胀的速度。
最近在做一个多智能体协作系统项目代号就叫 agency-agents。这个系统本质上是一个把多个AI智能体组合起来、让它们像一个小团队一样分工协作的框架。它解决的问题很直接单个智能体在处理复杂任务时经常顾此失彼——又要理解用户意图又要调用工具还要管理上下文最后输出质量很难保证。而通过“多个智能体各管一段、互相配合”的方式可以把复杂任务拆碎让每个智能体只做自己最擅长的事再由一个调度层统一协调。这篇文章我会把这个系统的设计思路、核心代码结构、实测数据和踩过的坑完整记录下来给正在做多智能体方案的开发者一个可参考的样本。1. 项目整体设计与方案选型1.1 多智能体系统要解决的三个核心矛盾构建多智能体系统之前得先想清楚它到底解决什么问题。如果只是做一个带工具调用的单智能体应用那没必要引入多智能体的复杂度。我在实际项目中遇到的核心矛盾有三个。第一个是上下文长度有限。单个智能体的上下文窗口再大也扛不住长时间任务积累的对话历史、中间结果和工具返回数据。比如让一个智能体同时处理数据清洗、特征工程、模型调参三个环节它在第三个环节时早就忘了第一步的具体细节输出质量断崖式下降。第二个是职责混杂导致指令跟随变差。同一个智能体既当数据分析师又当代码审查员还要负责用户沟通它的系统提示词会变得极其臃肿。我在测试中发现超过两到三层职责叠加后智能体经常在错误的时间调用错误的工具。它开始“自由发挥”明明该做数据统计它却跑去生成图表。第三个是失败定位困难。单智能体应用一旦输出错误你很难判断是提示词问题、模型问题还是工具调用的问题。多智能体把任务切成独立模块后每个环节都可以单独验证出错时可以快速定位到具体某个智能体。正是这三个矛盾让我决定采用多智能体协作架构。agency-agents 的设计目标不是做一个“更聪明的单体大脑”而是做一个“能协调多个专业大脑的中枢系统”。1.2 为什么选择四层架构而不是完全去中心化在架构选型阶段我对比了三种主流方案完全去中心化的对等网络架构、中心化编排架构、以及分层协作架构。完全去中心化听起来很酷每个智能体都能主动发起对话、请求其他智能体帮忙但实际跑起来问题很多——消息风暴难以控制循环调用经常发生调试时根本看不清消息在智能体之间怎么流转。我最终选择了四层架构接入层、编排调度层、智能体执行层、基础设施层。这个结构有点像一个公司接入层是前台接收用户请求编排调度层是项目经理负责拆解任务、分派给不同执行人智能体执行层是干活的具体员工基础设施层是共享的办公资源比如数据库、消息队列、工具API。选择这个结构的关键考量是“可观测性优先”。每一层职责清晰消息流向固定日志可以按层追踪。即使某个环节出了问题也能快速判断是编排层分派错误还是执行层某个智能体能力不足。牺牲了一点灵活性换来了可控性在多智能体系统的早期阶段这是更稳妥的选择。1.3 技术栈选型与核心依赖技术栈的选择直接决定了开发效率和系统稳定性。我在这套系统里核心依赖如下模型层接入多个大语言模型基础任务用轻量模型复杂推理用强模型。通信层异步消息队列负责智能体之间的消息传递和任务状态同步。工具层统一函数调用协议每个工具注册成标准化接口。存储层向量数据库存长期记忆关系型数据库存任务元数据。可观测层结构化日志加分布式追踪记录每个智能体的输入输出。选型时有个重要的取舍通信层最开始我只用内存队列后来因为任务量上升、消息积压才换成持久化的消息队列组件。如果你的任务是纯同步、量级也不大内存队列完全够用但一旦涉及长时间运行的任务持久化队列几乎是必须的否则进程一重启所有状态全部丢失。2. 核心模块拆解与实现细节2.1 智能体定义用统一的Agent协议封装所有能力agency-agents 里的每个智能体都遵循同一个抽象协议。我定义了一个基础类所有具体智能体都继承自这个类。协议的关键字段包括agent_id、角色描述、可用工具列表、目标函数、以及上下文缓冲区。class BaseAgent: def __init__(self, agent_id: str, role_desc: str, tools: list, model_config: dict): self.agent_id agent_id self.role_desc role_desc self.tools {tool.name: tool for tool in tools} self.model_config model_config self.context_buffer [] # 保存当前任务相关的上下文 async def run(self, task_input: dict) - dict: 执行当前任务返回结构化结果 raise NotImplementedError async def call_tool(self, tool_name: str, params: dict) - dict: 调用注册过的工具 if tool_name not in self.tools: return {status: error, message: ftool {tool_name} not found} return await self.tools[tool_name].execute(**params)这里有个特别容易被忽略的细节role_desc 不只是给智能体看的提示词它同时也是调度器做任务分派时的匹配依据。我在实际项目里用向量化方式把 task_input 和所有智能体的 role_desc 做相似度匹配实现初步路由。这个设计的好处是不需要硬编码分派规则新加一个智能体时只要注册它的角色描述调度器自动就能识别它适合处理什么任务。2.2 编排调度层任务分解与路由策略编排层是整个系统最核心的部分。它接收用户请求后执行三个步骤解析任务目标、分解子任务、分派给对应智能体。任务分解我采用的是“意图拆分加依赖分析”的方式。先用一个轻量模型快速判断任务的整体类型再基于预设的分解模板生成子任务列表。每个子任务都包含任务ID、依赖的任务ID列表、目标智能体类型、输入参数、预期输出格式。class Orchestrator: def __init__(self, agents: dict, decomposer, queue): self.agents agents self.decomposer decomposer self.queue queue async def process(self, user_request: str) - str: # 1. 任务分解 plan await self.decomposer.decompose(user_request) # 2. 按依赖关系构建DAG并调度 task_id_map {} for sub_task in plan.sub_tasks: if sub_task.dependencies: await self.queue.wait_for_dependencies(sub_task.dependencies) target_agent self.route(sub_task) result await self.agents[target_agent].run(sub_task.input) task_id_map[sub_task.task_id] result # 3. 汇总结果 return self.aggregate(task_id_map, plan.output_template)路由策略我用了两种模式组合基于角色描述的语义匹配作为主路由基于关键词规则的硬匹配作为兜底。比如包含“计算”或“统计”的任务优先路由到数据分析智能体包含“代码”或“函数”的路由到代码生成智能体。语义匹配处理未知场景硬匹配保证高频场景不跑偏。2.3 工具注册与权限控制工具是多智能体系统的能力底座工具层设计得不好整个系统就会变成“一个会聊天但什么都干不了的空壳”。我在系统里实现了统一的工具注册协议每个工具都包含名称、描述、参数Schema、执行函数、权限级别五个要素。register_tool( namedatabase_query, description执行只读SQL查询适合数据提取和统计, params_schema{ type: object, properties: { query: {type: string, description: 完整的SELECT语句} } }, permission_levelread_only ) async def database_query(query: str): # 连接只读副本不允许任何写操作 async with get_readonly_connection() as conn: result await conn.fetch(query) return {rows: result}权限控制是这里不可忽视的部分。多智能体环境下因为任务被拆分某个智能体可能只看到任务片段这反而降低了误操作风险。但如果权限设置不当也容易造成更严重的问题——一个原本只该做数据读取的智能体在生成代码时可能试图执行系统命令。我的做法是给每个工具打上权限标签并在智能体执行层加上“只能调用自身权限范围内工具”的硬校验而不是完全依赖模型自觉。2.4 记忆与上下文管理方案多智能体系统的记忆管理比单智能体复杂得多因为每个智能体都有自己的上下文缓冲同时还要共享一部分全局记忆。我把记忆分成三层第一层是各智能体的私有记忆。比如数据分析智能体在处理当前数据集时特征分布、缺失值情况这些信息只保存在它自己的上下文缓冲里。第二层是任务级共享记忆同一任务下的多个智能体可以访问用于传递中间结果。第三层是全局长期记忆存在向量数据库里记录历史任务的经验、用户的偏好、常见问题解决方案。实现上我用了“双写策略”智能体执行完一个子任务后把结果同时写入任务级存储和长期记忆库。写长期记忆库前会做一次摘要压缩避免存储爆炸。这个方案看起来简单但实际调整了好几次。最开始所有中间结果都被完整存入长期记忆库没过多久向量数据库的规模就失控了检索相关性也急剧下降。后来改成只存摘要加关键指标情况才好转。3. 实操过程与核心环节实现3.1 从零构建一个可运行的多智能体Demo如果你也想搭建自己的多智能体协作系统我建议从一个非常窄的场景入手。我当初选的第一个场景是“自动生成数据分析报告”。任务链路是用户提供一个CSV文件路径和几个分析问题系统自动完成数据加载、统计计算、图表生成、报告撰写四个环节最终输出一份Markdown格式的分析报告。这个场景非常适合用来练手因为四个环节边界清晰每个环节都可以分配一个独立智能体数据加载智能体、统计计算智能体、图表生成智能体、报告撰写智能体。它们之间的依赖关系也很明确——数据加载完才能统计统计完才能画图和写报告。实现步骤是这样的定义四个智能体的角色描述和工具列表。实现编排器的任务分解逻辑把用户请求拆成四个子任务。为每个智能体编写具体的提示词模板和执行流程。实现消息队列处理智能体间的数据传递。搭建仪表盘观察各智能体的执行状态和延迟。第一步到第三步用了一天时间第四步到第五步花了两天。大部分时间都耗在调试智能体之间的数据格式不一致上——统计智能体输出的JSON结构图表智能体期望的输入结构两者之间需要做一层字段映射。这个在单智能体系统里完全不是问题但在多智能体系统里就变成了日常摩擦。3.2 数据加载智能体的实现细节数据加载智能体是整个任务链的起点。它接收用户提供的文件路径检查文件存在性和格式读取数据后做基础清洗然后输出一个标准化数据结构。class DataLoaderAgent(BaseAgent): def __init__(self, config): tools [ FileReaderTool(), DataProfilerTool(), DataCleanerTool() ] super().__init__( agent_iddata_loader, role_desc负责读取数据文件检查数据质量执行基础清洗操作, toolstools, model_configconfig ) async def run(self, task_input: dict) - dict: file_path task_input[file_path] # 读取文件 raw_data await self.call_tool(file_reader, {path: file_path}) # 数据画像输出列名、类型、缺失值分布 profile await self.call_tool(data_profiler, {data: raw_data}) # 基础清洗剔除全空行、统一日期格式 cleaned await self.call_tool(data_cleaner, { data: raw_data, profile: profile }) return { data_reference: cleaned[storage_id], column_types: profile[column_types], missing_counts: profile[missing_counts], row_count: cleaned[row_count] }一个比较隐蔽的坑是不要把原始数据直接作为返回值传给下一个智能体。数据量一大消息队列就会吃不住。我在实现中使用共享存储ID数据清洗后存到对象存储返回值里只带一个引用ID下一个智能体通过这个ID去拉取数据。3.3 编排器的DAG执行引擎编排器内部我用了一个有向无环图来管理子任务依赖。最开始的版本是线性的一个子任务执行完接着执行下一个但碰到某些场景会卡死比如报告撰写智能体需要同时等待统计结果和图表结果如果线性执行图表子任务必须等统计子任务完全结束才开始两者之间白白浪费了一次串行等待。改成DAG执行引擎后可以并行处理没有依赖关系的子任务。DAG引擎的核心逻辑class DAGExecutor: def __init__(self, tasks: dict): self.tasks tasks # task_id - TaskDefinition self.dependencies {tid: set(deps) for tid, deps in tasks.items()} self.dependents defaultdict(set) for tid, deps in self.dependencies.items(): for dep in deps: self.dependents[dep].add(tid) async def execute(self, execute_func): ready [tid for tid, deps in self.dependencies.items() if not deps] result_store {} while ready: current ready.pop() results await asyncio.gather(*[ execute_func(self.tasks[tid], result_store) for tid in current ]) for tid, res in zip(current, results): result_store[tid] res # 清理已完成依赖解锁下一批任务 new_ready set() for tid in self.dependencies: self.dependencies[tid] - set(current) if not self.dependencies[tid]: new_ready.add(tid) ready list(new_ready) return result_store并行执行带来一个额外的好处就是整体任务耗时显著下降。在四个子任务的场景里原来的串行耗时是每个子任务耗时之和现在最短路径计算后图表生成和统计计算可以同时跑总耗时约等于“数据加载统计计算报告撰写”的链路耗时。3.4 报告撰写智能体汇总与格式化报告撰写智能体是任务链的最后一环它负责把前面的所有结果转成自然语言报告。这个智能体的系统提示词里最重要的约束是禁止编造数据。所有结论必须引用上游智能体返回的JSON字段直接取数、转述、归纳。我在提示词里专门加了一段强约束你是一个专业的报告撰写智能体。你的任务是把结构化分析结果转化为清晰易懂的Markdown报告。 可用的数据字段包括data_summary、statistical_results、chart_reference。 每条结论必须直接基于上述字段中的实际值不得补充外部知识或自行推断。 如果字段中有异常值或缺失值请在报告中明确指出不要掩盖。 输出格式要求包含标题、数据概览、统计结论、附录图表引用四个部分。实测下来加了这个约束之后报告幻觉率从原先的30%以上降到了5%以内。多智能体系统里每个智能体的输出质量直接影响下游智能体的输入所以对输出幻觉的容忍度更低必须在提示词层面就做严格限制。3.5 实测数据与效果对比我用一组模拟的销售数据做了完整测试数据包含3万行、12个字段分析目标是月度销售趋势和区域分布。实测数据如下总耗时约18秒其中数据加载及清洗用了4.2秒统计计算用了7.8秒图表生成和报告撰写各用了1.5秒和4.5秒。相比单智能体直接处理同样任务总耗时节省了约25%主要节省在统计和图表环节的并行执行上。更关键的是输出质量。单智能体在同一个任务中报告里出现了两个数据错误——它把季度汇总数值直接当成了月度值还漏掉了一个区域的缺失值说明。多智能体场景下统计计算智能体严格返回JSON数值报告撰写智能体只能照实转述数据错误率明显降低。4. 常见问题与排查技巧4.1 消息风暴与递归调用问题多智能体系统里最危险的问题之一就是消息风暴。A智能体请求B帮忙B处理不了转给CC又觉得应该让A来处理于是消息在智能体之间无休止循环消耗大量token还拖垮系统。我在最初测试时踩过这个坑。两个智能体因为角色描述存在重叠调度器把同一个任务反复路由给它们双方都在等待对方给出“最终结果”消息积压了上千条。解决办法有两个层面。第一是路由层加“最大跳数”限制任何任务最多经过三个智能体就会终止并交给兜底智能体。第二是执行层加“任务指纹去重”如果检测到同一个任务指纹在短时间内反复出现直接拒绝执行并报告循环。class LoopProtector: def __init__(self, max_hops3, max_repeats2): self.max_hops max_hops self.task_fingerprints defaultdict(list) def can_execute(self, task_fingerprint: str, current_hops: int) - bool: if current_hops self.max_hops: return False timestamps self.task_fingerprints[task_fingerprint] # 清除一分钟前的记录 recent [t for t in timestamps if time.time() - t 60] self.task_fingerprints[task_fingerprint] recent if len(recent) self.max_repeats: return False self.task_fingerprints[task_fingerprint].append(time.time()) return True4.2 上下文污染与数据错位多智能体系统中不同智能体共享任务级存储时很容易产生上下文污染。比如数据分析智能体在任务中更新了数据集字段报告撰写智能体如果读取的是初始数据引用就会产生数据错位。我遇到过一个典型案例统计计算智能体在执行过程中进行了异常值剔除但报告撰写智能体引用的还是数据加载阶段的统计描述两份数据不一致导致报告出现矛盾结论。排查思路是给每个中间结果加上版本号下游智能体读取时必须指定版本。同时在日志里记录每个智能体的数据输入来源出问题时可以快速追溯是哪一步产生了数据变更。4.3 并发限制与超时处理多智能体系统天然具有并发特性但并发也带来新问题。多个智能体同时调用同一个外部API时可能触发接口限流某个智能体响应缓慢时会阻塞整个任务链路。我处理的策略包括为每个外部API维护独立的并发信号量限制同时请求数量为每个子任务设置超时时间超时后调度器会标记该子任务失败继续执行后续可完成的子任务最后在汇总阶段标注哪些环节失败了。超时时间不能设置太短因为大模型推理本身速度较慢。根据我的测试普通分析类子任务超时设置在30到60秒比较合理涉及长时间工具调用的可以放宽到120秒。4.4 调试多智能体系统的心得调试多智能体系统最大的困难是“看不到全局”。单智能体的输出错误很容易重现多智能体的错误可能依赖当时的历史上下文、并发时序、中间结果状态复现难度高。我的调试方法有三个。第一是给每次任务生成唯一的trace_id所有日志都带上这个ID方便串联完整执行链路。第二是记录每个智能体输入输出的JSON快照存到独立的日志文件不会因为控制台日志刷屏而丢失关键信息。第三是提供一个“单步执行模式”可以手动指定某个子任务单独运行任意一个智能体传入构造好的测试数据快速验证它在隔离环境下的行为。单步执行模式非常关键。我在排查一个图表生成智能体的问题时一开始完全摸不着头脑——它在完整链路里经常生成错误图表单独跑却总是正常。后来发现是上游统计智能体返回的JSON里某个字段在极端情况下会缺失图表智能体没有做边界处理。用单步模式配合构造的异常输入几分钟就定位到了问题。4.5 性能瓶颈与扩展建议多智能体系统跑久了性能瓶颈通常出现在三个地方。第一个是LLM API的调用延迟和并发配额这个受限于服务商只能通过缓存和请求合并来缓解。第二个是消息队列的积压如果子任务执行时间很长队列会持续堆积需要增加消费者实例。第三个是向量数据库的检索延迟长期记忆不断膨胀后检索相关性和速度都会下降需要定期压缩和索引优化。我建议在系统上线初期就做好压测核心目标是弄清楚单个任务平均会派生出多少个子任务、每条消息的平均体积、智能体之间的平均往返次数。这三个指标直接决定了系统能支撑的最大并发任务量。5. 落地经验与后续扩展5.1 从Demo到生产的几个关键改造如果只是跑通Demo内存队列加进程内状态就够了。但要落地到生产环境我做了几个改造。第一是把内存队列替换成持久化消息队列保证应用重启时任务状态不丢。第二是引入配置中心把模型参数、工具开关、路由规则全部外部化不再硬编码。第三是增加监控告警当某个智能体的错误率超过阈值或者任务平均耗时异常升高时能及时收到通知。这些改造听起来都是老生常谈但在多智能体系统里的意义更为重大。因为系统天然是一门“分布式系统”任何单一节点的不可靠都会被多级链路放大。消息队列持久化保证某个执行节点崩溃后下游智能体能够重新拉取任务配置外部化让新智能体上线、旧智能体下线不用重启整个服务。5.2 智能体能力的复用与沉淀随着系统跑得更久我越来越觉得多智能体系统的核心资产不是代码而是已经调试好的智能体能力库和工具库。一个调试好的数据分析智能体换一套提示词可以服务完全不同的业务场景一个工具如果写得好在多个智能体间复用收益是指数级的。所以我在代码之外维护了两类文档。一类是智能体卡片记录每个智能体的适用场景、输入输出格式、已知限制和调试历史。另一类是工具规范文档记录每个工具的调用方式、参数边界、权限级别和常见错误。这两类文档对新加入的开发者非常友好也避免了自己半年后回来看代码时的认知断层。5.3 下一步规划自适应调度与能力学习目前的编排层还依赖预先设定的任务分解模板和角色描述匹配规则扩展性有明显上限。遇到完全没见过的任务类型时拆分质量会下降调度器可能把任务分派给不合适的智能体。下一步准备尝试两个方向。一个是基于历史路由结果动态调整路由权重让调度器在反复试错中学会哪些智能体更适合处理哪类任务。另一个是引入“能力自评”机制每个智能体在执行完任务后对自身结果做一个置信度评估调度器汇总这些置信度来动态调配更优质的智能体资源。这两个方向都在实验阶段效果还不敢打包票但整体思路是明确的多智能体系统的上限拼的不是单个模型多强而是调度层有多智能。5.4 写在最后的几点感受做完agency-agents这个项目我最大的感受是多智能体系统的开发模式跟单智能体完全不同。单智能体是“写提示词调模型”多智能体是“设计协作机制管理复杂度”。前者的思维重心在文本后者的思维重心在架构。过程中踩过消息风暴的坑也吃过上下文污染的亏但看到最终多个智能体像一个小团队一样协同完成任务时那种成就感还是相当真实的。如果你正准备上手多智能体系统我的建议是不要一开始就搭复杂的框架先把一个两到三个智能体的极简场景跑通再逐步增加智能体数量。每一步都保持系统可观测、可调试你的复杂度控制能力就会跟上架构膨胀的速度。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
