向量检索在推荐系统中的应用用户行为向量与内容向量的融合排序一、深度引言与场景痛点大家好我是赵咕咕。今年年初我们给内部知识库加了个推荐模块。需求很标准用户浏览了某篇文章在侧边栏推荐相关内容。第一版直接用基于标签的协同过滤效果还行——点击率提升了 12%。但有个问题一直让我不舒服推荐列表里的内容质量全看标签打得准不准。两篇都打了Python标签的文章一篇是《Python 入门教程》一篇是《Python GIL 深度解析》对刚入门用户的推荐价值完全不一样。标签分不出这个。后来我们尝试了向量检索方案——把用户行为序列和文章内容都向量化在同一个向量空间里做检索。效果立即提升点击率从 12% 跳到了 24%页面停留时长增加了 40%。这篇文章我把用户行为向量和内容向量在推荐系统中的融合方案从头梳理一遍。二、底层机制与原理深度剖析2.1 为什么只靠内容向量不够内容向量content embedding能表达这两篇文章语义相近但它不知道用户现在处于什么阶段。一个刚学 Python 的用户看了《Python 变量和数据类型》和一位正在做性能优化的用户看了同一篇文章——推荐需求完全不同。这就是协同过滤信号的价值用户行为序列反映了用户的兴趣发展阶段。把行为向量和内容向量结合起来推荐系统才能做到在合适的时间推荐合适的内容。2.2 用户行为向量化我们用的方法很直接——行为序列编码收集用户最近 N 次N50的浏览/点击行为。对每篇浏览过的文章做内容 embedding。按时间衰减权重最近浏览的权重大一周前的权重小。加权平均 → 用户兴趣向量。这个简单方法在实践中效果惊人地好。复杂模型如 Transformer 编码用户序列的提升幅度只有 5-8%但推理延迟增加了 10 倍。2.3 双塔融合排序架构这个双塔架构的核心思想是计算分离用户塔离线计算用户行为序列的处理、embedding 加权平均都是离线完成的结果缓存在 Redis 中。用户每次请求直接从缓存读取用户向量。内容塔实时/近实时新内容入库后异步生成 embedding 写入向量索引。内容更新频率远低于用户请求频率可以用批处理。检索层多路召回用户向量 KNN 召回语义相关 标签协同召回冷启动内容 热门补盲。三路互补。2.4 时间衰减的重要性如果不用时间衰减一个三个月前集中看了大量Django文章的用户用户向量会被 Django 内容主导——但他可能已经转去做 FastAPI 了。时间衰减的公式很简单user_vector Σ (content_vector_i * e^(-λ * Δt_i)) / Σ (e^(-λ * Δt_i))其中 λ 控制衰减速度。我们设定 λ0.05意味着 7 天前的行为权重是当前行为的一半。这个参数需要根据你的内容更新频率调整——日报类内容 λ 可以大一些0.1知识库类内容 λ 可以小一些0.01。三、生产级代码实现import asyncio import logging import math import time from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Any import numpy as np import redis.asyncio as aioredis logger logging.getLogger(__name__) dataclass class UserBehavior: 用户行为记录。 content_id: str action_type: str # view / click / like / share timestamp: float # Unix timestamp duration_ms: int 0 # 停留时长毫秒 class UserVectorService: 用户行为向量服务。 负责收集行为 → 构建用户兴趣向量 → 缓存。 def __init__( self, embedding_model: Any, redis_client: aioredis.Redis, decay_lambda: float 0.05, # 时间衰减系数 max_history: int 50, # 最多保留的行为数 cache_ttl: int 3600, # 缓存有效期秒 ): self._embedder embedding_model self._redis redis_client self._decay decay_lambda self._max_history max_history self._cache_ttl cache_ttl async def get_user_vector( self, user_id: str ) - np.ndarray | None: 获取用户兴趣向量优先从缓存读取。 cache_key fuser_vec:{user_id} # 尝试读缓存 cached await self._redis.get(cache_key) if cached: return np.frombuffer(cached, dtypenp.float32) # 缓存未命中重新计算 vector await self._build_user_vector(user_id) if vector is not None: # 写入缓存 await self._redis.setex( cache_key, self._cache_ttl, vector.tobytes() ) return vector async def _build_user_vector( self, user_id: str ) - np.ndarray | None: 从行为序列构建用户兴趣向量。 # 获取最近行为从消息队列或数据库 behaviors await self._get_recent_behaviors(user_id) if not behaviors: return None now time.time() weighted_sum None total_weight 0.0 for behavior in behaviors: # 获取内容 embedding content_emb await self._get_content_embedding( behavior.content_id ) if content_emb is None: continue # 时间衰减 delta_days (now - behavior.timestamp) / 86400.0 decay math.exp(-self._decay * delta_days) # 行为类型加权 action_weight self._action_weight(behavior.action_type) weight decay * action_weight if weighted_sum is None: weighted_sum content_emb * weight else: weighted_sum content_emb * weight total_weight weight if weighted_sum is None or total_weight 0: return None # 加权平均 L2 归一化 user_vec weighted_sum / total_weight norm np.linalg.norm(user_vec) if norm 0: user_vec user_vec / norm return user_vec.astype(np.float32) staticmethod def _action_weight(action_type: str) - float: 不同行为类型的权重。 weights { share: 3.0, # 分享说明高度认可 like: 2.0, # 喜欢说明兴趣强烈 click: 1.5, # 点击说明有意向 view: 1.0, # 浏览是最弱信号 } return weights.get(action_type, 0.5) async def _get_recent_behaviors( self, user_id: str ) - list[UserBehavior]: 获取用户最近的行为序列。 生产环境从 Kafka/数据库读取这里用 mock。 # 实际应查询用户行为存储 return [] async def _get_content_embedding( self, content_id: str ) - np.ndarray | None: 获取内容 embedding从缓存或重新计算。 cache_key fcontent_emb:{content_id} cached await self._redis.get(cache_key) if cached: return np.frombuffer(cached, dtypenp.float32) # 重新计算简化实际应查向量数据库 return None class HybridRecommender: 混合推荐引擎。 用户向量 KNN 标签协同 热门 候选集 → 精排 → Top-N。 def __init__( self, user_vector_service: UserVectorService, vector_store: Any, # FAISS/Milvus redis_client: aioredis.Redis, llm_client: Any, mmr_lambda: float 0.5, # MMR 多样性系数 ): self._user_service user_vector_service self._vector_store vector_store self._redis redis_client self._llm llm_client self._mmr_lambda mmr_lambda async def recommend( self, user_id: str, top_n: int 20, exclude_ids: set[str] | None None, timeout: float 3.0, ) - list[dict[str, Any]]: 主推荐入口。 exclude exclude_ids or set() try: return await asyncio.wait_for( self._recommend_impl(user_id, top_n, exclude), timeouttimeout, ) except asyncio.TimeoutError: logger.error(推荐超时, user%s, user_id) # 降级返回热门内容 return await self._get_trending(top_n) async def _recommend_impl( self, user_id: str, top_n: int, exclude: set[str] ) - list[dict[str, Any]]: # ── 第一路用户向量 KNN 召回 ── user_vec await self._user_service.get_user_vector(user_id) knn_candidates [] if user_vec is not None: try: knn_results self._vector_store.search( user_vec.tolist(), k100 ) knn_candidates [ {id: r[id], score: float(r[score]), source: knn} for r in knn_results if r[id] not in exclude ] except Exception as e: logger.error(KNN 召回失败: %s, e) # ── 第二路标签协同过滤 ── tag_candidates await self._tag_cf_recall(user_id, exclude) # ── 第三路热门补盲 ── trending_candidates await self._get_trending(50) trending_candidates [ {id: c[id], score: float(c[score]), source: trending} for c in trending_candidates if c[id] not in exclude ] # ── 多路合并去重 ── seen: set[str] set() merged [] for candidate in knn_candidates tag_candidates trending_candidates: if candidate[id] not in seen: seen.add(candidate[id]) merged.append(candidate) # ── MMR 多样性重排 ── diverse_results self._mmr_rerank(merged, top_n * 2) # ── 最终精排 ── final await self._rerank(user_id, diverse_results[:top_n * 2]) return final[:top_n] async def _tag_cf_recall( self, user_id: str, exclude: set[str] ) - list[dict[str, Any]]: 标签协同过滤召回。 # 简化实现从 Redis 读取用户最近交互内容的标签 # 然后推荐同标签的热门内容 try: recent_tags await self._redis.smembers( fuser_tags:{user_id} ) except Exception: return [] candidates [] for tag in recent_tags: tag tag.decode() if isinstance(tag, bytes) else tag try: tag_contents await self._redis.zrevrange( ftag:{tag}:hot, 0, 9, withscoresTrue ) for content_id, score in tag_contents: cid content_id.decode() if isinstance(content_id, bytes) else content_id if cid not in exclude: candidates.append({ id: cid, score: float(score) * 0.6, # 权重略低于 KNN source: tag_cf, }) except Exception: continue return candidates async def _get_trending( self, top_n: int 20 ) - list[dict[str, Any]]: 获取热门内容降级兜底。 try: results await self._redis.zrevrange( trending:24h, 0, top_n - 1, withscoresTrue ) return [ { id: (r[0].decode() if isinstance(r[0], bytes) else r[0]), score: float(r[1]), source: trending, } for r in results ] except Exception: return [] def _mmr_rerank( self, candidates: list[dict[str, Any]], top_n: int, ) - list[dict[str, Any]]: MMR (Maximal Marginal Relevance) 多样性重排。 平衡相关性和多样性避免推荐列表中全是相似内容。 if len(candidates) 1: return candidates selected: list[dict[str, Any]] [] remaining list(candidates) # 先选得分最高的 best max(remaining, keylambda x: x[score]) selected.append(best) remaining.remove(best) while remaining and len(selected) top_n: mmr_scores {} for candidate in remaining: # 相关性得分 relevance candidate[score] # 与已选内容的最大相似度惩罚项 max_similarity 0.0 for sel in selected: # 简化版用内容 ID hash 的差异模拟相似度 # 生产环境应该用真实的 embedding 余弦相似度 sim abs( hash(candidate[id]) - hash(sel[id]) ) / (2 ** 32) max_similarity max(max_similarity, 1.0 - sim * 0.1) mmr_scores[candidate[id]] ( self._mmr_lambda * relevance - (1 - self._mmr_lambda) * max_similarity ) # 选 MMR 最高的 next_id max(mmr_scores, keymmr_scores.get) next_candidate next( c for c in remaining if c[id] next_id ) selected.append(next_candidate) remaining.remove(next_candidate) return selected async def _rerank( self, user_id: str, candidates: list[dict[str, Any]] ) - list[dict[str, Any]]: LLM 精排轻量版——对候选做最终排序。 if len(candidates) 10: return candidates # 生产环境这里会用更轻量的排序模型如 WideDeep # 而不是 LLM。LLM 仅用于 A/B 实验。 return sorted( candidates, keylambda x: x[score], reverseTrue, ) async def main(): # Mock 示例 import redis.asyncio as aioredis r aioredis.from_url(redis://localhost:6379) recommender HybridRecommender( user_vector_serviceNone, # 简化 vector_storeNone, redis_clientr, llm_clientNone, ) results await recommender.recommend(user_123, top_n10) for i, item in enumerate(results, 1): print(f{i}. {item[id]} (score{item[score]:.3f}, source{item[source]})) if __name__ __main__: asyncio.run(main())代码设计要点三路召回互补KNN语义 标签CF冷启动 热门兜底。每路最多贡献候选集中的一部分避免某一路过度影响。MMR 多样性在最终排序中加入 MMR 算法抑制相似内容扎堆。mmr_lambda控制相关性和多样性的平衡。时间衰减公式exp(-λ * Δt)简单有效。λ 值需要根据业务特性调整——内容更新越快λ 应该越大。缓存分层用户向量缓存 1 小时用户行为变化较慢内容向量缓存长期内容本身不变。分层缓存减少 embedding 计算开销。四、边界分析与架构权衡4.1 冷启动的三种策略策略优点缺点适用场景热门推荐零成本用户接受度高缺乏个性化新用户前 3 天注册信息兴趣标签有一定个性化用户可能瞎选有注册流程的产品探索-利用Epsilon-Greedy自动学习偏好初期体验差有足够流量的产品我们采用热门 逐步个性化——前 5 次交互混入 50% 热门之后每次减少 10%第 10 次交互后完全个性化。用户对前几次推荐质量不敏感因为还没建立预期这个策略的留存数据比直接个性化好 15%。4.2 用户向量的更新频率用户行为向量需要定期更新。更新越频繁推荐越及时但计算成本越高。平衡点高频用户日活 5 次每 15 分钟更新一次。中频用户日活 1-5 次每次请求时增量更新只合并最近行为。低频用户周活不需要频繁更新缓存 24 小时。4.3 向量维度与检索性能Faiss 在 1024 维 HNSW 索引上100 万条数据的 KNN 检索延迟约 5-10ms。如果维度降到 256延迟降到 2-3ms但召回率下降 3-5%。对于推荐场景2-3ms 和 5-10ms 都是可接受的推荐不是实时搜索。所以选择 1024 维的通用 embedding 模型保持高召回率。4.4 什么时候向量推荐不适用用户行为数据太少用户只有 1-2 次交互时行为向量几乎是噪声不如直接用内容标签推荐。内容更新极快新闻、短视频等场景新内容来不及生成 embedding 就已经过期了。实时性比向量检索更重要。强实时性要求搜索广告等场景用户vector → 内容检索 → 重排的链路延迟可能超过 100ms 的 SLA。五、总结向量检索在推荐系统中的价值在于它把协同过滤的用户-内容交互矩阵变成了一个可检索的向量空间。用户和内容在同一个向量空间里推荐就变成了一个 KNN 搜索问题。关键设计决策用户塔离线内容塔在线——分离计算降低线上延迟。时间衰减 行为加权——让用户向量反映最近的、最强烈的兴趣而不是历史堆砌。多路召回 MMR 重排——KNN 不能独立工作需要标签和热门互补MMR 保证多样性。分层缓存——用户向量缓存 1 小时内容向量缓存长期。缓存命中率 95%embedding 计算开销降到可忽略。向量推荐不是万能方案但在内容质量差异化大、标签不可靠、语义理解很重要的场景下它比传统协同过滤有明显的优势。代价是工程复杂度高了一个量级——你需要维护 embedding 服务、向量数据库、用户行为序列存储。值不值看你的推荐场景是否真的需要语义级别的理解。下一篇预告系统巡检报告的可视化用监控图、拓扑图和趋势图替代文字流水账。