Keras自定义损失函数：分组最大平均二元交叉熵或张量75分位数

针对你想要实现的两种自定义损失函数，我来一步步帮你解决问题：

一、优先方案：分组计算最大平均二元交叉熵 #

你面临的核心问题是分组信息无法传入默认的损失函数，因为Keras原生损失只接收y_true和y_pred。解决思路是把分组ID作为训练时的辅助信息，通过自定义训练循环来传入，避免让模型学习分组ID的特征（我们只需要它来计算损失）。

1. 实现步骤 #

第一步：定义基础模型

不管你用Sequential还是Functional API，模型只需要接收特征输入即可，不需要把分组ID作为特征：

import tensorflow as tf
from tensorflow.keras import Sequential, layers

# 示例模型，根据你的实际任务调整结构
model = Sequential([
    layers.Dense(64, activation='relu', input_shape=(num_features,)),
    layers.Dense(32, activation='relu'),
    layers.Dense(1, activation='sigmoid')
])

第二步：定义分组损失函数

这个函数会接收y_true、y_pred和group_ids，计算每个组的平均BCE后返回最大值：

def group_max_bce(y_true, y_pred, group_ids):
    # 计算每个样本的二元交叉熵
    per_sample_bce = tf.keras.losses.binary_crossentropy(y_true, y_pred)
    # 获取当前批量中的唯一分组ID
    unique_groups = tf.unique(tf.squeeze(group_ids))[0]
    
    # 定义单个组的平均BCE计算逻辑
    def calc_group_avg(group_id):
        # 筛选当前组的样本
        group_mask = tf.equal(tf.squeeze(group_ids), group_id)
        group_bce = tf.boolean_mask(per_sample_bce, group_mask)
        # 处理空组边界情况（实际训练中不会出现）
        return tf.cond(
            tf.size(group_bce) > 0,
            lambda: tf.reduce_mean(group_bce),
            lambda: 0.0
        )
    
    # 遍历所有唯一分组，计算每组平均BCE
    group_avg_bces = tf.map_fn(calc_group_avg, unique_groups, dtype=tf.float32)
    # 返回最大的组平均BCE
    return tf.reduce_max(group_avg_bces)

第三步：自定义训练循环（推荐，更灵活）

用TensorFlow的自定义训练循环，我们可以直接把分组ID传入损失函数，同时完全控制训练流程，适配K-fold交叉验证：

# 准备训练数据集：打包特征、分组ID、标签
train_dataset = tf.data.Dataset.from_tensor_slices(
    (X_train, group_ids_train, y_train)
).shuffle(1000).batch(32)

# 定义优化器
optimizer = tf.keras.optimizers.Adam(learning_rate=1e-3)

# 训练步骤函数
@tf.function
def train_step(x_batch, groups_batch, y_batch):
    with tf.GradientTape() as tape:
        y_pred = model(x_batch, training=True)
        loss = group_max_bce(y_batch, y_pred, groups_batch)
    
    # 更新模型参数
    grads = tape.gradient(loss, model.trainable_variables)
    optimizer.apply_gradients(zip(grads, model.trainable_variables))
    return loss

# 开始训练
epochs = 10
for epoch in range(epochs):
    total_loss = 0.0
    batch_count = 0
    for x_batch, groups_batch, y_batch in train_dataset:
        loss = train_step(x_batch, groups_batch, y_batch)
        total_loss += loss
        batch_count += 1
    print(f"Epoch {epoch+1}, Average Loss: {total_loss / batch_count:.4f}")

K-fold交叉验证适配

在K-fold中，只需要在每个fold划分数据时，同时划分特征、分组ID和标签，然后用上述训练循环即可，完全不影响流程。

二、替代方案：75分位数BCE损失 #

你之前的代码错误在于用了Python的索引方式操作Tensor，而且len(y_true)在TensorFlow中是张量，不能直接做算术运算。以下是修正后的实现：

方法1：用TensorFlow原生quantile函数 #

def percentile_75_bce(y_true, y_pred):
    per_sample_bce = tf.keras.losses.binary_crossentropy(y_true, y_pred)
    # 计算75分位数，axis=0表示对当前批量的所有样本计算
    # keepdims=False返回标量损失
    return tf.quantile(per_sample_bce, q=0.75, axis=0, keepdims=False)

方法2：排序后取对应位置（更直观） #

如果tf.quantile结果不符合预期，可以手动排序后取索引：

def percentile_75_bce(y_true, y_pred):
    per_sample_bce = tf.keras.losses.binary_crossentropy(y_true, y_pred)
    # 对BCE值升序排序
    sorted_bce = tf.sort(per_sample_bce)
    # 计算75分位数的索引（处理整数边界）
    batch_size = tf.shape(sorted_bce)[0]
    idx = tf.cast(tf.math.ceil(0.75 * tf.cast(batch_size, tf.float32)) - 1, tf.int32)
    # 避免索引为负的边界情况
    idx = tf.maximum(idx, 0)
    return sorted_bce[idx]

这个损失函数可以直接在模型编译时使用：

model.compile(optimizer='adam', loss=percentile_75_bce)
model.fit(X_train, y_train, batch_size=32, epochs=10)

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