简介这份资源面向推荐算法入门与进阶学习者提供基于Python实现的协同过滤完整参考代码涵盖基于物品与基于用户两条技术路线可作为课程设计、大作业、工程实训或毕设项目的起步素材。压缩包共4个文件以2个py脚本为核心分别对应Item_CF与User_CF两套算法实现另附1个csv数据文件用于读取用户-物品评分矩阵以及1个gitignore配置项整体约6KB体量轻巧便于快速阅读与调试。目前已有449人学习下载说明该方向具备一定关注度。读者可借此理解相似度计算、邻居选取与评分预测等协同过滤关键环节对照代码梳理算法流程并在此基础上自行调整参数、替换数据集或扩展功能。需注意代码仅作参考需具备一定Python基础能自行调试报错并修改逻辑不宜直接照搬。1. 从零手写 UserCF 与 ItemCF两套协同过滤到底该怎么选很多人第一次接触推荐系统都是从「基于用户的协同过滤」和「基于物品的协同过滤」这两个词开始的。它们听起来像一对双胞胎实际在工程里差别极大UserCF 找的是「和你口味相似的人喜欢什么」ItemCF 找的是「你喜欢的东西和哪些东西相似」。前者依赖用户群体的实时行为后者依赖物品之间的稳定关联。如果你手上只有一份用户-物品评分表想用 Python 把这两套算法都跑通并且搞清楚什么场景该用哪个那这篇笔记就是写给你的。我会从数据结构、相似度计算、邻居选取、评分预测一路写到评测和踩坑代码可以直接抄参数可以照着调。适合刚入门推荐系统、想动手复现的开发者也适合已经用过现成库、但想弄明白底层逻辑的熟手。2. 数据准备与相似度计算把评分矩阵变成可计算的输入2.1 用字典还是用 NumPy两种数据结构的取舍协同过滤的输入本质上是一张用户-物品评分表。最直观的存法是嵌套字典{user: {item: rating}}优点是稀疏友好用户没评过的物品根本不占位置缺点是每次算相似度都要遍历字典键向量化困难。另一种是 NumPy 二维数组优点是可以用矩阵运算加速缺点是稀疏场景下内存浪费严重。我一般会分两步走数据加载和邻居搜索用字典相似度矩阵计算用 NumPy。下面是一个最小可用的数据构造和两种结构的转换代码。import numpy as np from collections import defaultdict # 原始评分数据用户, 物品, 评分 raw [ (u1, i1, 5), (u1, i2, 3), (u1, i3, 4), (u2, i1, 4), (u2, i3, 5), (u2, i4, 2), (u3, i2, 5), (u3, i3, 3), (u3, i4, 4), (u4, i1, 3), (u4, i2, 4), (u4, i4, 5), ] # 字典结构user - {item: rating} user_item defaultdict(dict) item_user defaultdict(dict) for u, i, r in raw: user_item[u][i] r item_user[i][u] r # 构建 ID 映射方便转 NumPy users sorted(user_item.keys()) items sorted(item_user.keys()) u_idx {u: k for k, u in enumerate(users)} i_idx {i: k for k, i in enumerate(items)} # 转成评分矩阵未评分填 0 mat np.zeros((len(users), len(items))) for u, i, r in raw: mat[u_idx[u], i_idx[i]] r print(mat)这段代码做了三件事把原始三元组拆成用户视角和物品视角的字典建立 ID 到索引的映射最后生成一个稠密矩阵。参数上唯一需要注意的是未评分填 0 这个选择——如果评分范围是 1 到 50 可以安全地表示「没评过」但如果你的评分本身包含 0就得换成 NaN 或者加一个 mask 矩阵。常见做法是额外维护一个mask (mat 0)的布尔矩阵后续计算只在 mask 为 True 的位置进行。2.2 余弦相似度与皮尔逊相似度公式背后的适用边界相似度是协同过滤的核心。余弦相似度看的是两个向量的夹角对评分绝对值不敏感皮尔逊相似度先减去各自均值再算余弦能消除用户打分偏严或偏松的影响。下面把两种相似度都实现一遍。def cosine_sim(vec_a, vec_b): 余弦相似度输入为两个一维数组 dot np.dot(vec_a, vec_b) norm_a np.linalg.norm(vec_a) norm_b np.linalg.norm(vec_b) if norm_a 0 or norm_b 0: return 0.0 return dot / (norm_a * norm_b) def pearson_sim(vec_a, vec_b): 皮尔逊相似度只考虑共同评分过的维度 mask (vec_a 0) (vec_b 0) if mask.sum() 2: return 0.0 a vec_a[mask] b vec_b[mask] a_mean a.mean() b_mean b.mean() num np.sum((a - a_mean) * (b - b_mean)) den np.sqrt(np.sum((a - a_mean) ** 2)) * np.sqrt(np.sum((b - b_mean) ** 2)) if den 0: return 0.0 return num / den余弦相似度适合物品数量远大于用户数量的场景因为物品向量通常比较稳定皮尔逊适合用户评分尺度差异大的场景比如有人习惯打 3 分有人习惯打 5 分。参数上皮尔逊要求至少有两个共同评分项否则分母为 0代码里直接返回 0 是一种保守处理。实际工程中还会遇到修正余弦相似度它在余弦基础上减去用户均值效果介于两者之间适合评分数据比较稠密的情况。2.3 相似度矩阵的批量计算与内存控制逐个调用相似度函数在数据量大时会非常慢。常见做法是用矩阵乘法一次性算出余弦相似度矩阵。对于 UserCF用户相似度矩阵是mat mat.T对于 ItemCF物品相似度矩阵是mat.T mat。但直接乘会得到未归一化的点积还需要除以模长。def cosine_matrix(mat): 批量计算余弦相似度矩阵 norm np.linalg.norm(mat, axis1, keepdimsTrue) norm[norm 0] 1e-9 # 防止除零 normalized mat / norm sim normalized normalized.T np.fill_diagonal(sim, 0) # 自己和自己不算相似 return sim user_sim cosine_matrix(mat) # 用户相似度 item_sim cosine_matrix(mat.T) # 物品相似度 print(user_sim.shape, item_sim.shape)这里的关键参数是1e-9这个极小值用来避免零向量导致的除零错误。np.fill_diagonal(sim, 0)把对角线置零因为推荐时不应该把用户自己或物品自己算作邻居。内存方面如果用户数超过 10 万用户相似度矩阵就是 100 亿个浮点数直接爆内存。这时候要么用稀疏矩阵要么改用基于局部敏感哈希的近似最近邻。我一般会在用户数超过 5 万时切换到 ItemCF因为物品数量通常远小于用户数量物品相似度矩阵更可控。3. UserCF 实现找相似用户预测未知评分3.1 邻居选取TopK 的 K 到底设多少UserCF 的第一步是为每个用户找到最相似的 K 个邻居。K 太小邻居覆盖不够推荐结果不稳定K 太大不相似的用户混进来推荐质量下降。经验值通常在 20 到 100 之间数据越稀疏K 应该越大。def topk_neighbors(sim_matrix, k): 为每个用户返回 TopK 邻居索引和相似度 neighbors {} for u in range(sim_matrix.shape[0]): scores sim_matrix[u] # argsort 默认升序取最后 k 个再反转 idx np.argsort(scores)[-k:][::-1] neighbors[u] [(i, scores[i]) for i in idx if scores[i] 0] return neighbors user_neighbors topk_neighbors(user_sim, k3) for u, nbrs in user_neighbors.items(): print(users[u], [(users[i], round(s, 3)) for i, s in nbrs])这段代码对每个用户做一次全排序时间复杂度是 O(N log N)。如果用户数很大可以用np.argpartition先做部分排序把复杂度降到 O(N)。参数 K 的选择没有银弹我通常会在验证集上从 10 开始以 10 为步长试到 100看 RMSE 或 Recall 的变化曲线选拐点附近的値。注意代码里过滤了相似度小于等于 0 的邻居因为负相似度意味着口味相反拉进来会起反作用。3.2 评分预测加权平均与均值中心化的差别有了邻居就可以预测目标用户对未评分物品的评分。最朴素的做法是用邻居的评分做加权平均权重就是相似度。但更稳的做法是先减去邻居的评分均值预测后再加回目标用户的均值这就是均值中心化。def predict_usercf(user, item, user_item, user_neighbors, user_sim, users, items): 预测 user 对 item 的评分 if item in user_item[user]: return user_item[user][item] u users.index(user) neighbors user_neighbors.get(u, []) if not neighbors: return 0.0 num, den 0.0, 0.0 user_mean np.mean(list(user_item[user].values())) if user_item[user] else 0 for nbr_idx, sim in neighbors: nbr users[nbr_idx] if item in user_item[nbr]: nbr_mean np.mean(list(user_item[nbr].values())) num sim * (user_item[nbr][item] - nbr_mean) den abs(sim) if den 0: return user_mean return user_mean num / den # 预测 u1 对 i4 的评分 score predict_usercf(u1, i4, user_item, user_neighbors, user_sim, users, items) print(u1 对 i4 的预测评分:, round(score, 3))均值中心化的好处是消除了用户打分尺度的影响。比如一个用户从不打 5 分另一个用户动不动就打 5 分直接加权平均会偏向后者。减去均值后大家都是在「高于自己平均水平多少」这个尺度上比较。参数上den用绝对值求和是为了处理负相似度的情况——如果邻居相似度为负加权时应该反向贡献。实际工程中还会对相似度做幂次调整比如sim ** 1.5让高相似度邻居的权重更大。3.3 生成推荐列表从预测评分到 TopN 排序预测完所有未评分物品后按分数排序取前 N 个就是推荐列表。这里要注意过滤掉用户已经评过分的物品以及处理冷启动用户——如果一个用户没有任何评分UserCF 无法为他找邻居。def recommend_usercf(user, user_item, user_neighbors, user_sim, users, items, topn3): 为用户生成 TopN 推荐 if user not in user_item or not user_item[user]: return [] # 冷启动无法推荐 scored [] for item in items: if item in user_item[user]: continue score predict_usercf(user, item, user_item, user_neighbors, user_sim, users, items) scored.append((item, score)) scored.sort(keylambda x: x[1], reverseTrue) return scored[:topn] recs recommend_usercf(u1, user_item, user_neighbors, user_sim, users, items) print(u1 的推荐:, recs)这段代码对每个未评分物品都调用一次预测函数时间复杂度是 O(物品数 × 邻居数)。如果物品上万可以先用物品相似度矩阵做粗排再精排。冷启动是 UserCF 的硬伤新用户没有行为数据找不到相似邻居只能退化成热门推荐。常见做法是给新用户先推全局热门物品等积累了几个行为后再切换到个性化推荐。4. ItemCF 实现找相似物品复用用户历史4.1 物品相似度的计算细节活跃用户惩罚ItemCF 的核心是物品相似度矩阵。直接算余弦相似度会有一个问题活跃用户评过很多物品的人会人为拉高很多物品之间的相似度。比如一个用户评了 500 部电影那这 500 部电影两两之间都会因为他的存在而相似度偏高。解决办法是给活跃用户降权常见公式是1 / log(1 len(user_item[u]))。def item_similarity_with_penalty(item_user, items, users): 带活跃用户惩罚的物品相似度 n_items len(items) sim np.zeros((n_items, n_items)) item_idx {it: k for k, it in enumerate(items)} for i in range(n_items): for j in range(i 1, n_items): item_a, item_b items[i], items[j] users_a set(item_user[item_a].keys()) users_b set(item_user[item_b].keys()) common users_a users_b if not common: continue num, den_a, den_b 0.0, 0.0, 0.0 for u in common: w 1.0 / np.log(1 len(item_user[item_a]) len(item_user[item_b])) num w for u in users_a: den_a 1.0 / np.log(1 len(item_user[item_a])) for u in users_b: den_b 1.0 / np.log(1 len(item_user[item_b])) if den_a 0 and den_b 0: sim[i][j] sim[j][i] num / np.sqrt(den_a * den_b) return sim item_sim_penalty item_similarity_with_penalty(item_user, items, users) print(np.round(item_sim_penalty, 3))这里的惩罚权重1 / log(1 物品被评次数)是 ItemCF 的经典做法来自 ItemCF 的原始论文。参数上log 的底数通常用自然对数 e也可以用 2影响不大。注意代码里对每个共同评分用户都加了权重而不是简单计数这样活跃用户的贡献被压缩了。实际工程中还会对相似度矩阵做归一化让每行最大值为 1避免热门物品的相似度普遍偏高。4.2 基于物品相似度的评分预测ItemCF 预测用户对某个物品的评分时看的是这个用户评过分的物品中哪些和目标物品相似。用这些相似物品的评分做加权平均权重是物品相似度。def predict_itemcf(user, item, user_item, item_sim, items): 基于物品相似度预测评分 if user not in user_item or not user_item[user]: return 0.0 if item in user_item[user]: return user_item[user][item] item_idx {it: k for k, it in enumerate(items)} j item_idx[item] num, den 0.0, 0.0 for rated_item, rating in user_item[user].items(): i item_idx[rated_item] sim item_sim[i][j] if sim 0: num sim * rating den sim if den 0: return 0.0 return num / den score predict_itemcf(u1, i4, user_item, item_sim_penalty, items) print(ItemCF 预测 u1 对 i4 的评分:, round(score, 3))和 UserCF 的均值中心化不同ItemCF 这里用的是原始评分加权平均。原因是物品的评分尺度通常比用户稳定一部电影的平均分不会因为某个用户而剧烈变化。参数上只取相似度大于 0 的物品参与预测负相似度物品直接忽略。如果用户评过的物品很多可以只取相似度最高的 K 个物品减少计算量。4.3 推荐列表生成与实时性优化ItemCF 的推荐列表生成和 UserCF 类似但有一个重要优势物品相似度矩阵可以离线算好线上只需要查表。用户新产生一个行为立刻就能根据他历史评分的物品找到相似物品实时性很好。def recommend_itemcf(user, user_item, item_sim, items, topn3): 基于 ItemCF 生成 TopN 推荐 if user not in user_item or not user_item[user]: return [] scores defaultdict(float) item_idx {it: k for k, it in enumerate(items)} for rated_item, rating in user_item[user].items(): i item_idx[rated_item] for j, sim in enumerate(item_sim[i]): if sim 0: continue target items[j] if target in user_item[user]: continue scores[target] sim * rating ranked sorted(scores.items(), keylambda x: x[1], reverseTrue) return ranked[:topn] recs recommend_itemcf(u1, user_item, item_sim_penalty, items) print(ItemCF 推荐:, recs)这段代码遍历用户评过的每个物品再遍历该物品的相似物品累加得分。时间复杂度是 O(用户评过物品数 × 物品数)如果物品上万可以用倒排索引只遍历相似度 TopK 的物品。参数上sim * rating是常见的加权方式也可以用sim * (rating - user_mean)做中心化。ItemCF 的线上服务通常会把物品相似度矩阵存成 KV 结构每个物品对应一个相似物品列表查询时直接取。5. 避坑与排查协同过滤落地时最容易翻车的五个地方5.1 现象推荐结果全是热门物品长尾物品从不出现原因相似度计算时没有对热门物品做惩罚热门物品因为被评次数多和几乎所有物品都有较高相似度导致推荐列表被它们霸占。解决在物品相似度里加入活跃用户惩罚或者对最终得分做热度降权比如除以log(1 物品被评次数)。我一般会在排序阶段加一个score / np.log(1 item_popularity[item])的调整。5.2 现象离线评测 RMSE 很低线上点击率却很差原因RMSE 衡量的是评分预测准确度但推荐系统线上更关注排序质量。一个物品预测评分 4.2 还是 4.3 对用户点击影响很小但 TopN 列表里排第 1 还是第 10 影响很大。解决离线评测改用 RecallK、NDCGK 这类排序指标同时做 A/B 测试验证线上效果。参数上K 通常取 10 或 20和线上展示位数量对齐。5.3 现象新用户和新物品完全无法推荐原因UserCF 依赖用户历史行为找邻居ItemCF 依赖物品被评记录算相似度两者对冷启动都无能为力。解决新用户走热门推荐或基于内容的推荐新物品用属性相似度做兜底。常见做法是维护一个「冷启动池」新物品先随机曝光给部分用户积累几十个行为后再进入协同过滤。5.4 现象相似度矩阵计算时内存溢出原因用户数或物品数过大稠密矩阵超出内存。解决用稀疏矩阵存储或者只计算 TopK 相似邻居不保留完整矩阵。Python 里可以用scipy.sparse的csr_matrix计算相似度时用sklearn.metrics.pairwise.cosine_similarity它内部会处理稀疏格式。如果数据量实在太大就上 Spark 或者用 Faiss 做近似最近邻。5.5 现象推荐结果每次刷新都在变用户觉得不稳定原因相似度矩阵没有定期更新或者线上用了实时行为但没做平滑。解决物品相似度矩阵每天离线更新一次线上实时行为只做微调。参数上可以给离线相似度和实时相似度各一个权重比如 0.8 和 0.2避免实时噪声主导结果。另外推荐列表可以加一个缓存层同一个用户短时间内看到的结果保持一致。6. 评测与调参用 RecallK 和覆盖率验证两套算法6.1 离线评测的切分方式与指标计算协同过滤的离线评测通常把用户行为按时间切分前 80% 做训练后 20% 做测试。对每个用户用训练集生成推荐列表看测试集里的物品有多少出现在推荐列表中这就是 RecallK。def recall_at_k(recommend_fn, user_item, test_item, users, items, k3): 计算 RecallK hits, total 0, 0 for user in users: if user not in test_item or not test_item[user]: continue recs recommend_fn(user, user_item, items, topnk) rec_items set(r[0] for r in recs) true_items set(test_item[user]) hits len(rec_items true_items) total len(true_items) return hits / total if total 0 else 0.0 # 构造一个简单测试集 test_item {u1: [i4], u2: [i2], u3: [i1], u4: [i3]} def rec_itemcf_wrapper(user, user_item, items, topn): return recommend_itemcf(user, user_item, item_sim_penalty, items, topn) print(ItemCF Recall3:, round(recall_at_k(rec_itemcf_wrapper, user_item, test_item, users, items, 3), 3))这段代码把推荐函数作为参数传入方便对比 UserCF 和 ItemCF。参数 K 要和线上展示位一致如果线上只展示 3 个就测 Recall3。除了召回率覆盖率也很重要——如果所有用户推荐的都是同样几个物品覆盖率会很低说明算法多样性不足。计算覆盖率就是推荐过的物品去重后除以总物品数。6.2 UserCF 与 ItemCF 的选型对照表两套算法没有绝对优劣关键看场景。下面这张表是我在实际项目中总结的选型依据。维度UserCFItemCF相似度对象用户之间物品之间实时性较差用户行为变化快较好物品相似度稳定冷启动新用户难推荐新物品难推荐可解释性弱「和你相似的人喜欢」强「因为你喜欢 A所以推荐 B」适用场景用户少、物品多、社交属性强用户多、物品少、物品关联稳定计算复杂度用户数平方物品数平方典型参数邻居数 K20~100相似物品数 K10~50从表里可以看出新闻推荐适合 UserCF因为新闻时效性强用户兴趣变化快电商推荐适合 ItemCF因为商品关联相对稳定而且「买了又买」的解释性很好。如果两种都想用可以做成混合推荐加权融合两边的得分权重用线上 A/B 测试调。6.3 相似度归一化与最终推荐融合ItemCF 有一个容易被忽略的细节物品相似度矩阵需要按物品做归一化否则热门物品的相似度普遍偏高推荐结果会偏向热门。归一化方式是对每一行除以该行最大值。def normalize_sim(sim): 按行归一化相似度矩阵 max_vals sim.max(axis1, keepdimsTrue) max_vals[max_vals 0] 1.0 return sim / max_vals item_sim_norm normalize_sim(item_sim_penalty) print(归一化后 i1 的相似度:, np.round(item_sim_norm[0], 3))归一化之后每个物品的相似度都在 0 到 1 之间热门物品和长尾物品的权重更均衡。如果要做 UserCF 和 ItemCF 的融合推荐可以分别算出两边的 TopN然后按排名做加权比如final_score 0.6 * itemcf_score 0.4 * usercf_score。权重怎么定我的习惯是先在离线用网格搜索找一组不错的再上线做小流量 A/B 测试以点击率为准微调。这套流程跑下来两套协同过滤算法基本就能在你的数据上落地了。希望帮到你。本文还有配套的精品资源点击获取