张芷铭的个人博客

#Python #Mixin #设计模式 #数据处理

Mixin 模式支持灵活组合模型算子，通过多继承动态注入功能，保持接口统一与资源可控。

基础架构

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

from abc import ABC, abstractmethod

class ModelMixin(ABC):
    """模型算子基类 Mixin"""
    @abstractmethod
    def initialize(self):
        """加载权重、分配资源"""
        pass

    @abstractmethod
    def process(self, data: dict) -> dict:
        """处理输入数据"""
        pass

class PipelineBase:
    """处理流水线基类"""
    def __init__(self):
        self.models = {}

    def run(self, input_data: dict) -> dict:
        return {name: model.process(input_data)
                for name, model in self.models.items()}

具体模型实现

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

class RAMModelMixin(ModelMixin):
    def initialize(self):
        self.ram_model = get_ram_model()
        self.models["ram"] = self

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

class GroundingDINOMixin(ModelMixin):
    def initialize(self):
        self.dino_model = load_dino_model()
        self.models["dino"] = self

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

动态流水线

1
2
3
4
5
6
7
8
9

class DynamicPipeline(PipelineBase):
    def __init__(self):
        super().__init__()
        self.initialize_models()

    def initialize_models(self):
        for base in type(self).__mro__:
            if issubclass(base, ModelMixin) and base != ModelMixin:
                base.initialize(self)

使用示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# RAM + ZeroDepth 组合
class SimplePipeline(DynamicPipeline, RAMModelMixin, ZeroDepthMixin):
    pass

pipeline = SimplePipeline()
result = pipeline.run({"image": "example.jpg"})

# 全模型组合
class FullVisionPipeline(DynamicPipeline, RAMModelMixin,
                         GroundingDINOMixin, RAFTMixin):
    pass

高级特性

延迟初始化：按需加载模型，减少内存占用

1
2
3
4
5

class LazyPipeline(PipelineBase):
    def run(self, input_data):
        for mixin in self.mixins:
            if not hasattr(mixin, "model"):
                mixin.initialize()

初始化顺序控制：通过 MRO 或显式排序

1
2

class OrderedPipeline(DynamicPipeline):
    __mixin_order__ = [GroundingDINOMixin, RAMModelMixin]

注意事项

  • 避免不同 Mixin 使用相同方法名
  • 依赖其他模型的 Mixin 需控制初始化顺序
  • 可结合工厂模式简化调用

Comments