以下參考課程 LLM Twin: Building Your Production-Ready AI Replica 撰寫

在探討完設計模式後，我們將聚焦第三堂課中提及的 Change Data Capture (CDC) 技術。在構建 GenAI 服務時，由於資料源會不斷更新，我們需要確保來自 LinkedIn、Medium 和 GitHub 等多個平台的資料和資料庫保持同步和最新，而 CDC 正是解決這一問題的方法。

一樣會分成三天的篇幅由淺入深的介紹。

CDC簡介

Change Data Capture (CDC) 是一種用於捕獲資料庫變更的技術。CDC 的主要功能包括：

• 識別資料變更：監控並記錄資料庫（如 MongoDB）的插入、更新和刪除操作。
• 生成變更日誌：將捕獲的變更轉換為結構化的日誌。
• 傳播變更訊息：將變更事件發送到消息隊列（如 RabbitMQ），供其他系統處理。

CDC 在現代數據系統中有多重應用和優勢。首先，它通過實時複製變更至目標資料庫，確保了服務間的一致性。其次，CDC 通過僅處理變更數據而非進行全局同步，達成系統效能最優化。而 CDC 也支持事件驅動架構，將數據變更視為觸發相應系統操作的事件，統一接收端介面。

在分佈式系統中，CDC 解決了一致性問題：它解耦了數據庫更新和消息發送，將數據庫作為事件源，並確保即使在應用程序故障的情況下也能可靠地發送消息。這些特性共同增強了系統的靈活性、效率和可靠性，使 CDC 特別適用於複雜的分佈式環境。

CDC 的工作流程

透過課程的圖片，我們可以拆解 CDC 的工作流程

1. 左上區塊（橘黃色部份，沒有 CDC 的情況）
   1. 應用程式觸發：透過由 REST 調用或事件來觸發。
   2. 更新資料庫：App 首先更新資料庫。
   3. 資料庫確認：資料庫向 App 確認更新完成。
   4. 發送訊息：App 嘗試透過訊息服務發送訊息（如 Kafka）
   5. 在沒有 CDC 的情況下，如果 App 在步驟3完成、步驟4成功之前失敗，就會導致數據不一致。
2. 右下區塊（藍色部份，有 CDC 的情況）
   1. 有了 CDC 機制後，會解耦資料庫的更新與訊息的傳送。
   2. CDC 會監控數據庫的變更日誌（如 MongoDB 的 oplog）。
   3. 當發現變更時，CDC 自動捕獲這個事件。
   4. CDC 將變更事件發送到消息服務（如 Kafka 或 RabbitMQ）。
   5. 而其他系統也可以訂閱這些消息，保持數據的一致性。

CDC 在 GenAI 系統中的應用

在 GenAI 系統中，特別是在實現檢索增強生成（RAG）時，CDC 扮演著關鍵角色：

1. 知識庫實時更新
   • 目的：確保 GenAI 系統能夠使用最新的知識來回答問題。
   • CDC 解決方案：
     • 監控文檔數據庫的變更。
     • 當檢測到變更時，自動生成新文檔的文本向量。
     • 將新生成的向量更新到向量數據庫中。

# 偽代碼範例
def on_document_change(change_event):
    doc = change_event['fullDocument']
    embeddings = generate_embeddings(doc['content'])
    vector_db.upsert(doc['id'], embeddings)

cdc_stream.watch(on_document_change)

2. 訓練數據同步
   • 目的：確保模型訓練過程能及時利用新的訓練數據。
   • CDC 解決方案：
     • 監控訓練數據的變更。
     • 當增加新的訓練數據時，將其推送到訓練隊列。
     • 當訓練數據累積到一定閾值時，自動觸發模型微調過程。

def on_training_data_change(change_event):
    data = change_event['fullDocument']
    training_queue.push(data)
    if training_queue.size > THRESHOLD:
        trigger_model_fine_tuning()

cdc_stream.watch(on_training_data_change)

3. 多模態數據整合
   • 目的：保持多模態數據（如圖片、音頻）與其文本描述的同步。
   • CDC 解決方案：
     • 監控多模態數據（如圖片）的變更。
     • 當檢測到新的或更新的多模態數據時，自動生成或更新其文字描述。
     • 將生成的描述更新到文本數據庫中。

def on_image_change(change_event):
    image = change_event['fullDocument']
    description = generate_image_description(image['url'])
    text_db.update(image['id'], {'description': description})

cdc_stream.watch(on_image_change)

最後是實際上應用的小建議：

1. 使用資料庫原生的 CDC 功能，如 MongoDB 的 Change Streams。
2. 在更複雜的情境下，考慮使用 Debezium 等專業的 CDC 工具。
3. 設計合理的錯誤處理和重試機制，確保數據同步的可靠性。
4. 監控 CDC 流程，設置適當的告警，及時發現並解決問題。