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): # ...
- 工厂从注册表获取类和它支持的 Metric 列表,然后负责实例化 Metric 并传递给 Evaluator。
- 优点: 注册时绑定关系,信息明确。
- 缺点: 注册逻辑变复杂。Metric 也需要被某种方式发现(可能 Metric 也有自己的注册表)。
- 如果使用了注册模式,可以在注册
选择哪种方式?
- 如果关系简单固定,方案 1 (Evaluator 自我声明) 可能最直接。
- 如果追求灵活性和配置化,方案 2 (配置驱动) 是工业界常见的做法。
- 如果已经用了注册模式,并且想让注册信息更丰富,方案 3 (关联注册) 是一个自然扩展。
通常,对于 Metrics 本身,也需要一个类似的发现/注册机制(可能用动态加载或注册表),这样 Evaluator 或工厂才能找到并实例化它们。
总结
要解决 Python 中跨文件查找子类导致 __subclasses__()
返回空列表的问题,核心在于确保子类所在的模块在使用 __subclasses__()
之前已经被加载 。常用方法包括:
- 手动在包的
__init__.py
中导入 :简单但需手动维护。 - 动态扫描并加载模块 :自动化程度高,更符合开闭原则,但代码稍复杂。
- 显式注册模式(通常配合前两者之一使用) :控制力强,解耦好,但需要在子类添加装饰器。
选择哪种方案取决于项目规模、团队习惯以及对自动化和灵活性的要求。对于 Metric 与 Evaluator 的关联问题,可以通过类内声明、外部配置或扩展注册信息等方式来管理。
选择合适的模式能让你的代码库在未来扩展时更加从容。