Django构建生产级音乐推荐系统：从模型到上线的完整实践

简介本资源是一个面向计算机专业本科生与深度学习初学者的课程设计级音乐推荐系统项目基于Python Django框架实现聚焦于音频与歌词多模态特征建模及个性化推荐算法落地。项目融合自动编码器与CNN提取非线性特征并结合协同过滤、KNN与词向量等技术构建混合推荐机制有效支撑从数据预处理、模型训练到Web端展示的完整闭环。压缩包共226个文件含65个核心Python源码含Django视图、模型与推荐逻辑、22张界面与架构示意图jpg/png、13个真实音频样本m4a/mp3、11个CSS样式文件如play.css、ranking.css等体现前端交互细节以及pkl模型文件和sqlite3数据库整体95.32MB结构清晰、模块可拆解。目前已有175人学习下载读者可直接运行调试后端服务、复现深度学习训练流程、分析推荐结果逻辑并参考配套文档与论文理解技术选型依据与实验对比思路。1. 为什么用 Django 搭音乐推荐系统不是 Flask也不是 FastAPI更不是纯 PyTorch Web API你手头有一堆用户听歌行为日志、千万级歌曲元数据、几十万首带标签的音频特征MFCC、Chroma、Spectral Contrast想做一个能上线、能查日志、能配权限、能接真实用户登录的音乐推荐服务——这时候硬套一个flask torch.jit.script的轻量方案三个月后你会在凌晨两点对着 500 错误和 CSRF token 失效抓狂而直接扔个uvicorn pydantic的 ASGI 接口连用户收藏夹增删都得自己写事务回滚逻辑。Django 不是“过重”而是把音乐推荐系统真正落地时绕不开的脏活用户体系、权限控制、后台管理、异步任务调度、数据库迁移、日志审计——全给你焊死在框架里。这个项目标题里的.zip不是摆设它代表一个可部署、可维护、带 admin 后台、支持多角色普通用户/运营/管理员的完整闭环。它不教你怎么推导注意力权重但教你如何让推荐结果真正在生产环境里稳定跑满 7×24 小时——比如当用户点击“跳过”时这条负样本怎么原子写入 PostgreSQL 并触发增量训练队列比如当新歌入库如何用 Django Signal 自动提取特征并更新 Faiss 索引。适合已经跑通 LightGBM 或双塔模型、正卡在“模型怎么变成产品”的中级工程师也适合需要交毕设答辩、但拒绝交一个只能本地python app.py的同学。2. 从零搭起推荐系统骨架Django 项目结构与核心模块拆解2.1 初始化 Django 项目并规划推荐域模块先创建虚拟环境并安装基础依赖注意 Python 版本需 ≥3.8Django ≥4.2因后续要用async视图和django.contrib.postgres的 JSONB 字段python -m venv venv_musicrec source venv_musicrec/bin/activate # Windows 用 venv_musicrec\Scripts\activate pip install django4.2.16 djangorestframework3.14.0 psycopg2-binary2.9.7 django-admin startproject musicrec . python manage.py startapp users python manage.py startapp songs python manage.py startapp recommendations关键点在于模块职责隔离users继承AbstractUser扩展字段如preference_vector存储用户 embedding、last_active_at用于活跃度衰减songs定义Song模型含audio_featuresJSONField 存 MFCC 等 128 维向量、genre_tagsArrayField 存 [rock, indie]、popularity_score实时计算的热度分recommendations核心业务模块含RecommendationLog记录每次推荐的user_id,song_ids,trigger_reason,timestamp和TrainingJob管理离线训练任务状态。提示不要把推荐算法逻辑写进views.py所有模型推理、特征工程、相似度计算必须封装在recommendations/services.py中视图只负责参数校验、调用 service、序列化响应。这是避免后期算法迭代导致整个 Django 请求链路崩塌的底线。2.2 数据库设计为推荐场景优化的 PostgreSQL 表结构音乐推荐对查询模式有强约束高频单用户召回100ms、批量冷启动新用户首次请求、实时负反馈写入。PostgreSQL 的jsonb和gin索引比 MySQL 更适配# songs/models.py from django.contrib.postgres.fields import ArrayField, JSONField from django.contrib.postgres.indexes import GinIndex class Song(models.Model): title models.CharField(max_length200) artist models.CharField(max_length150) duration_sec models.PositiveSmallIntegerField() audio_features JSONField() # {mfcc: [0.12, -0.45, ...], chroma: [...]} genre_tags ArrayField(models.CharField(max_length30), defaultlist) popularity_score models.FloatField(default0.0) class Meta: indexes [ GinIndex(fields[genre_tags]), # 加速 tag-based filtering models.Index(fields[popularity_score]), # 热度排序 ]users/models.py中扩展用户画像# users/models.py from django.contrib.auth.models import AbstractUser import numpy as np class CustomUser(AbstractUser): # 用户 embedding 向量128维存为 JSON 数组而非二进制便于 debug 和 admin 查看 preference_vector JSONField(defaultlist) # [0.23, -0.11, ..., 0.87] last_active_at models.DateTimeField(auto_nowTrue) def save(self, *args, **kwargs): # 确保 vector 长度固定避免前端传错导致 DB 写入失败 if self.preference_vector and len(self.preference_vector) ! 128: raise ValueError(preference_vector must be exactly 128 elements) super().save(*args, **kwargs)2.3 推荐服务层封装双塔模型推理与混合策略recommendations/services.py是整个系统的“心脏”它屏蔽了模型细节暴露统一接口# recommendations/services.py import numpy as np from sklearn.metrics.pairwise import cosine_similarity from .models import Song, CustomUser class RecommendationService: staticmethod def get_user_embedding(user_id: int) - np.ndarray: 从 DB 读取用户 embedding若为空则返回默认向量 try: user CustomUser.objects.get(iduser_id) vec np.array(user.preference_vector or [0.0] * 128) return vec / np.linalg.norm(vec) # L2 归一化 except CustomUser.DoesNotExist: return np.zeros(128) staticmethod def _get_song_embeddings(song_ids: list) - np.ndarray: 批量获取歌曲 embedding实际应从 Faiss 或 Redis 缓存读 songs Song.objects.filter(id__insong_ids).values(id, audio_features) vectors [] for s in songs: # audio_features 是 dict取 embedding 字段训练时存的 128维向量 vec np.array(s[audio_features].get(embedding, [0.0]*128)) vectors.append(vec / np.linalg.norm(vec)) return np.vstack(vectors) if vectors else np.empty((0, 128)) staticmethod def hybrid_recommend(user_id: int, top_k: int 20) - list: 混合推荐主路径用向量相似度fallback 用热度标签匹配 返回 song_id 列表非序列化数据由视图层处理 user_vec RecommendationService.get_user_embedding(user_id) # 主策略向量召回需提前构建 Faiss index此处简化为 DB 查询 candidate_songs Song.objects.exclude( id__inSong.objects.filter( id__inRecommendationLog.objects.filter( user_iduser_id ).values_list(song_ids, flatTrue) ) )[:500] # 先取 500 首未交互过的候选 if len(candidate_songs) 0: return Song.objects.order_by(-popularity_score)[:top_k].values_list(id, flatTrue) # 计算余弦相似度生产环境应替换为 Faiss GPU search cand_vecs RecommendationService._get_song_embeddings([s.id for s in candidate_songs]) if len(cand_vecs) 0: return Song.objects.order_by(-popularity_score)[:top_k].values_list(id, flatTrue) scores cosine_similarity([user_vec], cand_vecs)[0] scored_songs sorted( zip([s.id for s in candidate_songs], scores), keylambda x: x[1], reverseTrue ) # 取 top_k但插入 20% 热度 fallback避免信息茧房 top_ids [sid for sid, _ in scored_songs[:int(0.8*top_k)]] hot_ids Song.objects.order_by(-popularity_score)[:int(0.2*top_k)].values_list(id, flatTrue) return list(set(top_ids list(hot_ids)))[:top_k]这个 service 层的设计意图很明确把算法当作黑盒插件而不是嵌入业务逻辑的胶水代码。后期换成 BERT4Rec 或 LightGCN只需重写hybrid_recommend内部逻辑所有视图、路由、admin 页面完全不用动。3. 深度学习模型接入PyTorch 模型如何安全加载到 Django 请求流3.1 模型加载时机避免每次请求都torch.load()Django 的 WSGI/ASGI 进程模型决定了模型不能在views.py里torch.load()—— 这会导致每个请求都反序列化一次模型CPU 占用飙升且无法共享显存。正确做法是在 Django 应用初始化时加载apps.py# recommendations/apps.py from django.apps import AppConfig import torch class RecommendationsConfig(AppConfig): default_auto_field django.db.models.BigAutoField name recommendations def ready(self): # 仅在主进程加载模型避免多进程重复加载 if runserver in sys.argv or uwsgi in sys.argv: from .ml_models import DualTowerModel # 加载预训练模型假设已导出为 TorchScript self.model DualTowerModel() self.model.load_state_dict(torch.load(models/dual_tower_v2.pt)) self.model.eval() # 关键必须设为 eval 模式 self.model.to(cuda if torch.cuda.is_available() else cpu)然后在services.py中通过apps.get_app_config(recommendations).model获取实例# recommendations/services.py from django.apps import apps def get_recommendation_model(): return apps.get_app_config(recommendations).model注意如果使用 CUDA务必确认uwsgi或gunicorn启动时指定--enable-threads且torch.set_num_threads(1)避免多线程竞争显存。CPU 环境下torch.set_num_threads(os.cpu_count() // 2)更稳妥。3.2 特征工程管道从原始行为日志到模型输入推荐模型的输入不是 raw log而是结构化特征。我们用pandasscikit-learn构建可复用 pipeline# recommendations/feature_engineering.py import pandas as pd from sklearn.preprocessing import StandardScaler, LabelEncoder import numpy as np class UserFeaturePipeline: def __init__(self): self.scaler StandardScaler() self.genre_encoder LabelEncoder() self.is_fitted False def fit_transform(self, df: pd.DataFrame) - np.ndarray: df columns: [play_count, skip_rate, avg_duration, fav_genres] 返回 128维用户向量实际中应接 MLP 压缩 # 数值特征标准化 numeric_cols [play_count, skip_rate, avg_duration] scaled self.scaler.fit_transform(df[numeric_cols]) # 类别特征编码fav_genres 是 list需展平后编码 all_genres [g for genres in df[fav_genres] for g in genres] self.genre_encoder.fit(all_genres) # 聚合 genre embedding简化版取均值 genre_vectors [] for genres in df[fav_genres]: if not genres: genre_vectors.append(np.zeros(32)) else: encoded self.genre_encoder.transform(genres) # 此处应查预训练 genre embedding 表这里用随机模拟 vecs np.random.randn(len(genres), 32) genre_vectors.append(vecs.mean(axis0)) genre_emb np.vstack(genre_vectors) self.is_fitted True return np.hstack([scaled, genre_emb]) # shape: (n_samples, 33235) → 后续接 MLP 扩维到 128 def transform(self, df: pd.DataFrame) - np.ndarray: if not self.is_fitted: raise RuntimeError(Pipeline not fitted. Call fit_transform first.) # 同上但只 transform该 pipeline 在离线训练时拟合在线上服务时只transform保证特征一致性。血泪经验线上 inference 的 scaler 参数必须和训练时完全一致否则模型输出全乱。我们把scaler和genre_encoder的state_dict一起存为joblib文件部署时同步拷贝。3.3 推理加速TorchScript ONNX 双轨部署策略PyTorch 模型直接model(input)在高并发下延迟不可控。必须导出为 TorchScript 或 ONNX# 导出 TorchScript推荐兼容性好 example_input torch.randn(1, 128) # user vector traced_model torch.jit.trace(model.user_tower, example_input) traced_model.save(models/user_tower_traced.pt) # 导出 ONNX便于跨平台如 C 服务 torch.onnx.export( model.song_tower, torch.randn(1, 128), models/song_tower.onnx, input_names[input], output_names[output], dynamic_axes{input: {0: batch}, output: {0: batch}} )Django 中加载 TorchScript 模型比原生模型快 3~5 倍# recommendations/ml_models.py import torch class DualTowerModel: def __init__(self): self.user_tower torch.jit.load(models/user_tower_traced.pt) self.song_tower torch.jit.load(models/song_tower_traced.pt) def predict(self, user_vec: torch.Tensor, song_vecs: torch.Tensor) - torch.Tensor: user_emb self.user_tower(user_vec) song_embs self.song_tower(song_vecs) return torch.cosine_similarity(user_emb.unsqueeze(1), song_embs.unsqueeze(0), dim2)4. 推荐效果验证与 AB 测试闭环不只是跑通而是跑赢4.1 离线评估用历史日志构造 RecallK 和 NDCGK不能只看 loss 下降必须用真实业务指标验证。我们用过去 7 天用户行为日志构造测试集# evaluations/offline_eval.py import numpy as np from sklearn.metrics import ndcg_score def calculate_ndcg_at_k(y_true: np.ndarray, y_pred_scores: np.ndarray, k: int 10) - float: y_true: binary relevance vector (1clicked, 0not clicked) y_pred_scores: predicted scores for same items, descending order # 取 top-k 索引 top_k_indices np.argsort(y_pred_scores)[::-1][:k] y_true_topk y_true[top_k_indices] # sklearn ndcg_score 要求 2D 输入 return ndcg_score([y_true_topk], [np.arange(len(y_true_topk), 0, -1)]) # 示例对 1000 个用户计算平均 NDCG10 ndcg_scores [] for user_id in test_user_ids: true_interactions get_true_interactions(user_id) # 从日志取真实点击 pred_scores model.predict(user_id, candidate_song_ids) # 模型打分 ndcg calculate_ndcg_at_k(true_interactions, pred_scores) ndcg_scores.append(ndcg) print(fMean NDCG10: {np.mean(ndcg_scores):.4f})关键点RecallK 只反映“有没有推荐对”NDCGK 才反映“推荐顺序对不对”。对音乐推荐后者更重要——用户大概率只听前 3 首第 1 名和第 3 名的点击率差 3 倍。4.2 在线 AB 测试Django 中实现流量分流与指标埋点AB 测试不是前端切流量而是后端决策。我们在recommendations/views.py中注入实验逻辑# recommendations/views.py from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt import random from .services import RecommendationService csrf_exempt def recommend_api(request): if request.method ! POST: return JsonResponse({error: Only POST allowed}, status405) user_id int(request.POST.get(user_id)) # 10% 流量进入新模型A/B 实验 exp_group control if random.random() 0.9 else treatment if exp_group treatment: # 调用新模型 song_ids RecommendationService.hybrid_recommend_v2(user_id) exp_name dual_tower_v2 else: # 调用旧模型协同过滤 song_ids RecommendationService.collab_filtering(user_id) exp_name cf_baseline # 记录 AB 日志写入 PostgreSQL供数仓分析 from .models import ABTestLog ABTestLog.objects.create( user_iduser_id, exp_nameexp_name, groupexp_group, recommended_songssong_ids, timestamptimezone.now() ) return JsonResponse({songs: list(song_ids), exp_group: exp_group})配套的ABTestLog模型需支持高效聚合查询# recommendations/models.py class ABTestLog(models.Model): user_id models.IntegerField() exp_name models.CharField(max_length50) group models.CharField(max_length20) # control or treatment recommended_songs ArrayField(models.IntegerField()) timestamp models.DateTimeField() class Meta: indexes [ models.Index(fields[exp_name, group, timestamp]), ]提示AB 测试必须持续至少 7 天且要按用户维度分流不是请求维度否则同一用户在 control/treatment 组来回切换指标不可信。Django 中用user_id % 10 1比random()更稳定。4.3 埋点与监控用 Django Middleware 抓取推荐转化漏斗光有 AB 不够要知道“推荐→播放→收藏→分享”每一步的转化率。自定义 Middleware# recommendations/middleware.py from django.utils.deprecation import MiddlewareMixin from django.urls import resolve import logging logger logging.getLogger(recommendation) class RecommendationMetricsMiddleware(MiddlewareMixin): def process_view(self, request, view_func, view_args, view_kwargs): if view_func.__name__ recommend_api: # 记录请求开始 request.start_time time.time() def process_response(self, request, response): if hasattr(request, start_time): latency time.time() - request.start_time # 记录关键指标 logger.info( fRECOMMEND_LATENCY user{getattr(request, user_id, unknown)} flatency_ms{latency*1000:.0f} fstatus{response.status_code} ) return response配合LOGGING配置将日志推送到 ELK 或 Grafana# settings.py LOGGING { version: 1, handlers: { recommendation_file: { level: INFO, class: logging.handlers.RotatingFileHandler, filename: /var/log/django/recommendation.log, maxBytes: 1024*1024*5, # 5MB } }, loggers: { recommendation: { handlers: [recommendation_file], level: INFO, } } }5. 避坑指南那些让音乐推荐系统上线即翻车的 4 个致命细节5.1 现象用户首次登录后推荐全是冷门小众歌点击率暴跌 60%原因新用户preference_vector为空cosine_similarity计算时用零向量导致与所有歌曲相似度为 0fallback 到纯热度排序——但热度榜被僵尸账号刷榜实际是低质内容。解决在get_user_embedding中加入冷启动策略若preference_vector为空返回基于注册信息生成的初始向量如[0.1, 0.1, ..., 0.1]genre_bias同时强制 fallback 到“新歌发现”池人工标注的 1000 首高质量新歌而非全局热度榜。5.2 现象Django Admin 后台编辑歌曲时audio_features字段显示为{}无法修改原因JSONField在 Django Admin 中默认渲染为只读文本框且未配置widget。解决在songs/admin.py中自定义 ModelAdminfrom django import forms from django.contrib import admin from .models import Song class SongAdminForm(forms.ModelForm): class Meta: model Song fields __all__ widgets { audio_features: forms.Textarea(attrs{rows: 5, cols: 80}), } admin.register(Song) class SongAdmin(admin.ModelAdmin): form SongAdminForm list_display [title, artist, popularity_score]5.3 现象uWSGI 启动后GPU 显存占用 0%模型推理全在 CPU 上跑QPS 不足 5原因uWSGI 默认多进程模式下每个 worker 进程独立加载模型但torch.cuda.is_available()在子进程中返回FalseCUDA 上下文未正确继承。解决启动 uWSGI 时加参数--enable-threads --master --processes 2 --threads 4在apps.py的ready()中显式调用torch.cuda.set_device(0)并检查torch.cuda.current_device()更稳妥方案改用gunicorngevent并设置GEVENT_SUPPORT1。5.4 现象用户点击“不喜欢”后该歌曲仍频繁出现在后续推荐中原因负反馈未实时更新向量相似度计算仅写入RecommendationLog但召回阶段未排除已跳过歌曲。解决在hybrid_recommend中增加实时过滤# recommendations/services.py def hybrid_recommend(user_id: int, top_k: int 20) - list: # ... 前面逻辑 ... # 实时排除已跳过歌曲最近 24 小时 skipped_ids RecommendationLog.objects.filter( user_iduser_id, trigger_reasonskip, timestamp__gtetimezone.now() - timedelta(hours24) ).values_list(song_ids, flatTrue) candidate_songs Song.objects.exclude( id__inlist(skipped_ids) # 关键实时过滤 ).exclude( id__inRecommendationLog.objects.filter( user_iduser_id ).values_list(song_ids, flatTrue) )[:500] # ... 后续逻辑 ...6. 生产就绪技巧让推荐系统扛住突发流量与长期迭代6.1 缓存策略Redis 分层缓存设计不是简单 set/get音乐推荐的缓存不能只缓存最终结果要分层L1本地内存单进程内缓存user_id → embedding用functools.lru_cache(maxsize1000)L2Redis缓存user_id:embedding和song_id:featuresTTL 设为 1 小时特征变化慢L3Redis Sorted Set为每个用户维护user:{id}:rec_queue存song_id → score用ZREVRANGE快速取 top-kTTL 10 分钟保证新鲜度。# recommendations/cache_utils.py import redis import json from django.conf import settings redis_client redis.Redis( hostsettings.REDIS_HOST, portsettings.REDIS_PORT, db0, decode_responsesTrue ) def cache_user_embedding(user_id: int, vector: list, ttl: int 3600): redis_client.setex( fuser:{user_id}:embedding, ttl, json.dumps(vector) ) def get_cached_user_embedding(user_id: int) - list: data redis_client.get(fuser:{user_id}:embedding) return json.loads(data) if data else None def cache_recommendation_queue(user_id: int, song_scores: dict, ttl: int 600): song_scores: {song_id: score} pipe redis_client.pipeline() for song_id, score in song_scores.items(): pipe.zadd(fuser:{user_id}:rec_queue, {song_id: score}) pipe.expire(fuser:{user_id}:rec_queue, ttl) pipe.execute()6.2 异步训练Celery Django-Q 实现模型自动迭代推荐模型不能手动 retrain。用 Celery 定时触发# recommendations/tasks.py from celery import shared_task from .ml_models import train_dual_tower_model shared_task def train_recommendation_model(): 每日凌晨 2 点训练新模型 try: new_model_path train_dual_tower_model() # 替换线上模型原子操作 import shutil shutil.move(new_model_path, models/dual_tower_latest.pt) # 通知 Django 重新加载 from django.core.cache import cache cache.set(model_version, new_model_path.split(_)[-1].split(.)[0], timeout3600) except Exception as e: # 发送告警 send_alert(fModel training failed: {e}) # schedules.py from celery.schedules import crontab CELERY_BEAT_SCHEDULE { train-daily: { task: recommendations.tasks.train_recommendation_model, schedule: crontab(hour2, minute0), # 每天 2:00 }, }6.3 权限与审计Django Guardian 实现细粒度推荐策略管控运营人员需要动态调整某类用户的推荐权重如“学生群体增加校园民谣曝光”但不能直接改代码。用django-guardianpip install django-guardian# settings.py INSTALLED_APPS [guardian] AUTHENTICATION_BACKENDS [guardian.backends.ObjectPermissionBackend] # recommendations/admin.py from guardian.admin import GuardedModelAdmin admin.register(Song) class SongAdmin(GuardedModelAdmin): list_display [title, artist, popularity_score] # 运营可在 admin 页面为某首歌分配 can_boost_for_student 权限然后在hybrid_recommend中读取权限from guardian.shortcuts import get_objects_for_user def hybrid_recommend(user_id: int, top_k: int 20): user CustomUser.objects.get(iduser_id) # 获取该用户有权限 boost 的歌曲 boosted_songs get_objects_for_user( user, recommendations.can_boost_for_student, Song ) # 在召回结果中提升这些歌曲的分数 # ...我坚持一个习惯每次上线新推荐策略必先在 Django Shell 里手动跑通全流程——从User.objects.get(id123).preference_vector到RecommendationService.hybrid_recommend(123)再到ABTestLog.objects.filter(user_id123).last()。这 3 分钟的手动验证能避开 80% 的配置错误和路径问题。模型再 fancy如果settings.py里DEBUGFalse但ALLOWED_HOSTS漏了域名整个服务就是 500。希望帮到你。本文还有配套的精品资源点击获取