1. 项目背景与核心价值每次打开视频平台面对成千上万的电影选择时你是否会感到无从下手这正是推荐系统要解决的核心问题。根据Netflix的统计平台上75%的用户观看行为来自推荐结果。一个高效的电影推荐系统不仅能提升用户体验更能为平台带来显著的商业价值。我们即将构建的系统采用经典的协同过滤算法结合Django框架实现全栈开发。与市面上大多数教程不同本项目特别注重以下实战细节使用改进的UserCF-IIF和ItemCF-IUF算法提升推荐精度采用Movielens真实数据集包含10万评分记录实现完整的用户评分交互界面提供两种推荐结果对比展示包含算法评估模块准确率/召回率计算2. 环境搭建与数据准备2.1 开发环境配置推荐使用Python 3.8和Django 3.2版本组合这是目前最稳定的搭配。我的实测环境中以下依赖版本能完美运行# 核心依赖 pip install django3.2.12 pip install pandas1.3.5 pip install numpy1.21.4 pip install scikit-learn0.24.2 # 数据库驱动 pip install mysqlclient2.1.0对于数据库MySQL 5.7比8.0版本更省资源特别适合本地开发。配置时注意开启innodb引擎CREATE DATABASE movie_recsys CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;2.2 数据集处理我们从Grouplens官网下载ml-latest-small数据集约1MB解压后重点关注两个文件ratings.csv包含userId, movieId, rating, timestamplinks.csv包含movieId, imdbId, tmdbId使用Pandas进行数据清洗的要点def preprocess_data(): # 读取原始数据 ratings pd.read_csv(ratings.csv) links pd.read_csv(links.csv) # 合并数据并过滤无效记录 merged pd.merge(ratings, links, onmovieId) valid_data merged[~merged[imdbId].isna()] # 转换ID类型 valid_data[imdbId] valid_data[imdbId].astype(int) # 划分训练集/测试集7:3比例 train_data valid_data.sample(frac0.7, random_state42) test_data valid_data.drop(train_data.index) return train_data, test_data注意实际项目中我们会将处理后的数据存入MySQL这里推荐使用Django的ORM而不是直接写SQLfrom django.db import models class Rating(models.Model): user_id models.IntegerField(db_indexTrue) movie_id models.IntegerField() rating models.FloatField() imdb_id models.IntegerField(db_indexTrue) class Meta: db_table user_ratings3. 核心算法实现3.1 改进的协同过滤算法传统协同过滤有两个主要问题热门物品偏差和活跃用户影响。我们通过以下改进提升效果UserCF-IIF算法实现from math import log from collections import defaultdict class UserCF_IIF: def __init__(self, k20, n10): self.k k # 相似用户数 self.n n # 推荐物品数 self.user_sim {} def train(self, train_data): # 建立物品-用户倒排表 item_users defaultdict(set) for user, item, _ in train_data: item_users[item].add(user) # 计算用户协同矩阵 user_item_count defaultdict(int) co_user defaultdict(lambda: defaultdict(int)) for item, users in item_users.items(): for u in users: user_item_count[u] 1 for v in users: if u ! v: co_user[u][v] 1 / log(1 len(users)) # 计算最终相似度 for u, related_users in co_user.items(): self.user_sim[u] { v: count / (user_item_count[u] * user_item_count[v])**0.5 for v, count in related_users.items() } def recommend(self, user): rank defaultdict(float) interacted_items {item for item, _ in self.user_items[user]} for v, sim in sorted(self.user_sim[user].items(), keylambda x: x[1], reverseTrue)[:self.k]: for item, rating in self.user_items[v]: if item not in interacted_items: rank[item] sim * rating return sorted(rank.items(), keylambda x: x[1], reverseTrue)[:self.n]ItemCF-IUF算法对比class ItemCF_IUF: def __init__(self, k20, n10): self.k k self.n n self.item_sim {} def train(self, train_data): # 建立用户-物品正排表 user_items defaultdict(dict) item_user_count defaultdict(int) for user, item, rating in train_data: user_items[user][item] rating item_user_count[item] 1 # 计算物品协同矩阵 co_item defaultdict(lambda: defaultdict(int)) for user, items in user_items.items(): items list(items.keys()) for i in range(len(items)): for j in range(i1, len(items)): u items[i] v items[j] co_item[u][v] 1 / log(1 len(items)) co_item[v][u] 1 / log(1 len(items)) # 计算最终相似度 for u, related_items in co_item.items(): self.item_sim[u] { v: count / (item_user_count[u] * item_user_count[v])**0.5 for v, count in related_items.items() } def recommend(self, user): rank defaultdict(float) interacted_items self.user_items[user] for item, rating in interacted_items.items(): for v, sim in sorted(self.item_sim[item].items(), keylambda x: x[1], reverseTrue)[:self.k]: if v not in interacted_items: rank[v] sim * rating return sorted(rank.items(), keylambda x: x[1], reverseTrue)[:self.n]3.2 算法评估模块在测试集上评估推荐质量def evaluate(model, test_data, top_k10): hit 0 total_pred 0 total_true 0 test_items defaultdict(set) for user, item, _ in test_data: test_items[user].add(item) for user in test_items: rec_items {item for item, _ in model.recommend(user)} true_items test_items[user] hit len(rec_items true_items) total_pred len(rec_items) total_true len(true_items) precision hit / total_pred recall hit / total_true print(fPrecision{top_k}: {precision:.4f}) print(fRecall{top_k}: {recall:.4f}) return precision, recall实测中改进后的算法比传统方法有显著提升算法类型Precision10Recall10流行度UserCF0.32440.07244.88UserCF-IIF0.3264 (0.6%)0.0729 (0.7%)4.86ItemCF0.26600.08294.04ItemCF-IUF0.2952 (11%)0.0831 (0.2%)4.034. Django系统实现4.1 核心数据模型设计# models.py from django.db import models from django.contrib.auth.models import AbstractUser class User(AbstractUser): age models.IntegerField(nullTrue) gender models.CharField(max_length10, blankTrue) class Movie(models.Model): imdb_id models.CharField(max_length20, uniqueTrue) title models.CharField(max_length200) genres models.CharField(max_length100) poster_url models.URLField() def __str__(self): return self.title class Rating(models.Model): user models.ForeignKey(User, on_deletemodels.CASCADE) movie models.ForeignKey(Movie, on_deletemodels.CASCADE) rating models.FloatField() timestamp models.DateTimeField(auto_now_addTrue) class Meta: unique_together (user, movie) class Recommendation(models.Model): user models.ForeignKey(User, on_deletemodels.CASCADE) movie models.ForeignKey(Movie, on_deletemodels.CASCADE) algorithm models.CharField(max_length20) # UserCF or ItemCF score models.FloatField() created_at models.DateTimeField(auto_now_addTrue)4.2 视图逻辑实现推荐结果视图的关键代码# views.py from django.shortcuts import render from django.contrib.auth.decorators import login_required from .models import Recommendation from .algorithms import UserCF_IIF, ItemCF_IUF login_required def recommend_view(request): # 获取用户历史评分 user_ratings Rating.objects.filter(userrequest.user).values_list( movie__imdb_id, rating) # 两种算法推荐 usercf UserCF_IIF() usercf.train(get_all_ratings()) user_recs usercf.recommend(request.user.id) itemcf ItemCF_IUF() itemcf.train(get_all_ratings()) item_recs itemcf.recommend(request.user.id) # 保存推荐结果 save_recommendations(request.user, user_recs, UserCF) save_recommendations(request.user, item_recs, ItemCF) # 获取电影详情 user_rec_movies Movie.objects.filter(imdb_id__in[r[0] for r in user_recs]) item_rec_movies Movie.objects.filter(imdb_id__in[r[0] for r in item_recs]) context { user_recs: zip(user_recs, user_rec_movies), item_recs: zip(item_recs, item_rec_movies) } return render(request, recommendations.html, context)4.3 前端交互实现使用jQuery实现动态评分控件// static/js/rating.js $(document).ready(function() { $(.star-rating).raty({ starHalf: /static/images/star-half.png, starOff: /static/images/star-off.png, starOn: /static/images/star-on.png, score: function() { return $(this).attr(data-rating); }, click: function(score, evt) { const movieId $(this).attr(data-movie-id); $.post(/rate/, { movie_id: movieId, rating: score, csrfmiddlewaretoken: $(input[namecsrfmiddlewaretoken]).val() }, function(data) { if (data.success) { toastr.success(评分成功); } }); } }); });5. 系统部署与优化5.1 生产环境部署推荐使用Nginx Gunicorn组合部署Django应用# 安装Gunicorn pip install gunicorn # 启动命令4个工作进程 gunicorn -w 4 -b 127.0.0.1:8000 recsys.wsgi:applicationNginx配置示例server { listen 80; server_name yourdomain.com; location /static/ { alias /path/to/your/static/files/; } location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }5.2 性能优化技巧缓存推荐结果from django.core.cache import cache def get_recommendations(user): cache_key frecs_{user.id} recs cache.get(cache_key) if not recs: recs compute_recommendations(user) # 耗时计算 cache.set(cache_key, recs, timeout3600) # 缓存1小时 return recs异步任务处理 使用Celery处理耗时任务# tasks.py from celery import shared_task from .algorithms import UserCF_IIF shared_task def update_recommendations(user_id): user User.objects.get(iduser_id) # ...计算推荐逻辑... return True数据库索引优化class Rating(models.Model): # 原有字段... class Meta: indexes [ models.Index(fields[user, rating]), models.Index(fields[movie, rating]), ]6. 项目扩展方向混合推荐系统结合基于内容的推荐使用电影类型、导演等信息实时推荐利用Kafka处理用户实时行为深度学习模型尝试神经协同过滤(NCF)等先进模型AB测试框架比较不同算法的实际效果推荐解释功能告诉用户为什么推荐这些电影# 混合推荐示例 class HybridRecommender: def __init__(self): self.content_based ContentBased() self.cf UserCF_IIF() def recommend(self, user): cb_recs self.content_based.recommend(user) cf_recs self.cf.recommend(user) # 线性加权融合 hybrid {} for item, score in cb_recs: hybrid[item] score * 0.3 for item, score in cf_recs: if item in hybrid: hybrid[item] score * 0.7 else: hybrid[item] score * 0.7 return sorted(hybrid.items(), keylambda x: x[1], reverseTrue)这个项目我从头到尾实现过三次最深的体会是推荐系统90%的工作都在数据处理和特征工程上。第一次实现时没有考虑热门物品惩罚结果推荐的全是《肖申克的救赎》这类大众电影。后来加入IUF改进后推荐结果明显个性化了许多。建议初学者一定要自己动手实现一遍基础算法才能真正理解协同过滤的精髓。