heterogeneous_cf.py

1"""2Heterogeneous CF 召回 —— 异构行为协同过滤 / EHCF（第10章 §10.1.7）34核心思想：用户反馈是异构的（点击 -> 点赞 -> 收藏 -> 完播，漏斗逐级稀疏）。传统 CF 只用单一5    行为建模，既浪费监督信号又高度依赖负采样。Heterogeneous CF 将多种行为统一建模，并通过6    跨行为知识迁移（Knowledge Transfer）把高频行为学到的知识迁移给低频行为，从而在无需负7    采样的情况下学到更准确的用户/物品 Embedding。8参考论文：Chen et al., "Efficient Heterogeneous Collaborative Filtering without Negative9    Sampling for Recommendation", AAAI 2020（EHCF）10一、HeteroItemCF（统计派）：按行为类型构造加权共现矩阵，再做多行为融合；召回支持11    「先融合相似度再召回」与「分行为召回后加权融合」两种模式。12        C^(k)(i,j) = Σ_u (α_k·x^(k)_ui)(α_k·x^(k)_uj) / log(1+|N_k(u)|)13        S^(k)(i,j) = C^(k)(i,j) / sqrt( W_k(i)·W_k(j) )，  S(i,j) = Σ_k λ_k·S^(k)(i,j)14二、EHCFModel（模型派，PyTorch 实现，autograd 自动求导）：Embedding + Transfer-based15    Prediction Layer，无负采样的加权平方回归损失（借鉴 WMF），并按 λ_k 做多任务联合优化。16    正文中 EHCF 把 R̂^2 展开以解耦用户侧/物品侧二阶统计量来降低复杂度；本文件数据规模很小，17    直接遍历全物品集即可。18        z_uv = p_u ⊙ q_v（Hadamard 积），R̂^(k)_uv = h_k^T z_uv（第 k 个行为的线性预测层）19        h_k  = Σ_{t∈N(k)} ( h_t M_tk + r_tk )（跨行为知识迁移）20        L_k = Σ_{u∈B} Σ_{v∈V} c^(k)_uv ( R^(k)_uv - R̂^(k)_uv )^2，  L = Σ_k λ_k L_k21"""2223import math24import random25from collections import defaultdict2627import torch28import torch.nn as nn29import torch.optim as optim3031BEHAVIOR_ALPHA = {"click": 1.0, "like": 3.0, "collect": 5.0, "finish": 8.0}       # 置信度权重 α_k32BEHAVIOR_LAMBDA = {"click": 0.2, "like": 0.25, "collect": 0.25, "finish": 0.3}    # 融合权重 λ_k33BEHAVIOR_CHAIN = ["click", "like", "collect", "finish"]                           # 行为漏斗顺序343536def _dot(a: list[float], b: list[float]) -> float:37    return sum(x * y for x, y in zip(a, b))383940def _cosine(a: list[float], b: list[float]) -> float:41    na, nb = math.sqrt(_dot(a, a)), math.sqrt(_dot(b, b))42    return _dot(a, b) / (na * nb) if na > 0 and nb > 0 else 0.0434445# ══ 第一部分：加权共现矩阵的异构 ItemCF ══════════════════════════════════════46class HeteroItemCF:47    """按行为类型分别构造加权共现矩阵，再做多行为融合的异构协同过滤（纯统计，无待学习参数）。"""4849    def __init__(self, alpha: dict[str, float] | None = None, lam: dict[str, float] | None = None):50        self.alpha = alpha or dict(BEHAVIOR_ALPHA)51        self.lam = lam or dict(BEHAVIOR_LAMBDA)52        self.sim_k: dict[str, dict[int, dict[int, float]]] = {}   # S^(k)53        self.sim_fused: dict[int, dict[int, float]] = defaultdict(dict)54        self.hist: dict[str, dict[int, dict[int, float]]] = {}    # 行为 -> 用户 -> {物品: α_k}5556    def fit(self, logs: list[tuple[int, int, str]]) -> None:57        """从异构行为日志 [(user, item, behavior), ...] 构建加权共现相似度。"""58        for uid, iid, beh in logs:59            if beh in self.alpha:60                self.hist.setdefault(beh, defaultdict(dict))[uid][iid] = self.alpha[beh]6162        for beh, user_map in self.hist.items():63            co: dict[int, dict[int, float]] = defaultdict(lambda: defaultdict(float))64            w_sum: dict[int, float] = defaultdict(float)65            for item_w in user_map.values():66                # IUF 惩罚：行为过多的活跃用户对共现的贡献打折67                iuf = 1.0 / math.log(1.0 + len(item_w)) if len(item_w) > 1 else 1.068                for i, wi in item_w.items():69                    w_sum[i] += wi * wi70                    for j, wj in item_w.items():71                        if i != j:72                            co[i][j] += wi * wj * iuf73            sim: dict[int, dict[int, float]] = defaultdict(dict)74            lam = self.lam.get(beh, 0.0)75            for i, row in co.items():76                for j, c in row.items():77                    denom = math.sqrt(w_sum[i] * w_sum[j])78                    sim[i][j] = c / denom if denom > 0 else 0.079                    # 融合：S(i,j) = Σ_k λ_k S^(k)(i,j)80                    self.sim_fused[i][j] = self.sim_fused[i].get(j, 0.0) + lam * sim[i][j]81            self.sim_k[beh] = sim8283    def similar_items(self, item: int, behavior: str | None = None, top_k: int = 5) -> list[tuple[int, float]]:84        """查询相似物品；behavior=None 表示使用多行为融合相似度 S(i,j)。"""85        table = self.sim_fused if behavior is None else self.sim_k.get(behavior, {})86        return sorted(table.get(item, {}).items(), key=lambda x: (-x[1], x[0]))[:top_k]8788    def recall(self, uid: int, top_n: int = 5, mode: str = "fused") -> list[tuple[int, float]]:89        """异构召回两种模式：mode="fused" 先融合得到 S(i,j) 再用全部行为历史（取最高置信度90        作触发权重）召回；mode="per_behavior" 按行为分别召回再按 λ_k 加权融合（更常用）。"""91        seen = {i for users in self.hist.values() for i in users.get(uid, {})}92        scores: dict[int, float] = defaultdict(float)93        if mode == "fused":94            trigger: dict[int, float] = {}95            for users in self.hist.values():96                for i, w in users.get(uid, {}).items():97                    trigger[i] = max(trigger.get(i, 0.0), w)98            for i, w in trigger.items():99                for j, s in self.similar_items(i, None, 10):100                    if j not in seen:101                        scores[j] += w * s102        else:103            for beh, users in self.hist.items():104                for i in users.get(uid, {}):105                    for j, s in self.similar_items(i, beh, 10):106                        if j not in seen:107                            scores[j] += self.lam.get(beh, 0.0) * s108        return sorted(scores.items(), key=lambda x: (-x[1], x[0]))[:top_n]109110111# ══ 第二部分：EHCF（Embedding + Transfer-based Prediction Layer）═════════════112class EHCFModel(nn.Module):113    """多行为联合建模 + 跨行为知识迁移 + 无负采样的加权平方回归优化（PyTorch autograd）。"""114115    def __init__(self, n_users: int, n_items: int, dim: int = 16,116                 reg: float = 1e-4, c_pos: float = 1.0, c_neg: float = 0.02):117        super().__init__()118        self.dim = dim119        self.reg = reg120        self.c_pos, self.c_neg = c_pos, c_neg          # 置信权重 c^+ / c^-121        self.behaviors, self.lam = list(BEHAVIOR_CHAIN), dict(BEHAVIOR_LAMBDA)122123        self.p = nn.Embedding(n_users, dim)  # 用户 Embedding124        self.q = nn.Embedding(n_items, dim)  # 物品 Embedding125        nn.init.xavier_normal_(self.p.weight)126        nn.init.xavier_normal_(self.q.weight)127128        # N(k) = {k 的上游行为, k 本身}：上游行为更稠密，知识沿漏斗向下迁移129        self.parents = {k: ([self.behaviors[i - 1]] if i > 0 else []) + [k]130                        for i, k in enumerate(self.behaviors)}131        self.h_base = nn.ParameterDict({132            k: nn.Parameter(torch.randn(dim) * 0.1) for k in self.behaviors133        })134        # 迁移矩阵 M_tk（近似单位矩阵初始化，加小噪声）与迁移偏置 r_tk135        self.m = nn.ParameterDict()136        self.r = nn.ParameterDict()137        for k in self.behaviors:138            for t in self.parents[k]:139                init = torch.eye(dim) + torch.randn(dim, dim) * 0.01140                self.m[f"{t}|{k}"] = nn.Parameter(init)141                self.r[f"{t}|{k}"] = nn.Parameter(torch.zeros(dim))142143    def effective_h(self, k: str) -> torch.Tensor:144        """h_k = (1/|N(k)|) Σ_{t∈N(k)} ( M_tk^T h_t + r_tk )，实现跨行为知识迁移。"""145        parents = self.parents[k]146        out = torch.zeros(self.dim)147        for t in parents:148            ht = self.h_base[t]149            mat = self.m[f"{t}|{k}"]150            rr = self.r[f"{t}|{k}"]151            out = out + ht @ mat + rr152        return out / len(parents)153154    def predict(self, uid, iid, behavior: str, h: torch.Tensor | None = None):155        """R̂^(k)_uv = h_k^T (p_u ⊙ q_v)。"""156        h = h if h is not None else self.effective_h(behavior)157        pu = self.p(uid)   # (B, dim)158        qv = self.q(iid)   # (B, dim)159        return (h * pu * qv).sum(dim=-1)160161    def train_model(self, label: dict[str, dict[int, set[int]]], n_items: int, epochs: int = 6,162                    lr: float = 0.02) -> None:163        """无负采样训练：对每个用户遍历全部物品做加权平方回归（正样本 c^+，未观测 c^-）。164        使用 PyTorch autograd，梯度由反向传播自动完成。"""165        optimizer = optim.Adam(self.parameters(), lr=lr)166167        all_items = torch.arange(n_items, dtype=torch.long)  # 全部物品 ID168        y_all = torch.zeros(n_items)169        for epoch in range(epochs):170            total, cnt = 0.0, 0171            order = list(range(self.p.weight.shape[0]))172            random.shuffle(order)173174            optimizer.zero_grad()175            loss_sum = None176            for uid in order:177                u_t = torch.tensor([uid], dtype=torch.long)178                for k in self.behaviors:179                    pos = label.get(k, {}).get(uid, set())180                    h = self.effective_h(k)181                    lam = self.lam.get(k, 1.0)182183                    # R：正样本为 1，未观测为 0；c：正样本 c^+，未观测 c^-184                    y = y_all.clone()185                    if pos:186                        y[list(pos)] = 1.0187                    c = torch.where(y > 0, torch.full_like(y, self.c_pos),188                                    torch.full_like(y, self.c_neg))189190                    pred = self.predict(u_t, all_items, k, h)  # (n_items,)191                    err = pred - y192                    loss_k = (lam * c * err * err).sum()       # L_k = Σ_v λ_k c_v (R̂ - R)^2193                    loss_sum = loss_k if loss_sum is None else loss_sum + loss_k194                    total += loss_k.item()195                    cnt += n_items196197            loss_sum.backward()   # 迁移层 / Embedding 梯度均由 autograd 自动求导198            optimizer.step()199            print(f"  epoch {epoch}  加权平方损失/样本 = {total / cnt:.6f}")200201    def recommend(self, uid: int, n_items: int, exclude: set[int], top_k: int = 5,202                  behavior: str | None = None) -> list[tuple[int, float]]:203        """behavior=None 时按 λ_k 加权融合各行为的预测得分。"""204        hs = {k: self.effective_h(k) for k in self.behaviors}205        all_items = torch.arange(n_items, dtype=torch.long)206        u_t = torch.tensor([uid] * n_items, dtype=torch.long)207        with torch.no_grad():208            if behavior is None:209                s = sum(self.lam[k] * self.predict(u_t, all_items, k, hs[k]) for k in self.behaviors)210            else:211                s = self.predict(u_t, all_items, behavior, hs[behavior])212            scores = s.tolist()213        out = [(iid, scores[iid]) for iid in range(n_items) if iid not in exclude]214        return sorted(out, key=lambda x: -x[1])[:top_k]215216217def _fmt(pairs: list[tuple[int, float]]) -> str:218    return "  ".join(f"item{j:02d}({s:+.4f})" for j, s in pairs) or "(无结果)"219220221# ── 快速验证 ──────────────────────────────────────────────────────────────────222if __name__ == "__main__":223    random.seed(42)224    torch.manual_seed(42)225    n_users, n_items, dim = 8, 20, 16226    # 合成异构行为日志：模拟「点击 -> 点赞 -> 收藏 -> 完播」的漏斗衰减227    funnel = {"like": 0.55, "collect": 0.35, "finish": 0.2}228    logs: list[tuple[int, int, str]] = []229    label: dict[str, dict[int, set[int]]] = {k: defaultdict(set) for k in BEHAVIOR_CHAIN}230    for uid in range(n_users):231        for iid in random.sample(range(n_items), 8):          # 每个用户的偏好物品子集232            if random.random() >= 0.85:233                continue234            logs.append((uid, iid, "click"))235            label["click"][uid].add(iid)236            for beh, rate in funnel.items():237                if random.random() < rate:238                    logs.append((uid, iid, beh))239                    label[beh][uid].add(iid)240    stat = {k: sum(1 for _, _, b in logs if b == k) for k in BEHAVIOR_CHAIN}241    print("异构行为日志统计（漏斗越深越稀疏）：\n  " + "  ".join(242        f"{k}={stat[k]}条(α={BEHAVIOR_ALPHA[k]:.0f},λ={BEHAVIOR_LAMBDA[k]:.2f})" for k in BEHAVIOR_CHAIN))243244    print("\n【第一部分】加权共现矩阵 + 多行为融合相似度")245    hicf = HeteroItemCF()246    hicf.fit(logs)247    for beh in BEHAVIOR_CHAIN:248        print(f"  item00 在 {beh:8s} 行为下的 Top-3：{_fmt(hicf.similar_items(0, beh, 3))}")249    print(f"  item00 多行为融合后的 Top-3：{_fmt(hicf.similar_items(0, None, 3))}")250    print("\n  两种召回模式对比（用户 0）：")251    print(f"    先融合相似度再召回：{_fmt(hicf.recall(0, 5, 'fused'))}")252    print(f"    分行为召回后融合  ：{_fmt(hicf.recall(0, 5, 'per_behavior'))}")253254    print("\n【第二部分】EHCF 多行为联合训练（无负采样 + Transfer-based Prediction Layer）")255    model = EHCFModel(n_users, n_items, dim=dim, reg=1e-4, c_pos=1.0, c_neg=0.02)256    model.train_model(label, n_items, epochs=6, lr=0.02)257    with torch.no_grad():258        hs = {k: model.effective_h(k) for k in BEHAVIOR_CHAIN}259    hs_np = {k: v.tolist() for k, v in hs.items()}260    print("\n  相邻行为预测头 h_k 的余弦相似度（迁移矩阵使其参数彼此靠近）：\n    " + "  ".join(261        f"cos(h_{a},h_{b})={_cosine(hs_np[a], hs_np[b]):+.4f}" for a, b in zip(BEHAVIOR_CHAIN, BEHAVIOR_CHAIN[1:])))262263    seen = {iid for u, iid, _ in logs if u == 0}264    print("\n  用户 0 分行为 Top-3 召回：")265    for beh in BEHAVIOR_CHAIN:266        print(f"    {beh:8s}: {_fmt(model.recommend(0, n_items, seen, 3, beh))}")267    print(f"\n  用户 0 多行为加权融合 Top-5 召回：{_fmt(model.recommend(0, n_items, seen, 5, None))}")