返回

Python工厂模式: __subclasses__()为空? 自动发现子类妙招

python

Python 工厂模式进阶:如何自动发现不同文件中的子类

搞代码扩展性的时候,工厂模式是个常客。通常我们会定义一个抽象基类(Abstract Base Class, ABC),然后在不同的 Python 文件里实现具体的子类。工厂(Factory)的作用就是根据需要创建这些子类的实例。但问题来了:怎么让工厂自动“知道”所有散落在不同文件里的子类呢?总不能每加一个子类就手动去工厂代码里 import 一次吧?那样也太不“自动”了。

特别是,你可能会遇到像下面这位朋友一样的情况:在工厂里调用基类的 __subclasses__() 方法,想获取所有子类,结果却得到一个空列表 []

# evaluator_factory.py
from evaluation.evaluators.evaluator import Evaluator
import importlib
import inspect

# ... (省略部分代码)

class EvaluatorFactory:
    # ...
    def _create_evaluators(self):
        evaluators = []
        # 期望这里能拿到 ScoreEvaluator, FeedbackEvaluator 等
        evaluator_classes = Evaluator.__subclasses__()
        print(evaluator_classes) # <--- 打印出来是 [],问题所在!

        # 尝试用 importlib + inspect,结果可能还是空
        evaluator_module = importlib.import_module("evaluation.evaluators.evaluator")
        evaluators_dict = {
            name: cls
            for name, cls in inspect.getmembers(evaluator_module, inspect.isclass)
            if issubclass(cls, Evaluator) and cls is not Evaluator # 修正了原始代码的逻辑
        }
        print(evaluators_dict) # <--- 很可能还是 {}

        # ... (后续代码)

代码结构大概是这样:

/evaluation
│── metrics/
│   │── metric.py          # 指标抽象类
│   │── QWK.py
│   │── PCC.py
│── evaluators/            # 评估器目录 (这是个 Python 包)
│   │── __init__.py
│   │── evaluator.py       # 评估器抽象基类 Evaluator
│   │── score_evaluator.py # 子类 ScoreEvaluator
│   │── feedback_evaluator.py # 子类 FeedbackEvaluator
│── evaluator_factory.py   # 工厂类 EvaluatorFactory

咋回事呢? __subclasses__() 怎么就失灵了?

一、问题分析:__subclasses__() 为何返回空列表?

这背后的原因其实挺直白:Python 不知道那些子类文件的存在,因为它们还没被加载。

__subclasses__() 这个特殊方法,只会返回当前 Python 解释器内存中 已经加载并识别 的、直接继承自该基类的所有子类。

evaluator_factory.py 中,你 import evaluation.evaluators.evaluator,这仅仅加载了 evaluator.py 文件,定义了 Evaluator 这个基类。但此时,Python 解释器并不知道 score_evaluator.pyfeedback_evaluator.py 这两个文件里定义了 Evaluator 的子类。除非这些文件因为某种原因被 import 了,否则它们的类定义对 __subclasses__() 来说就是“隐形”的。

同理,importlib.import_module("evaluation.evaluators.evaluator") 再加上 inspect.getmembers 也只能检查 evaluator.py 这个模块本身包含的类,自然找不到定义在其他文件里的子类。

所以,核心问题在于:必须先确保包含子类定义的模块被加载到内存中,然后 __subclasses__() 或类似的查找机制才能工作。

二、解决方案:让子类“现身”

知道了原因,解决起来就有方向了。我们需要想办法在调用 __subclasses__() 之前,把那些包含子类的模块给 import 进来。

方案一:在包的 __init__.py 中手动导入 (用户已发现)

这是最常见也比较直观的方法。利用 Python 包的 __init__.py 文件。当一个包(比如这里的 evaluation.evaluators)被导入时,它的 __init__.py 文件会自动执行。我们可以在这里手动导入所有需要的子类模块。

evaluation/evaluators/__init__.py

# 显式导入所有具体的 Evaluator 子类模块
# 这样,当 evaluators 包被导入时,这些子类就会被加载进内存
from .score_evaluator import ScoreEvaluator
from .feedback_evaluator import FeedbackEvaluator

# (可选)使用 __all__ 控制 `from evaluation.evaluators import *` 的行为
__all__ = ["ScoreEvaluator", "FeedbackEvaluator", "Evaluator"] # 把基类也加进去可能更好

# 注意:上面的 from .evaluator import Evaluator 通常不需要在这里做
# 因为 Evaluator 可能会被其他地方直接导入,或者根本不需要暴露在包的顶层
# 但如果你希望用户可以直接 from evaluation.evaluators import Evaluator,那就加上
# 这里为了演示方便,假定 Evaluator 不是包的公共接口的一部分
# 更新: 根据 __all__ 的常见用法,包含公开的类比较好

修改后的 evaluator_factory.py

在工厂代码执行前,需要确保 evaluation.evaluators 包被导入。这可以在工厂文件的开头完成,或者在使用工厂之前完成。

# evaluator_factory.py
import importlib
import inspect
import evaluation.evaluators # <--- 关键:导入这个包,触发 __init__.py 的执行
from evaluation.evaluators.evaluator import Evaluator

class EvaluatorFactory:
    # ...
    def _create_evaluators(self):
        # 现在,由于 __init__.py 已经执行,子类已被加载
        evaluator_classes = Evaluator.__subclasses__()
        print(evaluator_classes) # <--- 这次应该能看到 [ScoreEvaluator, FeedbackEvaluator] 了

        evaluators = []
        for cls in evaluator_classes:
            # 根据需要实例化,传递配置等
            # 这里假设子类的 __init__ 签名与基类一致或兼容
            evaluators.append(cls(self.framework_config, self.evaluation_config, **self.kwargs))

        return evaluators

    # ... (省略 evaluate 方法等)

原理与作用:

  • 利用 Python 包机制,import evaluation.evaluators 会执行 evaluators/__init__.py
  • __init__.py 中的 from .score_evaluator import ScoreEvaluator 等语句会加载相应的模块,从而将 ScoreEvaluator 等子类的定义载入内存。
  • 后续调用 Evaluator.__subclasses__() 时,就能找到这些已加载的子类了。

优点:

  • 简单直接,容易理解。
  • 明确控制哪些子类是“公开”的(通过 __init__.py 导出)。

缺点:

  • 每次添加新的子类文件,都需要手动 修改 evaluators/__init__.py 文件,增加对应的 import 语句。项目大了或者子类多了,维护起来有点麻烦,也容易忘记。

安全建议:

  • 使用 __all__ 可以明确指定 from package import * 时导入哪些名字,避免不必要的内部实现细节泄露。

方案二:动态扫描与加载模块

如果你不想每次添加子类都去改 __init__.py,可以写代码来自动扫描 evaluators 目录下的 Python 文件,并动态导入它们。

evaluator_factory.py (或一个辅助工具函数中)

import os
import pkgutil
import importlib
import inspect
from evaluation.evaluators.evaluator import Evaluator
import evaluation.evaluators # 仍然需要导入包本身,以获取路径等信息

def load_subclass_modules(package):
    """
    动态加载指定包路径下的所有模块,以便注册子类。
    """
    package_path = package.__path__
    package_name = package.__name__

    # 使用 pkgutil 比 os.listdir 更健壮,能处理 zipapp 等情况
    for _, module_name, _ in pkgutil.walk_packages(package_path, prefix=package_name + '.'):
        try:
            # 动态导入模块
            importlib.import_module(module_name)
            # print(f"Dynamically loaded module: {module_name}")
        except Exception as e:
            print(f"Warning: Failed to import module {module_name}: {e}")
            # 根据需要决定是否要处理导入失败的情况

class EvaluatorFactory:
    def __init__(self, framework_config, evaluation_config, **kwargs):
        self.framework_config = framework_config
        self.evaluation_config = evaluation_config
        self.kwargs = kwargs

        # 在初始化工厂时,先动态加载所有可能的子类模块
        load_subclass_modules(evaluation.evaluators)

        self.evaluators = self._create_evaluators()

    def _create_evaluators(self):
        # 此时,所有 evaluators 包下的模块(包括子类定义)应该都加载了
        evaluator_classes = Evaluator.__subclasses__()
        print(evaluator_classes) # <--- 应该能正确找到所有子类

        evaluators = []
        for cls in evaluator_classes:
            # 确保只实例化非抽象的子类
            if not inspect.isabstract(cls):
                try:
                    evaluators.append(cls(self.framework_config, self.evaluation_config, **self.kwargs))
                except Exception as e:
                    print(f"Error instantiating {cls.__name__}: {e}")
                    # 可能需要更精细的错误处理

        return evaluators

    async def evaluate(self):
        # ... (保持不变)
        pass

原理与作用:

  • pkgutil.walk_packages (或 pkgutil.iter_modules) 可以遍历一个包路径下的所有模块(及其子包内的模块)。
  • importlib.import_module(module_name) 动态地加载找到的每个模块。
  • 一旦模块被加载,其中定义的类(包括 Evaluator 的子类)就进入了内存。
  • 加载完成后,再调用 Evaluator.__subclasses__() 就能找到所有子类了。

优点:

  • 自动化: 添加新的 evaluator_xxx.py 文件后,无需修改任何现有代码,工厂能自动发现并使用它们(只要新文件放在 evaluators 目录下)。
  • 更符合“开闭原则”(对扩展开放,对修改封闭)。

缺点:

  • 代码稍微复杂一点点。
  • 需要注意扫描路径和模块命名规则。如果目录里有非子类的 Python 文件,也会被加载,虽然通常没什么副作用,但可能会加载不必要的代码。
  • 导入错误需要处理,否则一个坏掉的模块可能影响整个加载过程。

安全/使用建议:

  • 确保扫描的目录 (evaluation.evaluators 的路径) 是正确的。
  • 可以在 load_subclass_modules 中加入文件名过滤逻辑,比如只加载 _evaluator.py 结尾的文件,或者排除 __init__.py
  • pkgutil.walk_packages 默认会递归遍历子目录,如果你的 evaluators 包内还有子包,并且子包里也有 Evaluator 子类,它们也会被找到。如果只想扫描顶层目录,可以用 pkgutil.iter_modules

方案三:显式注册模式 (Decorator)

这是一种更“主动”的方式。让每个子类在定义时就通过某种方式“登记”到中央注册表(通常是一个字典)。工厂不再需要去“发现”子类,而是直接从注册表中获取。

1. 创建注册表和注册函数 (可以放在 evaluator.py 或单独的 registry.py)

evaluation/evaluators/registry.py

from evaluation.evaluators.evaluator import Evaluator # 确保 Evaluator 定义可用

EVALUATOR_REGISTRY = {} # 全局注册表

def register_evaluator(name):
    """
    一个装饰器工厂,用于注册 Evaluator 子类。
    使用 @register_evaluator("unique_name") 来装饰子类。
    """
    def decorator(cls):
        if not issubclass(cls, Evaluator):
            raise TypeError(f"Only subclasses of Evaluator can be registered. {cls.__name__} is not.")
        if name in EVALUATOR_REGISTRY:
            print(f"Warning: Evaluator name '{name}' already exists. Overwriting {EVALUATOR_REGISTRY[name].__name__} with {cls.__name__}")
        EVALUATOR_REGISTRY[name] = cls
        return cls # 保持装饰器行为,返回原类
    return decorator

# 注意:这个 registry.py 文件本身,或者引用它的文件,
# 必须确保在某个时间点被导入,以便 EVALUATOR_REGISTRY 被创建。

2. 在子类定义处使用装饰器注册

evaluation/evaluators/score_evaluator.py

from .evaluator import Evaluator
from .registry import register_evaluator # 导入注册器

@register_evaluator("score") # <--- 使用装饰器注册,给它一个唯一的名字
class ScoreEvaluator(Evaluator):
    def __init__(self, evaluation_config, framework_config, **kwargs):
        super().__init__(evaluation_config, framework_config, **kwargs)
        # ...
    
    def load_evaluation_data(self, data_path):
        # ...
        pass

    async def __call__(self, *args, **kwargs):
        # ...
        pass

evaluation/evaluators/feedback_evaluator.py

from .evaluator import Evaluator
from .registry import register_evaluator

@register_evaluator("feedback") # <--- 用不同的名字注册
class FeedbackEvaluator(Evaluator):
    # ... 实现 ...
    pass

3. 修改工厂,从注册表获取类

evaluator_factory.py

# evaluator_factory.py
import evaluation.evaluators # 仍然需要确保子类模块被加载,可以通过 __init__.py 或动态加载
# 或者更好的方式是: 让加载模块的逻辑与注册表的使用解耦
# 比如,有一个统一的入口点负责加载所有插件模块

# 导入注册表
from evaluation.evaluators.registry import EVALUATOR_REGISTRY

class EvaluatorFactory:
    def __init__(self, framework_config, evaluation_config, **kwargs):
        self.framework_config = framework_config
        self.evaluation_config = evaluation_config
        self.kwargs = kwargs

        # 加载模块的操作最好在这里之前完成,例如:
        # load_subclass_modules(evaluation.evaluators) # 调用之前的动态加载函数
        # 或者,确保 `import evaluation.evaluators` 已经执行(如果使用__init__.py方案)

        self.evaluators = self._create_evaluators()

    def _create_evaluators(self):
        evaluators = []
        print(f"Available evaluators in registry: {list(EVALUATOR_REGISTRY.keys())}")

        # 直接从注册表获取类
        for name, cls in EVALUATOR_REGISTRY.items():
            # 这里可以根据名字或配置进行过滤,比如只实例化需要的 evaluator
            print(f"Creating instance for evaluator: {name} ({cls.__name__})")
            try:
                evaluators.append(cls(self.framework_config, self.evaluation_config, **self.kwargs))
            except Exception as e:
                print(f"Error instantiating {cls.__name__} (registered as '{name}'): {e}")

        return evaluators

    async def evaluate(self):
        evaluation_results = {}
        print(f"Factory created {len(self.evaluators)} evaluator instances.")
        for evaluator in self.evaluators:
            # 获取注册时的名字可能需要反向查找或在实例上存一个属性
            # 假设 evaluator 实例能提供一个 name 属性,或者我们用类名
            evaluator_name = evaluator.__class__.__name__ # 简单示例
            try:
                results = await evaluator()
                evaluation_results[evaluator_name] = results
            except Exception as e:
                print(f"Error during evaluation by {evaluator_name}: {e}")
                evaluation_results[evaluator_name] = {"error": str(e)}

        return evaluation_results

关键点: 注册模式本身并不解决模块加载问题 。你仍然需要确保定义了 @register_evaluator 的那些 Python 文件(如 score_evaluator.py)被执行过一次。所以,注册模式通常需要与 方案一 (手动 __init__.py)方案二 (动态扫描加载) 结合使用。加载模块是为了让 @register_evaluator 这个装饰器代码有机会运行,从而完成注册动作。

原理与作用:

  • 定义一个中心化的注册表(EVALUATOR_REGISTRY)。
  • 提供一个装饰器 (@register_evaluator),子类用它来“报名登记”到注册表,通常附带一个唯一标识符(如 "score")。
  • 工厂直接查询注册表来获取所有可用的子类,而不是依赖 __subclasses__()

优点:

  • 显式控制: 非常清楚哪些类被注册了,以及用什么名字注册的。可以防止意外注册不想要的类。
  • 解耦: 工厂与子类的物理位置(哪个文件)进一步解耦。只要模块被加载且类被注册,工厂就能找到它。
  • 灵活性: 注册时可以传递更多元数据(比如版本号、依赖项等),不仅仅是类本身。

缺点:

  • 需要在每个子类定义处添加装饰器代码。
  • 仍然需要一个机制来确保包含注册代码的模块被加载(结合方案一或方案二)。
  • 需要管理注册名称的唯一性。

进阶使用技巧:

  • 装饰器可以更复杂,比如检查类的接口是否完整,或者自动根据类名生成注册名。
  • 注册表可以设计得更复杂,支持分组、按需加载等。

三、关于第二个问题:Metrics 与 Evaluator 的关联

用户还提到一个设计问题:有些 Metric(指标类,可能也有类似的文件结构)只适用于特定的 Evaluator。如何设计才能让团队成员在添加新 Metric 时,方便地将其关联到正确的 Evaluator,并且易于扩展?

这个问题有几种常见的处理方式:

  1. 约定优于配置 / Evaluator 自我声明:

    • Evaluator 子类中定义一个类属性或方法,用来声明它支持或需要的 Metric
    • 例如,在 ScoreEvaluator 中:
      # evaluation/evaluators/score_evaluator.py
      from ..metrics.QWK import QWK
      from ..metrics.PCC import PCC
      
      @register_evaluator("score") # 假设用了注册模式
      class ScoreEvaluator(Evaluator):
          SUPPORTED_METRICS = ["QWK", "PCC"] # 或者直接用 Metric 类: [QWK, PCC]
      
          def __init__(self, ...):
              super().__init__(...)
              # 可能在这里根据 SUPPORTED_METRICS 来加载/实例化指标
              self.metrics = self._load_metrics(self.SUPPORTED_METRICS)
      
          def _load_metrics(self, metric_names_or_classes):
              # 这里需要一个 Metric 的工厂或注册表来查找并实例化
              # ...
              pass
          # ...
      
    • 优点: 关联关系在 Evaluator 内部定义,逻辑内聚。
    • 缺点: 如果一个 Metric 被多个 Evaluator 使用,信息会分散。需要一个机制(比如 Metric 注册表/工厂)来根据名字或类获取 Metric 实例。
  2. 配置驱动:

    • 使用配置文件(如 YAML, JSON)来定义 Evaluator 和 Metric 的关系。
    • 例如 config.yaml:
      evaluators:
        score:
          class: ScoreEvaluator # 或者让工厂根据 key 'score' 找到类
          metrics:
            - QWK
            - PCC
        feedback:
          class: FeedbackEvaluator
          metrics:
            - SomeFeedbackMetric
      
    • 工厂读取配置,根据配置决定创建哪个 Evaluator 实例,并为其注入所需的 Metric 实例。
    • 优点: 配置集中,修改关系不需要改代码。非常灵活。
    • 缺点: 增加了配置文件的维护成本。运行时依赖配置文件的正确性。
  3. 关联注册:

    • 如果使用了注册模式,可以在注册 Evaluator 时就提供其关联的 Metric 信息。
    • 修改注册器和注册表:
      # evaluation/evaluators/registry.py
      EVALUATOR_REGISTRY = {} # 存储: {'name': {'class': cls, 'metrics': [metric_name_or_class]}}
      
      def register_evaluator(name, supported_metrics=None):
          supported_metrics = supported_metrics or []
          def decorator(cls):
              # ... (之前的检查) ...
              EVALUATOR_REGISTRY[name] = {'class': cls, 'metrics': supported_metrics}
              return cls
          return decorator
      
      # evaluation/evaluators/score_evaluator.py
      from ..metrics.QWK import QWK
      from ..metrics.PCC import PCC
      
      @register_evaluator("score", supported_metrics=[QWK, PCC]) # 注册时声明支持的Metric类
      class ScoreEvaluator(Evaluator):
          # ...
      
    • 工厂从注册表获取类和它支持的 Metric 列表,然后负责实例化 Metric 并传递给 Evaluator。
    • 优点: 注册时绑定关系,信息明确。
    • 缺点: 注册逻辑变复杂。Metric 也需要被某种方式发现(可能 Metric 也有自己的注册表)。

选择哪种方式?

  • 如果关系简单固定,方案 1 (Evaluator 自我声明) 可能最直接。
  • 如果追求灵活性和配置化,方案 2 (配置驱动) 是工业界常见的做法。
  • 如果已经用了注册模式,并且想让注册信息更丰富,方案 3 (关联注册) 是一个自然扩展。

通常,对于 Metrics 本身,也需要一个类似的发现/注册机制(可能用动态加载或注册表),这样 Evaluator 或工厂才能找到并实例化它们。

总结

要解决 Python 中跨文件查找子类导致 __subclasses__() 返回空列表的问题,核心在于确保子类所在的模块在使用 __subclasses__() 之前已经被加载 。常用方法包括:

  1. 手动在包的 __init__.py 中导入 :简单但需手动维护。
  2. 动态扫描并加载模块 :自动化程度高,更符合开闭原则,但代码稍复杂。
  3. 显式注册模式(通常配合前两者之一使用) :控制力强,解耦好,但需要在子类添加装饰器。

选择哪种方案取决于项目规模、团队习惯以及对自动化和灵活性的要求。对于 Metric 与 Evaluator 的关联问题,可以通过类内声明、外部配置或扩展注册信息等方式来管理。

选择合适的模式能让你的代码库在未来扩展时更加从容。