AI および LLM エンジニアのための Python のデザイン パターン: 実践ガイド

AI エンジニアにとって、複雑なシステムを構築する場合は特に、クリーンかつ効率的で保守しやすいコードを作成することが重要です。

デザインパターン ソフトウェア設計における一般的な問題に対する再利用可能なソリューションです。 AIおよび大規模言語モデル（LLM）エンジニアデザインパターンは、複雑なワークフローを効率的に処理する堅牢でスケーラブルで保守性の高いシステムの構築に役立ちます。この記事では、Pythonのデザインパターンについて、AIと LLMベースのシステム。それぞれのパターンを、実践的なAIユースケースとPythonコード例を用いて解説します。

AI と機械学習のコンテキストで特に役立ついくつかの主要な設計パターンと、Python の例を見てみましょう。

AIエンジニアにとってデザインパターンが重要な理由

AI システムには、多くの場合、次のものが含まれます。

1. 複雑なオブジェクトの作成 (例: モデルの読み込み、データ前処理パイプライン)。
2. コンポーネント間の相互作用の管理 (例: モデル推論、リアルタイム更新)。
3. 変化する要件に対するスケーラビリティ、保守性、柔軟性を処理します。

デザイン パターンはこれらの課題に対処し、明確な構造を提供し、アドホックな修正を減らします。デザイン パターンは主に 3 つのカテゴリに分類されます。

• 創造的パターン: オブジェクトの作成に焦点を当てます。(シングルトン、ファクトリー、ビルダー)
• 構造パターン: オブジェクト間の関係を整理します。(アダプタ、デコレータ)
• 行動パターン: オブジェクト間の通信を管理します。(戦略、オブザーバー)

1. シングルトンパターン

当学校区の シングルトンパターン クラスにインスタンスが 1 つだけ存在することを保証し、そのインスタンスへのグローバル アクセス ポイントを提供します。これは、構成設定、ログ システム、モデル インスタンスなどの共有リソースを冗長性なく一貫して管理する必要がある AI ワークフローで特に役立ちます。

いつ使用するか

• グローバル構成（例：モデルのハイパーパラメータ）の管理。
• 複数のスレッドまたはプロセス間でリソースを共有する（例： GPU メモリ).
• 単一の 推論エンジン またはデータベース接続。

製品の導入

AI モデルの構成を管理するために Python でシングルトン パターンを実装する方法は次のとおりです。

class ModelConfig:
    """
    A Singleton class for managing global model configurations.
    """
    _instance = None  # Class variable to store the singleton instance

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            # Create a new instance if none exists
            cls._instance = super().__new__(cls)
            cls._instance.settings = {}  # Initialize configuration dictionary
        return cls._instance

    def set(self, key, value):
        """
        Set a configuration key-value pair.
        """
        self.settings[key] = value

    def get(self, key):
        """
        Get a configuration value by key.
        """
        return self.settings.get(key)

# Usage Example
config1 = ModelConfig()
config1.set("model_name", "GPT-4")
config1.set("batch_size", 32)

# Accessing the same instance
config2 = ModelConfig()
print(config2.get("model_name"))  # Output: GPT-4
print(config2.get("batch_size"))  # Output: 32
print(config1 is config2)  # Output: True (both are the same instance)

説明

1. 当学校区の __new__ 方法: これにより、クラスのインスタンスが 1 つだけ作成されます。インスタンスがすでに存在する場合は、既存のインスタンスが返されます。
2. 共有状態： 両方 config1 および config2 同じインスタンスを指すため、すべての構成がグローバルにアクセス可能になり、一貫性が保たれます。
3. AIユースケース: このパターンを使用して、データセットへのパス、ログ構成、環境変数などのグローバル設定を管理します。

2. ファクトリーパターン

当学校区の ファクトリパターン オブジェクトの作成をサブクラスまたは専用のファクトリ メソッドに委任する方法を提供します。AI システムでは、このパターンは、コンテキストに基づいてさまざまな種類のモデル、データ ローダー、またはパイプラインを動的に作成するのに最適です。

いつ使用するか

• ユーザー入力またはタスク要件に基づいてモデルを動的に作成します。
• 複雑なオブジェクト作成ロジック (例: 複数ステップの前処理パイプライン) を管理します。
• 柔軟性を向上させるために、オブジェクトのインスタンス化をシステムの残りの部分から分離します。

製品の導入

テキスト分類、要約、翻訳など、さまざまな AI タスク用のモデルを作成するためのファクトリーを構築しましょう。

class BaseModel:
    """
    Abstract base class for AI models.
    """
    def predict(self, data):
        raise NotImplementedError("Subclasses must implement the `predict` method")

class TextClassificationModel(BaseModel):
    def predict(self, data):
        return f"Classifying text: {data}"

class SummarizationModel(BaseModel):
    def predict(self, data):
        return f"Summarizing text: {data}"

class TranslationModel(BaseModel):
    def predict(self, data):
        return f"Translating text: {data}"

class ModelFactory:
    """
    Factory class to create AI models dynamically.
    """
    @staticmethod
    def create_model(task_type):
        """
        Factory method to create models based on the task type.
        """
        task_mapping = {
            "classification": TextClassificationModel,
            "summarization": SummarizationModel,
            "translation": TranslationModel,
        }
        model_class = task_mapping.get(task_type)
        if not model_class:
            raise ValueError(f"Unknown task type: {task_type}")
        return model_class()

# Usage Example
task = "classification"
model = ModelFactory.create_model(task)
print(model.predict("AI will transform the world!"))
# Output: Classifying text: AI will transform the world!

説明

1. 抽象基本クラス： BaseModel クラスはインターフェースを定義します（predict) をすべてのサブクラスで実装して一貫性を確保する必要があります。
2. ファクトリーロジック： ModelFactory タスクの種類に基づいて適切なクラスを動的に選択し、インスタンスを作成します。
3. 拡張性: 新しいモデルタイプを追加するのは簡単です。新しいサブクラスを実装し、ファクトリーの task_mapping.

AIユースケース

タスクに基づいて異なる LLM (BERT、GPT、T5 など) を選択するシステムを設計しているとします。ファクトリー パターンを使用すると、既存のコードを変更することなく、新しいモデルが利用可能になったときにシステムを簡単に拡張できます。

3. ビルダーパターン

当学校区の ビルダーパターン 複雑なオブジェクトの構築とその表現を分離します。オブジェクトの初期化や構成に複数のステップが含まれる場合に便利です。

いつ使用するか

• 複数ステップのパイプラインの構築 (例: データの前処理)。
• 実験またはモデルトレーニングの構成を管理します。
• 多くのパラメータを必要とするオブジェクトを作成し、可読性と保守性を確保します。

製品の導入

ビルダー パターンを使用してデータ前処理パイプラインを作成する方法は次のとおりです。

class DataPipeline:
    """
    Builder class for constructing a data preprocessing pipeline.
    """
    def __init__(self):
        self.steps = []

    def add_step(self, step_function):
        """
        Add a preprocessing step to the pipeline.
        """
        self.steps.append(step_function)
        return self  # Return self to enable method chaining

    def run(self, data):
        """
        Execute all steps in the pipeline.
        """
        for step in self.steps:
            data = step(data)
        return data

# Usage Example
pipeline = DataPipeline()
pipeline.add_step(lambda x: x.strip())  # Step 1: Strip whitespace
pipeline.add_step(lambda x: x.lower())  # Step 2: Convert to lowercase
pipeline.add_step(lambda x: x.replace(".", ""))  # Step 3: Remove periods

processed_data = pipeline.run("  Hello World. ")
print(processed_data)  # Output: hello world

説明

1. 連鎖メソッド： add_step このメソッドを使用すると、パイプラインを定義するときに、直感的でコンパクトな構文を連鎖させることができます。
2. ステップバイステップの実行: パイプラインは、各ステップを順番に実行してデータを処理します。
3. AIユースケース: ビルダー パターンを使用して、複雑で再利用可能なデータ前処理パイプラインまたはモデル トレーニング セットアップを作成します。

4. 戦略パターン

当学校区の 戦略パターン 交換可能なアルゴリズムのファミリーを定義し、各アルゴリズムをカプセル化して、実行時に動作を動的に変更できるようにします。これは、同じプロセス (推論やデータ処理など) でコンテキストに応じて異なるアプローチが必要になる可能性がある AI システムで特に役立ちます。

いつ使用するか

• 異なる 推論 戦略（例：バッチ処理とストリーミング）。
• さまざまなデータ処理技術を動的に適用します。
• 利用可能なインフラストラクチャに基づいてリソース管理戦略を選択します。

製品の導入

戦略パターンを使用して、バッチ推論とストリーミング推論という 2 つの異なる推論戦略を AI モデルに実装してみましょう。

class InferenceStrategy:
    """
    Abstract base class for inference strategies.
    """
    def infer(self, model, data):
        raise NotImplementedError("Subclasses must implement the `infer` method")

class BatchInference(InferenceStrategy):
    """
    Strategy for batch inference.
    """
    def infer(self, model, data):
        print("Performing batch inference...")
        return [model.predict(item) for item in data]

class StreamInference(InferenceStrategy):
    """
    Strategy for streaming inference.
    """
    def infer(self, model, data):
        print("Performing streaming inference...")
        results = []
        for item in data:
            results.append(model.predict(item))
        return results

class InferenceContext:
    """
    Context class to switch between inference strategies dynamically.
    """
    def __init__(self, strategy: InferenceStrategy):
        self.strategy = strategy

    def set_strategy(self, strategy: InferenceStrategy):
        """
        Change the inference strategy dynamically.
        """
        self.strategy = strategy

    def infer(self, model, data):
        """
        Delegate inference to the selected strategy.
        """
        return self.strategy.infer(model, data)

# Mock Model Class
class MockModel:
    def predict(self, input_data):
        return f"Predicted: {input_data}"

# Usage Example
model = MockModel()
data = ["sample1", "sample2", "sample3"]

context = InferenceContext(BatchInference())
print(context.infer(model, data))
# Output:
# Performing batch inference...
# ['Predicted: sample1', 'Predicted: sample2', 'Predicted: sample3']

# Switch to streaming inference
context.set_strategy(StreamInference())
print(context.infer(model, data))
# Output:
# Performing streaming inference...
# ['Predicted: sample1', 'Predicted: sample2', 'Predicted: sample3']

説明

1. 抽象戦略クラス： InferenceStrategy すべての戦略が従う必要があるインターフェースを定義します。
2. 具体的な戦略: 各戦略（例： BatchInference, StreamInference) は、そのアプローチに固有のロジックを実装します。
3. ダイナミックスイッチング： InferenceContext 実行時に戦略を切り替えることができるため、さまざまなユースケースに柔軟に対応できます。

いつ使用するか

• 切り替え バッチ推論 オフライン処理用 ストリーミング推論 リアルタイムアプリケーション向け。
• タスクまたは入力形式に基づいて、データ拡張または前処理手法を動的に調整します。

5. オブザーバーパターン

当学校区の オブザーバーパターン オブジェクト間に 1 対多の関係を確立します。1 つのオブジェクト (サブジェクト) の状態が変化すると、そのすべての従属オブジェクト (オブザーバー) に自動的に通知されます。これは、リアルタイム監視、イベント処理、またはデータ同期を行う AI システムで特に役立ちます。

いつ使用するか

• モデルのトレーニング中に精度や損失などのメトリックを監視します。
• ダッシュボードまたはログのリアルタイム更新。
• 複雑なワークフロー内のコンポーネント間の依存関係を管理します。

製品の導入

オブザーバー パターンを使用して、AI モデルのパフォーマンスをリアルタイムで監視してみましょう。

class Subject:
    """
    Base class for subjects being observed.
    """
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        """
        Attach an observer to the subject.
        """
        self._observers.append(observer)

    def detach(self, observer):
        """
        Detach an observer from the subject.
        """
        self._observers.remove(observer)

    def notify(self, data):
        """
        Notify all observers of a change in state.
        """
        for observer in self._observers:
            observer.update(data)

class ModelMonitor(Subject):
    """
    Subject that monitors model performance metrics.
    """
    def update_metrics(self, metric_name, value):
        """
        Simulate updating a performance metric and notifying observers.
        """
        print(f"Updated {metric_name}: {value}")
        self.notify({metric_name: value})

class Observer:
    """
    Base class for observers.
    """
    def update(self, data):
        raise NotImplementedError("Subclasses must implement the `update` method")

class LoggerObserver(Observer):
    """
    Observer to log metrics.
    """
    def update(self, data):
        print(f"Logging metric: {data}")

class AlertObserver(Observer):
    """
    Observer to raise alerts if thresholds are breached.
    """
    def __init__(self, threshold):
        self.threshold = threshold

    def update(self, data):
        for metric, value in data.items():
            if value > self.threshold:
                print(f"ALERT: {metric} exceeded threshold with value {value}")

# Usage Example
monitor = ModelMonitor()
logger = LoggerObserver()
alert = AlertObserver(threshold=90)

monitor.attach(logger)
monitor.attach(alert)

# Simulate metric updates
monitor.update_metrics("accuracy", 85)  # Logs the metric
monitor.update_metrics("accuracy", 95)  # Logs and triggers alert

1. サービスカテゴリ: オブザーバーのリストを管理し、状態が変わったときに通知します。この例では、 ModelMonitor クラスはメトリックを追跡します。
2. オブザーバー: 通知を受けたときに特定のアクションを実行します。たとえば、 LoggerObserver ログメトリクスは、 AlertObserver しきい値を超えた場合にアラートを発します。
3. 分離設計: オブザーバーとサブジェクトは疎結合されており、システムはモジュール化され拡張可能になっています。

AI エンジニアと従来のエンジニアの設計パターンの違い

デザイン パターンは普遍的に適用可能ですが、従来のソフトウェア エンジニアリングと比較して、AI エンジニアリングに実装すると独特の特性を帯びます。違いは、AI システムに固有の課題、目標、ワークフローにあり、多くの場合、従来の用途を超えてパターンを適応または拡張する必要があります。

1. オブジェクトの作成: 静的ニーズと動的ニーズ

• 伝統的なエンジニアリング: ファクトリやシングルトンなどのオブジェクト作成パターンは、構成、データベース接続、またはユーザー セッション状態を管理するためによく使用されます。これらは通常、静的であり、システム設計時に明確に定義されます。
• AIエンジニアリング: オブジェクトの作成には多くの場合 動的ワークフロー、のような：
  • ユーザー入力やシステム要件に基づいて、即座にモデルを作成します。
  • 翻訳、要約、分類などのタスク用にさまざまなモデル構成を読み込みます。
  • データセットの特性 (表形式と非構造化テキストなど) によって異なる複数のデータ処理パイプラインをインスタンス化します。

例:AI では、ファクトリー パターンはタスクの種類とハードウェアの制約に基づいてディープラーニング モデルを動的に生成する可能性がありますが、従来のシステムでは、単純にユーザー インターフェイス コンポーネントを生成する可能性があります。

2. パフォーマンスの制約

• 伝統的なエンジニアリング: デザイン パターンは通常、Web サーバー、データベース クエリ、UI レンダリングなどのアプリケーションのレイテンシとスループットに合わせて最適化されます。
• AIエンジニアリングAIにおけるパフォーマンス要件は モデル推論の遅延、 グラフィックプロセッサ 利用、および メモリの最適化パターンは次のものに対応する必要があります:
  • 中間結果のキャッシュ 冗長な計算を削減します (デコレータまたはプロキシ パターン)。
  • システム負荷またはリアルタイム制約に基づいて、レイテンシと精度のバランスをとるためにアルゴリズムを動的に切り替えます (戦略パターン)。

3. データ中心の性質

• 伝統的なエンジニアリング: パターンは多くの場合、固定された入出力構造 (フォーム、REST API 応答など) で動作します。
• AIエンジニアリング: パターンは処理する必要がある データの変動性 構造と規模の両面において、以下を含みます。
  • リアルタイム システム用のストリーミング データ。
  • 柔軟な処理手順を備えたパイプラインを必要とするマルチモーダル データ (テキスト、画像、ビデオなど)。
  • 多くの場合、Builder や Pipeline などのパターンを使用して、効率的な前処理と拡張パイプラインを必要とする大規模なデータセット。

4. 実験と安定性

• 伝統的なエンジニアリング: パターンによって一貫したパフォーマンスと信頼性が保証される、安定した予測可能なシステムの構築に重点が置かれています。
• AIエンジニアリング: AIワークフローは多くの場合 実験的 そして以下を含みます:
  • さまざまなモデル アーキテクチャまたはデータ前処理手法を繰り返し実行します。
  • システム コンポーネントを動的に更新します (例: モデルの再トレーニング、アルゴリズムの交換)。
  • 多くの場合、Decorator や Factory などの拡張可能なパターンを使用して、生産パイプラインを中断することなく既存のワークフローを拡張します。

例:AI のファクトリーは、モデルをインスタンス化するだけでなく、事前にロードされた重みを添付したり、オプティマイザーを構成したり、トレーニング コールバックをリンクしたりすることもすべて動的に実行します。