求助：基于Keras回调实现未加权召回率((recall class1+recall class2)/2)自定义指标

Hey there! Let's tackle your two questions one by one.

Most custom Keras callbacks for metrics work by hooking into the on_epoch_end method, which runs after each training epoch. Here's the typical flow for an F1 callback that also returns accuracy:

• It uses the validation data you pass in to make predictions with the current model state.
• It calculates key values from the confusion matrix: True Positives (TP), True Negatives (TN), False Positives (FP), False Negatives (FN).
• Alongside computing the F1 score (2*TP/(2*TP + FP + FN)), it calculates accuracy using the formula (TP + TN)/(TP + TN + FP + FN).
• Finally, it adds this accuracy value to the logs dictionary, which Keras uses to display metrics during training.

Here's a simplified snippet of what that might look like in the callback:

def on_epoch_end(self, epoch, logs=None):
    logs = logs or {}
    # Get predictions and true labels
    y_pred = self.model.predict(self.val_data[0], verbose=0)
    y_true = self.val_data[1]
    
    # Calculate confusion matrix values
    tp = np.sum((y_true == 1) & (y_pred > 0.5))
    tn = np.sum((y_true == 0) & (y_pred <= 0.5))
    fp = np.sum((y_true == 0) & (y_pred > 0.5))
    fn = np.sum((y_true == 1) & (y_pred <= 0.5))
    
    # Compute F1 and accuracy
    f1 = 2*tp/(2*tp + fp + fn) if (2*tp + fp + fn) !=0 else 0.0
    acc = (tp + tn)/(tp + tn + fp + fn) if (tp + tn + fp + fn) !=0 else 0.0
    
    # Add to logs for training output
    logs['val_f1'] = f1
    logs['val_custom_acc'] = acc
    print(f"\nEpoch {epoch+1}: F1 = {f1:.4f}, Accuracy = {acc:.4f}")

Your goal is to compute unweighted recall as (recall_class1 + recall_class2)/2 — this is exactly macro recall for binary classification, where each class's recall is averaged equally regardless of class size.

Below are two robust implementations: one that calculates metrics manually (no external dependencies) and another that uses scikit-learn for brevity.

Option 1: Manual Calculation (No Scikit-Learn) #

This version handles different label formats (integer or one-hot encoded) and avoids division-by-zero errors:

import numpy as np
from keras.callbacks import Callback

class UnweightedRecallCallback(Callback):
    def __init__(self, val_data, threshold=0.5):
        super().__init__()
        self.val_data = val_data  # Expected format: (X_validation, y_validation)
        self.threshold = threshold  # Threshold for converting probabilities to class labels

    def on_epoch_end(self, epoch, logs=None):
        logs = logs or {}
        X_val, y_val = self.val_data
        
        # Get model predictions
        y_pred = self.model.predict(X_val, verbose=0)
        
        # Convert true labels to integer format (handle one-hot encoded)
        if y_val.ndim == 2 and y_val.shape[1] == 2:
            y_true = np.argmax(y_val, axis=1)
        else:
            y_true = y_val.flatten()
        
        # Convert predictions to integer class labels
        if y_pred.ndim == 2:
            if y_pred.shape[1] == 2:
                # One-hot predictions: take argmax
                y_pred_classes = np.argmax(y_pred, axis=1)
            else:
                # Single probability output: apply threshold
                y_pred_classes = (y_pred > self.threshold).astype(int).flatten()
        else:
            # 1D prediction array: apply threshold
            y_pred_classes = (y_pred > self.threshold).astype(int).flatten()
        
        # Calculate recall for each class
        # Recall for Class 0: True Negatives / (True Negatives + False Positives)
        tp0 = np.sum((y_true == 0) & (y_pred_classes == 0))
        fn0 = np.sum((y_true == 0) & (y_pred_classes == 1))
        recall0 = tp0 / (tp0 + fn0) if (tp0 + fn0) != 0 else 0.0
        
        # Recall for Class 1: True Positives / (True Positives + False Negatives)
        tp1 = np.sum((y_true == 1) & (y_pred_classes == 1))
        fn1 = np.sum((y_true == 1) & (y_pred_classes == 0))
        recall1 = tp1 / (tp1 + fn1) if (tp1 + fn1) != 0 else 0.0
        
        # Compute unweighted recall
        unweighted_recall = (recall0 + recall1) / 2
        
        # Add metric to logs for training display
        logs['val_unweighted_recall'] = unweighted_recall
        print(f"\nEpoch {epoch+1}: Unweighted Recall = {unweighted_recall:.4f}")

Option 2: Using Scikit-Learn (Simpler) #

If you're okay with using scikit-learn, this version is more concise and leverages the built-in recall_score function with average='macro':

import numpy as np
from keras.callbacks import Callback
from sklearn.metrics import recall_score

class UnweightedRecallCallback(Callback):
    def __init__(self, val_data, threshold=0.5):
        super().__init__()
        self.val_data = val_data
        self.threshold = threshold

    def on_epoch_end(self, epoch, logs=None):
        logs = logs or {}
        X_val, y_val = self.val_data
        y_pred = self.model.predict(X_val, verbose=0)
        
        # Convert true labels to integer format
        y_true = np.argmax(y_val, axis=1) if y_val.ndim == 2 else y_val.flatten()
        
        # Convert predictions to integer class labels
        if y_pred.ndim == 2:
            y_pred_classes = np.argmax(y_pred, axis=1) if y_pred.shape[1] == 2 else (y_pred > self.threshold).astype(int).flatten()
        else:
            y_pred_classes = (y_pred > self.threshold).astype(int).flatten()
        
        # Compute macro recall (unweighted) with zero division handling
        unweighted_recall = recall_score(y_true, y_pred_classes, average='macro', zero_division=0)
        
        logs['val_unweighted_recall'] = unweighted_recall
        print(f"\nEpoch {epoch+1}: Unweighted Recall = {unweighted_recall:.4f}")

How to Use the Callback #

Simply pass your validation data when initializing the callback, then add it to your callbacks list during training:

# Assume you have X_val, y_val as your validation dataset
unweighted_recall_cb = UnweightedRecallCallback(val_data=(X_val, y_val))

# Train your model with the callback
model.fit(
    X_train, y_train,
    epochs=20,
    validation_data=(X_val, y_val),
    callbacks=[unweighted_recall_cb]
)

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