item_cf.py

1"""2ItemCF 召回 —— 基于物品的协同过滤（第10章 §10.1.1）34核心思想：统计物品间共现用户数，用余弦相似度衡量物品相似性；5召回时查询用户交互物品的相似物品列表并累加得分。6"""78import math9from collections import defaultdict101112def build_item_similarity(user_items: dict[str, list[str]], top_k: int = 20) -> dict[str, list[tuple[str, float]]]:13    """14    离线计算物品间相似度。1516    Args:17        user_items: {user_id: [item_id, ...]}，每个用户的交互物品列表18        top_k: 每个物品保留的最相似邻居数1920    Returns:21        {item_i: [(item_j, sim), ...]} 按相似度降序排列22    """23    # 统计物品交互用户数 N(i) 以及共现次数 |N(i) ∩ N(j)|24    item_user_count: dict[str, int] = defaultdict(int)   # N(i)25    co_occur: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))2627    for _, items in user_items.items():28        for i in items:29            item_user_count[i] += 130        # 枚举用户行为物品对31        for i in items:32            for j in items:33                if i != j:34                    co_occur[i][j] += 13536    # 余弦相似度：|N(i) ∩ N(j)| / sqrt(N(i) * N(j))37    item_sim: dict[str, list[tuple[str, float]]] = {}38    for i, neighbors in co_occur.items():39        sims = []40        for j, cnt in neighbors.items():41            sim = cnt / math.sqrt(item_user_count[i] * item_user_count[j])42            sims.append((j, sim))43        sims.sort(key=lambda x: x[1], reverse=True)44        item_sim[i] = sims[:top_k]45    return item_sim464748def recall_itemcf(49    user_id: str,50    user_items: dict[str, list[str]],51    item_sim: dict[str, list[tuple[str, float]]],52    top_n: int = 10,53) -> list[tuple[str, float]]:54    """55    在线召回：根据用户历史行为和物品相似度表打分并排序。5657    得分：p(u, i) = Σ_{j ∈ N(u)} S(i, j)58    """59    history = set(user_items.get(user_id, []))60    score: dict[str, float] = defaultdict(float)61    for j in history:62        for i, sim in item_sim.get(j, []):63            if i not in history:64                score[i] += sim65    ranked = sorted(score.items(), key=lambda x: x[1], reverse=True)66    return ranked[:top_n]676869# ── 快速验证 ──────────────────────────────────────────────────────────────────70if __name__ == "__main__":71    user_items = {72        "u1": ["i1", "i2", "i3"],73        "u2": ["i1", "i2", "i4"],74        "u3": ["i2", "i3", "i5"],75        "u4": ["i1", "i3", "i5"],76        "u5": ["i4", "i5"],77    }7879    item_sim = build_item_similarity(user_items, top_k=5)80    print("物品相似度（Top-3）：")81    for item in ["i1", "i2", "i3"]:82        print(f"  {item}: {item_sim.get(item, [])[:3]}")8384    result = recall_itemcf("u1", user_items, item_sim, top_n=5)85    print("\nu1 的 ItemCF 召回结果：")86    for item, score in result:87        print(f"  {item}: {score:.4f}")