1. 项目概述：当推荐算法遇上短视频平台
去年帮朋友改造一个濒临倒闭的短视频平台时，我深刻体会到推荐系统对内容平台的重要性。这个基于协同过滤算法的短视频分享系统，采用Python+Django+Vue3技术栈实现，上线后用户停留时长直接提升了3倍。不同于简单的内容列表展示，系统能根据用户行为模式智能推荐可能感兴趣的视频内容。
核心架构分为三部分：后端使用Django处理业务逻辑和算法运算，前端用Vue3构建响应式界面，而协同过滤算法作为系统的"智能大脑"持续学习用户偏好。这种技术组合既保证了算法的高效执行，又提供了流畅的用户体验。
提示：协同过滤算法在资源占用和实时性方面需要特别注意，建议采用混合推荐策略平衡效果与性能
2. 核心技术解析
2.1 协同过滤算法实现
系统采用基于用户的协同过滤(UserCF)算法，核心公式如下：
code复制用户相似度 = Σ(用户A评分项 × 用户B评分项) / (√Σ用户A评分² × √Σ用户B评分²)
Python实现关键代码：
python复制def user_similarity(user1, user2):
# 获取共同评分项
common_items = set(user1['ratings'].keys()) & set(user2['ratings'].keys())
# 计算余弦相似度
numerator = sum(user1['ratings'][item] * user2['ratings'][item] for item in common_items)
denominator = (sqrt(sum(pow(user1['ratings'][item], 2) for item in user1['ratings'])) *
sqrt(sum(pow(user2['ratings'][item], 2) for item in user2['ratings'])))
return numerator / denominator if denominator != 0 else 0
实际应用中我们发现几个优化点：
- 对观看时长超过80%的视频自动记为5星
- 用户冷启动阶段采用热门内容填充
- 相似度计算采用离线批处理+实时增量更新
2.2 Django后端设计
模型层关键字段设计：
python复制class Video(models.Model):
title = models.CharField(max_length=200)
uploader = models.ForeignKey(User, on_delete=models.CASCADE)
tags = models.ManyToManyField('Tag')
view_count = models.IntegerField(default=0)
average_rating = models.FloatField(default=0)
class UserBehavior(models.Model):
user = models.ForeignKey(User, on_delete=models.CASCADE)
video = models.ForeignKey(Video, on_delete=models.CASCADE)
watch_duration = models.FloatField() # 观看时长占比
rating = models.FloatField(null=True, blank=True)
timestamp = models.DateTimeField(auto_now_add=True)
API接口采用DRF框架实现，关键端点包括：
/api/recommend/获取推荐列表/api/behavior/提交用户行为数据/api/video/{id}/视频详情获取
2.3 Vue3前端实现
推荐流组件核心逻辑：
javascript复制const fetchRecommendations = async () => {
try {
const response = await axios.get('/api/recommend/', {
params: {
limit: 10,
exclude: viewedVideos.value
}
})
videoList.value = response.data.results
} catch (error) {
console.error('获取推荐失败:', error)
// 降级方案：加载热门视频
loadFallbackContent()
}
}
性能优化技巧：
- 使用Intersection Observer实现懒加载
- 视频预加载3个后续内容
- 本地缓存已看过的视频ID
3. 系统实现关键步骤
3.1 数据准备与处理
构建推荐系统需要三类核心数据：
- 用户画像数据（注册信息、偏好设置）
- 内容特征数据（视频标签、分类、时长）
- 行为数据（播放、点赞、分享、评论）
我们使用Pandas进行数据预处理：
python复制def prepare_rating_matrix():
# 从数据库获取原始数据
behaviors = UserBehavior.objects.all().values()
df = pd.DataFrame.from_records(behaviors)
# 转换评分标准
df['rating'] = df.apply(lambda x:
x['rating'] if not pd.isnull(x['rating'])
else min(5, x['watch_duration'] * 5),
axis=1)
# 生成评分矩阵
rating_matrix = df.pivot_table(
index='user_id',
columns='video_id',
values='rating',
fill_value=0
)
return rating_matrix
3.2 推荐引擎实现
完整的推荐流程包含：
- 相似用户计算（离线，每日更新）
- 推荐生成（实时）
- 结果过滤（去重、多样性控制）
核心推荐类实现：
python复制class Recommender:
def __init__(self):
self.similarity_matrix = None
def train(self, rating_matrix):
"""计算用户相似度矩阵"""
n_users = rating_matrix.shape[0]
self.similarity_matrix = np.zeros((n_users, n_users))
for i in range(n_users):
for j in range(i+1, n_users):
sim = cosine_similarity(
rating_matrix.iloc[i].values,
rating_matrix.iloc[j].values
)
self.similarity_matrix[i][j] = sim
self.similarity_matrix[j][i] = sim
def recommend(self, user_id, k=5):
"""生成推荐"""
similar_users = np.argsort(
self.similarity_matrix[user_id])[::-1][1:11]
recommendations = defaultdict(float)
for sim_user in similar_users:
for item in rating_matrix.columns:
if rating_matrix.iloc[user_id][item] == 0 and \
rating_matrix.iloc[sim_user][item] > 0:
recommendations[item] += (
self.similarity_matrix[user_id][sim_user] *
rating_matrix.iloc[sim_user][item]
)
return sorted(recommendations.items(),
key=lambda x: x[1], reverse=True)[:k]
3.3 系统集成与部署
技术栈组合：
- 前端：Vue3 + Vite + Pinia
- 后端：Django + Django REST framework
- 数据库：PostgreSQL + Redis缓存
- 算法：Python + NumPy + Pandas
部署架构：
code复制客户端 → Nginx(静态文件) → Gunicorn(Django) → PostgreSQL
↘ Redis(缓存)
性能优化配置：
python复制# settings.py
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
}
}
}
# 推荐结果缓存1小时
RECOMMENDATION_CACHE_TIMEOUT = 3600
4. 实战经验与优化技巧
4.1 冷启动解决方案
新用户推荐策略：
- 基于内容特征推荐（使用视频标签匹配）
- 混合热门内容（周榜TOP100）
- 引导用户选择兴趣标签
实现代码：
python复制def cold_start_recommend(user=None, tags=[]):
if user and user.is_authenticated:
# 已登录但数据不足的用户
if len(user.behaviors.all()) < 5:
return hybrid_recommend(user, tags)
# 完全新用户
popular = Video.objects.filter(
created_at__gte=timezone.now()-timedelta(days=7)
).order_by('-view_count')[:10]
if tags:
content_based = Video.objects.filter(
tags__name__in=tags
).distinct().order_by('-view_count')[:5]
return list(chain(popular, content_based))
return popular
4.2 实时行为处理
使用Django Signals实现实时行为追踪：
python复制@receiver(post_save, sender=UserBehavior)
def update_recommendations(sender, instance, created, **kwargs):
if created and instance.watch_duration > 0.7:
# 重要行为触发实时更新
cache_key = f'rec_{instance.user_id}'
cache.delete(cache_key)
# 异步更新相似度矩阵
update_user_similarity.delay(instance.user_id)
Celery任务示例：
python复制@app.task
def update_user_similarity(user_id):
current_user = get_user(user_id)
similarity_model = load_similarity_model()
# 获取最近交互的20个用户
recent_partners = get_interaction_users(user_id, limit=20)
for partner in recent_partners:
new_sim = calculate_similarity(current_user, partner)
similarity_model.update(user_id, partner.id, new_sim)
similarity_model.save()
4.3 效果评估指标
我们监控的核心指标：
- 点击通过率（CTR）
- 平均观看时长
- 推荐内容多样性
- 用户留存率
评估代码示例：
python复制def calculate_diversity(recommendations):
"""计算推荐列表的标签多样性"""
tags = set()
for video in recommendations:
tags.update(tag.name for tag in video.tags.all())
return len(tags) / len(recommendations)
def evaluate_recommendation(user_id, recommendations):
ctr = UserAction.objects.filter(
user_id=user_id,
video_id__in=[v.id for v in recommendations],
action_type='click'
).count() / len(recommendations)
avg_duration = UserBehavior.objects.filter(
user_id=user_id,
video_id__in=[v.id for v in recommendations]
).aggregate(avg=Avg('watch_duration'))['avg'] or 0
diversity = calculate_diversity(recommendations)
return {
'ctr': ctr,
'avg_duration': avg_duration,
'diversity': diversity
}
5. 典型问题与解决方案
5.1 数据稀疏性问题
常见现象：
- 新用户得不到准确推荐
- 长尾视频很少被推荐
我们的解决方案：
- 混合推荐策略（协同过滤+内容过滤）
- 基于标签的相似视频扩展
- 适当引入随机探索机制
实现代码：
python复制def hybrid_recommend(user, k=10):
# 协同过滤结果
cf_recs = collaborative_filtering(user, k//2)
# 内容过滤补充
viewed_tags = Tag.objects.filter(
videos__behaviors__user=user
).distinct()
content_recs = Video.objects.filter(
tags__in=viewed_tags
).exclude(
behaviors__user=user
).order_by('-view_count')[:k//2]
# 合并并去重
combined = list(cf_recs) + list(content_recs)
return sorted(
list({v.id: v for v in combined}.values()),
key=lambda x: x.view_count,
reverse=True
)[:k]
5.2 系统性能优化
遇到的瓶颈：
- 用户增长后相似度计算变慢
- 实时推荐响应时间波动
优化措施：
- 分块计算相似度矩阵
- 引入近似最近邻(ANN)算法
- 分级缓存策略
优化后的相似度计算：
python复制def approximate_similarity(user_ids, rating_matrix, n_trees=10):
"""使用LSH森林加速相似用户查找"""
from sklearn.neighbors import LSHForest
lshf = LSHForest(n_estimators=n_trees)
lshf.fit(rating_matrix)
distances, indices = lshf.kneighbors(
rating_matrix.loc[user_ids],
n_neighbors=20
)
return {
user_id: [(indices[i][j], 1-distances[i][j])
for j in range(len(indices[i]))]
for i, user_id in enumerate(user_ids)
}
5.3 推荐多样性提升
常用技术：
- 基于聚类的多样性筛选
- 滑动窗口多样性控制
- 多目标优化排序
实现示例：
python复制def diversify_recommendations(recommendations, window_size=3):
"""滑动窗口多样性控制"""
final_recs = []
tags_in_window = set()
for i, rec in enumerate(recommendations):
rec_tags = {t.name for t in rec.tags.all()}
# 计算与当前窗口的标签重复度
overlap = len(rec_tags & tags_in_window) / len(rec_tags) if rec_tags else 0
if overlap < 0.5 or not final_recs:
final_recs.append(rec)
tags_in_window.update(rec_tags)
# 维护滑动窗口
if len(final_recs) > window_size:
removed_tags = {t.name for t in final_recs[-window_size-1].tags.all()}
tags_in_window -= removed_tags
if len(final_recs) >= len(recommendations):
break
return final_recs
在项目上线后的三个月内，我们持续优化推荐算法，最终实现了以下改进：
- 推荐内容点击率提升210%
- 用户平均停留时长从1.2分钟增加到3.8分钟
- 长尾视频曝光量增加5倍