You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

抽象类中修饰抽象__call__方法的日志装饰器仅外层代码执行,wrapper未触发的问题排查

抽象类中修饰抽象__call__方法的日志装饰器仅外层代码执行,wrapper未触发的问题排查

这个问题我之前也踩过坑!你看到print("this is called")执行了,但wrapper里的内容没触发,核心原因是子类重写__call__方法时,完全覆盖了父类被装饰后的抽象方法——装饰器其实只在父类定义抽象方法的时候执行了外层逻辑,子类的__call__根本没带上这个装饰器的包装。

问题根源拆解

当你在抽象类的@abstractmethod上叠加@log_results时,装饰器的外层代码(print("this is called"))会在类定义阶段执行,但因为__call__是抽象方法,子类必须重写它。子类重写时直接定义了新的__call__,完全和父类被装饰后的方法没关系,所以实际调用子类实例的__call__时,根本不会进入你写的wrapper。


解决方案一:模板方法模式(最稳妥的面向对象实现)

放弃在抽象方法上直接装饰,而是在父类实现一个非抽象的__call__,把日志逻辑固定在这里,让子类实现一个具体的业务方法(比如_run_evaluation)。这样所有子类都会强制走日志流程,不用手动加装饰器。

修改后的抽象类代码:

from functools import wraps
from abc import ABC, abstractmethod
from typing import List

class Evaluator(ABC):
    def __init__(self, evaluation_config, framework_config, **kwargs):
        self.evaluation_config = evaluation_config
        self.framework_config = framework_config
        self.logger = WandbLogger()

    @abstractmethod
    def _create_metrics(self) -> List[Metric]:
        pass

    @abstractmethod
    def load_evaluation_data(self, data_path):
        pass

    # 固定的模板方法,包含日志逻辑
    async def __call__(self, *args, **kwargs):
        print("this is called too!")
        # 调用子类实现的具体业务逻辑
        result = await self._run_evaluation(*args, **kwargs)
        self.logger.log(evaluation_result=result, config=self.framework_config)
        return result

    # 子类必须实现的抽象方法
    @abstractmethod
    async def _run_evaluation(self, *args, **kwargs):
        pass

子类修改(仅重命名方法即可):

class ScoreEvaluator(Evaluator):
    # ... 其他原有代码不变 ...

    # 把原来的__call__改名为_run_evaluation
    async def _run_evaluation(self, *args, **kwargs):
        evaluation_data_path = os.getenv("EVALUATION_DATA_PATH")
        evaluation_data = self.load_evaluation_data(evaluation_data_path)
        evaluation_data = {key: value for key, value in evaluation_data.items()}

        prediction_dict = {
            key: await self.score_generator(
                question=[item["question"] for item in evaluation_data[key]],
                essay=[item["essay"] for item in evaluation_data[key]],
                image=(
                    [item["image"] for item in evaluation_data[key]]
                    if TaskName.TASK_1.value == key
                    else None
                ),
                task_name=key,
                do_sample=False,
                temperature=0,
            )
            for key in evaluation_data.keys()
        }
        print(prediction_dict)

        evaluation_results = {}
        for key, predictions in prediction_dict.items():
            predictions = pd.DataFrame(predictions)
            golds = pd.DataFrame([item["human_rate"] for item in evaluation_data[key]])
            evaluation_results[key] = {
                criterion: self.evaluate_criterion(
                    predictions=predictions[criterion].tolist(), golds=golds[criterion].tolist()
                )
                for criterion in golds.columns
            }

        return evaluation_results

解决方案二:元类自动装饰子类的__call__方法

如果你坚持要保留__call__的命名,用元类可以自动给所有子类的__call__加上装饰器。注意你的__call__是async方法,所以装饰器的wrapper也要改成async兼容版本!

修正后的装饰器和元类代码:

from functools import wraps
from abc import ABC, ABCMeta

# 修正为支持async方法的装饰器
def log_results(func):
    """Decorator to log evaluation results."""
    print("this is called")

    @wraps(func)
    async def wrapper(self, *args, **kwargs):
        print("this is called too!")
        result = await func(self, *args, **kwargs)  # async方法需要await
        self.logger.log(evaluation_result=result, config=self.framework_config)
        return result
    return wrapper

class EvaluatorMeta(ABCMeta):
    def __init__(cls, name, bases, dct):
        super().__init__(name, bases, dct)
        # 只给Evaluator的子类(排除父类本身)的__call__加装饰器
        if bases and Evaluator in bases and "__call__" in dct:
            cls.__call__ = log_results(cls.__call__)

class Evaluator(ABC, metaclass=EvaluatorMeta):
    def __init__(self, evaluation_config, framework_config, **kwargs):
        self.evaluation_config = evaluation_config
        self.framework_config = framework_config
        self.logger = WandbLogger()

    @abstractmethod
    def _create_metrics(self) -> List[Metric]:
        pass

    @abstractmethod
    def load_evaluation_data(self, data_path):
        pass

    # 父类保留抽象__call__,子类必须重写
    @abstractmethod
    async def __call__(self, *args, **kwargs):
        pass

这样子类的__call__在定义时会被元类自动装饰,调用时就会进入wrapper执行日志逻辑了。


方案对比

  • 模板方法模式:逻辑更清晰,明确区分父类固定流程和子类业务逻辑,后期维护成本更低,推荐优先使用。
  • 元类方案:更灵活,不需要修改子类方法命名,但可读性稍差,适合必须保留__call__命名的场景。

备注:内容来源于stack exchange,提问作者Dinosaur

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.14 11:08:10