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

2025-04-02 20:28:05

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.py 或 feedback_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，并且易于扩展？

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

约定优于配置 / 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 实例。

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

关联注册：

如果使用了注册模式，可以在注册 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):
    # ...