08 · 深度学习推荐模型

  用户看过: [篮球鞋, 口红, 笔记本电脑, 键盘, 面膜]
                    ↓ 平均
             一个"四不像"的用户向量
                    ↓
  现在要判断他会不会点【机械键盘】
  → 口红和面膜的信息只是噪声，还稀释了键盘的信号

import torch
import torch.nn as nn

class AttentionUnit(nn.Module):
    """DIN 的核心：算历史行为对当前候选物品的注意力权重"""
    def __init__(self, emb_dim, hidden=[64, 32]):
        super().__init__()
        # 输入是 4 段拼接：候选、历史、差、积 → 显式提供交互信号
        layers, in_dim = [], emb_dim * 4
        for h in hidden:
            layers += [nn.Linear(in_dim, h), nn.PReLU()]
            in_dim = h
        layers.append(nn.Linear(in_dim, 1))
        self.mlp = nn.Sequential(*layers)

    def forward(self, query, keys, mask):
        """
        query: (B, D)      候选物品 Embedding
        keys:  (B, T, D)   历史行为 Embedding 序列
        mask:  (B, T)      1=真实行为, 0=padding
        """
        B, T, D = keys.shape
        q = query.unsqueeze(1).expand(-1, T, -1)          # (B,T,D)
        # ⭐ 拼接四种交互形式，让 MLP 自己学怎么算相关性
        x = torch.cat([q, keys, q - keys, q * keys], dim=-1)
        scores = self.mlp(x).squeeze(-1)                   # (B,T)

        # padding 位置置为极小值，softmax 后接近 0
        scores = scores.masked_fill(mask == 0, -1e9)
        weights = torch.softmax(scores, dim=1)             # (B,T)

        # 加权求和 → 针对该候选物品的用户兴趣向量
        return torch.bmm(weights.unsqueeze(1), keys).squeeze(1)   # (B,D)


class DIN(nn.Module):
    def __init__(self, n_items, n_cats, emb_dim=32):
        super().__init__()
        self.item_emb = nn.Embedding(n_items, emb_dim, padding_idx=0)
        self.cat_emb  = nn.Embedding(n_cats,  emb_dim, padding_idx=0)
        self.attn = AttentionUnit(emb_dim * 2)
        self.mlp = nn.Sequential(
            nn.Linear(emb_dim * 2 * 3, 200), nn.PReLU(),
            nn.Linear(200, 80), nn.PReLU(),
            nn.Linear(80, 1),
        )

    def forward(self, hist_items, hist_cats, cand_item, cand_cat, mask):
        hist = torch.cat([self.item_emb(hist_items), self.cat_emb(hist_cats)], -1)
        cand = torch.cat([self.item_emb(cand_item),  self.cat_emb(cand_cat)], -1)

        interest = self.attn(cand, hist, mask)             # 动态兴趣向量
        # 拼接：候选 + 兴趣 + 交互
        x = torch.cat([cand, interest, cand * interest], dim=-1)
        return torch.sigmoid(self.mlp(x)).squeeze(-1)

  【离线，每天跑一次】
     所有 1 亿个物品过物品塔 → 1 亿个向量 → 存入向量索引 (Faiss/HNSW)

  【线上，每次请求】
     只跑一次用户塔  → 一个用户向量 u   (~5ms)
     在向量索引里找和 u 最近的 1000 个 → ANN 检索  (~5ms)

     总共 10ms，从 1 亿里选出 1000 个！ ✅

import torch
import torch.nn as nn
import torch.nn.functional as F

class TwoTower(nn.Module):
    def __init__(self, n_users, n_items, emb_dim=64, out_dim=64):
        super().__init__()
        self.user_emb = nn.Embedding(n_users, emb_dim)
        self.item_emb = nn.Embedding(n_items, emb_dim)
        self.user_tower = nn.Sequential(
            nn.Linear(emb_dim, 128), nn.ReLU(), nn.Linear(128, out_dim))
        self.item_tower = nn.Sequential(
            nn.Linear(emb_dim, 128), nn.ReLU(), nn.Linear(128, out_dim))

    def encode_user(self, u):
        # ⭐ L2 归一化：让点积等价于余弦相似度，训练更稳定
        return F.normalize(self.user_tower(self.user_emb(u)), dim=-1)

    def encode_item(self, i):
        return F.normalize(self.item_tower(self.item_emb(i)), dim=-1)


def in_batch_softmax_loss(model, users, pos_items, temperature=0.05,
                          item_freq=None):
    """
    In-batch 负采样：同一批次里其他人的正样本，就是我的负样本
    好处：不用单独采负样本，一个 batch 就能产生 B×(B-1) 个负样本对
    """
    u = model.encode_user(users)          # (B, D)
    v = model.encode_item(pos_items)      # (B, D)

    logits = u @ v.T / temperature        # (B, B) 每个用户对所有物品的分

    # ⭐ 关键修正：LogQ 校正
    # 热门物品更容易出现在 batch 里，会被过度当成负样本 → 减去 log(采样概率)
    if item_freq is not None:
        logits = logits - torch.log(item_freq[pos_items]).unsqueeze(0)

    labels = torch.arange(len(users), device=users.device)  # 对角线是正样本
    return F.cross_entropy(logits, labels)