Bradley-Terry损失函数在推荐系统中的应用与优化实战

1次阅读

没有评论

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

在传统推荐系统的 pointwise 方法中，我们通常将推荐问题简化为对单个物品的评分预测。但这种方法存在两个主要问题：

用户行为数据本质上是相对偏好（用户选择 A 而非 B），而 pointwise 方法强制转换为绝对评分
隐式反馈场景下，未点击物品不代表负面评价（可能是未曝光），直接当作负样本会导致严重的采样偏差

以电影推荐为例，当用户观看了《肖申克的救赎》但没看《阿甘正传》时，我们无法确定用户是否真的不喜欢后者——可能只是没机会看到。Pairwise 方法通过比较物品对的相对偏好关系，更符合实际业务场景的数据特性。

Bradley-Terry 模型将用户 u 对物品 i 和 j 的偏好概率建模为：

$$
P(i >_u j) = \frac{\exp(f_u(i))}{\exp(f_u(i)) + \exp(f_u(j))}
$$

其中 $f_u(i)$ 是模型对用户 u 和物品 i 的匹配度评分。这个形式实际上是逻辑函数的扩展版本，具有以下特点：

当 $f_u(i) \gg f_u(j)$ 时，概率趋近于 1
当 $f_u(j) \gg f_u(i)$ 时，概率趋近于 0
两者分数接近时，概率在 0.5 附近波动

对应的负对数似然损失函数为：

$$
\mathcal{L} = -\sum_{(u,i,j) \in D} \log \sigma(f_u(i) – f_u(j))
$$

其中 $D$ 是所有用户 - 正样本 - 负样本三元组构成的数据集，$\sigma$ 是 sigmoid 函数。这种形式天然适合用随机梯度下降优化。

import torch
import torch.nn as nn
from typing import Tuple

class BradleyTerryLoss(nn.Module):
    """
    实现带负采样的 Bradley-Terry 损失
    Args:
        neg_ratio: 每个正样本对应的负样本数
    """
    def __init__(self, neg_ratio: int = 3):
        super().__init__()
        self.neg_ratio = neg_ratio
        self.bce = nn.BCEWithLogitsLoss()

    def forward(self, 
                user_emb: torch.Tensor, 
                pos_emb: torch.Tensor, 
                neg_emb: torch.Tensor) -> torch.Tensor:
        """
        计算 mini-batch 损失
        Args:
            user_emb: (batch_size, emb_dim)
            pos_emb: (batch_size, emb_dim)
            neg_emb: (batch_size, neg_ratio, emb_dim)
        """
        batch_size = user_emb.size(0)

        # 计算正样本得分 [batch_size, 1]
        pos_score = (user_emb * pos_emb).sum(dim=1, keepdim=True)

        # 计算负样本得分 [batch_size, neg_ratio]
        neg_score = torch.bmm(neg_emb, user_emb.unsqueeze(2)).squeeze(2)

        # 构造标签和扩展正样本分数
        labels = torch.cat([torch.ones_like(pos_score),
            torch.zeros_like(neg_score)
        ], dim=1)

        scores = torch.cat([pos_score.expand_as(neg_score),
            neg_score
        ], dim=1)

        return self.bce(scores, labels)

# 示例训练循环
model = YourRecommendationModel()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = BradleyTerryLoss(neg_ratio=3)

for epoch in range(10):
    for user, pos_item, neg_items in train_loader:  # neg_items.shape=[B,3,emb_dim]
        optimizer.zero_grad()
        user_emb = model.user_encoder(user)
        pos_emb = model.item_encoder(pos_item)
        neg_emb = model.item_encoder(neg_items)

        loss = criterion(user_emb, pos_emb, neg_emb)
        loss.backward()
        optimizer.step()

当用户和物品量达到百万级时，需要特别注意以下性能瓶颈：