這三天總共會介紹以下六種設計模式在 GenAI 的應用，今天進入到第二天：

觀察者模式 (Observer)

1. 介紹：是一種行為設計模式，允許多個觀察者（如：數據處理流程）自動響應被觀察者（如：資料庫）的變化。這種設計支持鬆耦合的系統架構，適合如事件驅動的數據處理系統。
2. Gen AI 應用情境：
   • 即時模型更新：當取得新的訓練數據時，自動觸發模型的 RAG 或 fine-tune 過程。
   • 動態特徵工程：根據數據變化自動調整特徵提取策略，以適應不同的數據分佈。
   • 多模態模型協同：在多模態AI系統中，當一個模態的數據更新時，自動更新其他相關模態的模型。
   • 數據質量監控：實時監控輸入數據的質量，當發現異常時自動觸發警報或數據清理流程。
   • 自動化提示詞優化：根據模型輸出的反饋，自動調整和優化提示策略。
   • 分布式訓練協調：在分布式訓練環境中，協調不同節點間的狀態更新和同步。
   • 資源自動擴縮：根據工作負載的變化，自動調整計算資源的分配。
3. 案例與說明：以下使用觀察者模式實現 GenAI 數據處理流程。

class Database:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        if observer not in self._observers:
            self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self, data):
        for observer in self._observers:
            observer.update(data)

    def upload_data(self, data):
        print(f"上傳新數據到資料庫: {data}")
        self.notify(data)

class DataProcessor:
    def update(self, data):
        pass

class DataCleaningPipeline(DataProcessor):
    def update(self, data):
        print(f"數據清洗流程處理數據: {data}")

class FeatureExtractionPipeline(DataProcessor):
    def update(self, data):
        print(f"特徵提取流程處理數據: {data}")

class MachineLearningPipeline(DataProcessor):
    def update(self, data):
        print(f"機器學習流程處理數據: {data}")

def main():
    database = Database()
    
    cleaning_pipeline = DataCleaningPipeline()
    feature_extraction = FeatureExtractionPipeline()
    ml_pipeline = MachineLearningPipeline()

    database.attach(cleaning_pipeline)
    database.attach(feature_extraction)
    database.attach(ml_pipeline)

    database.upload_data("用戶行為數據")
    print("\n---\n")
    database.upload_data("銷售交易數據")

if __name__ == "__main__":
    main()

1. Database 類作為被觀察者，管理觀察者列表並在數據更新時通知它們。
2. DataProcessor 是觀察者的抽象基類，定義了更新接口。
3. DataCleaningPipeline、FeatureExtractionPipeline 和 MachineLearningPipeline 是具體的觀察者，實現了各自的數據處理邏輯。
4. 在 main 函數中，我們創建了數據庫和多個處理流程，並將處理流程附加到數據庫上。
5. 當新數據上傳時，所有附加的處理流程都會自動得到通知並執行其處理邏輯。

模板方法模式 (Template Method)

1. 介紹：是一種行為設計模式，在一個類中先定義了算法框架，再將某些步驟的實現延遲到子類中。模板方法讓子類可以在不改變算法結構的情況下，重新定義算法中的某些步驟。
2. Gen AI 應用情境：
   • 標準化的數據預處理流程：定義通用的數據清理、標準化、擴展等步驟，同時允許針對特定數據類型另制處理。
   • 模型訓練流程：建立統一的模型訓練框架，包括數據加載、模型初始化、訓練循環、評估等步驟，允許針對不同模型架構進行特定實現。
   • 特徵工程流程：定義通用的特徵提取和選擇流程，但允許針對不同的資料型態或任務進行特定的特徵工程。
   • 模型評估框架：建立標準化的模型評估流程，包括測試集準備、指標計算、結果可視化等，同時允許另外制定的評估指標或方法。
   • 模型部署流程：建立統一的模型部署框架，包括模型導出、優化、打包等步驟，同時允許針對不同的部署環境或硬件進行特定優化。
3. 案例與說明

from abc import ABC, abstractmethod

class GenAIPipelineTemplate(ABC):
    def process_data(self):
        self.load_data()
        self.preprocess()
        self.feature_engineering()
        self.train_model()
        self.evaluate_model()

    def load_data(self):
        print("加載數據集")

    @abstractmethod
    def preprocess(self):
        pass

    @abstractmethod
    def feature_engineering(self):
        pass

    def train_model(self):
        print("訓練模型")

    def evaluate_model(self):
        print("評估模型性能")

class TextClassificationPipeline(GenAIPipelineTemplate):
    def preprocess(self):
        print("文本數據預處理：分詞、去停用詞")

    def feature_engineering(self):
        print("文本特徵工程：TF-IDF 轉換")

class ImageClassificationPipeline(GenAIPipelineTemplate):
    def preprocess(self):
        print("圖像數據預處理：縮放、標準化")

    def feature_engineering(self):
        print("圖像特徵工程：卷積特徵提取")

# 使用範例
def main():
    print("文本分類流程：")
    text_pipeline = TextClassificationPipeline()
    text_pipeline.process_data()

    print("\n---\n")

    print("圖像分類流程：")
    image_pipeline = ImageClassificationPipeline()
    image_pipeline.process_data()

if __name__ == "__main__":
    main()

1. GenAIPipelineTemplate 類定義了 GenAI 數據處理的通用流程，包括數據加載、預處理、特徵工程、模型訓練和評估等步驟。
2. preprocess 和 feature_engineering 方法被定義為抽象方法，要求子類必須實現這些步驟。
3. TextClassificationPipeline 和 ImageClassificationPipeline 是具體的子類，它們實現了特定於文本和圖像數據的預處理和特徵工程步驟。
4. 主程序展示了如何使用這些管道來處理不同類型的數據。

最後兩個案例會在明天繼續介紹：）

ref.
Refactoring.Guru | design-patterns