Python实战：从零构建在线电影推荐系统，协同过滤与Flask接口全解析-北京尧图网络

简介这份资源是基于Python的在线电影推荐系统完整项目包面向计算机相关专业的毕业设计、课程设计学生以及希望入门推荐算法的开发者。项目围绕用户观看历史与电影内容展开分析涵盖数据爬取、数据预处理、特征提取、相似度计算、推荐算法与结果可视化等核心模块可帮助读者理解基于内容推荐与协同过滤的实现思路。压缩包共392个文件约9.76MB包含28个py源码文件、43个pyc编译文件、62个js脚本、35个css样式、9个html页面以及大量jpg、gif、png图片素材和字体文件另附sql数据库脚本整体结构完整便于直接运行与二次开发。目前已有255人学习下载。对于需要完成推荐系统类课题的读者可参考其模块划分、算法流程与前后端组织方式快速搭建可演示的个性化电影推荐应用。1. 在线电影推荐系统从零搭一套能跑起来的 Python 方案很多人第一次接触推荐系统都是被协同过滤四个字劝退的——公式一堆代码一跑就报维度不匹配。但如果你手上只有一份用户对电影的评分数据想快速搭出一个能给出猜你喜欢结果的在线电影推荐系统Python 其实是最省事的路径。这个标题背后要解决的核心问题很具体给定用户-电影评分矩阵预测某个用户没看过的电影里他最可能喜欢哪几部并且能通过一个接口实时返回结果。适合谁适合刚学完 pandas 和 numpy、想找一个完整项目练手的人也适合需要给内部系统快速加一个推荐模块的后端工程师。整套方案不依赖深度学习框架一台普通笔记本就能跑通重点在于把数据清洗、相似度计算、推荐生成、接口暴露这条链路走完整。2. 数据准备与评分矩阵构建推荐系统的地基怎么打推荐系统的效果上限在数据进入模型之前就已经被决定了。电影推荐场景里最常见的原始数据是三列用户 ID、电影 ID、评分值。听起来简单但真正动手时你会发现一堆问题——同一个用户对同一部电影重复评分、评分值是字符串而不是数字、电影 ID 是标题而不是整数。这些不处理干净后面算相似度全是脏结果。2.1 用 pandas 加载并清洗评分数据假设你手上是一份 CSV字段为 userId、movieId、rating、timestamp。下面这段代码做三件事去重、类型转换、过滤低频用户和低频电影。import pandas as pd import numpy as np # 加载原始评分数据 df pd.read_csv(ratings.csv) # 去重同一用户对同一电影只保留最新一条评分 df df.sort_values(timestamp).drop_duplicates( subset[userId, movieId], keeplast ) # 类型转换防止评分被当成字符串 df[rating] pd.to_numeric(df[rating], errorscoerce) df df.dropna(subset[rating]) # 过滤评分次数少于 5 次的用户、被评分少于 5 次的电影直接剔除 user_counts df[userId].value_counts() movie_counts df[movieId].value_counts() df df[df[userId].isin(user_counts[user_counts 5].index)] df df[df[movieId].isin(movie_counts[movie_counts 5].index)] print(f清洗后剩余 {len(df)} 条评分{df[userId].nunique()} 个用户{df[movieId].nunique()} 部电影)逻辑说明drop_duplicates配合sort_values保证同一对用户-电影只留最新评分避免重复数据放大某个用户的影响。errorscoerce把无法转换的值变成 NaN 再删掉比直接报错更稳。过滤阈值设成 5 是经验值——低于这个数的用户和电影提供的协同信息太少留着只会增加矩阵稀疏度拖慢计算还拉低准确率。如果你的数据量很大这个阈值可以调到 10 甚至 20。参数方面keeplast依赖 timestamp 排序如果你的数据没有时间戳字段改成keepfirst也行但要去掉sort_values那行。过滤阈值不是固定的数据量大就调高数据量小就调低核心原则是保证每个用户至少能和其他用户产生交集。2.2 构建用户-电影评分矩阵清洗完之后要转成矩阵形式。推荐系统里最常用的结构是行是用户、列是电影、值是评分的二维矩阵。pandas 的 pivot_table 一步就能搞定。# 构建用户-电影评分矩阵缺失值填 0 表示未评分 rating_matrix df.pivot_table( indexuserId, columnsmovieId, valuesrating, fill_value0 ) print(f矩阵形状{rating_matrix.shape}) # 稀疏度计算 sparsity (rating_matrix 0).sum().sum() / (rating_matrix.shape[0] * rating_matrix.shape[1]) print(f矩阵稀疏度{sparsity:.2%})逻辑说明pivot_table默认对重复的 index-column 组合取均值但前面已经去重了所以这里不会出问题。fill_value0把未评分位置填 0这是基于用户的协同过滤的标准做法——0 表示没有交互不表示评了 0 分。稀疏度这个指标很关键电影推荐场景下通常超过 95%意味着矩阵里绝大多数格子是空的。如果你的稀疏度低于 90%说明数据质量不错基于邻域的方法效果会比较好如果超过 99%就要考虑降维或者换用矩阵分解了。提示矩阵规模超过 10000×10000 时pivot_table 会吃掉大量内存。这时候建议用 scipy.sparse 的 csr_matrix 来存后面算相似度时也更快。3. 协同过滤核心实现相似度计算与推荐生成数据准备好了接下来是推荐系统真正干活的部分。基于用户的协同过滤思路很直白找到和目标用户口味相似的一批人把他们喜欢但目标用户没看过的电影推过来。听起来简单但相似度怎么算、邻居选几个、评分怎么加权每个环节都有讲究。3.1 余弦相似度与皮尔逊相似度的选择两种最常用的相似度度量余弦相似度和皮尔逊相关系数。余弦相似度看的是两个用户评分向量的方向是否一致对绝对值不敏感皮尔逊相似度先去均值再算相关能消除不同用户打分尺度差异——有人习惯全打 4 分有人习惯打 2 分皮尔逊能纠正这个偏差。from sklearn.metrics.pairwise import cosine_similarity # 基于评分矩阵计算用户间余弦相似度 user_sim cosine_similarity(rating_matrix) # 转成 DataFrame 方便按 userId 索引 user_sim_df pd.DataFrame( user_sim, indexrating_matrix.index, columnsrating_matrix.index ) # 查看某个用户最相似的 10 个邻居 target_user rating_matrix.index[0] neighbors user_sim_df[target_user].sort_values(ascendingFalse)[1:11] print(f用户 {target_user} 的 Top10 相似邻居) print(neighbors)逻辑说明cosine_similarity直接接收矩阵返回的是 n×n 的相似度矩阵对角线是 1自己和自己的相似度。[1:11]跳过自己取前 10 个邻居。这里有个容易翻车的点——如果某个用户只评了一部电影他和别人的余弦相似度要么是 0 要么是 1这种极端值会污染推荐结果。所以前面清洗时过滤低频用户不是可选项是必须做的。参数方面邻居数量 K 一般取 10 到 50 之间。K 太小推荐结果不稳定容易受个别邻居影响K 太大计算量上去不说还会引入不相关的用户稀释推荐精度。我一般先用 20 试然后根据离线评估的准确率微调。如果你的用户量只有几百K 取 10 就够了上万用户时 K 可以到 50。3.2 基于邻居评分加权生成推荐列表有了相似度矩阵下一步就是给目标用户生成推荐。核心公式是预测评分相似度加权平均的邻居评分。下面这段代码实现了完整的推荐生成流程。def recommend_for_user(user_id, rating_matrix, user_sim_df, top_k20, top_n10): 基于用户的协同过滤推荐 user_id: 目标用户 top_k: 选取的邻居数量 top_n: 返回的推荐电影数量 # 获取目标用户的评分向量 user_ratings rating_matrix.loc[user_id] # 找到已评分的电影 rated_movies user_ratings[user_ratings 0].index # 取相似度最高的 top_k 个邻居排除自己 sim_scores user_sim_df[user_id].drop(user_id) top_neighbors sim_scores.sort_values(ascendingFalse).head(top_k) # 只保留相似度大于 0 的邻居 top_neighbors top_neighbors[top_neighbors 0] if len(top_neighbors) 0: return pd.Series(dtypefloat) # 加权预测每个邻居的评分乘以相似度再除以相似度之和 neighbor_ratings rating_matrix.loc[top_neighbors.index] weighted_sum neighbor_ratings.T.dot(top_neighbors) sim_sum top_neighbors.sum() predicted weighted_sum / sim_sum # 去掉用户已经看过的电影 predicted predicted.drop(rated_movies, errorsignore) # 按预测评分降序返回 top_n return predicted.sort_values(ascendingFalse).head(top_n) # 调用示例 target_user rating_matrix.index[0] recommendations recommend_for_user(target_user, rating_matrix, user_sim_df) print(f为用户 {target_user} 推荐的电影) print(recommendations)逻辑说明neighbor_ratings.T.dot(top_neighbors)这一步是矩阵乘法把邻居评分矩阵转置后和相似度向量做点积等价于对每个邻居的评分按相似度加权求和。除以sim_sum做归一化保证预测评分在合理范围内。drop(rated_movies)去掉已看过的电影这是推荐系统的基本礼仪——推一部用户已经看过的电影体验直接归零。参数方面top_k控制邻居数量top_n控制返回结果数量。实际部署时top_n一般设 10 到 20太多用户看不过来。相似度阈值 0这个条件很重要负相似度意味着口味相反他们的喜好参考价值是负的直接排除比加权更安全。注意如果某个用户的所有邻居相似度都是 0函数会返回空结果。线上系统里这种情况要兜底比如返回热门电影列表不能让接口直接报错。4. 避坑与排查推荐系统上线前必须过的五道坎代码跑通和系统能用之间隔着至少五个坑。下面这些是我在实际部署中踩过的每一条都对应真实故障。现象一推荐结果全是冷门电影热门片一部都不推。原因低频电影因为评分人数少在相似度计算中反而容易获得高权重导致推荐列表被小众片占据。解决在生成推荐时加一个最小评分人数阈值比如只推荐被至少 20 个用户评分过的电影。在predicted结果上过滤即可predicted predicted[predicted.index.isin(movie_counts[movie_counts 20].index)]。现象二新用户进来后推荐接口超时。原因新用户没有任何评分记录rating_matrix.loc[user_id]直接 KeyError或者相似度计算时全为 0 导致空结果循环。解决对没有评分记录的用户走冷启动逻辑直接返回全局热门电影 Top20同时在前端引导用户完成至少 5 部电影的评分。现象三相似度矩阵计算一次要几分钟接口响应超过 5 秒。原因每次请求都重新算cosine_similarity用户量上千时矩阵乘法开销巨大。解决相似度矩阵离线算好存成文件或缓存线上只做查表和加权。用户评分更新时再触发重算不需要实时计算。现象四推荐结果里出现用户刚看过的电影。原因drop(rated_movies)只去掉了评分矩阵里有记录的电影但用户可能通过其他渠道看过但没评分。解决维护一份用户观看历史表推荐时同时过滤评分记录和观看记录。如果只有评分数据至少保证rated_movies的过滤逻辑正确别把errorsignore当成万能药。现象五不同用户收到的推荐结果高度雷同。原因邻居数量 K 设得太大或者相似度矩阵区分度不够导致所有用户都指向同一批热门电影。解决降低 K 值或者在相似度计算前对评分做标准化减去用户均分让相似度更能反映口味差异而不是打分习惯。5. 从离线脚本到在线服务Flask 接口封装与性能优化前面所有代码都是在 Jupyter 里跑的离线脚本要变成在线推荐系统最后一步是暴露 HTTP 接口。Flask 是最轻量的选择十几行代码就能把推荐函数包装成 API。from flask import Flask, request, jsonify import pandas as pd import numpy as np from sklearn.metrics.pairwise import cosine_similarity app Flask(__name__) # 启动时加载数据并预计算相似度矩阵 df pd.read_csv(ratings.csv) rating_matrix df.pivot_table( indexuserId, columnsmovieId, valuesrating, fill_value0 ) user_sim cosine_similarity(rating_matrix) user_sim_df pd.DataFrame(user_sim, indexrating_matrix.index, columnsrating_matrix.index) # 预计算热门电影作为冷启动兜底 popular_movies df.groupby(movieId)[rating].agg([count, mean]) popular_movies popular_movies[popular_movies[count] 20] popular_movies popular_movies.sort_values(mean, ascendingFalse).head(20) app.route(/recommend/int:user_id, methods[GET]) def recommend(user_id): top_n request.args.get(top_n, 10, typeint) # 冷启动用户不在评分矩阵中 if user_id not in rating_matrix.index: return jsonify({ user_id: user_id, cold_start: True, recommendations: popular_movies.index.tolist()[:top_n] }) # 正常推荐流程 user_ratings rating_matrix.loc[user_id] rated_movies user_ratings[user_ratings 0].index sim_scores user_sim_df[user_id].drop(user_id) top_neighbors sim_scores.sort_values(ascendingFalse).head(20) top_neighbors top_neighbors[top_neighbors 0] if len(top_neighbors) 0: return jsonify({ user_id: user_id, cold_start: True, recommendations: popular_movies.index.tolist()[:top_n] }) neighbor_ratings rating_matrix.loc[top_neighbors.index] weighted_sum neighbor_ratings.T.dot(top_neighbors) predicted weighted_sum / top_neighbors.sum() predicted predicted.drop(rated_movies, errorsignore) predicted predicted.sort_values(ascendingFalse).head(top_n) return jsonify({ user_id: user_id, cold_start: False, recommendations: [ {movie_id: int(mid), score: round(float(score), 3)} for mid, score in predicted.items() ] }) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)逻辑说明相似度矩阵在应用启动时算一次之后所有请求复用这是把响应时间从秒级降到毫秒级的关键。冷启动逻辑单独处理用户不在矩阵里或者没有有效邻居时返回热门电影保证接口永远有结果。debugFalse是上线必须的调试模式会暴露堆栈信息且有性能开销。参数方面top_n通过 URL 参数传入默认 10。端口 5000 是 Flask 默认值生产环境建议用 gunicorn 加多 worker 部署命令是gunicorn -w 4 -b 0.0.0.0:5000 app:app4 个 worker 对一般规模足够。如果用户量上万相似度矩阵内存占用会比较大可以考虑用 numpy 的 float32 存储内存直接减半。验证方法很简单启动服务后访问http://localhost:5000/recommend/1?top_n5看返回的 JSON 里推荐列表是否合理。再访问一个不存在的用户 ID确认冷启动兜底生效。最后用ab或wrk压一下接口看 QPS 和响应时间是否满足预期。这套方案我前后改过三版第一版每次请求都重算相似度接口响应 8 秒被测试同学追着骂第二版加了缓存但没做冷启动新用户直接 500 报错第三版才把这两个问题都补上。如果你准备把这个方向做深下一步可以试试矩阵分解SVD替代基于邻域的方法稀疏矩阵下效果提升明显但可解释性会下降——这是另一个话题了。希望帮到你。本文还有配套的精品资源点击获取

Python实战：从零构建在线电影推荐系统，协同过滤与Flask接口全解析

相关推荐

AI Agent 可观测性实践：用 OpenTelemetry 让决策路径可追踪

diagram-design：语义化流程图驱动可执行系统契约

Linux线程深度剖析：轻量级进程、虚拟内存分页、进程线程资源对比 | 线程上篇

更多资讯

Docker本地部署Home Assistant：从零搭建私有智能家居平台

有源配电网SOP规划：为何必须考虑DG时序特性与MINLP建模

屏幕故障不用怕：黑屏、花屏、闪屏的定位排查全流程

ApplyPilot六阶段流水线全景解析：从职位发现到自动提交的完整工作原理

AI Agent工具调用模块和MCP：把MCP endpoint改到TaoToken的配置与验证

单片机计算机毕设之基于 51 单片机的小型燃气机房 CO 与温湿度阈值可调告警装置设计基于物联网的食堂后厨一氧化碳泄漏远程监测智能排风系统设计（030124）

今日更新

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？

相关推荐

AI Agent 可观测性实践：用 OpenTelemetry 让决策路径可追踪

diagram-design：语义化流程图驱动可执行系统契约

Linux线程深度剖析：轻量级进程、虚拟内存分页、进程线程资源对比 | 线程上篇

更多资讯

Docker本地部署Home Assistant：从零搭建私有智能家居平台

有源配电网SOP规划：为何必须考虑DG时序特性与MINLP建模

屏幕故障不用怕：黑屏、花屏、闪屏的定位排查全流程

ApplyPilot六阶段流水线全景解析：从职位发现到自动提交的完整工作原理

AI Agent工具调用模块和MCP：把MCP endpoint改到TaoToken的配置与验证

单片机计算机毕设之基于 51 单片机的小型燃气机房 CO 与温湿度阈值可调告警装置设计 基于物联网的食堂后厨一氧化碳泄漏远程监测智能排风系统设计（030124）

今日更新

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？

单片机计算机毕设之基于 51 单片机的小型燃气机房 CO 与温湿度阈值可调告警装置设计基于物联网的食堂后厨一氧化碳泄漏远程监测智能排风系统设计（030124）