推荐系统的极简实现：从协同过滤论文到可运行原型的转化路径 推荐系统的极简实现从协同过滤论文到可运行原型的转化路径一、引言提到推荐系统许多人想到复杂的深度学习模型和海量数据。但对于大多数创业产品一个简单的协同过滤就足够产生可感知的推荐效果。不需要GPU集群不需要分布式训练。本文不堆砌公式而是走通从白板到代码的完整路径。目标是让读者读完就能在自己的产品里嵌入基础推荐。行业案例某内容平台的推荐系统演进一家创业公司做技术内容推荐初期用户约5000人文章约2000篇。第一版推荐按发布时间排序。 newest first。数据用户平均阅读1.2篇文章/天留存率低。第二版推荐引入基于物品的协同过滤本文方案。做法记录用户的阅读、点赞、收藏行为。用本文的Item-Based协同过滤做推荐。上线后用户平均阅读量从1.2篇/天提升到2.1篇/天。7日留存率从31%提升到44%。关键发现协同过滤在新用户冷启动阶段效果差。新用户没有行为数据无法做个性化推荐。解决方案新用户用热门推荐全局点击TOP N积累10条以上行为数据后切换到协同过滤。第三版推荐引入内容特征文章标签、分类做混合推荐。做法协同过滤的推荐结果按内容相似度做重排序。避免推荐内容过于同质化。效果用户阅读多样性提升停留时间增加约25%。这个案例说明推荐系统不需要一开始就做深度学习。协同过滤对中小规模产品已经足够。等用户量到10万级以上再考虑更复杂的模型。二、原理协同过滤的两种范式两种范式的选择User-Based适合用户数 物品数的场景如付费产品。Item-Based适合物品数 用户数的场景如电商。混合模式先Item-Based做召回再按热度排序。本文选择Item-Based因为它计算量小冷启动友好。三、代码完整的协同过滤实现import math import heapq from dataclasses import dataclass, field from typing import List, Dict, Set, Tuple, Optional from collections import defaultdict dataclass class Rating: user_id: str item_id: str score: float # 1-5分或隐式反馈(0/1) timestamp: int class CollaborativeFilter: 基于物品的协同过滤 def __init__(self, k_neighbors: int 20): self.k k_neighbors # 用户→物品→评分 self.user_items: Dict[str, Dict[str, float]] defaultdict(dict) # 物品→用户→评分倒排索引 self.item_users: Dict[str, Dict[str, float]] defaultdict(dict) # 缓存的物品相似度 self._item_similarity: Dict[str, Dict[str, float]] {} def fit(self, ratings: List[Rating]): 训练构建用户-物品矩阵 for r in ratings: self.user_items[r.user_id][r.item_id] r.score self.item_users[r.item_id][r.user_id] r.score # 预计算物品相似度可选节省在线计算 print(f训练完成{len(self.user_items)}用户, f{len(self.item_users)}物品) def _cosine_similarity( self, item_a: str, item_b: str ) - float: 余弦相似度 users_a self.item_users.get(item_a, {}) users_b self.item_users.get(item_b, {}) common_users set(users_a.keys()) set(users_b.keys()) if len(common_users) 2: return 0.0 # 计算点积 dot_product sum( users_a[u] * users_b[u] for u in common_users ) # 计算模长 norm_a math.sqrt( sum(s * s for s in users_a.values()) ) norm_b math.sqrt( sum(s * s for s in users_b.values()) ) if norm_a 0 or norm_b 0: return 0.0 return dot_product / (norm_a * norm_b) def _adjusted_cosine_similarity( self, item_a: str, item_b: str ) - float: 修正余弦相似度减去用户平均分 users_a self.item_users.get(item_a, {}) users_b self.item_users.get(item_b, {}) common_users set(users_a.keys()) set(users_b.keys()) if len(common_users) 2: return 0.0 # 计算共同用户平均分 user_avg {} for u in common_users: ratings self.user_items[u].values() user_avg[u] sum(ratings) / len(ratings) if ratings else 0 num sum( (users_a[u] - user_avg[u]) * (users_b[u] - user_avg[u]) for u in common_users ) den_a math.sqrt(sum( (users_a[u] - user_avg[u]) ** 2 for u in common_users )) den_b math.sqrt(sum( (users_b[u] - user_avg[u]) ** 2 for u in common_users )) if den_a 0 or den_b 0: return 0.0 # 惩罚共同用户少的物品对 penalty min(len(common_users) / 10, 1.0) return (num / (den_a * den_b)) * penalty def get_similar_items( self, item_id: str, n: int None ) - List[Tuple[str, float]]: 获取Top N相似物品 if n is None: n self.k if item_id in self._item_similarity: return list(self._item_similarity[item_id].items())[:n] similarities [] for other in self.item_users: if other item_id: continue sim self._adjusted_cosine_similarity(item_id, other) if sim 0: similarities.append((other, sim)) # TopN top heapq.nlargest( n, similarities, keylambda x: x[1] ) self._item_similarity[item_id] dict(top) return top def recommend( self, user_id: str, n: int 10, exclude_interacted: bool True ) - List[Tuple[str, float]]: 为用户生成推荐 user_ratings self.user_items.get(user_id, {}) interacted set(user_ratings.keys()) candidates: Dict[str, float] defaultdict(float) # 遍历用户交互过的物品 for item_id, rating in user_ratings.items(): similar_items self.get_similar_items(item_id, self.k) for sim_item, similarity in similar_items: if exclude_interacted and sim_item in interacted: continue # 加权累积用户评分 × 物品相似度 candidates[sim_item] rating * similarity # 按得分降序 top heapq.nlargest( n, candidates.items(), keylambda x: x[1] ) return top def recommend_for_new_user( self, user_recent_items: List[str], n: int 10 ) - List[Tuple[str, float]]: 为新用户推荐冷启动 candidates: Dict[str, float] defaultdict(float) for item_id in user_recent_items: if item_id not in self.item_users: continue similar_items self.get_similar_items(item_id, 5) for sim_item, similarity in similar_items: candidates[sim_item] similarity top heapq.nlargest( n, candidates.items(), keylambda x: x[1] ) return top # 评估模块 class RecEvaluator: 推荐系统离线评估 staticmethod def precision_at_k( recommended: List[str], relevant: Set[str], k: int 10 ) - float: PrecisionK if not recommended: return 0.0 top_k recommended[:k] hits sum(1 for item in top_k if item in relevant) return hits / min(k, len(top_k)) staticmethod def recall_at_k( recommended: List[str], relevant: Set[str], k: int 10 ) - float: RecallK if not relevant: return 0.0 top_k recommended[:k] hits sum(1 for item in top_k if item in relevant) return hits / len(relevant) staticmethod def f1_at_k( recommended: List[str], relevant: Set[str], k: int 10 ) - float: F1K p RecEvaluator.precision_at_k(recommended, relevant, k) r RecEvaluator.recall_at_k(recommended, relevant, k) if p r 0: return 0.0 return 2 * p * r / (p r) staticmethod def hit_rate_at_k( recommended: List[str], relevant: Set[str], k: int 10 ) - float: HitRateK top_k recommended[:k] return 1.0 if any(item in relevant for item in top_k) else 0.0 # 部署封装 class RecommendationService: 推荐服务封装 def __init__(self, cf: CollaborativeFilter): self.cf cf self._cache: Dict[str, List[Tuple[str, float]]] {} def get_feed(self, user_id: str, page: int 1, size: int 20) - Dict: 获取用户信息流推荐 cache_key f{user_id}:{page}:{size} if cache_key in self._cache: return {items: self._cache[cache_key], source: cache} recs self.cf.recommend(user_id, nsize * page) start (page - 1) * size page_items recs[start:start size] self._cache[cache_key] page_items return {items: page_items, source: computed} def refresh(self, new_ratings: List[Rating]): 增量更新 self.cf.fit(new_ratings) self._cache.clear() # 清除缓存 # 使用示例 if __name__ __main__: # 构造训练数据 ratings [ Rating(u1, item_a, 5, 1000), Rating(u1, item_b, 3, 1001), Rating(u1, item_d, 4, 1002), Rating(u2, item_a, 4, 1003), Rating(u2, item_c, 5, 1004), Rating(u2, item_e, 2, 1005), Rating(u3, item_b, 4, 1006), Rating(u3, item_c, 3, 1007), Rating(u3, item_d, 5, 1008), Rating(u4, item_a, 5, 1009), Rating(u4, item_b, 4, 1010), Rating(u4, item_f, 4, 1011), ] cf CollaborativeFilter(k_neighbors10) cf.fit(ratings) # 为用户u1推荐 recs cf.recommend(u1, n5) print(推荐给 u1:) for item, score in recs: print(f {item}: {score:.3f}) # 相似物品 similar cf.get_similar_items(item_a, n3) print(f\n与 item_a 相似:) for item, sim in similar: print(f {item}: {sim:.3f}) # 评估 evaluator RecEvaluator() recommended_items [item for item, _ in recs] relevant {item_c, item_e, item_f} print(f\nPrecision5: {evaluator.precision_at_k(recommended_items, relevant):.3f}) print(fRecall5: {evaluator.recall_at_k(recommended_items, relevant):.3f})关键设计点修正余弦相似度减去用户平均分消除评分偏差。惩罚稀疏性共同用户少的物品对降低可信度。增量更新支持新数据追加缓存自动失效。完整评估Precision/Recall/F1/HitRate四指标。四、权衡协同过滤的边界数据稀疏问题。用户和物品很多但交互很少时矩阵极度稀疏。可以引入内容特征物品描述做混合推荐。冷启动问题。新物品没有交互数据无法被推荐。用热门推荐或内容相似度做冷启动策略。实时性 vs 计算量。全量重计算相似度耗时。生产环境可以在线用缓存相似度离线定时每天重算。可解释性。协同过滤是黑盒。如果产品需要推荐理由需要额外设计解释逻辑。取舍决策推荐系统方案的选择框架面对推荐需求如何选择方案根据用户规模和数据情况决定用户规模判断用户数1万协同过滤足够。本文方案可以直接用。用户数1万-10万协同过滤内容特征混合。效果比纯协同过滤好。用户数10万考虑深度学习方案如矩阵分解、神经网络。数据量够大模型才能训练好。数据情况判断行为数据稀疏人均交互5次协同过滤效果差。建议用内容推荐或热门推荐。行为数据充足人均交互20次协同过滤效果好。用户行为已经能反映偏好。有内容特征标签、分类做混合推荐。协同过滤捕捉协同信号内容特征捕捉相似度。实现成本判断低1-2周协同过滤。本文方案不到200行代码。中1-2个月协同过滤内容特征召回排序。需要特征工程。高3个月以上深度学习推荐系统。需要GPU、训练管道、模型服务。冷启动策略选择新用户热门推荐。用全局点击TOP N。新物品内容相似度。推荐和热门物品内容相似的物品。新系统先积累数据再做推荐。不要在没有数据时就上推荐系统。决策输出大多数创业产品本文的协同过滤方案已经足够。等用户量到10万级推荐效果遇到瓶颈再考虑更复杂的方案。不要为了用AI而用深度学习协同过滤对中小规模产品是最优解。五、总结协同过滤是实现推荐系统的最低门槛方案。本文从零实现了Item-Based协同过滤的完整链路包含训练、推荐、评估和服务封装。核心代码不到200行。生产化路线先用本文代码验证推荐效果1-2周。效果达标后不需要急着上深度学习。先优化数据质量、增加特征维度。只有协同过滤效果遇到瓶颈才值得投入更复杂的模型。