☰
多Agent编排系统核心机制:从消息总线到任务调度的源码级解析
2026/10/21 8:43:29 网站建设 项目流程

很多人第一次看到 oh-my-openagent 这个名字,第一反应都是:这不就是个插件合集么,跟 oh-my-zsh 一个路子,给 Agent 配上各种现成的扩展包,装上就能用。我最初也是这么想的,直到我把它的源码从头到尾翻了一遍,才发现这个判断错得离谱。它的目录结构、运行机制、以及 Agent 之间的协作方式,从头到尾都在做一件完全不同的事:它不是往某个 Agent 身上挂插件,而是在搭建一个能让多个 Agent 自主分工、互相通信、协同完成复杂任务的编排系统。

这篇文章我会从源码结构、消息通信、任务调度、技能扩展、状态管理这几个维度,拆解它到底是靠什么机制实现了"多 Agent 编排",以及这种设计思路和传统插件模式到底差在哪里。如果你正在选型多 Agent 框架,或者想理解 Agent 编排系统内部的工作方式,这篇应该能给你一个比较完整的参考。

1. 先看目录:项目骨架会暴露真实定位

1.1 插件世界的习惯性预期

用过 oh-my-zsh 这类项目的人,对"插件"的预期其实是固定的:项目根目录下会有一个 plugins 目录,里面每个子文件夹就是一个独立插件,彼此之间几乎不通信,只在宿主程序加载时被逐个挂载。这种架构的核心特征是宿主程序掌握全部控制权,插件只是被动等待调用的能力单元。

我带着这个预期打开 oh-my-openagent 的仓库时,首先找的就是 plugins 目录。结果根目录下根本没有这玩意儿,取而代之的是 core、intrinsics、runtime 这类目录。这个细节很说明问题——一个真正的插件项目,不会把 core 放在最显眼的位置,因为它的核心价值是"外围扩展"而不是"内核机制"。

1.2 目录结构的核心模块

实际翻下来,它顶层目录大概是这样组织的:

  • core:Agent 抽象、任务规划器、执行上下文、生命周期钩子
  • intrinsics:内置的技能实现,比如代码执行、文件读写、网络请求
  • runtime:消息总线、事件循环、会话存储、并发调度器
  • services:面向外部的接入层,比如 HTTP 服务、命令行入口
  • config:全局配置、Agent 角色声明、技能注册表

注意这个分层逻辑:core 定义"Agent 是什么",runtime 解决"Agent 之间怎么通信",intrinsics 才是一组可以被调用的基础技能。这个结构本身就是在回答一个架构问题:如果多个 Agent 是平等的协作单元,那么系统必须把"骨架"和"血肉"分开。插件模式把扩展点挂在宿主上,而这里把扩展点挂在了运行时上。

1.3 核心抽象暴露了设计意图

读 core 目录下的抽象类时,这个感受更明显。它里面定义了几个顶层概念:

  • Agent:独立的智能体单元,有身份、有记忆、有可调用的技能集
  • MessageBus:全局消息总线,负责事件发布与订阅
  • TaskPlanner:任务规划器,负责拆解复杂任务并指派给合适的 Agent
  • ExecutionLoop:每个 Agent 独立的执行循环

这四样东西凑在一起,已经是一个完整的分布式协作模型了。插件模式不会定义 MessageBus,也不会定义 TaskPlanner,因为这两个概念在"宿主加扩展"的世界里根本不存在。它们的出现只有一个解释:这个系统默认任务会被拆开、分发、并行处理,再汇聚结果——这就是编排系统的基本盘。

提示:判断一个项目是不是真正的多 Agent 编排系统,第一眼看目录就够。核心抽象不是"接口定义了一堆工具",而是"定义了运行时、通信、调度、生命周期"这些编排层必须的组件,那就是系统而非合集。

2. 消息总线:Agent 之间的协作不是函数调用

2.1 为什么插件不需要总线

传统的插件系统里,组件之间是"直接调用"的关系:主程序知道插件的接口,需要的时候直接调用函数。这种模式的好处是简单直观,坏处是耦合度高——一旦插件的数量上去了,依赖关系会迅速变成一团乱麻。

而 oh-my-openagent 里,Agent 之间的协作用的是发布订阅模型,核心就是 runtime 下的 MessageBus。代码里,Agent 不会直接 import 另一个 Agent 的方法,而是把消息发布到一个总线上,由总线根据事件类型路由给订阅了它的 Agent。我摘一段它核心逻辑的伪代码:

# runtime/message_bus.py 简化示意 class MessageBus: def __init__(self): self.subscribers = defaultdict(list) def subscribe(self, event_type, agent_id): self.subscribers[event_type].append(agent_id) def publish(self, event: Event): for agent_id in self.subscribers[event.type]: self._dispatch(agent_id, event) def _dispatch(self, agent_id, event): agent = self.agents[agent_id] agent.handle_event(event)

这段代码初看平平无奇,但它的存在直接决定了系统性质。Agent 之间只需要知道"什么事情发生了",而不需要知道"谁处理它"。比如一个 Agent 完成任务后发布了一个 task.completed 事件,另一个负责汇总结果的 Agent 会收到这个事件并主动开始工作。发起方根本不关心谁来接这个活,接收方也不需要知道消息从哪来。

2.2 消息体里全是编排语义

更关键的是消息本身的设计。它的 Event 对象不是简单的事件名加 payload,而是包含了一整套编排语义的字段:

  • message_id:全局唯一消息 ID,用于追踪完整调用链
  • channel:逻辑通道名,相当于"这个话题的讨论组"
  • intent:消息意图类型,用于接收方判断是否需要响应
  • context_window:上下文窗口引用,指向会话状态存储区
  • reply_to:回复目标消息 ID,用来串联对话关系

这些字段里的 context_window 和 reply_to 最值得注意。它们的存在意味着消息不是一次性投递就完事,而是参与了一个跨 Agent 的长期状态流。A Agent 在处理任务过程中产生的中途状态,可以通过 context_window 传递给 B Agent 继续处理;B Agent 的结果可以通过 reply_to 回到 A Agent 的上下文里。这不是回调,这是事件驱动的协作文档。

2.3 这样的设计解决了什么问题

用一个生活化的类比:插件模式像一个公司里所有员工都直接向老板汇报,老板安排 A 做完给 B,B 做完给 C,员工之间不直接说话。总线模式则像一个开放办公区,大家在一个共享频道里发布进展,谁有能力处理当前状态就接手,做完再发回频道里。

这种机制在多 Agent 场景里几乎是必须的。因为 Agent 的能力边界是不固定的,你没法在主程序里写死"这个任务一定要由某个 Agent 处理"。任务拆分的结果在运行前是不可预测的,所以只能靠事件总线动态匹配。同时,总线还天然支持了并行——不同 channel 上的事件互不阻塞,多个 Agent 可以同时处理各自的事件流。

注意:如果你在写多 Agent 系统时发现代码里全是 Agent 之间的直接函数调用,多半还没脱离单机程序的思维模式。真正的编排系统里,Agent 之间的关系应该是"通过事件互相感知"而不该是"直接引用互相调用"。

3. 任务编排:Orchestrator 如何拆解和分发任务

3.1 请求进来先过规划器

消息总线解决的是通信问题,但通信本身不会让系统变得更智能。真正的编排逻辑集中在 TaskPlanner 这一层。

当一个任务请求进入系统时,它不会像插件模式那样直接查表找到某个对应工具来执行,而是先被送到规划器。规划器的工作分三步:理解任务目标、拆解子任务、分配 Agent。这个分配不完全是写死的规则匹配,而是基于每个 Agent 的能力描述做动态选择。

我看到的核心调度逻辑是这样:

输入:用户请求 "帮我收集三个竞品的价格信息,汇总成对比表" 处理: 1. 意图识别:信息收集类任务 2. 子任务拆解: - 子任务A:搜索目标A的价格信息 - 子任务B:搜索目标B的价格信息 - 子任务C:搜索目标C的价格信息 - 子任务D:汇总A/B/C的结果并生成对比表 3. 依赖分析: - A/B/C 互不依赖,可以并行 - D 依赖 A/B/C 的完成 4. 技能匹配: - A/B/C 分配给具备网页搜索能力的 Agent - D 分配给具备表格生成能力的 Agent

3.2 依赖图的处理方式

这里最值得学的是它对子任务依赖图的管理。规划器输出的不是一张扁平的任务列表,而是一个有向无环图。runtime 内部维护了每个节点的状态:

  • ready:依赖已满足,等待执行
  • running:正在执行中
  • completed:已完成,结果已发布
  • failed:执行失败,等待重试或回滚

任务 D 的初始状态是 blocked,直到 A/B/C 全部变成 completed 才会解锁。这套机制从实现上讲并不复杂,但它让系统具备了承接复杂任务的能力。单 Agent 插件模式下,这种依赖调度根本无从谈起,因为所有执行路径都是预先写死的,任务之间只有顺序没有协作。

3.3 执行失败的恢复逻辑

编排系统还有个细节很打动我:对失败的恢复处理。当某个子任务执行失败时,规划器不会简单地把错误抛回给用户,而是会做一次重试评估。它检查失败原因,如果是外部 API 临时不可用,就换一个备用技能重试;如果是 Agent 本身状态异常,就把该 Agent 标记为降级,并把任务转交给有同等能力的邻居 Agent 继续。

这个"任务转移"的动作很有意思。它通过消息总线广播一个 task.reassign 事件,其他注册了同类技能能力的 Agent 收到事件后,会评估自己的负载并回复是否接收。总线在这里又成了资源调度的一部分,而不只是消息传递工具。

提示:判断编排系统成熟不成熟,看它失败时怎么处理就对了。只会重试 3 次的还算不上编排,能做到"任务转移、负载评估、邻居接管"的,才是真正考虑了多 Agent 协作复杂性的设计。

4. 技能扩展:Agent 的能力不是靠"加载插件"获得的

4.1 能力描述文件的意义

传统插件模式的 Agent 要获得一项新能力,就是"装一个插件",装完这个能力就永久性地挂在 Agent 身上。oh-my-openagent 里确实也有类似"技能"的概念,但做法完全不同。

每个技能包本质上是一份能力描述文件,里面写清楚:这个技能接受什么类型的输入、产出什么结构的数据、依赖哪些外部资源。这个描述不是给人看的文档,而是给规划器的"匹配依据"。规划器在拆解完子任务后,会根据能力描述做语义匹配,而不是靠插件 ID 精确匹配。

我举个例子。一个子任务需要"解析 PDF 并提取表格",技能库里有两个候选:一个是通用的文档解析工具,一个是专门的表格抽取工具。规划器比较两者的 capability 描述,发现后者在"表格结构还原"这个维度上更匹配,就会优先选择它。这种选择机制是运行时的动态决策,不是安装时写死的绑定关系。

4.2 Agent 之间也能互相"调用"技能

这个系统里还有一个有意思的设计:Agent 可以将自己的部分能力以"子技能"的形式暴露给总线上的其他 Agent。这意味着 Agent 之间的协作不只是"你做完给我做",还可以是"A Agent 的规划器发现某一步自己的技能解决不了,就发布一个 skill.query 事件,询问总线上是否有其他 Agent 提供匹配的子技能"。

这个机制让系统的能力边界变得很灵活。没有谁能永远拥有全部能力,但整个系统作为一个整体,能力是"涌现"的。你在配置里定义不同角色的 Agent 时,不需要精确设计每一次协作路径,只需要描述清楚每个 Agent 的能力边界,系统会在运行中自行探索出协作方式。

4.3 核心区别:注册能力还是安装功能

用一张表来对比这两种模式的差异应该更清楚:

对比维度插件模式编排系统的技能注册
能力绑定对象宿主程序Agent 角色
匹配方式插件 ID 精确匹配能力语义动态匹配
协作方式宿主代码写死调用链运行时事件路由
扩展粒度实现功能声明能力
失败恢复重试或崩溃任务转交邻居 Agent

这个区别的直接后果是:插件模式适合做"功能确定"的产品,比如编辑器、终端工具,功能边界清晰,扩展就是加功能。而编排系统的技能注册适合做"任务不确定"的系统,你不知道用户下一个任务会拆成什么样的子图,只能让系统在运行时动态决策。

5. 状态与记忆:多个 Agent 之间怎么共享上下文

5.1 状态存储不在 Agent 内部

插件模式里,状态通常跟着插件走——每个插件维护自己的配置和缓存。但多 Agent 场景下,一个任务的状态是横跨多个 Agent 的,如果状态各自维护,任务流转时就得做大量序列化和传输。oh-my-openagent 把状态抽离成了一个独立层,叫 SessionStore,所有 Agent 共享访问。

这个 SessionStore 不是简单的 key-value 存储,它按会话隔离,每个会话有独立的上下文空间。Agent 从总线收到事件时,会根据事件里的 context_window 字段找到对应上下文,读取这个任务之前的中间结果。这解决了多 Agent 协作里一个很关键的问题:A 做了一半的事,B 怎么知道当前做到哪了?

我在源码里看到 SessionStore 提供了三个层级的记忆:

  • 工作记忆:当前任务的中间变量,任务结束就清理
  • 会话记忆:整个会话的对话历史和决策轨迹,长期保存
  • 长期记忆:跨会话的通用知识,比如用户的偏好、常用的工具组合

这个分层设计非常实际。工作记忆保证了并发任务之间的隔离,不至于同一会话的多个并发子任务互相污染;长期记忆则保证了 Agent 在多次会话中表现一致,不用每次从零开始。

5.2 并发状态隔离的实现

我特别注意到它的并发隔离策略。当规划器把一个大任务拆成多个并行子任务时,每个子任务会被分配到不同的上下文子空间。子空间是读共享、写隔离的——所有子任务都能读到父任务提供的公共上下文,但各自写入的中间结果互不可见,直到子任务完成后由父节点统一合并。

这个机制让我想起分布式系统中的"工作窃取"模式和日志追踪里的 span 概念,但在多 Agent 编排场景里,它承担着更具体的职责:防止多个 Agent 同时修改同一个状态导致上下文错乱。

5.3 状态跟踪与回放

还有一个值得提的功能是状态快照。每个任务节点开始执行前,系统都会自动打一个快照。任务失败重试时,不是从零重跑,而是把状态回滚到最近一个快照,然后重新执行。这个处理和数据库事务的回滚语义很像,但发生在一个充满不确定性的 LLM 任务调度环境里,难度更大,含金量也更高。

我实际调试一个多 Agent 协作流程时,就经历过一次典型的回滚场景:A Agent 调用了外部 API 拿到了部分结果,但在生成最终输出时崩溃了。系统自动回滚到 A 任务开始的快照,把已经拿到结果的缓存保留在上下文里,重新恢复执行时,A Agent 直接使用缓存继续推进,而不是重新调用 API。这个恢复过程对外部 API 的调用次数控制,就是靠快照和状态缓存配合实现的。

注意:如果一个多 Agent 系统内部没有独立的状态管理层,只是靠 Agent 之间互传大段 JSON 来共享上下文,那一旦任务链路变长,光是序列化和传输成本就能拖垮整个系统。状态抽离成独立层,是编排系统走向成熟的必由之路。

6. 这套设计带来的实际开发体验差异

6.1 写"技能"的感受完全不同

我在上面这套机制里实际写过几个自定义技能,体会非常深。传统插件模式下,写一个插件本质上是实现一个接口:宿主调用你,你做事情,返回结果。整个生命周期由宿主控制,你不需要关心其他插件在想什么。

但在 oh-my-openagent 里写技能,你更像是在写一个"简历":你需要声明自己擅长什么、不擅长什么、需要什么资源、输出什么格式。系统不保证你的技能每次都会被调用,它只是在规划器做匹配时把能力描述当作候选之一。第一次跑通时,我感觉自己不是在开发功能模块,而是在为一个"虚拟员工"写岗位说明书。

整个开发流程变成了三步:

  1. 编写技能的核心实现逻辑
  2. 编写能力描述文件,尽量精确地描述输入输出边界
  3. 在测试环境里跑一些需要该技能的复杂任务,观察规划器是否在恰当的场景选中它

第二步是最大的心智转变。描述文件写偏了,技能就永远不会被匹配到;写得太泛,又会被匹配到一堆不合适的任务上。

6.2 查问题的路径更长,但更健康

插件模式出问题,排查路径通常很短:插件报错,看插件日志。编排系统出问题,排查路径就长多了:先确定是规划器拆错了,还是消息没送达,还是 Agent 状态异常,还是技能匹配错了。这听起来像是劣势,但实际用下来,它带来的好处是问题边界非常清晰。

我遇到过两次"任务失败但找不到原因"的情况,最终都靠链路追踪定位到了问题。系统内部记录了每个事件从发布到被订阅者消费的完整路径,包括路线经过了哪个 channel、消息在哪个节点被挂起、以及上下文状态在哪个环节发生了意外变更。插件模式下,因为所有逻辑都耦合在宿主代码里,这类问题反而更难定位——你得把整个执行过程翻一遍才能找到问题源头。

这套病态排查体验的本质区别在于:插件系统里程序是全部逻辑的载体,出了问题你检查的是程序;编排系统里逻辑分布在各个 Agent、总线、规划器、状态管理层里,出了问题你检查的是链路。

6.3 什么项目适合选这种架构

看完整套机制,我个人的判断是:如果你的产品是一个功能明确、调用链固定的工具型应用,插件模式仍然是效率最高的选择,引入编排系统反而是过度设计。但如果你做的场景需求多样、任务复杂、经常需要动态组合不同能力,比如说:

  • 企业级的智能助理(一个任务涉及信息检索、数据分析、报告生成)
  • 自动化研究助手(需要多轮搜索、阅读、验证、总结)
  • 复杂的业务流程自动化(需要多个领域的 Agent 协作完成一条完整业务线)

那 oh-my-openagent 这类多 Agent 编排框架会给后续扩展留下巨大的空间。前期多花一些时间理解它的消息流和调度机制,后期你需要新增 Agent 角色时,几乎不需要改动已有系统,只需要注册新角色的能力描述就行。

这种"可插拔 Agent 角色"的能力,最终才是它区别于插件集合的真正分水岭。

7. 我踩过的坑和摸索出来的实践建议

代码翻得越深,越发现这类编排系统真正难的不是单个组件,而是组件之间的配合。我在本地搭了一套最小环境,跑了几个真实的协作任务,踩了三个值得记录的坑。

第一个坑是技能描述写得过于宽泛。一个负责格式化输出的技能,我在能力描述里写了"文本处理",结果规划器把"总结长文本"的任务也分配给了它,执行结果完全跑偏。后来我把描述细化成"将结构化数据渲染为指定格式的文本",并且明确标注了不擅长语义级别的文本理解,调度马上就精准了。技能描述的本质是给规划器的"契约",写得太抽象,匹配就一定会出偏差。

第二个坑是 Agent 之间的上下文隔离没配置好。我在同一个 SessionStore 里给两个 Agent 开了同一个上下文空间,结果并发执行时出现了一个 Agent 覆盖另一个 Agent 中间数据的情况。后来严格按照"子任务独立子空间、父节点统一合并"的模式改造之后,问题才消失。多 Agent 系统里,上下文隔离永远是第一优先级。

第三个坑是盲信事件机制导致任务链路悬空。我最初设计了一个"A 发事件给 B 做后续处理"的流程,但没考虑到 B 因为负载过高可能拒绝接收。结果就是事件发布了,订阅了,但 B 评估后拒绝了,任务状态永远停在"已发布未处理"。后来我在发布端加了一层确认机制,如果一定时间内没有被任何 Agent 认领,规划器会重新做一次能力匹配,换一个候选 Agent 继续执行。

现在我个人的实践建议是:不管用什么编排框架,都要重视"可观测性"。多 Agent 系统运行起来就是一个分布式系统,如果没有完整的链路追踪和状态可视化,一旦任务链路变得复杂,调试几乎是盲人摸象。我在本地环境里把所有消息都打点输出,每个事件从发布到消费的路径都能用文字日志完整还原,这在初期调试阶段帮了大忙。

另外,编排系统的启动配置千万不能一把梭,要分批接入 Agent。我先只启一个基础角色,跑通最核心的执行链路,再逐步加上其他角色。这个习惯帮我节省了大量的排查时间——角色越多,变量越多,一次性全接进来出错了根本定位不到是哪个 Agent 的问题。

从"把它当插件用"到"把它当系统用",这个认知转变花了我不少时间。但一旦理解了它的编排本质,再看市面上别的多 Agent 工具,就能比较快地判断出哪些是真正在解决协作问题,哪些不过是给单个模型包了一层工具壳。如果你也在研究多 Agent 系统的设计,建议先别看各种高深的理论文章,直接找一套这类框架的源码,从消息总线和规划器入手翻一遍,收获会比读一百篇文章都大。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询