Apache RocketMQ × AI：面向 Multi-Agent 的事件驅動架構

介紹了如何基于 Apache RocketMQ 新特性構建異步化 Multi-Agent 系統，深入探討了 Agent 間的異步通信、上下文隔離、狀態恢復與任務編排機制，并通過實際案例展示如何利用 RocketMQ 實現 Multi-Agent 的任務調度。

作者：周禮

本文整理自阿里云智能集團高級技術專家周禮在 2025 全球機器學習技術大會上的精彩演講《Apache RocketMQ x AI：面向異步化 Agent 的事件驅動架構》，介紹了如何基于 Apache RocketMQ 新特性構建異步化 Multi-Agent 系統，深入探討了 Agent 間的異步通信、上下文隔離、狀態恢復與任務編排機制，并通過實際案例展示如何利用 RocketMQ 實現 Multi-Agent 的任務調度。

Multi-Agent 協同的核心：Agent 能力發現與任務閉環

隨著大模型能力提升與推理成本下降，MCP、A2A 等協作協議的成熟，AI 邁入了 Agentic AI 的時代，AI 應用也從“被動響應”進入了“主動決策、自主執行”階段。這一演進催生了 Multi-Agent 架構：任務由多個專業化 Agent 協同完成，不再依賴單一模型或固定流程，開發者得以在模型自主性與業務可控性之間實現平衡。

與傳統應用的固定編排不同，Agentic AI 具備自主規劃能力，可將目標拆解為動態步驟，但規劃又依賴每一步的結果反饋，依據完整的結果鏈路，Supervisor 才能掌握進展、評估成效，決定下一步行動。因此，要實現真正高效、可靠的協同，僅靠大模型的推理能力遠遠不夠，Agent 之間的協作有兩個關鍵點：能力發現與任務閉環。

1、Agent 能力發現

Agent 的能力發現有兩個主要功能：

1. 動態注冊 Agent 的能力（如“我能做數據分析”、“我擅長文案生成”）；

2. 支持 Supervisor Agent 在運行時查詢并選擇合適的 Sub Agent 執行任務。

若沒有 Agent 能力發現的功能，Agent 協作就只能依賴硬編碼，喪失了自主性與擴展性。這一機制可以類比傳統微服務中的服務發現與地址查找，調用方依賴注冊中心查詢服務地址，查找的條件是編碼的 ServiceName。Agent 能力的發現與之相似而又有不同，它是面向語義的能力和意圖驅動匹配，而能力識別和匹配又是交給大模型處理的，這個過程是實現智能分工的前提。

2、任務協同

在大模型（LLM）驅動的多 Agent 系統中，多個智能體（Agent）通過協作、競爭或分工完成復雜任務。尤其在 Supervisor Agent 架構中，Supervisor 作為系統的“大腦”，通過高層次的協調與管理，將多個專業化 Agent 組織成一個有機整體，從而完成單個 Agent 無法勝任的復雜任務。在 Agent 能力的聲明與發現的基礎上，為了實現 Supervisor 與其他 Agent 間的高效協同，需要設計合理的通信機制。不同的通信模式適用于不同場景，在靈活性、可擴展性、控制力和性能之間各有取舍。

• 輪詢式通信： 定期主動查詢其他 Agent 的狀態，獲取最新信息（如存儲于數據庫、Redis 中）。這種方式實現簡單，但是延遲高，頻繁輪詢浪費資源，難以處理動態拓撲變化。
• 點對點通信（Point-to-Point Invocation）： 主動調用另一個 Agent 的接口，等待響應，如 REST API、gRPC、函數調用等方式。這種方式控制流明確，支持強一致性交互，但耦合度高，難以動態調整，不利于異步處理。
• 發布-訂閱模式（Pub/Sub）： Agent 將消息發布到特定主題（topic），其他感興趣的 Agent 訂閱該主題接收消息，解耦調用和被調用者，易于擴展和水平伸縮。

我們考察基于發布/訂閱模式實現 Agent 間異步調用的場景，Sub Agent 接收任務并完成后，需要將結果反饋給 Supervisor，并附帶上下文唯一標識（如 Task_ID），以便 Supervisor 異步接收反饋并驅動下一步決策。但要保證 Supervisor Agent 節點異步獲取到上次任務的結果，需要在異步場景下實現反饋機制，常見方案如下：

1. 獨占隊列： 每個 Supervisor 實例綁定獨立 Queue 或者 Topic 來接收下游回寫的結果 —— 資源開銷大，管理復雜，存在性能瓶頸；
2. 廣播過濾： Supervisor 集群共享消費分組，所有實例接收全部消息并自行過濾 —— 產生大量無效流量，浪費資源且有穩定性風險；
3. 共享存儲： 結果存儲在數據庫或者緩存 —— 更靈活可靠，但每次 Supervisor Agent 作為發起方，需要不停進行輪詢以確定自己發起的 Task 是否已經有結果產生，增加了存儲成本與交互成本，實際類似于上述的輪詢式通信。

可以看到，基于發布/訂閱模式實現通信時比較復雜，其核心原因是主流的分布式消息中間件主要面向的是靜態編排的業務場景，采用“發完即忘”（Fire-and-Forget）模式，不關心下游的反饋，這使得通信鏈路難以完成閉環。

RocketMQ 面向 Agentic AI 的新特性

接下來，我們探討如何實現 Agent 的異步通信機制和動態決策，RocketMQ 在傳統模式的基礎上進行了針對性設計，推出帶語義的 Topic 和 Lite-Topic 的新特性：以 Topic 語義作為能力注冊與發現的基礎，解決調用誰的問題；以 Lite Topic 動態綁定任務并等待結果消息，解決調用后異步獲取結果的問題。 兩者結合，以更簡潔的方式實現需要反饋的異步任務驅動模式。

1、從數據通道到語義載體：Topic 的智能化演進

在傳統的消息系統中，Topic 僅作為數據傳輸的通道存在，它定義了“消息發往哪里”，但無法表達“為什么發”或“誰需要它”。然而，在 Multi-Agent 協同場景下，通信不再只是簡單的數據搬運，而是意圖驅動的智能協作過程。為此，我們重新定義了 Topic：它不僅是消息的主題命名和分類，更是業務意圖與能力語義的載體。

通過將自然語言描述與結構化元數據引入 Topic 定義，每個 Topic 不再只是一個主題名稱，而是一個具備“自我表達能力”的協作單元。例如，一個實現了 A2A 協議中 AppCard 標準的 Topic 格式如下：

這樣的設計使得 Topic 具備了可讀性、可發現性與可推理性。結合 Nameserver 的服務注冊與發現機制，這些帶有語義標簽的 Topic 可被統一索引和查詢。每一個 Agent 可以通過訂閱某個能描述自身能力的 Topic 來實現綁定關系，具備注冊和被發現的能力，上層 Agent（如 Supervisor Agent）可通過能力關鍵詞（如“數據分析”“內容生成”）動態發現并使用合適的 Topic 來異步驅動下游的 Agent，在任務編排過程中，Supervisor Agent 能夠像調用函數一樣選擇 Topic，實現基于語義理解的動態路由決策。

2、輕量級消費模式：Lite-Topic

Lite-Topic 是在 RocketMQ 百萬隊列基礎上設計的一種新類型的 Topic，它無需預創建 Topic 和訂閱關系，并且能自動管理生命周期，主要面向短期、小量消息傳輸、客戶端訂閱關系動態臨時變化、訂閱集合高度個性化的場景。這種輕量化消費模型天然就能支持粒度更細的資源隔離，從而支持異步場景下的結果反饋機制，保證 Sub Agent 回寫的結果能讓發起任務的 Supervisor Agent 獲取到。

為了維護這種“千人千面”的訂閱關系，我們提出一種去中心化 + 最終一致性的訂閱關系管理方式：將訂閱關系提前注冊到 Broker，避免每次請求重復傳輸，以增量注冊方式應對新的模型中 Lite Topic 訂閱的頻繁變化，以全量方式做到最終一致性。同時，在組織方式上不再以 Group 為維度，而是以 Client_ID 為維度管理的訂閱關系，可以稱之為每個客戶端的興趣集（InterestSet）。

不同客戶端的訂閱集合維護在服務側：

• 增量/全量同步：實時性和最終一致性；
• 存活檢測：Proxy 通過心跳判斷 Client 是否在線，下線時通知 Broker 清理 InterestSet；
• 分片機制：客戶端的完整的訂閱集合分片存儲。

在消息讀取方面：不再使用傳統 Pull / Pop 模型中客戶端針對每個 Topic 的每個 Queue 發起讀請求的模式，因為這會帶來數千個并發請求，連接與線程開銷會線性增長。讀請求不需攜帶 Topic，而只帶上自己的身份即可，這樣保證即使訂閱集合龐大，讀請求依然是輕量的。因為每個 Lite-Topic 的消息量并不大（幾條到幾百條），發送流量也不高甚至較為離散，但是 Topic 數量多，所以需要一種新的分發機制，在仍然保證低延遲的同時，降低 1:1 讀模式帶來的開銷。

我們引入一個事件驅動的消息分發方式，核心組件是 ReadySet（就緒集合），就緒事件集合是每個 Client_ID 的待讀取 Topic 隊列集合，維護客戶端的 Ready 事件，即“哪些 Topic 有消息可讀”，存放當前有消息可讀的 Lite-Topic。在 RocketMQ 中能觸發這個集合變更的事件如下：

事件源	描述
新消息寫入事件	Producer 發送消息 → 匹配 InterestSet → 加入對應 client 的 ReadySet
ACK 事件	消費者確認部分消息后，若隊列仍有未讀消息
OrderLock 釋放事件	順序消費鎖釋放后，下一消息變為可讀 → 觸發加入
訂閱上線事件	客戶端完成注冊，若有消息 → 立即喚醒

這種直接訪問 ReadySet → 僅處理活躍 topic 的讀取方式，避免了每次讀請求遍歷客戶端所有訂閱集合進而輪詢每個 Lite Topic 帶來的無效讀操作。換個角度看，這個模型是：Pull 模型 + Push 語義。

• 保留 Pull 模型的優勢：客戶端主動控制流控、避免過載；客戶端只需發起一個輕量 Poll 請求；
• 引入 Push 語義的效率：Broker 主動告知“哪些 Topic 有消息可讀”。

從整體上看，通過引入 InterestSet + ReadySet 的事件驅動模型，InterestSet 維護客戶端的訂閱關系，并在事件觸發后分發到對應的 ReadySet，將傳統“盲目輪詢”轉化為“精準喚醒”，最終實現在大規模個性化訂閱場景下的高效、低延遲消息分發。

基于 RocketMQ 構建異步 Multi-Agent 系統

首先，通過上述 Lite-Topic 的能力，我們可以在 Multi-Agent 中更簡潔地異步獲取 Sub Agent 的結果，Supervisor Agent 集群中，任何一個 Supervisor Agent 都可以通過動態訂閱 Lite-Topic（以 Task_id 命名）來接收下游任務結果，實現整個任務的閉環。

其次，再結合語義化 Topic 的 Agent 能力注冊與發現，我們構建了一套面向 Agentic AI 的高效異步協同架構。其核心業務流程如下：

1. 能力注冊與發現： 每個 Sub Agent 在啟動時創建與其業務職責對應的 Topic，并將協議規范、輸入輸出 Schema、自然語言描述等元數據注冊至 NameServer。通過持續訂閱該 Topic 接收任務，Sub Agent 不僅完成了通信接入，更實現了能力的主動暴露。
2. 語義驅動的任務編排： Supervisor Agent 基于用戶目標構建 Prompt 上下文，并動態查詢 NameServer 獲取當前可用的 Topic 列表，將其作為“可調用函數庫”注入大模型。LLM 由此可在真實、可觀測的能力空間中進行任務拆解與路徑規劃，避免了“幻覺式決策”，提升了執行的可行性與可控性。
3. 輕量級異步任務分發與反饋： 在執行階段，Supervisor 向目標 Topic 發送消息，同時為本次調用創建一個臨時的 Lite-Topic 作為專屬回調通道。該機制無需綁定具體實例，即可實現高并發下的結果路由，兼顧性能與靈活性。
4. 閉環驅動的持續決策： Supervisor 訂閱相關 Lite-Topic，異步聚合各子任務的執行結果，重新注入上下文，驅動下一輪推理與編排。整個過程形成一個以反饋為核心、動態演進的決策循環，真正實現了從“靜態流程”到“自主協作”的躍遷。

這套依托 RocketMQ 在發布/訂閱模型上的創新擴展而實現的架構，在保證系統松耦合和高擴展性的同時，有效支持了 Multi-Agent 場景下任務編排、結果反饋和多輪決策的需求。基于 RocketMQ 的這一實踐，為構建可靠、可控的異步智能體協作系統提供了一種可行的技術路徑。

展望未來，Apache RocketMQ 將持續在 AI 領域進行技術迭代與創新。誠邀您點擊此處參與問卷調研，反饋真實使用場景和痛點，幫助我們打造更符合 AI 時代需求的消息引擎。也歡迎釘釘搜索加入 RocketMQ for AI 用戶交流群（群號：110085036316），與我們交流探討～