作者 | 諳流科技魏祥臣

沒有意外，隨著模型規(guī)模的持續(xù)增長和應(yīng)用場景的日益復(fù)雜，AI Infra 也自然的從"單體架構(gòu)" -> "分布式架構(gòu)"進(jìn)行演進(jìn)，例如：

• 在大模型訓(xùn)練和推理階段，隨著模型規(guī)模的增長，需要通過多維度并行技術(shù)（數(shù)據(jù)并行、張量并行、流水線并行等）并發(fā)使用數(shù)百甚至數(shù)千個 GPU 才能滿足訓(xùn)練需求；

• 在智能體應(yīng)用階段，從能對話、寫文案的 Chatbot 到如今能自主規(guī)劃、工具調(diào)用、多 Agent 協(xié)作，工具越來越智能，調(diào)用鏈也越來越復(fù)雜；

• 再到各行業(yè)落地時，應(yīng)用的業(yè)務(wù)主路徑開始集成 AI 能力，也對部署架構(gòu)本身的高可靠、高可用及高性能提出了更多的要求

然而這個從單體架構(gòu)到分布式架構(gòu)的升級，最核心的變化就是通過消息中間件讓數(shù)據(jù)、模型、服務(wù)之間能夠異步、可靠、松耦合地協(xié)同工作，從而構(gòu)建可擴(kuò)展、可維護(hù)、可演進(jìn)的 AI 平臺的基礎(chǔ)設(shè)施。

Pulsar 作為消息中間件的中流砥柱，以其更鮮明的存算分離、云原生特性，可發(fā)揮著更大的價值。

1 多模態(tài)：讓 Pulsar 直接吞進(jìn) 超大消息，多模態(tài)訓(xùn)練“零”切片

傳統(tǒng)的單模態(tài)模型，如自然語言處理（NLP）模型僅處理文本，計算機視覺（CV）模型僅處理圖像，自動語音識別（ASR）模型僅處理音頻，它們彼此獨立。多模態(tài) AI 旨在讓機器能夠像人類一樣，通過融合和理解來自多種感官通道（如視覺、聽覺、語言）的信息來進(jìn)行感知、推理和交互。這個給多模態(tài)訓(xùn)練增加了不小的難度；

多模態(tài) AI 系統(tǒng)處理的數(shù)據(jù)類型遠(yuǎn)超傳統(tǒng)文本，包含了圖像、視頻、音頻、3D 點云等大體積的非結(jié)構(gòu)化數(shù)據(jù)。這些數(shù)據(jù)單個文件的大小就可能從幾 MB 到幾 GB 不等。其他的消息隊列系統(tǒng)往往對單條消息的大小有嚴(yán)格限制（例如，Kafka 默認(rèn)單條消息上限約 1 MB，調(diào)參后雖可放大，但需權(quán)衡副本同步壓力。），這迫使開發(fā)者在傳輸大文件時采用復(fù)雜的變通方案，如將文件存儲在對象存儲中，然后在消息中只傳遞文件的路徑或 URL。

這種方式雖然可行，但增加了系統(tǒng)的復(fù)雜性和處理延遲，并且無法充分利用消息隊列在數(shù)據(jù)流管理和處理方面的優(yōu)勢 。

然而 Pulsar 原生支持超大消息體，即 pulsar 的 chunk message，Pulsar 的 Chunk Message 是多模態(tài)訓(xùn)練的數(shù)據(jù)管道利器，它解決了大消息傳輸?shù)耐暾浴㈨樞蛐?、簡化性三大問題，可顯著降低多模態(tài)數(shù)據(jù)管道的工程負(fù)擔(dān)，使開發(fā)者聚焦模型邏輯而非傳輸細(xì)節(jié)。

Pulsar Chunk Message（分塊消息）是 Apache Pulsar 提供的一種用于透明處理超大消息（>5MB）的機制。它允許生產(chǎn)者端自動將大消息拆分為多個小塊傳輸，并在消費者端自動重組，業(yè)務(wù)層無需感知分塊細(xì)節(jié)。

2 多模態(tài)：用 Pulsar 把文本、圖像、音頻流綁定到一起

多模態(tài) AI 需要處理和融合的數(shù)據(jù)類型極其多樣化。系統(tǒng)需要同時處理文本（自然語言描述）、圖像（像素矩陣）、音頻（波形信號）、視頻（圖像序列和音頻流的組合）等多種異構(gòu)數(shù)據(jù)。

在許多場景中，不同模態(tài)的數(shù)據(jù)在時間上存在緊密的依賴關(guān)系。例如，在視頻理解任務(wù)中，音頻中的對話內(nèi)容需要與視頻中人物的口型、動作在時間上精確對齊；在自動駕駛場景中，激光雷達(dá)的點云數(shù)據(jù)、攝像頭的圖像數(shù)據(jù)和 GPS 的定位數(shù)據(jù)必須在同一時間點或時間窗口內(nèi)進(jìn)行融合，才能構(gòu)建出對周圍環(huán)境準(zhǔn)確的感知。因此，消息中間件不僅要能傳輸數(shù)據(jù)，還需要提供機制來保證跨模態(tài)數(shù)據(jù)的時間同步和順序性 。

利用 Pulsar 的 keyshare 消費模型，可以將同一 key 的數(shù)據(jù)總是被路由分配到同一實例完成聚合，方案如下：

• 時間同步：選定一個物理時鐘源（PTP/NTP/ 幀時鐘），所有模態(tài) Producer 在本地打時間桶 ID（t-bucket），粒度 = 1 ms 或 1 幀間隔。

• Produce 發(fā)送：每條消息把 t-bucket 放在 Pulsar 的事件時間（ eventTime ，SDK 原生字段）里，同時作為路由 Key。

• 消費者使用 Key_Shared 訂閱，Key = t-bucket，Pulsar 可保證相同 Key 的消息只會發(fā)給同一消費者實例

• 收到模態(tài) A、B、C 的同一桶消息后，再打包成一條 MultiModalFrame 喂給模型；

Key_Shared（鍵共享）是 Pulsar 的一種訂閱模式，它在 Shared 模式的基礎(chǔ)上增加了按消息 Key 的路由規(guī)則：相同 Key 的消息始終被分配到同一個消費者，而不同 Key 的消息可分布在多個消費者上并行處理，實現(xiàn)Key 級別的有序性與負(fù)載均衡。

3 模型訓(xùn)練：用好 Pulsar 壓縮 Topic，實現(xiàn) Checkpoint 秒級斷點續(xù)訓(xùn)

模型訓(xùn)練周期長、數(shù)據(jù)量大、集群規(guī)模大，出現(xiàn)中斷的概率顯著提高，且重啟代價高昂；

所以通常會使用Checkpoint機制來加速恢復(fù)的過程，但保存 Checkpoint 耗時較高，若存儲服務(wù)瞬時故障，寫入請求直接丟失，導(dǎo)致訓(xùn)練狀態(tài)丟失。

引入 pulsar 作為中間層后，可以將異常數(shù)據(jù)跳過、Checkpoint 異步緩存、任務(wù)級重試等操作都交給 pulsar 的特性來解決，方案如下：

Checkpoint 數(shù)據(jù)具有明顯的歷史消息無效的特性，如果發(fā)生積壓時，只有最新的一條 checkpoint 才有價值，這時可以使用 Pulsar 的壓縮 Topic(Compaction Topic)，壓縮 topic 將 Checkpoint Topic 從日志流變?yōu)?KV 存儲，僅保留每個 Key 的最新消息，自動清理歷史版本，這樣對比傳統(tǒng)方案（掃描 S3 文件列表 → 排序 → 下載）需要耗時 3-5 分鐘到直接接收最新 Key 的方案，耗時<1S;

Compaction Topic是 Apache Pulsar 提供的一種基于消息 Key 的日志壓縮機制，它會自動清理主題中每個 Key 的舊版本消息，僅保留最新版本，從而顯著減少主題體積、加速消費回溯，適用于"只關(guān)心最終狀態(tài)"的場景。

4 模型訓(xùn)練：以 Pulsar 為“輸油管”：優(yōu)化模型訓(xùn)練中的 GPU 饑餓

在大規(guī)模模型訓(xùn)練中，數(shù)據(jù)是驅(qū)動整個訓(xùn)練過程的“燃料”，特別是針對擁有數(shù)十億甚至萬億級參數(shù)的深度學(xué)習(xí)模型，能高效且穩(wěn)定的確保“燃料”能夠持續(xù)、穩(wěn)定地供應(yīng)給計算引擎（如 GPU 集群）是關(guān)鍵所在。

訓(xùn)練這些龐然大物需要海量的訓(xùn)練數(shù)據(jù)，這些數(shù)據(jù)通常以 TB 甚至 PB 計。數(shù)據(jù)加載和預(yù)處理的速度直接決定了 GPU 這一昂貴計算資源的利用率。有數(shù)據(jù)表明 I/O 延遲使 GPU 每步等待數(shù)百毫秒，空閑率可達(dá)30-50%。為了充分利用昂貴的計算資源，必須確保數(shù)據(jù)能夠以足夠快的速度被加載到每個計算節(jié)點的內(nèi)存中，如果數(shù)據(jù)供給速度跟不上模型消耗數(shù)據(jù)的速度，就會有大量時間浪費在等待數(shù)據(jù)上，即所謂的“數(shù)據(jù)饑餓”問題。

歷史的架構(gòu)中，數(shù)據(jù)預(yù)處理模塊與訓(xùn)練模塊存在耦合的情況，然而耦合的模塊可能相互影響從而降低了 GPU 的讀取效率；

這種架構(gòu)中，非常適合引入 Pulsar 在其中作為緩沖層，在數(shù)據(jù)平面預(yù)處理服務(wù)獨立擴(kuò)展，訓(xùn)練節(jié)點只專注消費，利用 Pulsar 的高吞吐特性，“喂養(yǎng)”GPU 的數(shù)據(jù)高速且穩(wěn)定；

并且當(dāng) GPU 消費慢時，還可以利用 Pulsar 的背壓機制，預(yù)處理消費時自動降低預(yù)取速率，避免 OOM，從而讓整個鏈路更加健壯；

不止如此，還可以繼續(xù)針對 topic 的消費進(jìn)行積壓監(jiān)控，如果出現(xiàn)積壓，輔以 K8S 的KEDA 機制+ Pulsar 的 Share 消費類型可以整個擴(kuò)縮容過程更加平滑和穩(wěn)定；

• 背壓（Backpressure）是 Pulsar 中用于防止生產(chǎn)者過載消費者的流量控制機制。當(dāng)消費者處理速度跟不上生產(chǎn)者發(fā)送速度時，系統(tǒng)通過多級反饋控制主動減緩上游生產(chǎn)速率，避免內(nèi)存溢出、數(shù)據(jù)丟失和系統(tǒng)崩潰。

• KEDA（Kubernetes Event-driven Autoscaling），是一種基于事件驅(qū)動的自動擴(kuò)容解決方案，支持通過外部事件源動態(tài)調(diào)整 Pod 副本數(shù)；

• Share 消費類型（Shared Subscription）是 Apache Pulsar 的一種訂閱模式，允許多個消費者綁定到同一個訂閱名上，消息通過輪詢機制分發(fā)給不同的消費者，每個消息僅會被分發(fā)給一個消費者，不保證消息順序，適合高吞吐、無需順序消費的場景。

5 智能體：利用 Pulsar 輕量化主題（non-persistent）解決 AI 應(yīng)用的異步通信難題

模型迭代日新月異，企業(yè)正在積極把 AI 能力嵌入業(yè)務(wù)流程。然而，企業(yè)應(yīng)用從調(diào)用傳統(tǒng)微服務(wù)應(yīng)用 API 接口 到 調(diào)用大模型“生成式”的 API 接口過程中，一個顯著的特征是任務(wù)處理時耗變的很長，傳統(tǒng)微服務(wù)應(yīng)用通常能實現(xiàn)毫秒級響應(yīng)，而 AI 應(yīng)用的處理周期跨度極大——從幾分鐘到數(shù)小時不等；

這就意味著原本微服務(wù)間的同步調(diào)用就不再適用，可將同步調(diào)用改為異步通知來解決長耗時的阻塞；改為異步通知后，那又如何能實現(xiàn)同步調(diào)用的即時通信吶，可以采取以下模型：

• Agent1 在啟動時注冊一個專屬于自己的用于接收回包的非持久化 Topic（non-persistent Topic），非持久化 topic 非常輕量化，數(shù)據(jù)不落盤存儲，生命周期可由 TTL 自動 或人工回收，Agent1 可使用獨占消費模型進(jìn)行消費該 topic

• 當(dāng) Agent1 有長耗時的調(diào)用模型請求時，向正常 topic 發(fā)送請求，并由模型處理模塊處理；該 Topic 為常規(guī) topic，具備消息持久化、消息回放、海量積壓等隊列特性

• 當(dāng) LLM 處理模塊完成后，根據(jù)請求包中的回包地址進(jìn)行回包投遞

基于此模型，可以利用 Pulsar 的 Persistent-topic，將長時耗任務(wù)進(jìn)行異步化處理，利用 pulsar 的高可用、低延時的特性來保障請求任務(wù)的可靠、解耦和削峰填谷；又可以利用 pulsar 的 Non-Persistent-topic 的輕量化，實現(xiàn)百萬級創(chuàng)建，快速回收等能力

Non-persistent Topic：是 pulsar 的一種 topic 類型，是“不落盤、純內(nèi)存” 的消息通道——數(shù)據(jù)不會寫入磁盤、不會做副本復(fù)制，Broker 宕機或進(jìn)程重啟即丟失，因此極致輕量、低延遲，適合“可丟、可重試、要快、要大量”的短時消息場景。

6 智能體：Pulsar 可為事件驅(qū)動的智能體提供“新基建”

AI Agent 的概念正在經(jīng)歷一場深刻的變革，從簡單的對話式 AI（Chatbot）向復(fù)雜的獨立實體轉(zhuǎn)變。AI Agent 就是將一個大模型（大腦）和一系列工具（感官與四肢）組裝起來，形成的一個能夠感知和改變外部環(huán)境的智能程序。

以創(chuàng)建一個營銷 Agent 為例，采用 ReAct 的模型，Agent 可能首先從 CRM 中提取客戶數(shù)據(jù)，使用 API 收集市場趨勢，并在新信息出現(xiàn)時不斷調(diào)整策略。通過通過記憶保留上下文并迭代查詢，Agent 能夠生成更準(zhǔn)確、更相關(guān)的輸出。

當(dāng)外部接口越來越豐富，Agent 需要不斷的擴(kuò)展收集信息來源，包括其他 Agent、工具和外部系統(tǒng)等等，以便做出更精準(zhǔn)的決策。

而這，從系統(tǒng)架構(gòu)設(shè)計的角度上講，就是一個分布式系統(tǒng)問題。這和微服務(wù)時代面臨的挑戰(zhàn)相似，因為在微服務(wù)中，各個組件必須高效地進(jìn)行通信，而不產(chǎn)生瓶頸或僵化的依賴關(guān)系。也和微服務(wù)架構(gòu)系統(tǒng)一樣，它們的輸出不僅僅應(yīng)該回流到 AI 應(yīng)用程序中，它們還應(yīng)該流入其他關(guān)鍵系統(tǒng)，如數(shù)據(jù)倉庫、CRM、CDP 和客戶成功平臺。所以完全可以將 Agent 理解為：有“大腦”的微服務(wù)；

從微服務(wù)的架構(gòu)演進(jìn)來看，Agent 的未來是事件驅(qū)動的，事件驅(qū)動的架構(gòu)需要一個高效的消息中間件作為“基建”，因為消息中間的特性可以很好的匹配事件驅(qū)動需要的橫向擴(kuò)展性、低延遲、松耦合、事件持久化等訴求

Pulsar 除了以上在消息中間件的優(yōu)勢外，還提供了 Function Mesh 的能力，利用 Function 的能力可以更近一步簡化 AI Agent 的架構(gòu)：

• ReAct 模式：ReAct（Reasoning and Action）是目前應(yīng)用最廣泛、最經(jīng)典的 AI Agent 運行模式之一 。其核心思想是模擬人類解決復(fù)雜問題的過程，通過一個 “思考（Thought）→ 行動（Action）→ 觀察（Observation）”的循環(huán)來逐步推進(jìn)任務(wù) 。

• Pulsar Function ：Pulsar 提供的輕量級、Serverless 流處理框架，定位是“用寫普通函數(shù)的代碼量，完成 ETL、過濾、聚合、打標(biāo)簽等實時計算”。它把“消息 → 計算 → 消息”的閉環(huán)直接跑在 Pulsar 集群內(nèi)部，簡單場景不需要額外部署 Flink、Storm 等重型流處理引擎。

7 智能體：具身智能需要“傳感器流 + 任務(wù)隊列”

在具身智能的場景中，既需要處理傳感器讀數(shù)流（連續(xù)、有序的數(shù)據(jù)），也需要處理獨立的指令或任務(wù)（這些任務(wù)需要獨立處理）；

例如：一個機器人 agent 在處理任務(wù)時，首先機器人的視覺或遙測傳感器持續(xù)發(fā)布事件流，這些事件需要按順序來處理或者來理解當(dāng)前所處環(huán)境的變化；然后當(dāng)機器人的 AI 決定采取行動，例如“拾取物體”或“導(dǎo)航到位置”時，這些任務(wù)會被添加到工作隊列中。這些行動消息可能需要多個執(zhí)行器模塊（消費者）會分擔(dān)這些任務(wù)。每個任務(wù)消息會被分配給一個執(zhí)行器，執(zhí)行器在完成任務(wù)后會進(jìn)行確認(rèn)。如果任務(wù)失敗，執(zhí)行器可以發(fā)送負(fù)向確認(rèn)（表示失敗），然后另一個實例可以重試。

我們回顧上述的過程，雖然都是利用消息管道進(jìn)行消息傳遞，但是這是兩種不同的數(shù)據(jù)類型：

• 類似傳感器流，生產(chǎn)者將數(shù)據(jù)追加到一個無界、有序的日志（即流）中。消費者隨后按順序從這個日志中讀取數(shù)據(jù)，并維護(hù)流中的偏移量（offset）。每個分區(qū)內(nèi)的順序是有保障的，消息在消費時不會被移除，這就是 kafka 專注的 stream（流）場景，它提供了高吞吐量和分區(qū)的嚴(yán)格排序，這使得它非常適合處理有序的事件流。

• 類似任務(wù)消息，生產(chǎn)者將消息發(fā)送到隊列，每條消息只由一個消費者處理（即使有多個消費者在監(jiān)聽）。消費者從隊列中拉取消息，并在處理完成后確認(rèn)每條消息，消息隨后會從隊列中移除。隊列擅長分發(fā)可以并行處理且無需全局排序要求的任務(wù)或工作。這就是 rabbitmq、rocketmq 專注的 queues（消息）場景，專注于每個消費者只處理一條消息，并具備消息重試和死信隊列等能力。

然后，在越來越多的 ai 場景，需要兩者兼具，因為 AI 代理在實時環(huán)境中觀測，同時必須執(zhí)行可靠的操作。這正是 pulsar 持續(xù)專注的方向，將流 + 消息進(jìn)行融合，Pulsar 原生支持多種消息語義。其靈活的訂閱模式（獨占、共享、故障轉(zhuǎn)移、鍵共享）讓你能在同一平臺上為不同任務(wù)選擇合適的工具。這意味著系統(tǒng)擴(kuò)展更少，組件間集成更簡單——這對復(fù)雜的 AI 代理架構(gòu)來說是一個很大的優(yōu)勢。

Pulsar 的流（Streaming）和消息（Messaging）場景結(jié)合，通過 Key-shard（鍵共享）、Failover（故障轉(zhuǎn)移）、Exclusive（獨占）、Shared（共享）四種訂閱模式來實現(xiàn)

參考

https://huggingface.co/spaces/nanotron/ultrascale-playbook

https://seanfalconer.medium.com/the-future-of-ai-agents-is-event-driven-9e25124060d6

https://www.linkedin.com/pulse/kafkas-role-powering-next-wave-event-driven-agentic-ai-jeyaraman-xq0kc

https://streamnative.io/blog/streams-vs-queues-why-your-agents-need-both--and-why-pulsar-protocol-delivers

https://dzone.com/articles/agentic-ai-using-apache-kafka-as-event-broker-with-agent2agent-protocol