如何在Keras中每个epoch获取scikit-learn指标报告并利用GPU

核心问题分析 #

你之前的错误是因为在图模式下，Keras的张量没有numpy()方法（仅Eager张量支持），强行开启Eager模式会禁用TensorFlow的图优化，导致训练速度暴跌且无法利用GPU加速。要解决这个问题，需要在不破坏图模式训练的前提下，在epoch结束时安全调用scikit-learn的指标。

方法一：自定义Keras Metric类（推荐，集成到训练流程） #

通过继承tf.keras.metrics.Metric，可以将scikit-learn指标无缝集成到Keras的训练循环中，每个epoch自动计算并输出指标，同时保留图模式的GPU加速优势。

以coverage_error为例，实现自定义指标：

import tensorflow as tf
from sklearn.metrics import coverage_error

class CoverageError(tf.keras.metrics.Metric):
    def __init__(self, name="coverage_error", **kwargs):
        super().__init__(name=name, **kwargs)
        # 用于存储每个batch的真实标签和预测得分
        self.y_true = []
        self.y_score = []

    def update_state(self, y_true, y_score, sample_weight=None):
        # 收集batch级别的张量（图模式下不会立即计算）
        self.y_true.append(y_true)
        self.y_score.append(y_score)

    def result(self):
        # 在epoch结束时才执行，此时可以安全转换为numpy数组
        y_true_np = tf.concat(self.y_true, axis=0).numpy()
        y_score_np = tf.concat(self.y_score, axis=0).numpy()
        # 调用scikit-learn指标
        err = coverage_error(y_true_np, y_score_np)
        # 转换为Tensor返回
        return tf.convert_to_tensor(err, dtype=tf.float32)

    def reset_state(self):
        # 每个epoch开始时重置存储的张量
        self.y_true = []
        self.y_score = []

使用方式：
在模型编译时直接传入自定义指标：

model.compile(
    optimizer="adam",
    loss="binary_crossentropy",
    metrics=[CoverageError()]  # 可以添加多个自定义指标
)

# 正常训练，无需开启Eager模式
model.fit(x_train, y_train, epochs=10, validation_data=(x_val, y_val))

方法二：自定义Callback（灵活适配多指标） #

如果需要同时计算多个scikit-learn指标，自定义Callback会更灵活。Callback在每个epoch结束时触发，此时可以安全获取真实标签和预测结果的numpy数组，完全不影响训练阶段的图模式GPU加速。

示例代码：

import tensorflow as tf
from sklearn.metrics import coverage_error, label_ranking_average_precision_score, label_ranking_loss

class SKLearnMetricsCallback(tf.keras.callbacks.Callback):
    def __init__(self, val_data):
        super().__init__()
        self.x_val, self.y_val = val_data

    def on_epoch_end(self, epoch, logs=None):
        # 获取验证集预测结果（图模式下predict会返回numpy数组）
        y_score = self.model.predict(self.x_val, verbose=0)
        y_true = self.y_val.numpy() if isinstance(self.y_val, tf.Tensor) else self.y_val
        
        # 计算多个scikit-learn指标
        coverage = coverage_error(y_true, y_score)
        lap = label_ranking_average_precision_score(y_true, y_score)
        ranking_loss = label_ranking_loss(y_true, y_score)
        
        # 打印指标结果，也可以存入logs字典让Keras自动记录
        print(f"\nEpoch {epoch+1} SKLearn Metrics:")
        print(f"Coverage Error: {coverage:.4f}")
        print(f"Label Ranking Average Precision: {lap:.4f}")
        print(f"Label Ranking Loss: {ranking_loss:.4f}")
        # 可选：将指标加入logs，方便TensorBoard等工具可视化
        if logs is not None:
            logs["val_coverage_error"] = coverage
            logs["val_lap"] = lap
            logs["val_ranking_loss"] = ranking_loss

使用方式：
训练时传入Callback：

# 假设val_data是(x_val, y_val)，可以是numpy数组或TensorFlow Dataset
callback = SKLearnMetricsCallback(val_data=(x_val, y_val))

# 正常训练，无需开启Eager模式
model.fit(
    x_train, y_train,
    epochs=10,
    validation_data=(x_val, y_val),
    callbacks=[callback]
)

关键说明 #

• 两种方法都保留了TensorFlow的图模式训练，训练阶段完全利用GPU加速，只有在epoch结束计算指标时才会转换为numpy数组调用scikit-learn。
• 自定义Metric更适合需要和Keras内置指标一样自动记录到训练日志的场景；自定义Callback则适合一次性计算多个指标，或需要自定义输出格式的需求。

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