如何在Keras中构建带逐样本权重自定义损失函数的神经网络

你代码中已经开启了run_eagerly=True，所有张量默认按eager模式执行，input、y_true、y_pred均为按批次存储的张量，第一维为批次大小，每一行对应同一个样本的输入、真实标签、预测结果。

1. 打印input、y_true和y_pred的取值 #

eager模式下直接用print或者tf.print均可输出张量的具体数值，示例代码：

def custom_loss(input):
    def loss(y_true, y_pred):
        # 直接打印，eager模式自动输出具体值
        print("批次输入：", input)
        print("真实标签：", y_true)
        print("预测结果：", y_pred)
        # 也可以用tf.print，兼容性更强，summarize=-1可打印全部内容，默认只打印前几个元素
        tf.print("批次输入：", input, summarize=-1)
        # 后续损失逻辑
    return loss

2. 将input转换为numpy ndarray格式 #

开启eager执行的前提下，直接调用张量的.numpy()方法即可转换：

def custom_loss(input):
    def loss(y_true, y_pred):
        input_np = input.numpy()
        y_true_np = y_true.numpy()
        y_pred_np = y_pred.numpy()
        # 此时input_np就是shape为(batch_size, 16)的numpy数组，可以直接遍历查询权重
        return ...
    return loss

如果后续要关闭eager提升运行效率，可以改用TensorFlow内置的哈希表查询权重，无需转numpy。

3. 实现分类错误按权重计算损失的逻辑 #

首先注意你当前模型存在问题：输出层为1个神经元，用softmax激活的话输出值永远为1，无法正常做二分类判断，建议将输出层激活改为sigmoid。
损失实现示例（假设输入的第0位为样本ID，提前构造好ID到权重的映射字典）：

# 提前构造样本ID到权重的映射，可替换为你的实际权重逻辑
sample_weight_map = {1: 0.5, 2: 1.2, 3: 2.0}

def custom_loss(input):
    def loss(y_true, y_pred):
        # 1. 判断每个样本是否分类正确
        # 二分类场景，sigmoid输出>0.5判定为正类
        y_pred_cls = tf.cast(y_pred > 0.5, tf.float32)
        # 正确为1，错误为0
        correct_flag = tf.cast(tf.equal(y_pred_cls, y_true), tf.float32)
        # 错误样本掩码：错误为1，正确为0
        error_mask = 1 - correct_flag

        # 2. 批量查询每个样本的权重
        input_np = input.numpy()
        batch_weights = []
        for i in range(input_np.shape[0]):
            sample_id = int(input_np[i][0])
            batch_weights.append(sample_weight_map.get(sample_id, 1.0)) # 不存在的ID默认权重为1
        batch_weights_tensor = tf.convert_to_tensor(batch_weights, dtype=tf.float32)

        # 3. 计算损失：正确样本损失为0，错误样本损失为对应权重，返回批次平均损失
        per_sample_loss = error_mask * batch_weights_tensor
        return tf.reduce_mean(per_sample_loss)
    return loss

如果想避免转numpy提升运行效率，可以用TensorFlow内置哈希表实现权重查询，无需依赖eager模式：

# 构造静态哈希表
keys = tf.constant([1,2,3], dtype=tf.int32)
values = tf.constant([0.5,1.2,2.0], dtype=tf.float32)
weight_table = tf.lookup.StaticHashTable(
    tf.lookup.KeyValueTensorInitializer(keys, values),
    default_value=1.0
)

def custom_loss(input):
    def loss(y_true, y_pred):
        y_pred_cls = tf.cast(y_pred > 0.5, tf.float32)
        correct_flag = tf.cast(tf.equal(y_pred_cls, y_true), tf.float32)
        error_mask = 1 - correct_flag
        # 直接用张量操作查权重，无需转numpy
        sample_ids = tf.cast(input[:, 0], tf.int32)
        batch_weights_tensor = weight_table.lookup(sample_ids)
        per_sample_loss = error_mask * batch_weights_tensor
        return tf.reduce_mean(per_sample_loss)
    return loss

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