如何在Keras中不开启run_eagerly将tensor转为numpy数组实现自定义指标

以下方案均适配keras=2.6、tensorflow=2.6版本环境，按优先级从高到低排序： #

方案1：TensorFlow原生API实现宏精确率（最优，无速度损失） #

不需要调用sklearn，完全基于TensorFlow算子实现，适配图执行模式，无需开启run_eagerly，逻辑完全对齐sklearn.metrics.precision_score(average='macro', zero_division=0)：

import tensorflow as tf
from tensorflow.keras import backend as K

class MacroPrecision(tf.keras.metrics.Metric):
    def __init__(self, num_classes, name='macro_precision', **kwargs):
        super(MacroPrecision, self).__init__(name=name, **kwargs)
        self.num_classes = num_classes
        # 初始化每个类别的TP、FP统计变量
        self.true_positives = self.add_weight(name='tp', shape=(num_classes,), initializer='zeros')
        self.false_positives = self.add_weight(name='fp', shape=(num_classes,), initializer='zeros')

    def update_state(self, y_true, y_pred, sample_weight=None):
        y_true = K.cast(y_true, 'int32')
        # 若y_pred为概率输出，先转为类别ID
        y_pred = K.argmax(y_pred, axis=-1)
        y_pred = K.cast(y_pred, 'int32')
        
        for cls in range(self.num_classes):
            cls_true = K.cast(K.equal(y_true, cls), 'float32')
            cls_pred = K.cast(K.equal(y_pred, cls), 'float32')
            tp = K.sum(cls_true * cls_pred)
            fp = K.sum((1 - cls_true) * cls_pred)
            self.true_positives[cls].assign_add(tp)
            self.false_positives[cls].assign_add(fp)

    def result(self):
        # 单类别精确率计算，处理除零场景
        precision_per_class = self.true_positives / (self.true_positives + self.false_positives + K.epsilon())
        # 对齐sklearn zero_division=0规则：无预测样本的类别精确率记为0
        mask = K.cast(self.true_positives + self.false_positives > 0, 'float32')
        precision_per_class = precision_per_class * mask
        # 求平均得到宏精确率
        return K.mean(precision_per_class)

    def reset_states(self):
        K.set_value(self.true_positives, K.zeros(self.num_classes))
        K.set_value(self.false_positives, K.zeros(self.num_classes))

使用时直接在model.compile中传入即可：

model.compile(
    optimizer='adam',
    loss='sparse_categorical_crossentropy', # 按实际损失调整
    metrics=[MacroPrecision(num_classes=你的实际类别数)]
)

方案2：tf.py_function包装sklearn逻辑（适配必须使用sklearn计算的场景） #

之前使用tf.numpy_function报错大概率是未正确包装为Keras指标类、返回值dtype不匹配，正确实现如下：

import tensorflow as tf
import numpy as np
from sklearn.metrics import precision_score

class SklearnMacroPrecision(tf.keras.metrics.Metric):
    def __init__(self, name='macro_precision', **kwargs):
        super(SklearnMacroPrecision, self).__init__(name=name, **kwargs)
        # 累积全epoch的标签和预测结果
        self.y_true_all = []
        self.y_pred_all = []

    def update_state(self, y_true, y_pred, sample_weight=None):
        def update_numpy(y_true_np, y_pred_np):
            self.y_true_all.append(y_true_np)
            y_pred_np = np.argmax(y_pred_np, axis=-1)
            self.y_pred_all.append(y_pred_np)
            return 0. # 占位返回
        
        tf.py_function(update_numpy, inp=[y_true, y_pred], Tout=tf.float32)

    def result(self):
        def calc_precision():
            y_true = np.concatenate(self.y_true_all, axis=0).ravel()
            y_pred = np.concatenate(self.y_pred_all, axis=0).ravel()
            return precision_score(y_true, y_pred, average='macro', zero_division=0).astype(np.float32)
        
        return tf.py_function(calc_precision, inp=[], Tout=tf.float32)

    def reset_states(self):
        self.y_true_all = []
        self.y_pred_all = []

该方案速度略低于原生API，但远快于开启run_eagerly=True的场景。 #

方案3：Epoch结束回调计算（最稳妥，适合离线验证场景） #

如果不需要训练过程中实时显示batch级指标，可以直接通过Keras Callback在每个epoch结束后计算全验证集的指标，完全规避图执行适配问题：

from tensorflow.keras.callbacks import Callback
from sklearn.metrics import precision_score

class MacroPrecisionCallback(Callback):
    def on_epoch_end(self, epoch, logs=None):
        logs = logs or {}
        val_pred = np.argmax(self.model.predict(self.validation_data[0], verbose=0), axis=-1)
        val_true = self.validation_data[1].ravel()
        val_precision = precision_score(val_true, val_pred, average='macro', zero_division=0)
        logs['val_macro_precision'] = val_precision
        print(f' - val_macro_precision: {val_precision:.4f}')

使用时在model.fit中传入回调即可：

model.fit(
    x_train, y_train,
    validation_data=(x_val, y_val),
    callbacks=[MacroPrecisionCallback()],
    epochs=10
)

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