张芷铭的个人博客

混入设计模式的一个示例

下面是为数据处理框架设计的Mixin混入模式代码框架，支持灵活组合模型算子（如RAM、GroundingDINO等）。核心思想是通过多重继承动态注入模型功能，同时保持接口统一和资源可控。

1. 基础接口与抽象类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from abc import ABC, abstractmethod

class ModelMixin(ABC):
    """所有模型算子的基类Mixin"""
    @abstractmethod
    def initialize(self):
        """模型初始化（加载权重、分配资源）"""
        pass

    @abstractmethod
    def process(self, data: dict) -> dict:
        """处理输入数据，返回结果字典"""
        pass

class PipelineBase:
    """处理流水线基类（组合Mixin的容器）"""
    def __init__(self):
        self.models = {}  # 存储已初始化的模型实例

    def run(self, input_data: dict) -> dict:
        """执行所有模型的process方法并返回整合结果"""
        result = {}
        for name, model in self.models.items():
            result[name] = model.process(input_data)
        return result

2. 具体模型实现（Mixin示例）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# RAM图像识别模型
class RAMModelMixin(ModelMixin):
    def initialize(self):
        from ram import get_ram_model  # 实际导入逻辑
        self.ram_model = get_ram_model()
        self.models["ram"] = self  # 注册到流水线

    def process(self, data):
        return self.ram_model.predict(data["image"])

# GroundingDINO目标检测
class GroundingDINOMixin(ModelMixin):
    def initialize(self):
        from grounding_dino import load_dino_model
        self.dino_model = load_dino_model()
        self.models["dino"] = self

    def process(self, data):
        return self.dino_model.detect(data["image"], data["text_prompt"])

# RAFT光流估计
class RAFTMixin(ModelMixin):
    def initialize(self):
        from raft import RAFT
        self.raft_model = RAFT()
        self.models["raft"] = self

    def process(self, data):
        return self.raft_model.estimate_flow(data["frame1"], data["frame2"])

# ZeroDepth深度估计
class ZeroDepthMixin(ModelMixin):
    def initialize(self):
        from zero_depth import ZeroDepth
        self.depth_model = ZeroDepth()
        self.models["zero_depth"] = self

    def process(self, data):
        return self.depth_model.estimate_depth(data["image"])

3. 动态流水线构建（Mixin组合核心）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

class DynamicPipeline(PipelineBase):
    """通过继承动态组合模型"""
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 初始化所有混入的模型
        self.initialize_models()

    def initialize_models(self):
        # 遍历类的MRO，调用所有ModelMixin子类的initialize
        for base_class in type(self).__mro__:
            if issubclass(base_class, ModelMixin) and base_class != ModelMixin:
                base_class.initialize(self)

4. 使用示例

场景1：仅使用RAM和ZeroDepth

1
2
3
4
5
6

class SimplePipeline(DynamicPipeline, RAMModelMixin, ZeroDepthMixin):
    pass

pipeline = SimplePipeline()
result = pipeline.run({"image": "example.jpg"})
# 输出: {"ram": RAM结果, "zero_depth": 深度图}

场景2：全模型组合（RAM + DINO + RAFT）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

class FullVisionPipeline(DynamicPipeline, RAMModelMixin, GroundingDINOMixin, RAFTMixin):
    pass

pipeline = FullVisionPipeline()
input_data = {
    "image": "scene.jpg",
    "text_prompt": "car",
    "frame1": "frame1.png",
    "frame2": "frame2.png"
}
result = pipeline.run(input_data)
# 输出: {"ram": 识别结果, "dino": 检测框, "raft": 光流图}

5. 高级特性扩展

延迟初始化（按需加载）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

class LazyPipeline(PipelineBase):
    def __init__(self, *mixins):
        super().__init__()
        self.mixins = [mixin() for mixin in mixins]  # 实例化但不初始化

    def run(self, input_data):
        for mixin in self.mixins:
            if not hasattr(mixin, "model"):  # 未初始化则触发
                mixin.initialize()
            # ...处理逻辑

依赖管理（通过MRO顺序控制初始化顺序）

1
2
3
4
5
6
7

class OrderedPipeline(DynamicPipeline):
    # 按列表顺序初始化：DINO → RAM → ZeroDepth
    __mixin_order__ = [GroundingDINOMixin, RAMModelMixin, ZeroDepthMixin]

    def initialize_models(self):
        for mixin in self.__mixin_order__:
            mixin.initialize(self)

设计优势与注意事项

  1. 灵活性

    • 通过继承组合任意模型（如 class MyPipeline(DynamicPipeline, RAM, RAFT)）
    • 新增模型只需实现 ModelMixin 接口，无需修改框架

  2. 资源控制

    • 每个模型独立初始化，避免加载未使用的模型
    • 支持延迟初始化（LazyPipeline）减少内存占用

  3. 注意事项

    • 命名冲突：避免不同Mixin使用相同方法名（如共用 preprocess() 时需重命名）
    • 初始化顺序：依赖其他模型的Mixin需通过MRO或显式排序控制
    • 类型检查：可用 @dataclass 或 TypeGuard 强化输入/输出类型约束

完整代码可参考 [Mixin模式详解] 和 [Python设计模式实战]。对于大型项目，建议结合工厂模式（如 PipelineFactory.create_pipeline(["ram", "dino"])）进一步简化调用。

💬 评论