多智能体协作系统架构设计与落地实践：从编排调度到DAG执行引擎

发布时间：2026/10/24 11:19:16
多智能体协作系统架构设计与落地实践：从编排调度到DAG执行引擎

最近在做一个多智能体协作系统，项目代号就叫 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": f"tool {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( name="database_query", description="执行只读SQL查询，适合数据提取和统计", params_schema={ "type": "object", "properties": { "query": {"type": "string", "description": "完整的SELECT语句"} } }, permission_level="read_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格式的分析报告。

这个场景非常适合用来练手，因为四个环节边界清晰，每个环节都可以分配一个独立智能体：数据加载智能体、统计计算智能体、图表生成智能体、报告撰写智能体。它们之间的依赖关系也很明确——数据加载完才能统计，统计完才能画图和写报告。

实现步骤是这样的：

  1. 定义四个智能体的角色描述和工具列表。
  2. 实现编排器的任务分解逻辑，把用户请求拆成四个子任务。
  3. 为每个智能体编写具体的提示词模板和执行流程。
  4. 实现消息队列，处理智能体间的数据传递。
  5. 搭建仪表盘，观察各智能体的执行状态和延迟。

第一步到第三步用了一天时间，第四步到第五步花了两天。大部分时间都耗在调试智能体之间的数据格式不一致上——统计智能体输出的JSON结构，图表智能体期望的输入结构，两者之间需要做一层字段映射。这个在单智能体系统里完全不是问题，但在多智能体系统里就变成了日常摩擦。

3.2 数据加载智能体的实现细节

数据加载智能体是整个任务链的起点。它接收用户提供的文件路径，检查文件存在性和格式，读取数据后做基础清洗，然后输出一个标准化数据结构。

class DataLoaderAgent(BaseAgent): def __init__(self, config): tools = [ FileReaderTool(), DataProfilerTool(), DataCleanerTool() ] super().__init__( agent_id="data_loader", role_desc="负责读取数据文件，检查数据质量，执行基础清洗操作", tools=tools, model_config=config ) 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处理不了转给C，C又觉得应该让A来处理，于是消息在智能体之间无休止循环，消耗大量token还拖垮系统。

我在最初测试时踩过这个坑。两个智能体因为角色描述存在重叠，调度器把同一个任务反复路由给它们，双方都在等待对方给出“最终结果”，消息积压了上千条。

解决办法有两个层面。第一是路由层加“最大跳数”限制，任何任务最多经过三个智能体就会终止并交给兜底智能体。第二是执行层加“任务指纹去重”，如果检测到同一个任务指纹在短时间内反复出现，直接拒绝执行并报告循环。

class LoopProtector: def __init__(self, max_hops=3, max_repeats=2): 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 True

4.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这个项目，我最大的感受是：多智能体系统的开发模式跟单智能体完全不同。单智能体是“写提示词调模型”，多智能体是“设计协作机制管理复杂度”。前者的思维重心在文本，后者的思维重心在架构。过程中踩过消息风暴的坑，也吃过上下文污染的亏，但看到最终多个智能体像一个小团队一样协同完成任务时，那种成就感还是相当真实的。

如果你正准备上手多智能体系统，我的建议是：不要一开始就搭复杂的框架，先把一个两到三个智能体的极简场景跑通，再逐步增加智能体数量。每一步都保持系统可观测、可调试，你的复杂度控制能力就会跟上架构膨胀的速度。