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

1次阅读

没有评论

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

在传统推荐系统中，pointwise 方法（如均方误差）直接预测用户对单个物品的评分。这种方式存在两个主要问题：

隐式反馈场景下，用户未交互的物品不一定代表负面偏好
忽略了物品之间的相对排序关系，而推荐本质上是排序问题

比如用户同时喜欢 A 和 B，但 pointwise 方法无法表达 ”A 比 B 更喜欢 ” 的细微差别。这就是我们需要 pairwise 方法的根本原因——通过比较物品对来建模相对偏好。

Bradley-Terry 模型的核心思想是将用户 u 对物品 i 的偏好概率定义为：

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

其中 $f_u(i)$ 是用户 u 对物品 i 的预测分数。这个形式实际上是一个 softmax 函数的二元特例，具有以下优良特性：

当 $f_u(i) \gg f_u(j)$ 时，概率接近 1
满足概率的规范性：$P(i>j) + P(j>i) = 1$
可微性：便于梯度下降优化

对应损失函数（负对数似然）为：

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

其中 $\mathcal{D}$ 是三元组数据集，$(u,i,j)$ 表示用户 u 更喜欢 i 而非 j。

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

    def forward(self, 
                user_emb: torch.Tensor, 
                item_emb: torch.Tensor,
                pos_pairs: torch.Tensor) -> torch.Tensor:
        """
        Args:
            user_emb: (batch_size, emb_dim)
            item_emb: (num_items, emb_dim)
            pos_pairs: (batch_size, 2) 每行是(user_idx, pos_item_idx)
        """
        batch_size = pos_pairs.size(0)

        # 正样本得分 [batch_size]
        pos_scores = (user_emb * item_emb[pos_pairs[:, 1]]).sum(dim=1)

        # 负采样 [batch_size * neg_ratio]
        neg_items = torch.randint(0, len(item_emb), (batch_size * self.neg_ratio,))
        neg_scores = (user_emb.repeat_interleave(self.neg_ratio, dim=0) * 
                      item_emb[neg_items]).sum(dim=1)

        # 计算 pairwise 差值并应用 sigmoid
        diff = pos_scores.repeat_interleave(self.neg_ratio) - neg_scores
        loss = -torch.log(torch.sigmoid(diff)).mean()

        return loss

# 示例训练循环
def train_one_epoch(model, optimizer, train_loader, device):
    model.train()
    total_loss = 0

    for batch in train_loader:
        users, pos_items = batch  # 已预处理为 (user, positive_item) 对
        users = users.to(device)
        pos_items = pos_items.to(device)

        optimizer.zero_grad()
        user_emb, item_emb = model()  # 假设模型返回嵌入

        loss = BradleyTerryLoss()(user_emb, item_emb, 
                                torch.stack([users, pos_items], dim=1))
        loss.backward()
        optimizer.step()

        total_loss += loss.item()

    return total_loss / len(train_loader)