5种实用推荐算法搞定新用户冷启动问题（附Python代码）

5种实用推荐算法搞定新用户冷启动问题（附Python代码）

新用户第一次打开你的App，面对琳琅满目的商品或海量的内容，推荐系统该给他看什么？这几乎是每个算法工程师和产品经理都会遇到的经典难题——新用户冷启动。没有点击历史，没有购买记录，甚至连浏览行为都为零，传统的协同过滤算法在这里完全失灵。对于资源有限的创业团队或中小型公司来说，这个问题尤为棘手：既没有海量数据去训练复杂的深度学习模型，又需要在第一时间抓住用户，避免其因“无感”而流失。

这篇文章就是为你准备的实战指南。我们不谈空洞的理论，只聚焦于能快速落地、直接复用的代码级解决方案。无论你是在搭建一个电商平台、一个内容社区，还是一个在线教育应用，下面这五种策略及其附带的Python代码，都能帮你迅速搭建起冷启动推荐的第一道防线，让新用户从“陌生人”变成“回头客”。

1. 冷启动的本质与我们的工具箱

在深入代码之前，我们得先统一认知：冷启动推荐的核心目标是什么？我认为，它不在于追求极致的个性化精度——这在初期既不可能也不经济——而在于快速建立用户与系统之间的有效连接。这个连接可以是让用户产生第一次点击、第一次收藏，或者仅仅是让他觉得“这个应用懂我”。

因此，我们的策略工具箱需要满足几个特性：实现简单、响应快速、可解释性强。基于这个思路，我为你筛选并实战了五种最常用、也最有效的策略。它们各有适用场景，你可以根据自身业务特点进行组合或选择。

注意：冷启动策略并非一成不变。随着用户产生初始行为，系统应能平滑过渡到更复杂的个性化推荐模型。因此，在设计之初就考虑好策略的“退出机制”和“数据埋点”至关重要。

下面这个表格快速对比了我们将要详述的五种策略的核心思想、优缺点及典型适用场景，方便你快速建立整体认知：

2. 策略一：万金油——全局热门推荐

当你对新用户一无所知时，把最受欢迎的东西给他看，总不会错得太离谱。这不仅是技术上的“保底”策略，更是产品逻辑上的必然选择：热门内容反映了社区的共识和潮流，能帮助新用户快速理解平台调性。

实现起来，它简单到令人发指。核心就是维护一个全局的物品热度榜。热度可以定义为过去一段时间内的点击量、播放量、购买量，或是综合了时间衰减的加权分数。对于冷启动，我们通常先从最简单的计数开始。

from collections import Counter, defaultdict from datetime import datetime, timedelta import pandas as pd class GlobalPopularityRecommender: """ 全局热门推荐器 维护一个基于时间窗口的热门物品列表 """ def __init__(self, time_window_hours=24): self.time_window = timedelta(hours=time_window_hours) # 使用字典记录每个物品在时间窗口内的交互事件 self.item_counter = Counter() # 使用队列记录事件的时间，用于过期清理（简单示例，生产环境可用Redis等） self.event_queue = [] # 元素为 (timestamp, item_id) def add_interaction(self, item_id, timestamp=None): """记录一次用户与物品的交互（点击、购买等）""" if timestamp is None: timestamp = datetime.now() self.event_queue.append((timestamp, item_id)) self.item_counter[item_id] += 1 # 清理过期事件（这里为简化逻辑，实际生产环境需要更高效的定期清理） self._clean_old_events(timestamp) def _clean_old_events(self, current_time): """清理超出时间窗口的旧事件""" cutoff_time = current_time - self.time_window while self.event_queue and self.event_queue[0][0] < cutoff_time: old_timestamp, old_item_id = self.event_queue.pop(0) self.item_counter[old_item_id] -= 1 if self.item_counter[old_item_id] == 0: del self.item_counter[old_item_id] def recommend(self, top_k=10, exclude_items=None): """为新用户生成热门推荐""" if exclude_items is None: exclude_items = set() # 获取top_k的热门物品，并排除用户可能已经看过的 valid_items = [(item, count) for item, count in self.item_counter.most_common() if item not in exclude_items] recommended = [item for item, _ in valid_items[:top_k]] return recommended # 模拟使用示例 if __name__ == "__main__": # 初始化推荐器，时间窗口设为1小时（模拟实时性要求高的场景） recommender = GlobalPopularityRecommender(time_window_hours=1) # 模拟一段时间内的用户交互数据流 mock_interactions = [ (1, datetime.now() - timedelta(minutes=30)), (2, datetime.now() - timedelta(minutes=25)), (1, datetime.now() - timedelta(minutes=20)), (3, datetime.now() - timedelta(minutes=15)), (2, datetime.now() - timedelta(minutes=10)), (3, datetime.now() - timedelta(minutes=5)), (4, datetime.now()), ] for item_id, ts in mock_interactions: recommender.add_interaction(item_id, ts) # 为新用户推荐 new_user_recs = recommender.recommend(top_k=3) print(f"基于全局热门的推荐结果: {new_user_recs}") # 输出可能为: [3, 2, 1] (根据实时计数排序)

这个类的设计考虑了实时性和可扩展性。time_window参数允许你定义热度的计算周期，比如“24小时热榜”或“实时热榜”。add_interaction方法模拟了线上实时日志的流入。在生产环境中，event_queue和item_counter应替换为Redis的Sorted Set等数据结构，以支持高并发和分布式场景。

几个关键点：

• 冷启动问题：系统刚上线时，item_counter为空。常见的解决方案是导入一批历史数据“预热”，或设置一个默认的流行物品列表作为fallback。
• 流行度偏见：长期只推荐热门物品会导致马太效应，让小众优质内容没有曝光机会。因此，它主要作为新用户的初始策略，一旦用户有了行为，就应降低其权重。

3. 策略二：初现个性——基于内容/元数据的推荐

如果用户在注册时勾选了“科技”、“健身”、“美妆”等兴趣标签，我们就有了撬动个性化推荐的第一个支点。基于内容的推荐（Content-Based Filtering）通过比较用户兴趣画像与物品的内容特征（元数据）来找到匹配项。

这里的关键在于特征表示和相似度计算。对于文本类元数据（如物品描述、标签集合），TF-IDF + 余弦相似度是经典且有效的组合。

import pickle from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity import numpy as np class ContentBasedRecommender: """ 基于内容/元数据的推荐器 适用于物品有文本描述、标签，且用户有显式兴趣画像的场景 """ def __init__(self): self.vectorizer = TfidfVectorizer(max_features=1000, stop_words='english') self.item_ids = None self.tfidf_matrix = None self.item_metadata = {} # item_id -> 文本元数据 def fit(self, item_metadata_dict): """ 训练TF-IDF向量化器 :param item_metadata_dict: 字典，{item_id: "标签1 标签2 描述文本"} """ self.item_metadata = item_metadata_dict self.item_ids = list(item_metadata_dict.keys()) corpus = [item_metadata_dict[iid] for iid in self.item_ids] # 学习词汇表并转换所有物品文本 self.tfidf_matrix = self.vectorizer.fit_transform(corpus) print(f"TF-IDF模型训练完成，词汇表大小: {len(self.vectorizer.vocabulary_)}") def recommend_for_user(self, user_profile_text, top_k=5, exclude_items=None): """ 为用户推荐物品 :param user_profile_text: 用户兴趣文本，如 "科技 编程 人工智能" :param top_k: 返回推荐数量 :param exclude_items: 需要排除的物品ID列表 :return: 推荐的物品ID列表 """ if exclude_items is None: exclude_items = set() if self.tfidf_matrix is None: raise ValueError("请先调用 fit() 方法训练模型。") # 将用户兴趣文本转换为TF-IDF向量 user_vector = self.vectorizer.transform([user_profile_text]) # 计算与所有物品的余弦相似度 cosine_sim = cosine_similarity(user_vector, self.tfidf_matrix).flatten() # 构建(物品ID, 相似度分数)的列表 item_score_pairs = list(zip(self.item_ids, cosine_sim)) # 按相似度降序排序，并排除指定物品 sorted_pairs = sorted(item_score_pairs, key=lambda x: x[1], reverse=True) recommendations = [] for item_id, score in sorted_pairs: if item_id not in exclude_items: recommendations.append(item_id) if len(recommendations) >= top_k: break return recommendations def save_model(self, filepath): """保存模型到文件""" with open(filepath, 'wb') as f: pickle.dump({ 'vectorizer': self.vectorizer, 'item_ids': self.item_ids, 'item_metadata': self.item_metadata }, f) @classmethod def load_model(cls, filepath): """从文件加载模型""" with open(filepath, 'rb') as f: data = pickle.load(f) recommender = cls() recommender.vectorizer = data['vectorizer'] recommender.item_ids = data['item_ids'] recommender.item_metadata = data['item_metadata'] # 需要重新计算TF-IDF矩阵 corpus = [recommender.item_metadata[iid] for iid in recommender.item_ids] recommender.tfidf_matrix = recommender.vectorizer.transform(corpus) return recommender # 实战示例 if __name__ == "__main__": # 模拟物品元数据（通常是来自数据库的标签或描述） items = { 101: "智能手机 科技 苹果 iOS 摄影", 102: "笔记本电脑 科技 编程 开发 轻薄", 103: "科幻小说 文学 太空 冒险", 104: "哑铃 健身 家庭 运动 器材", 105: "咖啡机 厨房 家电 美式 意式", 106: "蓝牙耳机 科技 音乐 降噪 无线", 107: "瑜伽垫 健身 女性 居家 运动", } # 初始化并训练推荐器 cbr = ContentBasedRecommender() cbr.fit(items) # 场景1：用户注册时选择了兴趣标签 user_tags = "科技 编程 开发" recs = cbr.recommend_for_user(user_tags, top_k=3) print(f"兴趣标签为'{user_tags}'的用户，推荐结果: {recs}") # 预期会推荐 102（笔记本电脑），可能还有101（手机）和106（耳机） # 场景2：用户填写了个人简介 user_bio = "我喜欢在家健身和阅读科幻小说" recs2 = cbr.recommend_for_user(user_bio, top_k=2) print(f"个人简介为'{user_bio}'的用户，推荐结果: {recs2}") # 预期会推荐 104（哑铃）、107（瑜伽垫）或103（科幻小说）

这个实现提供了一个可持久化的模型。fit方法通常在后台离线运行，处理全量物品元数据。线上服务时，recommend_for_user的计算非常快，因为TF-IDF矩阵是预先计算好的，只需要做一次向量化和一次矩阵乘法。

进阶思考：

• 特征工程：除了文本，如何处理类别特征（如品牌、颜色）和数值特征（如价格、评分）？你可以将它们进行one-hot编码或分桶后，与TF-IDF向量拼接，形成更丰富的物品向量。
• 冷启动中的冷启动：如果物品也是新的（没有交互数据），基于内容的推荐是物品冷启动的天然解决方案。
• 兴趣泛化：用户标签“科技”可能对应物品的“智能手机”、“笔记本电脑”、“智能手表”等多个相关特征，TF-IDF能很好地捕捉这种基于关键词的关联。

4. 策略三：拿来即用——基于注册信息的规则推荐

这是最直接、最快速的个性化方式。用户注册时填写的性别、年龄、城市、以及通过引导性问题收集的偏好（“您最喜欢的电影类型是？”），可以直接转化为过滤规则。

这种策略的本质是规则引擎。它不涉及复杂的模型，而是通过“如果-那么”的逻辑进行匹配。虽然简单，但在很多垂直领域效果立竿见影。

import pandas as pd from typing import Dict, Any, List class RuleBasedRecommender: """ 基于注册信息的规则推荐器 通过用户属性直接匹配物品属性 """ def __init__(self, items_dataframe: pd.DataFrame): """ :param items_dataframe: 包含物品所有属性的DataFrame。 例如列: ['item_id', 'category', 'sub_category', 'city', 'target_gender', 'min_age', 'max_age'] """ self.items_df = items_dataframe.copy() # 确保必要的列存在，这里只是示例 self._validate_columns() def _validate_columns(self): required_cols = {'item_id'} if not required_cols.issubset(self.items_df.columns): missing = required_cols - set(self.items_df.columns) raise ValueError(f"物品数据框缺少必要列: {missing}") def recommend(self, user_profile: Dict[str, Any], top_k: int = 10, rule_priority: List[str] = None) -> List[int]: """ 根据用户画像应用规则进行推荐 :param user_profile: 用户画像字典，如 {'city': '北京', 'gender': 'male', 'age': 25, 'preferred_category': '电子产品'} :param top_k: 返回推荐数量 :param rule_priority: 规则应用优先级列表，如 ['city', 'preferred_category', 'gender']，越靠前优先级越高。 :return: 推荐的物品ID列表 """ if rule_priority is None: rule_priority = ['preferred_category', 'city', 'gender', 'age'] # 默认优先级 filtered_df = self.items_df.copy() # 按优先级顺序应用过滤规则 for rule in rule_priority: user_value = user_profile.get(rule) if user_value is None: continue # 用户未提供此信息，跳过该规则 if rule == 'city': # 规则：推荐用户所在城市的物品（假设物品有'city'字段） if 'city' in filtered_df.columns: filtered_df = filtered_df[filtered_df['city'] == user_value] elif rule == 'gender': # 规则：推荐目标性别包含用户性别的物品 if 'target_gender' in filtered_df.columns: # target_gender 可能是 'all', 'male', 'female', 'unisex' filtered_df = filtered_df[ (filtered_df['target_gender'] == 'all') | (filtered_df['target_gender'] == user_value) ] elif rule == 'age': # 规则：推荐年龄区间包含用户年龄的物品 if all(col in filtered_df.columns for col in ['min_age', 'max_age']): filtered_df = filtered_df[ (filtered_df['min_age'] <= user_value) & (filtered_df['max_age'] >= user_value) ] elif rule == 'preferred_category': # 规则：推荐用户偏好类别的物品 if 'category' in filtered_df.columns: filtered_df = filtered_df[filtered_df['category'] == user_value] # 可以继续添加更多规则... # 如果过滤后结果已少于top_k，可以提前终止或降低后续规则严格度（这里简化处理） if len(filtered_df) <= top_k: break # 排序：可以按销量、评分、上架时间等排序，这里假设有'score'列 sort_column = 'score' if 'score' in filtered_df.columns else filtered_df.columns[1] recommended_df = filtered_df.sort_values(by=sort_column, ascending=False).head(top_k) return recommended_df['item_id'].tolist() # 模拟数据与实战 if __name__ == "__main__": # 模拟物品数据 data = { 'item_id': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 'category': ['电子产品', '服装', '电子产品', '图书', '服装', '电子产品', '图书', '运动', '运动', '美妆'], 'city': ['北京', '上海', '北京', '全国', '广州', '深圳', '全国', '北京', '上海', '全国'], 'target_gender': ['male', 'female', 'all', 'all', 'female', 'male', 'all', 'all', 'all', 'female'], 'min_age': [18, 16, 12, 10, 20, 18, 8, 15, 15, 18], 'max_age': [60, 50, 70, 80, 50, 60, 16, 50, 50, 50], 'score': [4.8, 4.5, 4.2, 4.9, 4.3, 4.7, 4.1, 4.6, 4.4, 4.0] # 综合评分 } items_df = pd.DataFrame(data) # 初始化推荐器 rule_rec = RuleBasedRecommender(items_df) # 新用户画像 new_user = { 'city': '北京', 'gender': 'male', 'age': 25, 'preferred_category': '电子产品' } # 进行推荐 recommendations = rule_rec.recommend(new_user, top_k=5) print(f"基于规则的用户推荐结果: {recommendations}") # 预期结果：会优先筛选出类别为‘电子产品’且城市为‘北京’的物品，然后考虑性别年龄，最后按评分排序。 # 可能是 [1, 3, 6, ...] 具体取决于评分

这个实现展示了规则的链式过滤和优先级概念。在实际业务中，规则可能更复杂，例如“一线城市的女性用户，优先推荐轻奢品牌，其次是快时尚品牌”。你可以通过扩展recommend方法中的条件判断来实现。

提示：规则推荐系统维护成本较高。当规则过多时，容易产生冲突或导致推荐池过小。建议将规则逻辑配置化，存储在数据库或配置文件中，便于管理和A/B测试。

5. 策略四：动态平衡——探索与利用（Bandit）算法

前面的策略都是“静态”的，我们假设热门、内容或规则就是最好的。但究竟哪种策略对新用户最有效？不同用户群体可能有不同偏好。多臂老虎机（Multi-Armed Bandit）算法就是为了解决这类“探索-利用”困境而生：是继续利用当前表现最好的策略（利用），还是尝试其他可能更好的策略（探索）？

Epsilon-Greedy是最直观的Bandit算法。它以ε的概率进行随机探索，以1-ε的概率选择当前点击率最高的策略。

import numpy as np import json from dataclasses import dataclass, asdict from typing import List, Dict @dataclass class StrategyStat: """记录每个推荐策略的统计信息""" name: str trials: int = 0 # 尝试次数（曝光） successes: int = 0 # 成功次数（点击、购买等） @property def success_rate(self): return self.successes / self.trials if self.trials > 0 else 0.0 class AdaptiveBanditRecommender: """ 基于Bandit算法的自适应推荐策略选择器 用于在多种冷启动策略中动态选择最优策略 """ def __init__(self, strategy_names: List[str], epsilon: float = 0.1, decay_factor: float = 0.99): """ :param strategy_names: 可用的策略名称列表，如 ['popularity', 'content', 'rule'] :param epsilon: 探索概率，0.1表示10%的时间随机选择策略 :param decay_factor: ε衰减因子，每次更新后 epsilon = epsilon * decay_factor，让系统后期更倾向于利用 """ self.strategies = {name: StrategyStat(name=name) for name in strategy_names} self.epsilon = epsilon self.initial_epsilon = epsilon self.decay_factor = decay_factor self.total_trials = 0 def select_strategy(self) -> str: """根据epsilon-greedy算法选择一个策略""" if np.random.random() < self.epsilon: # 探索：随机选择一个策略 chosen = np.random.choice(list(self.strategies.keys())) # print(f"[探索] 选择了策略: {chosen}") else: # 利用：选择成功率最高的策略 # 处理从未尝试过的策略（成功率视为0），给它们一些机会 rates = {name: stat.success_rate for name, stat in self.strategies.items()} # 如果所有策略都未尝试，随机选一个 if all(rate == 0 for rate in rates.values()): chosen = np.random.choice(list(self.strategies.keys())) else: chosen = max(rates, key=rates.get) # print(f"[利用] 选择了策略: {chosen} (成功率: {rates[chosen]:.3f})") return chosen def update_feedback(self, strategy_name: str, success: bool): """更新选定策略的反馈结果""" if strategy_name not in self.strategies: raise ValueError(f"未知策略: {strategy_name}") stat = self.strategies[strategy_name] stat.trials += 1 self.total_trials += 1 if success: stat.successes += 1 # 可选：随着试验次数增加，逐渐减小探索概率epsilon self.epsilon = self.initial_epsilon * (self.decay_factor ** self.total_trials) def get_statistics(self) -> Dict: """获取当前所有策略的统计信息""" return {name: asdict(stat) for name, stat in self.strategies.items()} def save_state(self, filepath: str): """保存当前状态到文件""" state = { 'strategies': {name: asdict(stat) for name, stat in self.strategies.items()}, 'epsilon': self.epsilon, 'total_trials': self.total_trials } with open(filepath, 'w') as f: json.dump(state, f, indent=2) def load_state(self, filepath: str): """从文件加载状态""" with open(filepath, 'r') as f: state = json.load(f) self.strategies = {name: StrategyStat(**data) for name, data in state['strategies'].items()} self.epsilon = state['epsilon'] self.total_trials = state['total_trials'] # 模拟线上A/B测试环境 if __name__ == "__main__": # 假设我们有三种冷启动策略 bandit = AdaptiveBanditRecommender(strategy_names=['热门推荐', '内容推荐', '规则推荐'], epsilon=0.3) # 模拟1000个新用户到来的决策过程 for user_id in range(1000): # 1. 为当前用户选择一个策略 chosen_strategy = bandit.select_strategy() # 2. 模拟使用该策略进行推荐，并得到用户反馈（这里用随机数模拟） # 假设不同策略的真实成功率不同：热门0.1， 内容0.15， 规则0.12 true_success_rate = {'热门推荐': 0.10, '内容推荐': 0.15, '规则推荐': 0.12} # 模拟反馈，有一定概率成功 feedback_success = np.random.random() < true_success_rate[chosen_strategy] # 3. 更新Bandit算法的统计信息 bandit.update_feedback(chosen_strategy, feedback_success) # 每100个用户打印一次统计信息 if (user_id + 1) % 100 == 0: print(f"\n--- 第 {user_id + 1} 个用户后 ---") stats = bandit.get_statistics() for s_name, s_data in stats.items(): rate = s_data['successes'] / s_data['trials'] if s_data['trials'] > 0 else 0 print(f" 策略 '{s_name}': 尝试 {s_data['trials']:>3d} 次, 成功 {s_data['successes']:>3d} 次, 成功率 {rate:.3f}") print(f" 当前探索概率 epsilon: {bandit.epsilon:.3f}") # 最终，算法应该会更多地选择真实成功率最高的“内容推荐”策略

这个AdaptiveBanditRecommender类可以作为一个独立的策略调度器。在实际系统中，它不直接生成推荐结果，而是决定对当前新用户使用哪一种底层推荐策略（热门、内容或规则）。然后，再调用对应的策略生成具体物品列表。

关键优势：

• 数据驱动：自动根据线上反馈调整策略选择，无需人工预设权重。
• 适应变化：如果“内容推荐”策略因为标签数据质量下降而效果变差，Bandit算法会逐渐减少对其的选择。
• 平滑探索：即使某个策略初期表现不好，ε机制仍会给予其少量曝光机会，以应对环境变化。

生产环境注意：上述简单模拟忽略了上下文信息（如用户设备、时间）。更高级的算法如Contextual Bandit(如LinUCB) 会考虑这些特征，为不同情境下的用户选择不同策略，效果更好但实现也更复杂。

6. 策略五：强强联合——混合加权推荐

单一策略总有局限。更稳健的做法是融合多种策略的结果，取长补短。混合推荐的核心是设计一个合理的加权打分机制，将不同策略产生的物品列表或分数进行合并排序。

常见的混合方式有：

1. 加权分数融合：每个策略给物品一个分数，加权求和后排序。
2. 级联混合：先用一个策略筛选出一批物品，再用另一个策略对这批物品进行排序或过滤。
3. 切换混合：根据条件（如用户是否有标签）决定使用哪种策略。

这里我们实现一个最通用的加权分数融合模型。

from typing import List, Dict, Callable, Tuple import numpy as np class HybridWeightedRecommender: """ 混合加权推荐器 融合多个基础推荐策略的结果 """ def __init__(self, base_recommenders: Dict[str, Tuple[Callable, float]]): """ :param base_recommenders: 字典，{策略名: (推荐函数, 权重)} 推荐函数签名：func(user_context, top_n) -> List[int] 或 List[Tuple[int, float]] """ self.recommenders = base_recommenders def recommend(self, user_context: Dict, top_k: int = 10, score_aggregation: str = 'weighted_sum') -> List[int]: """ 生成混合推荐结果 :param user_context: 用户上下文，包含用户ID、画像、地理位置等，传递给各个基础推荐器 :param top_k: 最终返回的推荐数量 :param score_aggregation: 分数聚合方式，'weighted_sum' 或 'rank_aggregation' :return: 混合排序后的物品ID列表 """ all_item_scores = {} for strategy_name, (rec_func, weight) in self.recommenders.items(): try: # 调用基础推荐器，获取其推荐结果 # 基础推荐器可以返回物品ID列表，或(物品ID, 分数)列表 raw_result = rec_func(user_context, top_n=top_k*2) # 让每个策略多返回一些 if not raw_result: continue # 标准化处理：将结果转换为(物品ID, 分数)的列表 if isinstance(raw_result[0], (int, np.integer)): # 如果只返回ID列表，则为其生成排名分数（倒数排名） item_ids = raw_result # 排名分数：第1名得1分，第2名得0.9分... 或使用其他衰减函数 scores = [1.0 / (i + 1) for i in range(len(item_ids))] strategy_items = list(zip(item_ids, scores)) else: # 假设返回的是(item_id, score)元组列表 strategy_items = raw_result # 根据聚合方式，将分数累加到总池 for item_id, raw_score in strategy_items: normalized_score = raw_score * weight # 简单加权 if score_aggregation == 'weighted_sum': all_item_scores[item_id] = all_item_scores.get(item_id, 0.0) + normalized_score # 可以扩展其他聚合方式，如最大值、最小值等 except Exception as e: print(f"策略 '{strategy_name}' 执行失败: {e}") continue # 按总分排序，取top_k sorted_items = sorted(all_item_scores.items(), key=lambda x: x[1], reverse=True) final_recommendations = [item_id for item_id, _ in sorted_items[:top_k]] return final_recommendations # 定义几个简单的基础推荐策略函数（模拟） def popularity_recommend(user_context, top_n): """模拟全局热门推荐""" # 这里应该是真实的逻辑，例如查询热门榜 # 假设返回物品ID列表 popular_items = [1001, 1002, 1003, 1004, 1005, 1006, 1007, 1008] return popular_items[:top_n] def content_based_recommend(user_context, top_n): """模拟基于内容的推荐""" user_tags = user_context.get('tags', '') # 这里应该是真实的CB逻辑，返回带分数的列表 # 模拟结果 if '科技' in user_tags: return [(2001, 0.9), (2002, 0.85), (2003, 0.8)] elif '运动' in user_tags: return [(3001, 0.95), (3002, 0.88)] else: return [(2001, 0.7), (3001, 0.6)] def rule_based_recommend(user_context, top_n): """模拟基于规则的推荐""" user_city = user_context.get('city', '') # 模拟规则推荐结果 if user_city == '北京': return [(4001, 0.9), (4002, 0.8), (4003, 0.7)] else: return [(5001, 0.9), (5002, 0.75)] # 实战组装与使用 if __name__ == "__main__": # 1. 定义基础推荐器及其权重 # 权重需要根据A/B测试或业务经验调整。例如，初期可能更依赖热门，后期更依赖个性化。 base_strategies = { 'popularity': (popularity_recommend, 0.3), # 热门策略，权重0.3 'content': (content_based_recommend, 0.5), # 内容策略，权重0.5 'rule': (rule_based_recommend, 0.2) # 规则策略，权重0.2 } # 2. 创建混合推荐器 hybrid_rec = HybridWeightedRecommender(base_strategies) # 3. 为新用户生成推荐 new_user_context_1 = {'tags': '科技 编程', 'city': '北京'} result_1 = hybrid_rec.recommend(new_user_context_1, top_k=5) print(f"用户1（科技，北京）的混合推荐: {result_1}") # 结果会融合：热门物品(100x)、科技内容物品(200x)、北京规则物品(400x) new_user_context_2 = {'tags': '', 'city': '上海'} # 无标签用户 result_2 = hybrid_rec.recommend(new_user_context_2, top_k=5) print(f"用户2（无标签，上海）的混合推荐: {result_2}") # 结果会更依赖热门和规则（上海） # 4. 动态调整权重（模拟） # 假设通过Bandit算法或线上监控，我们发现内容推荐效果很好，可以动态调高其权重 print("\n--- 动态调整权重后 ---") base_strategies['content'] = (content_based_recommend, 0.7) # 内容权重从0.5调到0.7 base_strategies['popularity'] = (popularity_recommend, 0.2) # 热门权重相应降低 hybrid_rec_updated = HybridWeightedRecommender(base_strategies) result_3 = hybrid_rec_updated.recommend(new_user_context_1, top_k=5) print(f"用户1的新混合推荐: {result_3}")

这个混合框架的灵活性极高。你可以轻松地插入我们前面实现的任何一个推荐器（GlobalPopularityRecommender,ContentBasedRecommender,RuleBasedRecommender）作为基础组件。权重的设置是核心调优点，可以通过离线评估（如准确率、覆盖率）或线上A/B测试来确定，甚至可以接入上一节的Bandit算法进行动态权重调整。

工程化建议：

• 分数标准化：不同策略输出的分数可能量纲不同（如热门是点击次数，内容推荐是相似度0-1）。在加权前需要进行标准化（如Min-Max Scaling或Z-score），否则权重会失真。
• 降级策略：确保至少有一个策略（如全局热门）总能返回结果，作为系统最后的保障。
• 缓存：对于计算成本较高的策略（如内容推荐），可以对用户画像进行哈希，缓存其推荐结果，在一定时间内复用。

在实际项目中，我通常会先部署一个简单的混合推荐（比如热门+规则），快速上线。然后逐步接入内容推荐和Bandit调度器，通过数据迭代优化。记住，冷启动推荐系统的成功，三分靠算法，七分靠对业务的理解和快速迭代。这些代码为你提供了坚实的起点，但真正的优化，始于你对用户行为的深入分析和持续的实验。