目录一个关于“猜你喜欢”的真实困境第一部分冷启动——当系统对你一无所知1.1 什么是冷启动问题1.2 常见的冷启动解决思路1.3 代码实战一个轻量级的冷启动方案1.4 前沿进展大模型时代的冷启动第二部分兴趣迁移——用户是会变的2.1 兴趣迁移的三种模式2.2 解决兴趣迁移的主流方法2.3 代码实战带遗忘机制的序列推荐一个关于“猜你喜欢”的真实困境你有没有这样的体验第一次打开一个新闻App推荐给你的内容莫名其妙明明是个科技爱好者首页却堆满了娱乐八卦或者你最近开始关注健身可算法似乎永远停留在上个月你喜欢的美食视频上。这不是算法在故意跟你作对。这背后是两个困扰了推荐系统领域多年的核心难题——用户冷启动和兴趣迁移。作为一个写过几年推荐系统代码的人我见过太多团队在这两个问题上反复踩坑。今天我想用一篇足够长的文章把这两个问题彻底讲透。我们会聊最新的技术方案也会贴出可以跑的代码。如果你正在做推荐系统或者只是好奇算法是怎么“认识”你的这篇应该对你有帮助。第一部分冷启动——当系统对你一无所知1.1 什么是冷启动问题冷启动Cold Start指的是系统在面对新用户或新物品时由于缺乏历史交互数据无法做出有效推荐的问题。在新闻推荐场景下冷启动尤为棘手。一个用户第一次打开你的App你没有任何点击、停留时长、分享、评论记录。你只知道他的设备信息、大致地理位置如果授权了可能还有一个通过第三方SDK拿到的模糊画像。零样本学习的尴尬传统协同过滤在新用户身上完全失效因为用户-物品共现矩阵里新用户那一行全是0。1.2 常见的冷启动解决思路我梳理一下目前工业界比较成熟的几类方案方案一热门推荐Most Popular最朴素的做法。新用户看到的是全站点击率最高的新闻。优点是简单、冷启动阶段点击率有保底缺点是个性化为零长期用会赶走用户。方案二探索与利用Explore Exploit给新用户推荐的内容里一部分是热门利用已知信息一部分是随机或不确定性较高的内容探索用户偏好。经典算法如汤普森采样Thompson Sampling、UCBUpper Confidence Bound。方案三元学习Meta Learning近几年比较受关注的方向。核心思想是让模型学会“如何快速学习”。通过在海量用户的历史行为上预训练一个元模型新用户只需要几次点击模型就能快速适配。MAMLModel-Agnostic Meta-Learning是代表工作。方案四利用内容特征Content-based Cold Start不依赖用户历史而是分析新闻本身的文本、标题、类别、关键词。新用户点击了某篇关于“人工智能”的新闻后立即推荐其他含相同关键词的文章。1.3 代码实战一个轻量级的冷启动方案我们来实现一个结合了内容特征和汤普森采样的冷启动模块。场景假设用户第一次访问我们给他推荐5篇文章收集点击反馈后更新偏好。pythonimport numpy as np import pandas as pd from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity from collections import defaultdict import random class ColdStartNewsRecommender: def __init__(self, news_corpus): news_corpus: list of dict, each with keys: id, title, category, content self.news_corpus news_corpus self.news_ids [item[id] for item in news_corpus] # 构建TF-IDF特征用于内容相似度 self.tfidf TfidfVectorizer(max_features500, stop_wordsenglish) self.news_vectors self.tfidf.fit_transform( [item[title] item.get(content, )[:200] for item in news_corpus] ) # 汤普森采样参数每个新闻的Beta分布 (alpha, beta) # 对于新用户所有新闻初始化为 (1, 1) 即均匀先验 self.alpha np.ones(len(news_corpus)) self.beta np.ones(len(news_corpus)) # 用户短期兴趣向量初始为0 self.user_interest_vector None def recommend_with_thompson(self, n5): 汤普森采样推荐从Beta分布采样选n个新闻 samples np.random.beta(self.alpha, self.beta) # 排除已经推荐过的简化起见这里不处理重复推荐 top_indices np.argsort(samples)[-n:][::-1] return [self.news_corpus[i] for i in top_indices] def recommend_with_content(self, clicked_news_ids, n5): 基于内容相似度的推荐找到与用户点击历史最相似的新闻 if not clicked_news_ids: return self.recommend_with_thompson(n) # 找到点击新闻的索引 clicked_indices [self.news_ids.index(nid) for nid in clicked_news_ids if nid in self.news_ids] if not clicked_indices: return self.recommend_with_thompson(n) # 计算用户兴趣向量点击新闻向量的平均 user_vec np.mean(self.news_vectors[clicked_indices].toarray(), axis0) self.user_interest_vector user_vec # 计算与所有新闻的相似度 similarities cosine_similarity([user_vec], self.news_vectors)[0] # 排除已经点击过的新闻 for idx in clicked_indices: similarities[idx] -1 top_indices np.argsort(similarities)[-n:][::-1] return [self.news_corpus[i] for i in top_indices] def update_feedback(self, news_id, clicked): 更新反馈clicked1表示点击0表示曝光未点击 idx self.news_ids.index(news_id) if clicked: self.alpha[idx] 1 else: self.beta[idx] 1 def hybrid_recommend(self, clicked_history, n5, explore_ratio0.3): 混合推荐一部分用内容相似度利用一部分用汤普森采样探索 n_explore int(n * explore_ratio) n_exploit n - n_explore exploit_recs [] if clicked_history: exploit_recs self.recommend_with_content(clicked_history, n_exploit) else: exploit_recs self.recommend_with_thompson(n_exploit) explore_recs self.recommend_with_thompson(n_explore) # 去重合并 seen_ids set([r[id] for r in exploit_recs]) final_recs exploit_recs.copy() for rec in explore_recs: if rec[id] not in seen_ids: final_recs.append(rec) seen_ids.add(rec[id]) if len(final_recs) n: break return final_recs[:n] # 模拟数据 sample_news [ {id: n1, title: Google launches new AI model Gemini Ultra, category: tech, content: Google announced...}, {id: n2, title: Apple Vision Pro review: spatial computing arrives, category: tech, content: The mixed reality headset...}, {id: n3, title: Champions League final: Real Madrid vs Bayern, category: sports, content: Match report...}, {id: n4, title: New study: Mediterranean diet reduces depression risk, category: health, content: Research shows...}, {id: n5, title: OpenAI releases Sora text-to-video model, category: tech, content: Generate realistic videos...}, {id: n6, title: Taylor Swift announces new album, category: entertainment, content: Pop star...}, {id: n7, title: Climate change: 2024 was hottest year on record, category: science, content: Global temperatures...}, ] recommender ColdStartNewsRecommender(sample_news) # 模拟新用户 print( 新用户首次访问 ) recs recommender.hybrid_recommend(clicked_history[], n5, explore_ratio0.6) for r in recs: print(f- {r[title]}) # 用户点击了第一篇文章 print(\n 用户点击了一篇AI新闻 ) recommender.update_feedback(n1, clickedTrue) recs recommender.hybrid_recommend(clicked_history[n1], n5, explore_ratio0.2) for r in recs: print(f- {r[title]}) # 用户又点击了另一篇科技新闻 print(\n 用户又点击了Vision Pro ) recommender.update_feedback(n2, clickedTrue) recs recommender.hybrid_recommend(clicked_history[n1, n2], n5, explore_ratio0.1) for r in recs: print(f- {r[title]})运行这段代码你会看到一开始推荐比较杂因为有探索在用户点击了两篇科技新闻后算法开始优先推荐其他科技类内容。这就是冷启动从“不认识”到“初步认识”的过程。1.4 前沿进展大模型时代的冷启动2024-2025年大语言模型LLM给冷启动带来了新思路。几个值得关注的方向零样本推荐直接让GPT-4或Claude根据新闻标题和用户画像即使很稀疏生成推荐。实验证明LLM在冷启动场景下的表现可以超过传统的MF和LightGCN。Prompt调优把冷启动视为一个少样本学习任务。构造prompt“用户之前点击了以下新闻[新闻标题列表]。请推荐5篇可能感兴趣的新新闻。”多模态冷启动结合新闻的图片、视频封面做特征提取。CLIP等视觉-语言模型可以直接计算用户偏好向量。第二部分兴趣迁移——用户是会变的如果说冷启动是“初识”的问题那么兴趣迁移就是“相守”的问题。它更难更隐蔽也更容易被忽视。2.1 兴趣迁移的三种模式我在实际系统中观察到的兴趣变化大致可以分成三类1. 短期波动Short-term fluctuation用户早上看财经新闻下午看科技新闻晚上看娱乐新闻。这只是不同场景下的正常切换不代表长期兴趣变了。2. 长期漂移Long-term drift一个用户从前喜欢看NBA最近半年开始追AI技术文章。这种变化缓慢但持久。3. 突发转变Sudden shift用户换了工作、开始备考、或者某个重大新闻事件发生后兴趣可能在几天内完全改变。传统推荐模型最大的问题就是过度依赖长期历史。一个训练好的模型会拼命给用户推半年前喜欢的内容因为那些交互数据权重高。2.2 解决兴趣迁移的主流方法时间衰减Time Decay最经典的方法。用户越久远的交互权重越低。一个简单的实现pythondef time_decay_weight(timestamp_hours_ago, half_life24): half_life: 半衰期小时24表示24小时前的交互权重为当前的一半 return 0.5 ** (timestamp_hours_ago / half_life)序列建模Sequential Modeling用RNN、LSTM、Transformer来建模用户行为序列。核心思想是下一个点击什么取决于最近点击了什么而不是三年前点击了什么。SASRecSelf-Attentive Sequential Recommendation和BERT4Rec是代表作。2024年很多团队开始用GPT风格的decoder-only架构做序列推荐。滑动窗口Sliding Window只保留最近K次交互用于训练放弃较老的数据。简单粗暴但在很多场景下效果不错。在线学习Online Learning模型实时更新。用户点击一篇新闻后几秒钟内模型参数就更新了。FTRLFollow The Regularized Leader和增量矩阵分解是常用方法。2.3 代码实战带遗忘机制的序列推荐下面实现一个轻量级的序列推荐模型融合了时间衰减和滑动窗口并用一个简单的注意力机制捕捉用户在窗口内的兴趣变化。pythonimport numpy as np import torch import torch.nn as nn import torch.nn.functional as F from collections import deque from datetime import datetime, timedelta import json class TimeAwareSequentialRecommender(nn.Module): 基于时间感知的序列推荐模型 def __init__(self, num_news, embedding_dim128, max_seq_len50, time_span_days7): super().__init__() self.num_news num_news self.embedding_dim embedding_dim self.max_seq_len max_seq_len self.time_span_days time_span_days # 只保留最近N天的交互 # 新闻嵌入 self.news_embeddings nn.Embedding(num_news, embedding_dim) # 时间衰减嵌入将时间间隔映射到衰减系数 self.time_decay_mlp nn.Sequential( nn.Linear(1, 16), nn.ReLU(), nn.Linear(16, 1), nn.Sigmoid() ) # 序列编码器Transformer Encoder encoder_layer nn.TransformerEncoderLayer( d_modelembedding_dim, nhead8, dim_feedforward256, batch_firstTrue ) self.transformer nn.TransformerEncoder(encoder_layer, num_layers2) # 输出层 self.output_proj nn.Linear(embedding_dim, num_news) def apply_time_decay(self, seq_embeddings, time_intervals_hours): seq_embeddings: [batch, seq_len, dim] time_intervals_hours: [batch, seq_len] 距离当前时间的小时数 decay_weights self.time_decay_mlp( (time_intervals_hours.unsqueeze(-1) / 24.0).clip(0, 30) # 归一化 ) # [batch, seq_len, 1] return seq_embeddings * decay_weights def forward(self, seq_indices, time_intervals, maskNone): # 获取序列嵌入 seq_emb self.news_embeddings(seq_indices) # [batch, seq_len, dim] # 应用时间衰减 seq_emb self.apply_time_decay(seq_emb, time_intervals) # Transformer编码 if mask is not None: seq_emb self.transformer(seq_emb, src_key_padding_maskmask) else: seq_emb self.transformer(seq_emb) # 取最后一个有效位置作为用户表示 # 简化起见取序列平均 user_repr seq_emb.mean(dim1) # [batch, dim] # 预测下一个点击 logits self.output_proj(user_repr) # [batch, num_news] return logits class UserInterestTracker: 用户兴趣追踪器维护滑动窗口内的历史支持兴趣变化的检测 def __init__(self, user_id, max_history100, time_window_days7): self.user_id user_id self.max_history max_history self.time_window_days time_window_days self.history deque(maxlenmax_history) # 每个元素为 (news_id, timestamp) # 兴趣分布类别级别 self.interest_distribution {} self.last_update_time None def add_interaction(self, news_id, category, timestamp): self.history.append({ news_id: news_id, category: category, timestamp: timestamp }) self.last_update_time timestamp self._update_interest_distribution() def _update_interest_distribution(self): 基于滑动窗口内的历史计算当前兴趣分布 now self.last_update_time or datetime.now() cutoff_time now - timedelta(daysself.time_window_days) # 过滤窗口内的交互 recent [h for h in self.history if h[timestamp] cutoff_time] category_counts {} for item in recent: cat item[category] # 时间衰减权重 hours_ago (now - item[timestamp]).total_seconds() / 3600 weight 0.5 ** (hours_ago / 24) # 24小时半衰期 category_counts[cat] category_counts.get(cat, 0) weight # 归一化 total sum(category_counts.values()) if total 0: self.interest_distribution { cat: cnt/total for cat, cnt in category_counts.items() } else: self.interest_distribution {} def detect_interest_shift(self, threshold0.3): 检测兴趣是否发生显著迁移 返回: (has_shifted, old_top_category, new_top_category) if len(self.history) 10: return False, None, None # 比较前半段和后半段的兴趣分布 mid_point len(self.history) // 2 old_half list(self.history)[:mid_point] new_half list(self.history)[mid_point:] def get_top_category(items, now_time, window_days): cutoff now_time - timedelta(dayswindow_days) recent [i for i in items if i[timestamp] cutoff] counts {} for item in recent: cat item[category] hours_ago (now_time - item[timestamp]).total_seconds() / 3600 weight 0.5 ** (hours_ago / 24) counts[cat] counts.get(cat, 0) weight return max(counts, keycounts.get) if counts else None now datetime.now() old_top get_top_category(old_half, now, self.time_window_days) new_top get_top_category(new_half, now, self.time_window_days) if old_top and new_top and old_top ! new_top: # 计算新旧分布的距离简化版 return True, old_top, new_top return False, None, None def get_current_interest_profile(self): 返回当前兴趣向量用于推荐排序 return self.interest_distribution # -------------------- 使用示例 -------------------- def simulate_user_interest_shift(): tracker UserInterestTracker(user_iduser123, max_history200, time_window_days3) # 模拟前5天看体育 base_time datetime(2025, 5, 1, 10, 0, 0) for i in range(30): tracker.add_interaction( news_idfsports_{i}, categorysports, timestampbase_time timedelta(hoursi) ) print(f5天后的兴趣分布: {tracker.get_current_interest_profile()}) # 检测兴趣迁移 shifted, old, new tracker.detect_interest_shift() print(f检测到兴趣迁移: {shifted}, 从 {old} 到 {new}) # 突然切换到科技兴趣迁移 print(\n 用户突然开始看科技新闻 ) for i in range(30): tracker.add_interaction( news_idftech_{i}, categorytech, timestampbase_time timedelta(days6, hoursi) ) print(f切换后的兴趣分布: {tracker.get_current_interest_profile()}) shifted, old, new tracker.detect_interest_shift(threshold0.3) print(f检测到兴趣迁移: {shifted}, 从 {old} 到 {new}) return tracker # 运行模拟 tracker simulate_user_interest_shift()