带自定义损失函数的Keras模型在Sklearn Pipeline中深拷贝失败求助

我定义了一个自定义损失函数：当预测值小于真实值时，将MAE（平均绝对误差）乘以2，否则直接返回MAE。在Sklearn Pipeline中训练该Keras模型时，尝试深拷贝包含模型与自定义对象的Pipeline，出现ValueError，提示无法恢复类型为_tf_keras_metric的自定义对象。必须使用Pipeline完成前置操作，需要解决深拷贝问题。

相关代码 #

自定义损失函数 #

def custom_loss(y_true, y_pred):
        mae = tf.keras.losses.MeanAbsoluteError()
        penalty = 2
        # penalize the loss heavily if the prediction is smaller than true
        loss = tf.where(
            condition=tf.greater(y_true, y_pred),
            x=mae(y_true, y_pred) * penalty,
            y=mae(y_true, y_pred)
        )
        return loss

深拷贝代码 #

regr = deepcopy(regr)
temp = RegressionRecords([], regr, r2_score(np.array(predict_df["true_data"]), np.array(predict_df["predictions"])), predict_df, None)

PredictionTransformer类 #

class PredictionTransformer(BaseEstimator, TransformerMixin):
    def __init__(self, estimator):
        self.estimator = estimator # Keras model passed in as estimator
    @property
    def history(self):
        return self.estimator.history

    @property
    def model(self):
        return self.estimator.model

    def fit(self, X, y):
        self.estimator.train(X, y)

    def predict(self, X):
        return self.estimator.transform(X)

错误信息 #

File "/usr/lib/python3.8/copy.py", line 172, in deepcopy
    y = _reconstruct(x, memo, *rv)
  File "/usr/lib/python3.8/copy.py", line 270, in _reconstruct
    state = deepcopy(state, memo)
  File "/usr/lib/python3.8/copy.py", line 146, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python3.8/copy.py", line 230, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python3.8/copy.py", line 146, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python3.8/copy.py", line 205, in _deepcopy_list
    append(deepcopy(a, memo))
  File "/usr/lib/python3.8/copy.py", line 146, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python3.8/copy.py", line 210, in _deepcopy_tuple
    y = [deepcopy(a, memo) for a in x]
  File "/usr/lib/python3.8/copy.py", line 210, in <listcomp>
    y = [deepcopy(a, memo) for a in x]
  File "/usr/lib/python3.8/copy.py", line 172, in deepcopy
    y = _reconstruct(x, memo, *rv)
  File "/usr/lib/python3.8/copy.py", line 270, in _reconstruct
    state = deepcopy(state, memo)
  File "/usr/lib/python3.8/copy.py", line 146, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python3.8/copy.py", line 230, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python3.8/copy.py", line 172, in deepcopy
    y = _reconstruct(x, memo, *rv)
  File "/usr/lib/python3.8/copy.py", line 270, in _reconstruct
    state = deepcopy(state, memo)
  File "/usr/lib/python3.8/copy.py", line 146, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python3.8/copy.py", line 230, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python3.8/copy.py", line 153, in deepcopy
    y = copier(memo)
  File "/usr/local/lib/python3.8/dist-packages/keras/engine/training.py", line 337, in __deepcopy__
    new = pickle_utils.deserialize_model_from_bytecode(
  File "/usr/local/lib/python3.8/dist-packages/keras/saving/pickle_utils.py", line 48, in deserialize_model_from_bytecode
    model = save_module.load_model(temp_dir)
  File "/usr/local/lib/python3.8/dist-packages/keras/utils/traceback_utils.py", line 67, in error_handler
    raise e.with_traceback(filtered_tb) from None
  File "/usr/local/lib/python3.8/dist-packages/keras/saving/saved_model/load.py", line 994, in revive_custom_object
    raise ValueError(
ValueError: Unable to restore custom object of type _tf_keras_metric. Please make sure that any custom layers are included in the `custom_objects` arg when calling `load_model()` and make sure that all layers implement `get_config` and `from_config`

方法1：修改自定义损失函数，避免实例化Metric类 #

当前损失函数直接实例化MeanAbsoluteError Metric类，导致序列化时无法正确恢复。改用TensorFlow基础操作计算MAE：

def custom_loss(y_true, y_pred):
    mae = tf.abs(y_true - y_pred)
    penalty = 2
    loss = tf.where(
        condition=tf.greater(y_true, y_pred),
        x=mae * penalty,
        y=mae
    )
    return tf.reduce_mean(loss)  # 确保返回标量损失

这种方式不依赖Metric实例，从根源避免自定义对象序列化问题。

方法2：为PredictionTransformer实现自定义深拷贝逻辑 #

重写__deepcopy__方法，手动处理模型的保存与加载，传入自定义损失函数：

class PredictionTransformer(BaseEstimator, TransformerMixin):
    def __init__(self, estimator):
        self.estimator = estimator # Keras model passed in as estimator
        
    @property
    def history(self):
        return self.estimator.history

    @property
    def model(self):
        return self.estimator.model

    def fit(self, X, y):
        self.estimator.train(X, y)

    def predict(self, X):
        return self.estimator.transform(X)
    
    def __deepcopy__(self, memo):
        # 手动保存模型到临时文件
        import tempfile
        from tensorflow.keras.models import load_model
        with tempfile.TemporaryDirectory() as temp_dir:
            self.model.save(temp_dir)
            # 加载模型时传入自定义损失
            copied_model = load_model(temp_dir, custom_objects={'custom_loss': custom_loss})
            # 根据你的estimator实际构造方式调整
            copied_estimator = type(self.estimator)(model=copied_model)
            copied_estimator.history = deepcopy(self.estimator.history, memo)
            return PredictionTransformer(copied_estimator)

方法3：手动复制Pipeline组件，替代deepcopy #

放弃整体深拷贝，逐个复制Pipeline组件，对Keras模型单独处理：

from sklearn.pipeline import Pipeline
import tempfile
from tensorflow.keras.models import load_model

copied_steps = []
for name, step in regr.steps:
    if isinstance(step, PredictionTransformer):
        # 单独复制Keras模型
        with tempfile.TemporaryDirectory() as temp_dir:
            step.model.save(temp_dir)
            copied_model = load_model(temp_dir, custom_objects={'custom_loss': custom_loss})
            copied_estimator = type(step.estimator)(model=copied_model)
            copied_estimator.history = deepcopy(step.estimator.history)
            copied_step = PredictionTransformer(copied_estimator)
    else:
        # Sklearn原生组件直接深拷贝
        copied_step = deepcopy(step)
    copied_steps.append((name, copied_step))

regr = Pipeline(copied_steps)

这种方式更可控，避免Pipeline整体深拷贝触发的Keras序列化问题。

内容的提问来源于stack exchange，提问作者Shen