简介这份Python推荐系统源码面向希望入门或进阶个性化推荐算法的开发者与学习者围绕数据处理、模型训练与结果评估的完整链路展开同时结合Python 3.x与Spark两种技术栈适合具备一定Python基础、想通过真实项目理解协同过滤与矩阵分解的读者。压缩包共70个文件约17.64MB以21个py脚本、6个scala程序、10个md文档为主另含csv、parquet等数据文件及_success、part-00000等Spark输出标记覆盖代码、数据与说明三类内容。资源内含数据测试用数据集、Spark MLlib的ALS实践、manual使用文档、论文阅读分享与推荐系统基础知识整理可帮助读者从数据清洗、特征工程一路走到模型构建与评估理解基于协同过滤、基于内容及混合推荐的实现思路。目前已有301人学习适合对照代码复现流程、积累推荐系统实战经验。1. 从一份 Python 推荐系统源码说起它到底能跑出什么结果很多做数据方向的朋友第一次接触推荐系统都是被协同过滤矩阵分解这些词吓住的觉得非得有海量用户行为数据、非得上一套分布式集群才能玩得转。其实不是。我手上这份 Python 推荐系统源码核心就是一套能本地跑通、也能往 Spark 上迁的完整链路从原始评分数据清洗到相似度计算再到 Top-N 推荐结果输出最后带一个能看效果的评估脚本。它解决的不是工业级千万 DAU的问题而是让你在单机上把推荐系统的每个环节都摸一遍知道哪一步在算什么、参数改了会怎样、结果为什么长这样。适合谁一是刚学完 Python 基础、想找个真实项目练手的人二是做后端或数据分析突然被要求加个推荐功能、需要快速搭原型的工程师三是课程设计或毕设选了推荐系统方向需要一份能读懂、能改、能讲清楚原理的源码。这份代码不依赖 GPU不依赖外部服务装好 Python 和几个常见库就能跑对新手友好对熟手也能当个干净的 baseline 来改。2. 环境准备与数据流拆解先让代码跑起来再谈调参2.1 依赖安装与 Python 环境配置拿到源码包后别急着python main.py先看requirements.txt。这类推荐系统源码通常依赖numpy、pandas、scikit-learn如果带 Spark 版本还会有pyspark。我一般会单独建虚拟环境避免和系统里的包打架。如果你之前搜过vscode python环境配置或python安装教程这一步其实是一样的逻辑解释器选对包装对后面少一半玄学问题。# 创建虚拟环境Python 3.8 都行推荐 3.9/3.10 python -m venv rec_env # 激活环境Windows 用 rec_env\Scripts\activate source rec_env/bin/activate # 安装核心依赖版本不用死磕但 numpy 别低于 1.21 pip install numpy pandas scikit-learn # 如果源码带 Spark 版本再加这一条 pip install pyspark这里有个细节pyspark在 Windows 上跑本地模式需要 Hadoop 的 winutils很多人卡在这一步以为代码有问题。其实如果只是跑单机小数据集完全可以用纯 Python 版本Spark 版本留着后面数据量上来了再切。虚拟环境的好处是你调崩了直接删掉重建不用重装整个 Python。2.2 数据格式与加载逻辑推荐系统源码里最常见的数据格式是user_id, item_id, rating, timestamp四列分隔符可能是逗号也可能是制表符。源码里一般会有一个data_loader.py或直接在main.py里用pandas.read_csv读。你要做的是先确认三件事列名对不对、有没有表头、评分范围是 1-5 还是 0-1。import pandas as pd # 常见加载方式注意 sep 和 header 参数 def load_ratings(path): # 如果文件没有表头headerNone然后手动指定列名 df pd.read_csv(path, sep,, header0, names[user_id, item_id, rating, timestamp]) # 去掉重复评分保留最新一条 df df.sort_values(timestamp).drop_duplicates( subset[user_id, item_id], keeplast) # 过滤掉评分过少的用户和物品这是推荐系统里常见的冷启动处理 user_counts df[user_id].value_counts() item_counts df[item_id].value_counts() df df[df[user_id].isin(user_counts[user_counts 5].index)] df df[df[item_id].isin(item_counts[item_counts 5].index)] return df.reset_index(dropTrue)这段代码的逻辑说明先去重保证同一个用户对同一个物品只有一条评分然后过滤掉交互少于 5 次的用户和物品。为什么要过滤因为推荐系统里长尾数据噪声大一个只评过 1 次分的用户你没法给他找相似用户强行算出来的结果也是垃圾。参数5不是固定的数据量大可以调到 10数据量小调到 3这个后面调参章节会细说。2.3 训练集/测试集划分与评估指标源码里一般会按时间戳切分而不是随机切分。随机切分在推荐系统里是个坑你会用未来的行为预测过去的行为离线指标好看上线就翻车。常见做法是每个用户最后一条评分做测试集其余做训练集。def train_test_split_by_time(df, test_size1): # 按用户分组取每个用户时间戳最大的 test_size 条做测试 df df.sort_values([user_id, timestamp]) test df.groupby(user_id).tail(test_size) train df.drop(test.index) return train, test评估指标源码里通常带precisionk、recallk、RMSE中的一两种。RMSE衡量评分预测准不准precisionk衡量 Top-N 推荐里有多少是用户真正喜欢的。看源码时注意它用的是哪种别把两个混在一起比较。参数k一般取 10 或 20取太小看不出差异取太大用户根本翻不到那么后面。3. 协同过滤核心实现相似度计算与推荐生成3.1 基于用户的协同过滤UserCF代码拆解UserCF 的逻辑很直白找到和你口味相似的人把他们喜欢但你没看过的物品推给你。源码里一般分三步构建用户-物品评分矩阵、计算用户相似度、生成推荐。这里用余弦相似度举例因为它在稀疏矩阵上表现稳定。import numpy as np from sklearn.metrics.pairwise import cosine_similarity def build_user_item_matrix(df): # 透视表行是用户列是物品值是评分缺失填 0 matrix df.pivot_table(indexuser_id, columnsitem_id, valuesrating).fillna(0) return matrix def user_similarity(matrix): # 余弦相似度输出是用户数 x 用户数的方阵 sim cosine_similarity(matrix) # 对角线是自己和自己置 0 避免推荐时把自己算进去 np.fill_diagonal(sim, 0) return sim def recommend_for_user(user_id, matrix, sim, top_k10, sim_users20): user_idx matrix.index.get_loc(user_id) # 取相似度最高的 sim_users 个用户 sim_scores list(enumerate(sim[user_idx])) sim_scores sorted(sim_scores, keylambda x: x[1], reverseTrue)[:sim_users] # 加权评分相似度 * 对方评分累加后排序 item_scores {} for idx, score in sim_scores: if score 0: continue rated_items matrix.iloc[idx].nonzero()[0] for item in rated_items: if matrix.iloc[user_idx, item] 0: # 只推没看过的 item_scores[item] item_scores.get(item, 0) score * matrix.iloc[idx, item] # 按分数排序取 top_k rec_items sorted(item_scores.items(), keylambda x: x[1], reverseTrue)[:top_k] return [matrix.columns[i] for i, _ in rec_items]逻辑说明build_user_item_matrix把长表转成宽表这是协同过滤的标准输入。user_similarity算余弦相似度注意fill_diagonal那一步不置 0 的话每个用户和自己相似度是 1推荐结果会变成推自己已经评过的这是新手改代码时最容易漏的。recommend_for_user里sim_users20表示只参考最相似的 20 个用户这个参数太大引入噪声太小推荐多样性差一般 20-50 之间试。3.2 基于物品的协同过滤ItemCF与适用场景ItemCF 的逻辑是你喜欢 A 物品那就推和 A 相似的物品给你。它比 UserCF 更稳定因为物品之间的相似度不会像用户兴趣那样频繁变。电商场景里 ItemCF 是主力源码里一般也会带一份。def item_similarity(matrix): # 转置后算物品之间的余弦相似度 sim cosine_similarity(matrix.T) np.fill_diagonal(sim, 0) return sim def recommend_by_item(user_id, matrix, item_sim, top_k10): user_idx matrix.index.get_loc(user_id) rated matrix.iloc[user_idx].nonzero()[0] scores {} for item in rated: # 取和当前物品最相似的 top 20 sim_items np.argsort(item_sim[item])[::-1][:20] for sim_item in sim_items: if matrix.iloc[user_idx, sim_item] 0: scores[sim_item] scores.get(sim_item, 0) item_sim[item][sim_item] * matrix.iloc[user_idx, item] rec sorted(scores.items(), keylambda x: x[1], reverseTrue)[:top_k] return [matrix.columns[i] for i, _ in rec]选型理由UserCF 适合用户数少于物品数的场景比如新闻推荐因为新闻更新快、物品多ItemCF 适合物品相对稳定的场景比如电商、视频。源码里两个都带的话你可以用同一份数据跑两遍对比precision10哪个高用哪个。别迷信哪个更先进数据说了算。3.3 矩阵分解SVD作为进阶替代如果协同过滤跑出来效果一般源码里通常还有一个svd_model.py或类似文件。矩阵分解把用户-物品矩阵拆成两个低维矩阵相乘能缓解稀疏性问题。scikit-learn里有TruncatedSVDsurprise库有专门的SVD源码用哪个看依赖。from sklearn.decomposition import TruncatedSVD def svd_recommend(matrix, n_components20, top_k10): svd TruncatedSVD(n_componentsn_components, random_state42) user_factors svd.fit_transform(matrix) item_factors svd.components_ # 重构评分矩阵 pred np.dot(user_factors, item_factors) # 对每个用户取 top_k排除已评分的 recs {} for i, user_id in enumerate(matrix.index): scores pred[i].copy() scores[matrix.iloc[i].nonzero()[0]] -np.inf top_items np.argsort(scores)[::-1][:top_k] recs[user_id] [matrix.columns[j] for j in top_items] return recs参数n_components是隐因子维度常见取值 20-200。太小欠拟合推荐结果趋同太大过拟合测试集指标反而降。我一般从 20 开始按 10 的步长往上加看RMSE什么时候不再降。random_state固定住不然每次跑结果不一样调参时你会怀疑人生。4. 避坑与排查跑推荐系统源码时最容易翻车的五个地方4.1 现象相似度矩阵全是 NaN 或 0原因通常是数据里有空值或者某个用户/物品在所有维度上都是 0。cosine_similarity遇到零向量会返回 0 或 NaN。解决在build_user_item_matrix之后加一步检查matrix matrix.loc[matrix.sum(axis1) 0, matrix.sum(axis0) 0]把全零行列剔掉。另外确认fillna(0)有没有漏有些源码用fillna(-1)那相似度计算就全乱了。4.2 现象推荐结果每次跑都不一样原因一般是没固定随机种子或者用了set做去重导致顺序不稳定。解决在numpy、random、sklearn相关代码前加np.random.seed(42)和random.seed(42)。如果是 Spark 版本还要设spark.conf.set(spark.sql.shuffle.partitions, 1)方便调试。别小看这个调参时结果不可复现你根本不知道是参数起作用还是随机波动。4.3 现象precisionk 高得离谱接近 1.0血泪经验大概率是数据泄漏。测试集里的物品在训练集里也出现了或者划分时没按时间切。解决检查train_test_split_by_time是不是真的按时间戳切的确认训练集里没有测试集的任何交互记录。另一个可能是评估时把用户已经评过分的物品也算进推荐列表了推荐前一定要排除已交互物品。4.4 现象Spark 版本跑起来报 Java 相关错误原因pyspark依赖 Java 8 或 11你机器上可能是 Java 17 或没装。解决java -version看一下不对就装个 Java 8/11 并设JAVA_HOME。如果只是本地小数据测试直接用纯 Python 版本别跟 Spark 环境较劲。等数据量真的到百万级交互再切 Spark那时候调环境也值了。4.5 现象内存溢出MemoryError原因用户-物品矩阵太大pivot_table直接撑爆内存。解决用scipy.sparse的csr_matrix替代稠密矩阵或者分批计算相似度。源码里如果用的是pandas透视表数据超过 10 万行就要警惕。常见做法是先用groupby聚合再转稀疏矩阵cosine_similarity本身支持稀疏输入。5. 从跑通到调优让推荐结果真正能看的几个技巧5.1 用网格搜索找相似度阈值和邻居数跑通之后别急着改模型先把sim_users和top_k这两个参数调明白。我一般写个小循环固定其他参数看precision10怎么变。results [] for sim_users in [10, 20, 30, 50]: for top_k in [5, 10, 20]: prec evaluate(sim_userssim_users, top_ktop_k) results.append((sim_users, top_k, prec)) # 按 precision 降序看 for r in sorted(results, keylambda x: x[2], reverseTrue)[:5]: print(fsim_users{r[0]}, top_k{r[1]}, precision{r[2]:.4f})这个表格不用画直接打印就行。重点看趋势sim_users从 10 加到 20 通常有提升加到 50 可能就平了甚至降。top_k取 10 是通用值但如果你做的是短视频推荐用户翻得快取 20 更合理。参数没有万能值看你的业务场景。5.2 混合推荐加权融合 UserCF 和 ItemCF单一算法总有短板源码里如果两个都有可以做个加权融合。常见做法是归一化后按 0.6/0.4 加权权重用验证集调。def hybrid_recommend(user_id, matrix, user_sim, item_sim, w0.6, top_k10): user_recs recommend_for_user(user_id, matrix, user_sim, top_k50) item_recs recommend_by_item(user_id, matrix, item_sim, top_k50) # 简单加权UserCF 结果权重 wItemCF 权重 1-w scores {} for i, item in enumerate(user_recs): scores[item] scores.get(item, 0) w * (1 - i / len(user_recs)) for i, item in enumerate(item_recs): scores[item] scores.get(item, 0) (1 - w) * (1 - i / len(item_recs)) return sorted(scores, keyscores.get, reverseTrue)[:top_k]这里的1 - i / len(recs)是位置衰减排前面的权重高。w取 0.6 是我在几个数据集上试出来的经验值你可以从 0.5 开始按 0.1 步长调。融合后precision10通常比单算法高 2-5 个百分点但别指望翻倍推荐系统没有银弹。5.3 冷启动用户的兜底策略新用户没有历史行为协同过滤直接歇菜。源码里一般会留个popular_items函数推全局最热的物品。我一般会再加一层按物品类别推用户注册时选过兴趣标签就按标签推没选就推热门。这部分代码通常不长但上线时能兜住 10%-20% 的流量。def cold_start_recommend(df, top_k10): # 按物品被评分次数排序取 top_k popular df[item_id].value_counts().head(top_k).index.tolist() return popular从那以后我每次拿到一份推荐系统源码都强制先跑一遍冷启动兜底逻辑确认新用户不会看到空白页。这个习惯帮我省过好几次线上事故。希望这份拆解能帮你把这份 Python 推荐系统源码真正用起来而不是躺在硬盘里吃灰。本文还有配套的精品资源点击获取