提到“django-flask基于python协同过滤算法的动漫视频在线平台”这个项目大多数人的第一反应是一个平台为什么要用两个Web框架Django和Flask不是二选一的关系吗这个项目最有意思的地方恰恰在这里——它用一套系统把两件事分开做了。Django管后台与内容面Flask管推荐与接口面中间用协同过滤算法把两者串联起来形成一套完整可跑的动漫视频推荐平台。这篇文章不打算只讲推荐算法公式的推导也不打算只讲框架怎么配而是完整复盘一个真实项目的搭建逻辑从数据模型设计、协同过滤选型与实现到Flask推荐服务落地再到线上效果验证的坑。适合正在做视频类网站、推荐系统或者只听说过协同过滤想把它真正用起来的开发者参考。里面所有的代码片段都来自我实际调试过的通用方案你可以直接拿去改。1. 双框架拆分同一个平台里 Django 和 Flask 谁在管什么1.1 管理面交给Django流量面交给Flask我先解释一下这个项目的名字逻辑。“django-flask”不是把两个框架揉进同一个进程里而是让它们各司其职、独立部署。这种架构在大部分中小型推荐类项目里其实非常舒服。Django负责的是“重”的那部分用户注册登录、后台管理、视频条目管理、分类标签管理、运营配置。为什么选Django因为它自带Admin后台、ORM、迁移机制、权限体系做内容管理系统几乎是开箱即用。在动漫视频平台里后台要经常上架新番、补充每部作品的标签和海报图、临时调整推荐位的置顶内容Django的Admin加上几个自定义ListFilter就能把这些运营操作全部覆盖比Flask手动搭后台省太多事。Flask负责的是“轻”的那部分推荐接口、相似动漫接口、用户行为埋点接收接口。它不关心后台长什么样只关心一件事——怎么把离线算好的推荐结果以最快的速度通过HTTP接口送给前端。Flask路由简洁、依赖少、启动快特别适合做这种功能单一的内部微服务。所以两条业务线天然分离Django团队可以按Django的方式迭代管理功能Flask服务则可以独立升级推荐逻辑而不牵连整个后台代码。两个服务通过HTTPJSON通信Django后台只负责写业务数据Flask推荐服务只负责读数据、算相似度、吐推荐结果数据层共用同一个MySQL库。1.2 这套架构真正解决的问题是什么如果只用Django把推荐逻辑写成一个View也完全能跑但你会被几个问题拖住第一Django启动时管理后台会加载一大堆中间件、信号、模板配置推荐接口每次请求都要穿一遍这层“外套”用Gunicorn压测时QPS上不去第二协同过滤模型每次重算后要重新加载如果和后台代码耦合在一起每一次模型更新都要重启整个后台服务风险很大第三如果以后推荐服务想加一个相似动漫的实时热搜池Flask这边只加一个路由就行完全不需要碰管理后台的代码。独立Flask服务在部署上的玩法也更灵活。Django服务用Gunicorn多worker跑在8001端口给运营和后台上传接口用Flask服务用轻量级Gunicorn跑在8002端口前面挂Nginx统一转发。两个端口互相不干扰推荐接口做缓存、限流、分版本A/B测试都只涉及Flask这一层。我还见过一种更彻底的做法把推荐服务做成独立的HTTP接口由Django后台在用户登录后通过服务器端请求去调用。也就是说前端的“猜你喜欢”请求打给DjangoDjango再转发给Flask。这个方案的好处是前端只需面对一个域名坏处是多了跳转延迟。我在项目里直接让前端请求Flask域名下的/recommend/api/v1/home绕过了Django因为推荐接口本身就是只读的没必要再经过后台鉴权层。如果是用户行为上报接口更要独立出来防止埋点流量把Django后台的慢查询拖垮。提示双框架不是必须的但一定要把“内容管理面”和“算法流量面”在逻辑上拆开。哪怕你只有一个Django项目也建议用不同的App、不同的数据库连接池来隔离这两部分。拆得早后面模型迭代和接口压测都轻松得多。2. 数据是第一位的动漫视频里的用户行为怎么建模2.1 三张核心表用户、视频条目、行为日志推荐系统里有句话叫“垃圾进垃圾出”。协同过滤算法本身不挑数据但你的数据如果没把“谁”、“看了什么”、“看得怎么样”记录完整后面所有计算都白搭。我设计的第一版数据结构包含三张表基本能覆盖动漫视频平台的所有需求。用户表在Django默认的User模型上扩展出user_id、nickname、signup_time就够真正重要的是行为表。视频条目表也不复杂除了基础标题、封面、播放地址之外一定要留几个关键字段category_id分类、tags标签建议用JSON存数组、publish_date上线时间、total_play_count总播放次数定期从统计表刷进去、item_embedding内容向量冷启动时会用到。用户行为表是推荐引擎的核心我设计成按天分区的日志表字段如下CREATE TABLE user_behaviors ( id BIGINT PRIMARY KEY AUTO_INCREMENT, user_id INT NOT NULL, item_id INT NOT NULL, behavior_type VARCHAR(16) NOT NULL, watch_ratio FLOAT DEFAULT 0.0, created_at DATETIME NOT NULL, KEY idx_user_time (user_id, created_at), KEY idx_item_time (item_id, created_at) );behavior_type这个字段非常关键我用它区分六种行为view播放、like点赞、collect收藏、share分享、comment评论、finish看到结尾。不要小看这一张表它决定了你能用什么算法、能算多复杂。为了避免多条行为记录膨胀我在Flask推荐服务里做实时点击流最终每秒写一次批量入库Django后台则提供一个同步接口把统计结果刷到total_play_count里。2.2 行为评分不是拍脑袋而是加权组合协同过滤算法要求输入一个“用户对物品的评分矩阵”但网站上几乎没有用户会主动给动漫打分。解决办法是把行为记录转换成0到100之间的隐式评分。我采用的评分公式是这样# 通过默认行为加权合成隐评分 def compute_rating(row): rating 0.0 if row[behavior_type] view: rating 30.0 * min(row[watch_ratio], 0.8) # 防滥用超过80%只记0.8 elif row[behavior_type] finish: rating 50.0 elif row[behavior_type] like: rating 45.0 elif row[behavior_type] collect: rating 60.0 elif row[behavior_type] share: rating 70.0 return min(rating, 100.0)为什么这么设计权重观看行为的权重最低但最容易产生适合做“广度”收藏和分享行为的成本更高说明用户确实喜欢权重相对高一些。考虑到大部分用户不会主动点赞所以我给collect和share定到60和70让重度用户的偏好信号能更明显地影响结果。watch_ratio乘0.8的上限是为了防止有人挂机播放刷数据这在实际线上几乎是必须处理的。2.3 真实调度链路和离线宽表光有行为日志表还不够我在Django后台新建了一个定时任务每小时做一次聚合把一天内的所有行为按“用户-物品”维度合成一张推荐宽表user_item_rating字段是user_id、item_id、rating、update_time。这张宽表就是离线协同过滤的输入。不要直接拿明细日志跑算法因为日志量大重复计算的时间和资源都不可控。先聚合出宽表再在下一次推荐模型重算前做快照是我跑下来最稳的路子。# Django管理命令里的聚合逻辑: python manage.py build_rating_matrix from django.core.management.base import BaseCommand from recommend.models import UserBehavior, UserItemRating from django.db.models import Sum class Command(BaseCommand): def handle(self, *args, **options): # 按最近30天窗口聚合窗口可配置 aggregated ( UserBehavior.objects .filter(created_at__gte2024-01-01) .values(user_id, item_id) .annotate(total_ratingSum(score)) ) UserItemRating.objects.all().delete() objs [ UserItemRating(user_idrow[user_id], item_idrow[item_id], ratingrow[total_rating]) for row in aggregated ] UserItemRating.objects.bulk_create(objs, batch_size2000)这个环节有个容易踩的坑直接把明细数据传给算法会导致内存占用巨大。我在实际项目中用了近三个月的行为数据明细表有几百万行但聚合后的宽表只有几十万行内存压力降了一个数量级。3. 为什么是协同过滤以及这边用哪个变种3.1 UserCF与ItemCF的取舍协同过滤大体分成两种基于用户的协同过滤UserCF和基于物品的协同过滤ItemCF。UserCF的思路是“和你口味相似的人看过什么就给你推荐什么”ItemCF的思路是“和你喜欢的物品像的物品就推荐给你”。在动漫视频平台里我的经验是ItemCF的推荐效果和工程可维护性都更好。原因是动漫内容本身是重口味偏好的用户群体可能出现很明显的圈层分化看热血番的人和看恋爱番的人重叠度很低。UserCF要求用户之间有足够多的共同行为才能算出相似性随着用户量增长用户相似矩阵是O(N^2)规模的线上维护成本很高。ItemCF计算的是物品间的相似度物品数量比用户数量小得多而且动漫内容相对稳定一个季度上线的番剧数量有限每周维护物品相似矩阵的开销很小。更实际的一点是ItemCF给出的推荐结果解释性很强“因为你喜欢《某战斗番》所以推荐《某同类番》”。这种逻辑直接展示在界面上用户更容易接受。很多人第一次写协同过滤一上来就写UserCF跑完看到结果却解释不清楚原理最终都不如ItemCF落地顺畅。3.2 从用户行为矩阵到相似度矩阵的计算我把离线训练流程拆成三个步骤读宽表、算物品相似度矩阵、生成Top-N推荐。第一步把宽表变成稀疏矩阵。用pivot_table很直观但真实数据里有几万个用户和几千个物品直接转成稠密矩阵会有大量NaN。建议用pandas的pivot_table矩阵然后填充0或者直接用scipy.sparse.csr_matrix来建矩阵。第二步对物品两两计算余弦相似度。第三步对用户已经看过的物品找出相似度最高的若干物品按加权得分排序输出。# 简化版物品协同过滤核心代码 import numpy as np import pandas as pd from sklearn.metrics.pairwise import cosine_similarity # df_rating: DataFrame, 列包括 [user_id, item_id, rating] def build_item_similarity(df_rating): pivot df_rating.pivot_table(indexuser_id, columnsitem_id, valuesrating).fillna(0) # item_sim形状: (n_items, n_items) item_sim cosine_similarity(pivot.T, dense_outputFalse) item_ids pivot.columns.tolist() sim_df pd.DataFrame(item_sim, indexitem_ids, columnsitem_ids) return sim_df def recommend_for_user(user_id, seen_items, sim_df, df_rating, top_n20): scores {} for item in seen_items: if item not in sim_df.index: continue sim_items sim_df[item].sort_values(ascendingFalse)[1: 51] # 每部只取最相似的50部 for sim_item, sim_score in sim_items.items(): if sim_item in seen_items: continue scores[sim_item] scores.get(sim_item, 0.0) sim_score ranked sorted(scores.items(), keylambda x: x[1], reverseTrue)[:top_n] return [item_id for item_id, _ in ranked]这段代码看起来逻辑很清晰但直接上线会有一个问题评分矩阵里填0的全是“缺失值”而不是“真实评分0”。比如用户从没看过的动漫会被当成负样本处理这会严重污染相似度计算。我实验后的修正方案是只在两个物品被至少一个共同用户评分过时才计算它们的相似度否则直接置为0。使用cosine_similarity和csr_matrix的朋友要注意这里在实际项目里我倾向于保留稀疏矩阵本身而不是把缺失位置变成0。不过这个简化版本做原型验证、跑通流程已经足够。实际项目中我会用scipy.sparse代替稠密矩阵使用sklearn的工具也需要留意稀疏矩阵的兼容性。3.3 矩阵分解SVD作为兜底方案纯粹基于共现物品的ItemCF遭遇数据稀疏时很多用户只对一两部动漫有行为共现矩阵太稀薄推荐结果会退化。我后来又接了一套基于隐语义的SVD方案思路是把用户-物品评分矩阵分解成用户隐向量矩阵和物品隐向量矩阵维度一般取80到120。SVD的优点是能挖掘用户和物品之间隐藏的关联哪怕没有直接共现也能通过共享的隐因子关联起来。在Python里实现SVD最省事的方式是用surprise库它内置了SVD模型原生支持训练集切分和RMSE评估from surprise import SVD, Dataset, Reader, accuracy from surprise.model_selection import train_test_split reader Reader(line_formatuser item rating, rating_scale(0, 100)) data Dataset.load_from_df( df_rating[[user_id, item_id, rating]].astype({ user_id: int, item_id: int }), reader ) trainset, testset train_test_split(data, test_size0.2) algo SVD(n_factors80, n_epochs20, lr_all0.005, reg_all0.02) algo.fit(trainset) predictions algo.test(testset) rmse accuracy.rmse(predictions, verboseFalse)SVD算出的预测分可以直接作为推荐分数也可以用于第二阶段的冷启动补充当ItemCF因为相似物品太少找不出结果时就用SVD的预测Top-N去填补。实际项目通常是两套并行跑然后做融合排序ItemCF候选集作为主体SVD预测分作为一个特征参与排序。最容易翻车的点是SVD对超参数比较敏感n_factors太大容易过拟合太小则丢失语义我在80-120之间测试下来效果最稳。4. 算法跑通只是开始冷启动、稀疏矩阵与热门偏差4.1 冷启动没有行为的新视频怎么推荐新上线的动漫在一周内几乎没有任何用户行为ItemCF和SVD都拿它没办法。解决冷启动不能靠协同过滤本身我采用的方案是“基于内容的相似补充”。在新番发布时引入两个字段类别、标签。比如一部新番标签为“战斗、热血、漫画改编”就把它和已有这些标签且历史上表现较好的动漫建立关联直接放进推荐候选集里。这部分逻辑在Django后台比较容易实现发布新番时表单里填写标签保存时自动计算一次“内容相似TopN”把相似结果写进content_similar表。Flask推荐接口发现协同过滤的候选集对某个用户不足20条时马上从内容相似表里拿项来补充。冷启动里还有一个被很多人忽略的小细节新番本身自带热度权重。我会在推荐排序时给新上线的物品加一个hot_bonus让它们有机会被用户发现这个值随时间衰减一周后归零。否则协同过滤推荐永远是老番的天下新内容几乎没有曝光机会。4.2 稀疏矩阵评分矩阵里全是空洞动漫视频平台里绝大多数用户只看过不到5部作品。这些用户共同评分过的作品交集非常小物品相似度计算出来往往是个位数。面对这样的矩阵硬算全院路线的余弦相似度就会遇到“共有0个共同评分”的尴尬。我采用的策略是“门槛过滤”构建相似度矩阵时只保留那些至少有3个共同评分用户的物品对。如果条件太严格可以降到2个但不要降到1个。只有1个共同用户时相似度波动极大体验很飘。对评分很少的用户比如少于3条行为的用户直接不要进入个性化推荐流程改用热门池和内容相似推荐。这样既保证了个性化推荐的稳定性也节约了推荐服务的计算资源。4.3 热门偏差推荐越来越像排行榜这是所有推荐系统都会遇到的问题。ItemCF天然倾向于把高热度物品和其他物品的相似度都算得很高导致推荐结果慢慢变成“大家都在看的榜单”失去了个性化。我在实验里发现如果不做任何处理在首页“猜你喜欢”里排第1名的推荐和用户历史喜欢的物品相似度很高但也很可能是近期全网播放量第一的番。处理手段有三种热度稀释、相似度归一化、以及推荐结果多样性约束。热度稀释的做法比较直接在排序阶段对每一个候选物品的最终得分乘以一个惩罚系数。punish_score score / (1 alpha * log(1 total_play_count))alpha取0.1到0.3之间。播放量越高得分被压低得越多。归一化则是把每行相似度除以行内最大值让相似度分量在0-1之间避免高分物品在某些物品上贡献过大。多样性约束更简单同一个系列或同一个标签的动漫在Top-N里最多出现2部确保推荐结果不局限于单一类型。实际线上我测得比较有效的是“打败30%的热门”也就是在排序前先把每部候选动漫的播放量分位数算出来把播放量打分前30%的候选稍微降权。5. 把推荐服务做成Flask接口的工程细节5.1 模型文件存哪里、什么时候加载离线阶段算完的物品相似度矩阵、SVD模型参数不能全部放在内存里算完就丢。我在Django端用一个管理命令把推荐模型文件导出为二进制格式例如item_sim.npz、svd_model.pklFlask服务启动时用模块级全局变量一次性加载到内存。第一次启动加载几十兆的矩阵可能需要几秒但加载之后每个请求都在内存里直接查表性能很好。保存相似度矩阵时我一般只保留每个物品最相似的50个邻居其他直接扔掉。一个5000物品的矩阵完整存下来可能几十MB只存Top50邻居之后文件能压缩到几MB加载速度也快得多。# 保存相似度Top50邻居 def save_sim_topk(sim_df, topk50): topk_dict {} for item_id in sim_df.index: row sim_df.loc[item_id].sort_values(ascendingFalse) row row.drop(indexitem_id) # 去掉自己和自己的相似度 topk_dict[item_id] row.iloc[:topk].to_dict() return topk_dict5.2 推荐接口设计与缓存策略Flask推荐接口不要做太重的事情。我的接口设计是GET /recommend/api/v1/home?user_id123首页个性化推荐返回20条动漫GET /recommend/api/v1/similar?item_id456视频详情页的“相似番剧推荐”POST /recommend/api/v1/behavior接收行为上报埋点首页推荐是重灾区。用户每次刷新都完整跑一遍候选加打分会很浪费我加了两层缓存。第一层是Redis缓存同一个user_id的推荐结果缓存5到10分钟命中就直接返回第二层是Flask进程内缓存用LRU字典缓存最近1000个用户的推荐结果TTL 60秒。from flask import Flask, jsonify, request from functools import lru_cache from redis import Redis app Flask(__name__) cache Redis.from_url(redis://localhost:6379/0) lru_cache(maxsize1000) def get_top_n_cached(user_id, top_n): # 这里从内存中的相似矩阵和用户评分里算推荐 return compute_recommendations(user_id, top_n) app.route(/recommend/api/v1/home) def recommend_home(): user_id request.args.get(user_id, typeint) if not user_id: return jsonify({error: missing user_id}), 400 cache_key frec:home:{user_id}:v1 cached cache.get(cache_key) if cached: return jsonify({recommendations: cached.split(,)}), 200 items get_top_n_cached(user_id, 20) cache.set(cache_key, ,.join(map(str, items)), ex600) return jsonify({recommendations: items}), 200这里有个挺容易忽略的点多worker进程下每个Flask进程都会各自加载一份模型字典Redis缓存是共享的但进程内缓存各算各的。如果只依赖进程内缓存后端的压力不会减太多。所以我把Redis当成唯一主缓存进程内LRU只是用来挡住同一秒内的重复请求。还有一点就是必须给缓存键加版本号例如v1模型重算之后可以平滑切新版本不用等待旧缓存过期。5.3 离线训练与在线推荐的节奏推荐模型不需要实时更新也不能一个月不更新。我定的节奏是这样的午夜1点用Django管理命令拉取最近30天行为日志聚合出宽表跑ItemCF和SVD生成Top50邻居矩阵和隐向量导出文件。凌晨3点Flask服务平滑重载模型文件。重载策略是读文件生成新字典然后通过全局变量一次性替换旧引用而不是重启进程这样可以避免服务中断。说起来容易第一次实现时还是踩了坑。我直接用的pickle序列化模型然后想在Flask运行中覆盖变量结果多个worker进程的旧变量还留在老进程里。后来我换成了文件监听机制Flask进程监控一个model_version.json发现版本号变化就重新加载新文件。这种方式我用下来最省心。6. 上线之后的验证路径从点击率到完播率6.1 第一版评估指标踩过的坑在实验室里我拿历史数据做切分算出SVD的RMSE只有20左右ItemCF的召回率也不错。实际上线后首页推荐位的点击率并没有比纯热门榜单好看多少。问题出在“评估指标和业务指标对不上”。协同过滤的离线指标主要衡量“预测分数准不准”但用户是不是真的喜欢推荐内容更适合看点击率、完播率、收藏转化率。一次点击只能说明封面吸引了用户不一定说明内容匹配。所以我在Flask推荐接口里做了埋点记录每次曝光、点击、播放完成三个事件统计三大指标指标计算方式业务含义点击率推荐位被点击次数 / 曝光次数推荐结果对用户的吸引力完播率看到结尾的播放次数 / 总播放次数内容质量和匹配度收藏率被收藏次数 / 点击次数用户对推荐结果的认可程度第一版推荐上线后点击率有提升但完播率低迷。后来测试发现很多高相似度的作品属于同一系列或类似画风用户点进去发现剧情不对胃口。后来我在排序阶段把“作品更偏大众向还是偏小众向”这个属性加了进去通过调低小众作品的权重完播率才慢慢升上来。6.2 A/B测试与调整方向做推荐系统一定要有A/B实验习惯。我用了一个很简单的分流方案按user_id模100分流比如尾号0到49的流量走新版推荐逻辑尾号50到99保持旧版。观察周期至少一周比较点击率和完播率再判断新版是否投入使用。我调整最多的参数是三个地方候选取样范围、相似度TopN数量、惩罚系数alpha。候选取样范围决定了每次推荐在多少部动漫里选太窄会让结果单一太宽会增加噪声TopN邻居的数量决定了推荐的多样性50到80之间的邻居数表现得比较平衡alpha则需要按用户活跃度分开调。整个项目从零到上线跑通框架层面没有特别高深的地方真正的复杂度都集中在数据构造和推荐结果质量的控制上。如果你也想复刻这个项目我建议第一步先把Django后台的用户行为和视频条目数据模型建好第二步用Jupyter把相似矩阵的可视化算出来确认推荐结果看起来合理再考虑上Flask接口。模型和框架都是工具数据要是乱了再好的算法也救不回来。
