简介：本资源是一个面向计算机专业本科生与深度学习初学者的课程设计级音乐推荐系统项目，基于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.8，Django ≥4.2，因后续要用async视图和django.contrib.postgres的 JSONB 字段）：

python -m venv venv_musicrec source venv_musicrec/bin/activate # Windows 用 venv_musicrec\Scripts\activate pip install django==4.2.16 djangorestframework==3.14.0 psycopg2-binary==2.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_features（JSONField 存 MFCC 等 128 维向量）、genre_tags（ArrayField 存 ['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_length=200) artist = models.CharField(max_length=150) duration_sec = models.PositiveSmallIntegerField() audio_features = JSONField() # {"mfcc": [0.12, -0.45, ...], "chroma": [...]} genre_tags = ArrayField(models.CharField(max_length=30), default=list) popularity_score = models.FloatField(default=0.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(default=list) # [0.23, -0.11, ..., 0.87] last_active_at = models.DateTimeField(auto_now=True) 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(id=user_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__in=song_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__in=Song.objects.filter( id__in=RecommendationLog.objects.filter( user_id=user_id ).values_list('song_ids', flat=True) ) )[:500] # 先取 500 首未交互过的候选 if len(candidate_songs) == 0: return Song.objects.order_by('-popularity_score')[:top_k].values_list('id', flat=True) # 计算余弦相似度（生产环境应替换为 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', flat=True) scores = cosine_similarity([user_vec], cand_vecs)[0] scored_songs = sorted( zip([s.id for s in candidate_songs], scores), key=lambda x: x[1], reverse=True ) # 取 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', flat=True) 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，而是结构化特征。我们用pandas+scikit-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(axis=0)) genre_emb = np.vstack(genre_vectors) self.is_fitted = True return np.hstack([scaled, genre_emb]) # shape: (n_samples, 3+32=35) → 后续接 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), dim=2)

4. 推荐效果验证与 AB 测试闭环：不只是跑通，而是跑赢

4.1 离线评估：用历史日志构造 Recall@K 和 NDCG@K

不能只看 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 (1=clicked, 0=not 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 个用户计算平均 NDCG@10 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(f"Mean NDCG@10: {np.mean(ndcg_scores):.4f}")

关键点：Recall@K 只反映“有没有推荐对”，NDCG@K 才反映“推荐顺序对不对”。对音乐推荐，后者更重要——用户大概率只听前 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'}, status=405) 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_id=user_id, exp_name=exp_name, group=exp_group, recommended_songs=song_ids, timestamp=timezone.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_length=50) group = models.CharField(max_length=20) # '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( f"RECOMMEND_LATENCY user={getattr(request, 'user_id', 'unknown')} " f"latency_ms={latency*1000:.0f} " f"status={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计算时用零向量，导致与所有歌曲相似度为 0，fallback 到纯热度排序——但热度榜被僵尸账号刷榜，实际是低质内容。
解决：在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中自定义 ModelAdmin：

from 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()在子进程中返回False（CUDA 上下文未正确继承）。
解决：

• 启动 uWSGI 时加参数--enable-threads --master --processes 2 --threads 4；
• 在apps.py的ready()中，显式调用torch.cuda.set_device(0)并检查torch.cuda.current_device()；
• 更稳妥方案：改用gunicorn+gevent，并设置GEVENT_SUPPORT=1。

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_id=user_id, trigger_reason='skip', timestamp__gte=timezone.now() - timedelta(hours=24) ).values_list('song_ids', flat=True) candidate_songs = Song.objects.exclude( id__in=list(skipped_ids) # 关键：实时过滤 ).exclude( id__in=RecommendationLog.objects.filter( user_id=user_id ).values_list('song_ids', flat=True) )[:500] # ... 后续逻辑 ...

6. 生产就绪技巧：让推荐系统扛住突发流量与长期迭代

6.1 缓存策略：Redis 分层缓存设计（不是简单 set/get）

音乐推荐的缓存不能只缓存最终结果，要分层：

• L1（本地内存）：单进程内缓存user_id → embedding，用functools.lru_cache(maxsize=1000)；
• L2（Redis）：缓存user_id:embedding和song_id:features，TTL 设为 1 小时（特征变化慢）；
• L3（Redis Sorted Set）：为每个用户维护user:{id}:rec_queue，存song_id → score，用ZREVRANGE快速取 top-k，TTL 10 分钟（保证新鲜度）。

# recommendations/cache_utils.py import redis import json from django.conf import settings redis_client = redis.Redis( host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0, decode_responses=True ) def cache_user_embedding(user_id: int, vector: list, ttl: int = 3600): redis_client.setex( f"user:{user_id}:embedding", ttl, json.dumps(vector) ) def get_cached_user_embedding(user_id: int) -> list: data = redis_client.get(f"user:{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(f"user:{user_id}:rec_queue", {song_id: score}) pipe.expire(f"user:{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], timeout=3600) except Exception as e: # 发送告警 send_alert(f"Model training failed: {e}") # schedules.py from celery.schedules import crontab CELERY_BEAT_SCHEDULE = { 'train-daily': { 'task': 'recommendations.tasks.train_recommendation_model', 'schedule': crontab(hour=2, minute=0), # 每天 2:00 }, }

6.3 权限与审计：Django Guardian 实现细粒度推荐策略管控

运营人员需要动态调整某类用户的推荐权重（如“学生群体增加校园民谣曝光”），但不能直接改代码。用django-guardian：

pip 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(id=user_id) # 获取该用户有权限 boost 的歌曲 boosted_songs = get_objects_for_user( user, 'recommendations.can_boost_for_student', Song ) # 在召回结果中提升这些歌曲的分数 # ...

我坚持一个习惯：每次上线新推荐策略，必先在 Django Shell 里手动跑通全流程——从User.objects.get(id=123).preference_vector到RecommendationService.hybrid_recommend(123)，再到ABTestLog.objects.filter(user_id=123).last()。这 3 分钟的手动验证，能避开 80% 的配置错误和路径问题。模型再 fancy，如果settings.py里DEBUG=False但ALLOWED_HOSTS漏了域名，整个服务就是 500。希望帮到你。

本文还有配套的精品资源，点击获取