面向 AI 和 LLM 工程师的 Python 设计模式：实用指南

作为人工智能工程师，编写干净、高效且可维护的代码至关重要，尤其是在构建复杂系统时。

设计模式 是软件设计中常见问题的可重复使用的解决方案。对于 人工智能及大型语言模型（LLM）工程师设计模式有助于构建强大、可扩展且可维护的系统，从而高效处理复杂的工作流。本文深入探讨 Python 中的设计模式，重点介绍其在 AI 和 LLM基于系统。我将使用实际的 AI 用例和 Python 代码示例来解释每种模式。

让我们探索一些在人工智能和机器学习环境中特别有用的关键设计模式以及 Python 示例。

为什么设计模式对人工智能工程师很重要

人工智能系统通常涉及：

1. 复杂对象创建（例如，加载模型、数据预处理管道）。
2. 管理组件之间的交互（例如，模型推理、实时更新）。
3. 处理可扩展性、可维护性和灵活性以满足不断变化的需求。

设计模式解决了这些挑战，提供了清晰的结构并减少了临时修复。它们分为三大类：

• 创作模式：专注于对象创建。（单例、工厂、建造者）
• 结构模式：组织对象之间的关系。（适配器、装饰器）
• 行为模式：管理对象之间的通信。（策略、观察者）

1.单例模式

- 单例模式 确保一个类只有一个实例，并提供该实例的全局访问点。这在 AI 工作流中尤其有价值，因为在 AI 工作流中，共享资源（如配置设置、日志系统或模型实例）必须一致地进行管理，且不存在冗余。

何时使用

• 管理全局配置（例如模型超参数）。
• 在多个线程或进程之间共享资源（例如， 显存).
• 确保一致访问单一 推理机 或数据库连接。

实施

以下是如何在 Python 中实现单例模式来管理 AI 模型的配置：

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__ 付款方式：这确保只创建该类的一个实例。如果实例已存在，则返回现有实例。
2. 共享状态：两个 config1 和 config2 指向同一个实例，使得所有配置都可以全局访问并且一致。
3. 人工智能用例：使用此模式来管理全局设置，如数据集路径、日志配置或环境变量。

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.

人工智能用例

假设您正在设计一个系统，该系统会根据任务选择不同的 LLM（例如 BERT、GPT 或 T5）。工厂模式使您可以在新模型可用时轻松扩展系统，而无需修改现有代码。

3.建造者模式

- 建造者模式 将复杂对象的构造与其表示分离开来。当对象需要多个步骤进行初始化或配置时，这种方法非常有用。

何时使用

• 构建多步骤管道（例如数据预处理）。
• 管理实验或模型训练的配置。
• 创建需要大量参数的对象，确保可读性和可维护性。

实施

下面展示了如何使用 Builder 模式创建数据预处理管道：

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. 人工智能用例：使用 Builder 模式创建复杂、可重复使用的数据预处理管道或模型训练设置。

4. 策略模式

- 策略模式 定义一系列可互换算法，封装每个算法并允许行为在运行时动态更改。这在 AI 系统中尤其有用，因为同一过程（例如推理或数据处理）可能需要根据上下文采用不同的方法。

何时使用

• 在不同 推理 策略（例如批处理与流式处理）。
• 动态应用不同的数据处理技术。
• 根据可用的基础设施选择资源管理策略。

实施

让我们使用策略模式为 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.观察者模式

- 观察者模式 在对象之间建立一对多关系。当一个对象（主体）改变状态时，其所有依赖者（观察者）都会自动收到通知。这在 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. 解耦设计：观察者和主体松散耦合，使得系统模块化、可扩展。

人工智能工程师与传统工程师的设计模式有何不同

设计模式虽然普遍适用，但与传统软件工程相比，在人工智能工程中实施时具有独特的特点。区别在于人工智能系统固有的挑战、目标和工作流程，这些挑战、目标和工作流程通常要求对模式进行调整或扩展，使其超出其传统用途。

1. 对象创建：静态需求与动态需求

• 传统工程：对象创建模式（如 Factory 或 Singleton）通常用于管理配置、数据库连接或用户会话状态。这些通常是静态的，并且在系统设计期间定义明确。
• 人工智能工程：对象创建通常涉及 动态工作流程，如：
  • 根据用户输入或系统要求动态创建模型。
  • 为翻译、总结或分类等任务加载不同的模型配置。
  • 实例化根据数据集特征（例如表格与非结构化文本）而变化的多个数据处理管道。

例如：：在人工智能中，工厂模式可能会根据任务类型和硬件约束动态生成深度学习模型，而在传统系统中，它可能只是生成用户界面组件。

2. 性能限制

• 传统工程：设计模式通常针对 Web 服务器、数据库查询或 UI 渲染等应用程序中的延迟和吞吐量进行优化。
• 人工智能工程：人工智能的性能要求延伸到 模型推理延迟， 图形处理器/TPU 利用率，以及 内存优化. 图案必须适应：
  • 缓存中间结果 减少冗余计算（装饰器或代理模式）。
  • 根据系统负载或实时约束动态切换算法（策略模式）来平衡延迟和准确性。

3.以数据为中心的特性

• 传统工程：模式通常对固定的输入输出结构进行操作（例如，表单、REST API 响应）。
• 人工智能工程：模式必须处理 数据可变性 在结构和规模上，包括：
  • 实时系统的流数据。
  • 多模式数据（例如文本、图像、视频）需要具有灵活处理步骤的管道。
  • 需要高效预处理和增强管道的大规模数据集，通常使用 Builder 或 Pipeline 等模式。

4. 实验与稳定性

• 传统工程：重点是构建稳定、可预测的系统，其中模式可确保一致的性能和可靠性。
• 人工智能工程：AI 工作流程通常 试验 并涉及：
  • 迭代不同的模型架构或数据预处理技术。
  • 动态更新系统组件（例如，重新训练模型、交换算法）。
  • 在不破坏生产流程的情况下扩展现有工作流程，通常使用装饰器或工厂等可扩展模式。

例如：：人工智能中的工厂不仅可以实例化模型，还可以附加预加载的权重、配置优化器和链接训练回调——所有这些都是动态的。