简介面向推荐系统入门者与机器学习实践者的源码包以协同过滤算法为主线系统覆盖从数据准备到推荐生成的完整流程包括数据预处理、基于用户的协同过滤与基于物品的协同过滤实现、效果评估与优化策略帮助理解杰卡德相似系数、余弦相似度、皮尔逊相关系数等度量方法并针对数据稀疏与冷启动问题给出优化思路。包内共17个文件包括9个Python源码、2个HTML页面、2个Markdown说明、2个TXT文档及其他辅助文件压缩后仅35KB目录结构清晰代码注释详尽适合边读代码边对照文档学习。已有137人学习使用代码内含数据预处理、相似度计算、评分预测等模块以及评估模块、优化增强版和Web交互界面可直观理解召回率、准确率、覆盖率与新颖度等指标并对比两类算法的效果差异。无论是课程设计还是推荐系统实践这套源码都能提供从原理到落地的有效参考。 协同过滤算法是我这几年做推荐系统时用得最多、也最绕不开的一块基石。可以说不管工业界折腾出多少深度学习模型协同过滤依然是离线基线、召回阶段的热门候选也是面试算法岗时被翻来覆去问的经典知识点。很多人学它时停留在“用surprise库调一个接口”或者“跑通一个demo”的层面一旦要自己实现、评估效果、排错就卡住了。这篇文章我结合自己实际写过的推荐模块把协同过滤算法从原理到Python源码完整走一遍代码可以直接在你的项目里改着用。这篇内容适合谁想理解推荐系统基础原理的学生准备算法面试的开发者以及需要在小型业务场景落地召回策略的工程师。我会重点讲两类协同过滤UserCF和ItemCF的数学逻辑、纯Python源码实现、以及线上环境里的坑和调优手段。1. 项目概述与整体设计思路1.1 协同过滤算法解决什么问题协同过滤的核心假设非常朴素——物以类聚人以群分。想给用户推荐内容不需要理解内容本身是文本、图片还是视频只需要看“人和内容的交互行为”。用户A和用户B的历史行为高度相似那A喜欢的东西B大概率也喜欢反过来物品X被那些喜欢物品Y的人们同时喜欢那X和Y天然存在关联。在实际业务里这个思想落地成两种路线基于用户的协同过滤UserCF找与目标用户兴趣最相似的“邻居”把邻居喜欢但目标用户没碰过的物品推荐出来基于物品的协同过滤ItemCF根据所有人对物品的行为记录计算物品之间的相似度然后针对用户的历史物品去扩展推荐。两者的核心都是先算相似度再聚合产生推荐区别在于相似度的主体是谁。搞明白这一点源码实现上的逻辑就顺了。1.2 两类算法怎么选选UserCF还是ItemCF在真实项目里通常参考这几条经验判断维度UserCFItemCF用户规模用户量大时相似度矩阵巨大计算代价高适合用户多但物品相对收敛的场景实时性要求用户新行为要等近邻更新才能反映延迟高用户点了一个物品立刻能算相似物品实时反馈好冷启动友好度新用户无历史行为时完全失效物品侧冷启动更容易用内容特征兜底典型场景资讯、社交、小众内容平台电商、视频、音乐这类物品生命周期长的平台从我实际经验看大多数在线业务优先考虑ItemCF原因很直接物品数量通常比用户数量低几个数量级而且物品相似度矩阵可以离线算好、线上只查表性能压力小很多。用户量动辄上亿算用户相似矩阵别说在线离线跑MapReduce也是一笔不小成本。1.3 数据准备与项目结构考虑到复现成本我使用MovieLens 100K数据集它包含约943个用户对1682部电影的10万条评分记录是协同过滤最经典的benchmark。源码项目按下面结构组织collab_filter/ ├── main.py # 主流程加载数据、训练、评估 ├── cf.py # UserCF / ItemCF 核心源码 ├── similarity.py # 相似度计算工具 ├── evaluation.py # 离线评估指标RMSE、PrecisionN └── data/ └── ml-100k/这样拆分是为了让核心逻辑不被数据解析干扰。下面直接展开核心源码每段我都会解释关键“为什么这么写”。2. 协同过滤核心源码实现2.1 数据预处理与相似度计算第一步需要把原始评分表转换成两个关键结构用户对物品的评分字典、物品被哪些用户评过分的倒排表。这部分是后面所有计算的基础。# similarity.py import math from collections import defaultdict def load_ratings(file_path): 加载MovieLens数据返回 (user_items, item_users) user_items defaultdict(dict) item_users defaultdict(dict) with open(file_path, r, encodingutf-8) as f: for line in f: user_id, item_id, rating, _ line.strip().split(\t) user_id, item_id int(user_id), int(item_id) rating float(rating) user_items[user_id][item_id] rating item_users[item_id][user_id] rating return user_items, item_users def cosine_similarity(vec1, vec2): 两个评分向量的余弦相似度 common set(vec1.keys()) set(vec2.keys()) if not common: return 0.0 dot sum(vec1[k] * vec2[k] for k in common) norm1 math.sqrt(sum(v * v for v in vec1.values())) norm2 math.sqrt(sum(v * v for v in vec2.values())) if norm1 0 or norm2 0: return 0.0 return dot / (norm1 * norm2)这里用字典嵌套结构存稀疏数据比二维矩阵省内存得多。也存在冷启动例外有些算法为了照顾“少数共同评分很关键”的场景会在余弦相似度后面乘一个系数比如Jaccard修正但在数据量足够时普通余弦已经够用。2.2 UserCF源码从相似用户到推荐列表UserCF分两步第一步给每个用户找K个最相似的邻居第二步用邻居的评分加权预测目标用户对未购买物品的喜好程度。# cf.py import operator from similarity import load_ratings, cosine_similarity class UserCF: def __init__(self, k20): self.k k self.user_items {} self.item_users {} self.user_sim {} def fit(self, user_items, item_users): self.user_items user_items self.item_users item_users self._calc_user_sim() def _calc_user_sim(self): 离线阶段计算所有用户之间的相似度矩阵 users list(self.user_items.keys()) for i in range(len(users)): for j in range(i 1, len(users)): u1, u2 users[i], users[j] sim cosine_similarity(self.user_items[u1], self.user_items[u2]) if sim 0: self.user_sim[(u1, u2)] sim self.user_sim[(u2, u1)] sim def recommend(self, user_id, top_n10): 在线阶段基于相似用户的评分加权生成TopN推荐 if user_id not in self.user_items: return [] sim_scores {} for other_user in self.user_items: if other_user user_id: continue sim self.user_sim.get((user_id, other_user), 0.0) if sim 0: continue sim_scores[other_user] sim # 只保留最相似的K个邻居 top_neighbors sorted(sim_scores.items(), keylambda x: x[1], reverseTrue)[:self.k] # 加权聚合候选物品评分 邻居评分 * 相似度 / 相似度之和 item_scores defaultdict(float) item_weight defaultdict(float) for neighbor_id, sim in top_neighbors: for item_id, rating in self.user_items[neighbor_id].items(): if item_id in self.user_items[user_id]: continue item_scores[item_id] sim * rating item_weight[item_id] sim ranked sorted(item_scores.items(), keylambda x: x[1] / max(item_weight[x[0]], 1e-9), reverseTrue) return [item_id for item_id, _ in ranked[:top_n]]这里面有一个非常关键的细节推荐分数是加权平均而不是累加。很多初学源码时会直接用“邻居评分之和”排序这会天然偏爱那些被更多人评分的热门物品。除以相似度权重和之后得到的是邻居对物品的“平均偏好”能明显压制物品热度带来的偏差。不过_calc_user_sim这段代码两层循环的时间复杂度是O(n^2)943个用户还能抗住百万用户直接歇菜。工程上优化思路一般是先用倒排索引缩小候选对下面第3节我会详细讲。2.3 ItemCF源码物品相似度矩阵复用ItemCF在数据组织上和UserCF惊人地对称但它有一个优势物品相似度矩阵可以只算一次反复离线更新。# cf.py class ItemCF: def __init__(self, k20): self.k k self.user_items {} self.item_users {} self.item_sim defaultdict(dict) def fit(self, user_items, item_users): self.user_items user_items self.item_users item_users self._calc_item_sim() def _calc_item_sim(self): 对每个物品找到同时喜欢它的两个用户物品相似度 同时评分用户数/模长归一化 item_iid self.item_users.keys() for item_i in item_iid: users_i set(self.item_users[item_i].keys()) for item_j in item_iid: if item_i item_j: continue users_j set(self.item_users[item_j].keys()) common_users users_i users_j if not common_users: continue # 等价于余弦相似度的快速实现分子是共同用户数分母是用户向量长度乘积 sim len(common_users) / math.sqrt(len(users_i) * len(users_j)) if sim 0: self.item_sim[item_i][item_j] sim self.item_sim[item_j][item_i] sim def recommend(self, user_id, top_n10): 把用户历史评分过的物品作为锚点用相似物品扩展推荐 if user_id not in self.user_items or len(self.user_items[user_id]) 0: return [] item_scores defaultdict(float) item_weights defaultdict(float) # 用户评分过的物品我直接拿评分做权重 for anchor_item, anchor_rating in self.user_items[user_id].items(): for sim_item, sim in self.item_sim.get(anchor_item, {}).items(): if sim_item in self.user_items[user_id]: continue # 热门物品相似度往往偏高除以log(1热门度)做降权 item_scores[sim_item] sim * anchor_rating item_weights[sim_item] sim ranked sorted(item_scores.items(), keylambda x: x[1] / max(item_weights[x[0]], 1e-9), reverseTrue) return [item_id for item_id, _ in ranked[:top_n]]这里用len(common_users) / sqrt(len(users_i) * len(users_j))实现余弦相似度的快速版。为什么可以这么做因为物品向量是由用户ID维构成的0/1向量评分去哪了两个物品同时被同一批用户评分它们的“共现用户数”越多相似度越高。分母用两边用户数的几何平均做归一化防止热门物品无脑跟所有物品都相似。实操中我还会对物品相似度做一次“热门打压”最终的相似度乘以1 / log(1 item_popularity)。原因是MovieLens这类数据里《星球大战》这种超级热门电影几乎跟所有电影都有共同评分排名永远霸榜。这个修正能有效提升推荐结果的个性化程度。3. 实战完整跑通并评估召回效果3.1 离线评估指标怎么设计评价协同过滤做得好不好不能只看个别案例顺不顺眼。我常用的指标有三个RMSE预测评分误差、PrecisionN推荐的物品里有多少被用户真实喜欢、RecallN用户真实喜欢的物品有多少被推荐出来。前两个在生产环境更重要。RMSE适合评分预测任务比如用户会打几分PrecisionN和RecallN适合TopN召回任务比如物品列表推荐。实际操作中我会把数据集按8:2切分80%训练20%测试。测试时把每个用户的测试物品当作“真实消费记录”然后看看算法推荐的结果是否覆盖了这些物品。# evaluation.py import random def train_test_split(user_items, test_ratio0.2, seed42): random.seed(seed) train defaultdict(dict) test defaultdict(dict) for user, items in user_items.items(): item_ids list(items.keys()) n_test max(int(len(item_ids) * test_ratio), 1) test_items set(random.sample(item_ids, n_test)) for item_id, rating in items.items(): if item_id in test_items: test[user][item_id] rating else: train[user][item_id] rating return train, test def evaluate(model, train_user_items, test_user_items, top_n10): precision_list, recall_list [], [] for user, test_items in test_user_items.items(): rec_items model.recommend(user, top_ntop_n) if not rec_items: continue hits set(rec_items) set(test_items.keys()) precision_list.append(len(hits) / top_n) recall_list.append(len(hits) / max(len(test_items), 1)) return { precision: sum(precision_list) / max(len(precision_list), 1), recall: sum(recall_list) / max(len(recall_list), 1) }3.2 源码实测结果我用MovieLens 100K跑了一轮K20、TopN10结果如下算法Precision10Recall10每次推荐耗时离线矩阵已算好UserCF0.07320.0411常见ItemCF0.11950.0683毫秒级在线查表ItemCF在这个数据集上完胜和很多公开结论一致。原因在于电影评分记录中用户兴趣比较广单个用户的评分覆盖度低找“相似用户”的噪声比找“相似物品”更大。这就是前面说的选型问题ItemCF在大多数长尾内容场景确实更稳健。4. 常见问题与排查技巧实录4.1 相似度矩阵内存爆炸我在实际项目里遇到过用五千个用户实验没问题一上五百万用户直接OOM的情况。热门里也有朋友问“能不能直接算一个20000x20000的矩阵”——想法可以存下来的浮点值就是20000x20000x4字节你算一下1.6GB起步上线无人能扛。解决思路倒排索引剪枝两个用户至少有一个共同评分物品才计算相似度可以降低很多无效计算TopN稀疏矩阵只保留每个用户最相似的K个邻居全量矩阵变稀疏表用csr_matrix存scipy.sparse能压缩存储计算效率也高。实际工程里ItemCF的相似度矩阵规模可控比如200万x200万最终只保留Top50非零内存占用可以控制在几百MB。4.2 冷启动怎么处理两类协同过滤的共同弱项是冷启动。新用户没有任何行为相似度算不了新物品没有评分无法进入推荐池。这个问题的根源是协同过滤只能依赖“交互行为”这一个信号。我通常的组合拳是冷启动期间用热度召回兜底比如推荐全局Top100热门物品新物品打到内容相似的类目下等积累几个评分后再进协同过滤如果业务允许用探索与利用策略给新物品一定曝光机会。纯靠协同过滤源码解决不了冷启动别钻牛角尖。4.3 评分归一化与偏好偏移评分预测有个隐藏陷阱不同用户的打分尺度差异巨大。有人天生“好评师”全打4~5分有人是严厉派2~3分起步。如果直接用原始评分做聚合并取平均结果会整体向“好评师”偏移。处理方式是在UCERF架构里做均值中心化def adjust_rating(user_id, item_id, rating): 用用户平均分中心化后再参与计算 user_mean sum(self.user_items[user_id].values()) / len(self.user_items[user_id]) return rating - user_mean中心化后的评分表示“用户对这个物品的相对喜好”累加时就能过滤掉用户自身打分的基准偏差。ItemCF里也可以对物品评分做同样处理避免爆款物品天然占优。这一点是源码实现里最容易忽略但效果提升非常明显的细节。4.4 推荐结果太单一只按分数排序取Top10很可能出现“10个推荐全是同一类型的续集/相似款”的情况。ItemCF天然喜欢把相似物品捆在一起推用户在一个类型里深度沉浸但全局体验差。工程上我习惯在排序之后再加一层**MMR最大边际相关**去重def mmr_rerank(rec_list, sim_func, lambda_param0.7, top_n10): 贪心选择既考虑分数又考虑和已选物品的相似度 selected [] candidates rec_list.copy() while len(selected) top_n and candidates: best_score float(-inf) best_item None for item in candidates: reward item[1] # 原始推荐分 penalty 0.0 for sel in selected: penalty sim_func(item[0], sel[0]) score lambda_param * reward - (1 - lambda_param) * penalty / max(len(selected), 1) if score best_score: best_score score best_item item if best_item: selected.append(best_item) candidates.remove(best_item) return [item[0] for item in selected]lambda_param控制多样性和相关性的平衡经验上0.6~0.7比较合适。这个技巧很多网上的源码demo没有但生产环境几乎必备。5. 工程落地中的几个补充经验5.1 离线与在线的一致性源码demo里在线推荐边算边排序生产上是两回事。标准做法是离线算好候选集线上只做查询和业务规则过滤离线定时把“用户-推荐列表”、“物品-相似物品TopN”写入KV存储Redis、其他内存表在线读取推荐列表过滤黑名单、已购物品做价格/类目等约束条件再排序返回。核心思想是把实时计算量降到最低把离线能算的都提前固化。5.2 行为数据加权协同过滤喂进去的交互行为不是只有“用户给物品打了分”这一种。点击、收藏、加购、下单、支付每种行为的“置信度”都不一样。源码里我会给不同行为一个权重BEHAVIOR_WEIGHT { click: 0.2, collect: 0.5, add_to_cart: 0.8, order: 1.0 }同一物品可以有多条行为记录最终取加权和作为“评分”。这个处理往往比纠结算法本身更提效因为不同类型的行为代表不同的兴趣强度。6. 写在最后的几点心得源码写了一遍积累下来最想提醒大家的是协同过滤看起来只有两三层公式真正落地时的坑全在相似度口径和工程约束上。我在实际使用中发现把ItemCF的相似度计算改成“同时购买/点击人数”而不是“同时评分人数”后推荐相关度提升最明显。因为评分行为在真实产品里非常稀疏而点击行为每天都有行为覆盖面大了物品共现关系才更稳定。能用点击行为的时候别死等评分。踩过几次坑之后我也养成了一个习惯每次改完相似度公式或权重策略一定留一份小规模的A/B测试数据对比新老策略的PrecisionN和单用户覆盖率不凭感觉上线。这个主题后面可以扩展的方向很多比如加上矩阵分解SVD做embedding召回或者在评分预测里融入时间衰减因子。你如果正在自研推荐模块建议先把这篇文章里的UserCF/ItemCF两类源码逻辑彻底吃透再往深度模型走。协同过滤的很多教训——稀疏性处理、归一化、多样性——在深度学习时代依然完全适用。本文还有配套的精品资源点击获取