pdn.py

1"""2PDN 路径召回（第10章 §10.6.1）34论文：Li et al., "Path-based Deep Network for Candidate Item Matching in Recommenders",5     SIGIR 202167核心思想：把传统的 U2I 匹配扩展为 U2I2I 的两跳路径建模  u → j → i，8其中 j 是用户历史交互过的 Trigger 物品，i 是待召回的目标物品：9    t_uj  = TriggerNet(z_u, a_uj, x_j)      第一跳：用户对历史物品 j 的触发强度10    s_ji  = SimNet(x_j, x_i, c_ji)          第二跳：历史物品 j 与候选物品 i 的相关性11    PATH_uji = t_uj + s_ji                   路径得分（用于 BCE 训练）12    d_ui  = DirectNet 双塔内积，泛化兴趣得分13    y_bias = BiasNet 曝光偏置（仅训练期使用）14    ŷ_ui  = d_ui + Σ_{j∈N(u)} (t_uj + s_ji) + y_bias   训练时的总得分 logit15    p_ui  = sigmoid(ŷ_ui)，用 BCEWithLogits 训练1617PDN 兼具两方面优势：每条路径 u→j→i 都能单独解释「因为你看过 j 所以推荐 i」，18保留了 ItemCF 的可解释性；同时 TriggerNet / SimNet 都是深度网络，可融合用户画像、19物品 side info、共现统计等丰富特征，具备深度模型的表达能力。20在线 Serving 时 SimNet 离线产出 I2I 倒排索引，TriggerNet 在线筛 Top-m Trigger，21把 O(|N(u)|·|I|) 的全量路径计算降到 m×K 规模，并移除 BiasNet 得到无偏打分。2223本实现：PyTorch 实现，y_ui = AGG_j(t_uj * s_ji)，BCE 训练，四个子网络联合优化。24"""2526import math27import random28import collections2930import torch31import torch.nn as nn32import torch.nn.functional as F33import torch.optim as optim343536class _MLP(nn.Module):37    """两层感知机 h = ReLU(W1 x + b1)，y = W2 h + b2。"""3839    def __init__(self, in_dim: int, hid_dim: int, out_dim: int = 1):40        super().__init__()41        self.fc1 = nn.Linear(in_dim, hid_dim)42        self.fc2 = nn.Linear(hid_dim, out_dim)43        nn.init.xavier_uniform_(self.fc1.weight)44        nn.init.zeros_(self.fc1.bias)45        nn.init.xavier_uniform_(self.fc2.weight)46        nn.init.zeros_(self.fc2.bias)4748    def forward(self, x: torch.Tensor) -> torch.Tensor:49        return self.fc2(F.relu(self.fc1(x)))505152class PDN(nn.Module):53    """54    简化版 PDN 召回模型，包含 TriggerNet / SimNet / DirectNet / BiasNet 四个模块。5556    特征约定：57        z_u  = [user_emb(u); 年龄归一化, 性别]      用户画像58        x_i  = [item_emb(i); cate_emb(c_i)]          物品特征（含 side info）59        a_uj = [点击次数归一化, 时间衰减]            用户对历史物品 j 的行为特征60        c_ji = [共现次数归一化, 是否同类目]          物品 j 与 i 的共现 / 关联特征61    """6263    def __init__(self, n_users: int, n_items: int, item_cate: list, n_cate: int,64                 emb_dim: int = 16, cate_dim: int = 4, hid: int = 8,65                 lr: float = 0.05, reg: float = 1e-4):66        super().__init__()67        self.n_users, self.n_items, self.item_cate = n_users, n_items, item_cate68        self.emb_dim, self.cate_dim = emb_dim, cate_dim69        s = math.sqrt(1.0 / emb_dim)7071        self.user_emb = nn.Embedding(n_users, emb_dim)72        self.item_emb = nn.Embedding(n_items, emb_dim)73        self.cate_emb = nn.Embedding(n_cate, cate_dim)74        for tab in (self.user_emb, self.item_emb, self.cate_emb):75            nn.init.uniform_(tab.weight, -s, s)7677        du, di = emb_dim + 2, emb_dim + cate_dim78        self.trigger_net = _MLP(du + 2 + di, hid, 1)   # (z_u, a_uj, x_j) -> t_uj79        self.sim_net     = _MLP(di + di + 2, hid, 1)   # (x_j, x_i, c_ji) -> s_ji80        self.user_tower  = _MLP(du, hid, emb_dim)      # DirectNet 用户塔 -> p_u81        self.item_tower  = _MLP(di, hid, emb_dim)      # DirectNet 物品塔 -> q_i82        self.bias_net    = _MLP(2, hid, 1)             # BiasNet（仅训练期）8384        self.cooc: dict = collections.defaultdict(float)  # 物品共现统计85        self.i2i:  dict = {}                               # 离线 I2I 倒排索引8687        self.optimizer = optim.Adam(self.parameters(), lr=lr, weight_decay=reg)8889    # ── 特征构造 ────────────────────────────────────────────────────────────9091    def _z_u(self, uid: int, profile: list) -> torch.Tensor:92        prof = torch.tensor(profile, dtype=torch.float32)93        return torch.cat([self.user_emb(torch.tensor(uid, dtype=torch.long)), prof])9495    def _x_i(self, iid: int) -> torch.Tensor:96        e = self.item_emb(torch.tensor(iid, dtype=torch.long))97        c = self.cate_emb(torch.tensor(self.item_cate[iid], dtype=torch.long))98        return torch.cat([e, c])99100    def _c_ji(self, j: int, i: int) -> torch.Tensor:101        """物品间共现特征：归一化共现次数 + 是否同类目。"""102        return torch.tensor(103            [self.cooc[(j, i)], 1.0 if self.item_cate[j] == self.item_cate[i] else 0.0],104            dtype=torch.float32)105106    # ── 四个子网络的前向 ────────────────────────────────────────────────────107108    def trigger(self, uid: int, profile: list, j: int, a_uj: list) -> torch.Tensor:109        """TriggerNet：t_uj = TriggerNet(z_u, a_uj, x_j)，第一跳「用户 → 历史物品」。"""110        a = torch.tensor(a_uj, dtype=torch.float32)111        x = torch.cat([self._z_u(uid, profile), a, self._x_i(j)])112        return self.trigger_net(x).squeeze(-1)          # scalar tensor113114    def sim(self, j: int, i: int) -> torch.Tensor:115        """SimNet：s_ji = SimNet(x_j, x_i, c_ji)，第二跳「历史物品 → 候选物品」。"""116        x = torch.cat([self._x_i(j), self._x_i(i), self._c_ji(j, i)])117        return self.sim_net(x).squeeze(-1)              # scalar tensor118119    def direct(self, uid: int, profile: list, i: int) -> torch.Tensor:120        """DirectNet：双塔内积 d_ui = p_u^T q_i，建模一跳泛化兴趣。"""121        p = self.user_tower(self._z_u(uid, profile))122        q = self.item_tower(self._x_i(i))123        return (p * q).sum()                            # scalar tensor124125    # ── 训练 ────────────────────────────────────────────────────────────────126127    def forward(self, uid: int, profile: list, hist: list,128                i: int, label: float, bias_feat: list) -> torch.Tensor:129        """130        一步训练（逐路径二分类 + 一跳 DirectNet 联合优化）。131132        与「把一条样本的所有路径加总再打一个标签」不同，这里对**每一条路径单独打标签**：133        目标是「同类目（或高共现）的物品往往被同一用户一起消费」，因此把历史中与目标 i134        同类目的物品记为**正路径** u→j→i，其余记为**负路径**，路径得分135            path_uji = t_uj + s_ji136        直接用 BCEWithLogits 训练。这样 SimNet 才知道「哪些 (j,i) 对是相关的」——否则137        路径和在一起时正负抵消，SimNet 学不到任何区分度（I2I 索引会退化成常数）。138        DirectNet 则用样本级标签，建模一跳泛化兴趣。139        """140        # 1) DirectNet 一跳兴趣：样本级 BCE141        d_ui = self.direct(uid, profile, i)142        lab = torch.tensor([label], dtype=torch.float32)143        loss = F.binary_cross_entropy_with_logits(d_ui.unsqueeze(0), lab)144145        # 2) 两跳路径：按「是否与目标同类目」逐路径打标签146        t_ujs = torch.stack([self.trigger(uid, profile, j, a) for j, a in hist])  # [|N(u)|]147        sim_ji = torch.stack([self.sim(j, i) for j, _ in hist])                   # [|N(u)|]148        path_label = torch.tensor(149            [1.0 if self.item_cate[j] == self.item_cate[i] else 0.0 for j, _ in hist],150            dtype=torch.float32)151        loss = loss + F.binary_cross_entropy_with_logits(152            t_ujs + sim_ji, path_label, reduction="mean")153154        # 3) BiasNet：吸收曝光位置偏置（仅训练期）155        y_bias = self.bias_net(torch.tensor(bias_feat, dtype=torch.float32)).squeeze(-1)156        loss = loss + F.binary_cross_entropy_with_logits(y_bias.unsqueeze(0), lab)157        return loss158159    def train_model(self, samples: list, epochs: int = 6) -> None:160        """samples: [(uid, profile, hist, item, label, bias_feat), ...]"""161        self.train()162        for epoch in range(epochs):163            random.shuffle(samples)164            total = 0.0165            for sp in samples:166                self.optimizer.zero_grad()167                loss = self(*sp)168                loss.backward()169                self.optimizer.step()170                total += loss.item()171            print(f"  epoch {epoch}  avg_loss={total / max(len(samples), 1):.4f}")172173    # ── 离线索引 + 在线召回 ─────────────────────────────────────────────────174175    def build_i2i_index(self, top_k: int = 4) -> None:176        """离线用 SimNet 给候选物品对打分，为每个 Trigger 物品保留 Top-K 关联物品。"""177        self.eval()178        self.i2i = {}179        with torch.no_grad():180            for j in range(self.n_items):181                scored = [(i, self.sim(j, i).item()) for i in range(self.n_items) if i != j]182                scored.sort(key=lambda x: x[1], reverse=True)183                self.i2i[j] = scored[:top_k]184185    def recommend(self, uid: int, profile: list, hist: list,186                  top_n: int = 5, top_m: int = 3) -> list:187        """188        在线召回：TriggerNet 选 Top-m Trigger → 查 I2I 索引得到候选 → 路径打分重排。189        线上不使用 BiasNet，因此得到的是去除曝光位置偏置后的兴趣打分。190        """191        self.eval()192        with torch.no_grad():193            trig_scores = [(j, self.trigger(uid, profile, j, a).item(), a) for j, a in hist]194            triggers = sorted(trig_scores, key=lambda x: x[1], reverse=True)[:top_m]195            seen = {j for j, _ in hist}196            cands = {i for j, _, _ in triggers for i, _ in self.i2i.get(j, []) if i not in seen}197            out = []198            for i in cands:199                score = self.direct(uid, profile, i).item()200                best_path = (None, -1e18)201                for j, t_uj, a in triggers:202                    # 路径得分：重新调用 trigger 以保持图一致（均在 no_grad 下）203                    path_score = (self.trigger(uid, profile, j, a) + self.sim(j, i)).item()204                    score += path_score205                    if path_score > best_path[1]:206                        best_path = (j, path_score)207                out.append((i, score, best_path[0], best_path[1]))208        out.sort(key=lambda x: x[1], reverse=True)209        return out[:top_n]210211212# ── 快速验证 ──────────────────────────────────────────────────────────────────213if __name__ == "__main__":214    random.seed(42)215    torch.manual_seed(42)216217    n_users, n_items, n_cate = 8, 16, 8218    item_cate = [i % n_cate for i in range(n_items)]  # 每个类目恰好 2 个物品219220    # 合成数据：用户偏好 4 个类目，历史行为在每个偏好类目里取 1 个物品，221    # 于是「正样本」恰好是某个历史物品的同类目兄弟 —— U2I2I 的路径信号非常干净222    profiles, hists, favs = {}, {}, {}223    for u in range(n_users):224        favs[u] = random.sample(range(n_cate), 4)225        picked = [random.choice([i for i in range(n_items) if item_cate[i] == c]) for c in favs[u]]226        profiles[u] = [random.random(), float(random.randint(0, 1))]227        hists[u] = [(j, [random.random(), random.random()]) for j in picked]228229    model = PDN(n_users, n_items, item_cate, n_cate, emb_dim=16, cate_dim=4, hid=8, lr=0.05)230    # 共现统计：同一用户历史内部的物品两两共现231    for u in range(n_users):232        ids = [j for j, _ in hists[u]]233        for a in ids:234            for b in ids:235                if a != b:236                    model.cooc[(a, b)] += 0.25237238    # 训练样本：目标物品落在偏好类目内记为点击；头部位置带来一部分虚假点击（曝光偏置）239    samples = []240    for u in range(n_users):241        seen = {j for j, _ in hists[u]}242        for i in range(n_items):243            if i in seen:244                continue245            pos = random.randint(0, 9)246            label = 1.0 if item_cate[i] in favs[u] else 0.0247            if label == 0.0 and pos <= 1 and random.random() < 0.3:248                label = 1.0249            samples.append((u, profiles[u], hists[u], i, label, [pos / 9.0, 1.0]))250251    n_pos = sum(1 for s in samples if s[4] > 0.5)252    print(f"用户数={n_users} 物品数={n_items} 类目数={n_cate} "253          f"训练样本数={len(samples)}（正样本 {n_pos}）")254    print("PDN 训练（TriggerNet + SimNet + DirectNet + BiasNet 联合优化）：")255    model.train_model(samples, epochs=6)256257    print("\n离线构建 I2I 倒排索引（SimNet 为每个 Trigger 物品保留 Top-3 关联物品）：")258    model.build_i2i_index(top_k=3)259    for j in range(4):260        pairs = "  ".join(f"i{i}(c{item_cate[i]}, s={s:+.3f})" for i, s in model.i2i[j])261        print(f"  trigger item {j:>2} (类目 {item_cate[j]}) -> {pairs}")262    sib_hit = sum(1 for j in range(n_items)263                  if item_cate[model.i2i[j][0][0]] == item_cate[j]) / n_items264    print(f"  SimNet Top-1 命中同类目兄弟物品的比例：{sib_hit:.3f}（共现 + side info 特征生效）")265266    u = 0267    print(f"\n用户 {u}：偏好类目={favs[u]}，历史行为={[j for j, _ in hists[u]]}"268          f"（类目 {[item_cate[j] for j, _ in hists[u]]}）")269    print("TriggerNet 对历史行为的触发强度 t_uj（在线只保留 Top-m 作为 Trigger）：")270    model.eval()271    with torch.no_grad():272        for j, a in hists[u]:273            t_uj = model.trigger(u, profiles[u], j, a).item()274            print(f"  item {j:>2} (类目 {item_cate[j]})  t_uj={t_uj:+.4f}")275276    recs = model.recommend(u, profiles[u], hists[u], top_n=5, top_m=3)277    print("\n在线召回 Top-5（已移除 BiasNet，附最大贡献路径以体现可解释性）：")278    for i, score, j_best, contrib in recs:279        hit = "√" if item_cate[i] in favs[u] else "×"280        print(f"  item {i:>2} (类目 {item_cate[i]}) {hit}  ŷ={score:.4f}"281              f"   主路径 u→{j_best}→{i}  贡献={contrib:.4f}")282    print("  每条推荐都能回溯到「因为你看过物品 j，所以推荐物品 i」——"283          "这正是 PDN 相比双塔模型多出来的可解释性。")