推荐系统中的BPR损失函数：原理剖析与工程实践优化

1次阅读

共计 2537 个字符，预计需要花费 7 分钟才能阅读完成。

在推荐系统的排序阶段，传统 pointwise 损失函数（如交叉熵）存在两个明显缺陷：

只关注单个物品的预测准确性，忽视了用户对不同物品的相对偏好关系
对未曝光物品的负样本处理简单粗暴（全部视为同等负面）

BPR(Bayesian Personalized Ranking)的创新在于将问题转化为 pairwise 排序任务，建模用户对正负样本对的偏好概率。例如电商场景中，用户点击过的商品（正样本）应比未点击的（负样本）获得更高评分，这正是 BPR 的核心思想。

BPR 的推导从贝叶斯后验最大化出发。定义用户 $u$ 对物品 $i$ 和 $j$ 的偏好关系为：

$$
\hat{x}{uij} = \hat{x}
$$} – \hat{x}_{uj

其中 $\hat{x}_{ui}$ 是模型对用户 $u$ 与物品 $i$ 的匹配度预测值。根据贝叶斯定理，最大化后验概率等价于：

$$
\prod_{u,i,j} P(i >_u j)^{\delta((u,i,j)\in D)} \cdot (1-P(i >_u j))^{\delta((u,i,j)\notin D)}
$$

使用 sigmoid 函数表示偏好概率后，得到最终损失函数：

$$
\text{BPR-OPT} = -\sum_{(u,i,j)\in D} \ln \sigma(\hat{x}_{uij}) + \lambda||\Theta||^2
$$

import tensorflow as tf

class BPRLoss(tf.keras.losses.Loss):
    def __init__(self, reg_lambda=0.01):
        super().__init__()
        self.reg = reg_lambda

    def call(self, y_true, y_pred):
        # y_pred shape: (batch_size, 3) -> [user_emb, pos_emb, neg_emb]
        user_emb = y_pred[:, 0]
        pos_emb = y_pred[:, 1]
        neg_emb = y_pred[:, 2]

        # 计算偏好分差
        x_uij = tf.reduce_sum(user_emb * pos_emb, axis=1) - \
                tf.reduce_sum(user_emb * neg_emb, axis=1)

        # 正则化项
        l2_norm = tf.reduce_sum(user_emb**2) + \
                  tf.reduce_sum(pos_emb**2) + \
                  tf.reduce_sum(neg_emb**2)

        return -tf.reduce_mean(tf.math.log(tf.sigmoid(x_uij))) + \
               self.reg * l2_norm

def build_sampler(data, num_neg=4, temperature=1.0):
    """
    data: 原始交互数据 DataFrame
    temperature: 采样温度系数(>1 时平滑分布，<1 时尖锐分布)
    """item_popularity = data['item_id'].value_counts()
    sample_weights = item_popularity ** (1/temperature)
    sample_weights = sample_weights / sample_weights.sum()

    def sampler(user_batch):
        pos_items = user_batch['pos_items']
        neg_items = np.random.choice(
            sample_weights.index, 
            size=(len(pos_items), num_neg),
            p=sample_weights.values
        )
        return {'user': user_batch['user'],
            'pos_items': pos_items,
            'neg_items': neg_items
        }

    return sampler

Hard Negative Mining：每轮训练后，筛选模型预测分数较高的未曝光物品作为难负样本
热度降权：对热门负样本进行降采样，缓解流行度偏差
温度系数调整 ：训练初期使用高温(>1) 探索，后期降低温度聚焦难样本

@tf.function
def train_step(batch):
    with tf.GradientTape() as tape:
        user_emb = model.user_embedding(batch['user'])
        pos_emb = model.item_embedding(batch['pos_items'])
        neg_emb = model.item_embedding(batch['neg_items'])

        inputs = tf.stack([user_emb, pos_emb, neg_emb], axis=1)
        loss = bpr_loss(None, inputs)

    gradients = tape.gradient(loss, model.trainable_variables)
    optimizer.apply_gradients(zip(gradients, model.trainable_variables))
    return loss

在电商推荐场景的 AB 测试结果显示：