RAG 检索结果去重与排序从向量相似度到多因子融合一、单一相似度排序在生活场景中的失败模式一个家庭食谱 RAG 系统的真实案例用户搜索西红柿炒鸡蛋的家常做法向量检索返回了 10 条结果其中 6 条是同一篇食谱的不同版本因为该食谱被作者修改过 6 次另外 4 条是含有西红柿和鸡蛋作为食材的其他菜谱。用户期待的最经典的西红柿炒鸡蛋食谱排在第三位前两位分别是西红柿鸡蛋面和西红柿鸡蛋馅饼。这个场景暴露了纯向量相似度排序的三个缺陷内容重复同一信息的多个版本挤占了检索结果的位置关键词匹配优先于语义西红柿鸡蛋的高频共现让无关菜谱排名靠前缺乏时效性考量3 年前的食谱和本周更新的食谱被平等对待。单一维度的评分无法满足用户对最相关、最权威、最新鲜的复合需求。需要引入多因子融合排序。二、多因子融合排序的架构将检索结果的质量评判从单一维度扩展为五个维度的加权融合graph TB Q[用户查询] -- VEC[向量检索br/返回 Top 50] VEC -- DEDUP[语义去重br/相似度 0.92 视为重复] DEDUP -- TOP30[保留 Top 30] TOP30 -- S1[相似度评分br/权重 0.40] TOP30 -- S2[关键词匹配br/权重 0.25] TOP30 -- S3[时效性评分br/权重 0.15] TOP30 -- S4[来源权威度br/权重 0.10] TOP30 -- S5[文本质量br/权重 0.10] S1 -- FUSION[加权融合] S2 -- FUSION S3 -- FUSION S4 -- FUSION S5 -- FUSION FUSION -- RANKED[重排序结果br/Top 10] RANKED -- LLM[送入大模型]流程分两个阶段第一阶段用宽松的向量检索拉取 50 条候选经过语义去重缩减到 30 条。第二阶段对 30 条候选计算五个维度的得分并加权融合最终取 Top 10 送入大模型。这种粗筛→精排的两阶段策略比直接在 50 条上精排更省计算资源。三、去重与多因子评分的实现from dataclasses import dataclass, field from typing import List, Tuple import numpy as np from datetime import datetime, timezone import re dataclass class SearchResult: 检索结果的统一数据结构 doc_id: str content: str # 向量检索返回的原始相似度 vector_score: float # 元信息——用于多因子计算 created_at: datetime updated_at: datetime source_authority: float # 0-1来源权威度 # 去重后的 group id——相同内容指向同一个 group dedup_group: str class RAGReranker: RAG 检索结果的重排序器 def __init__(self, similarity_threshold: float 0.92): self.similarity_threshold similarity_threshold def deduplicate( self, results: List[SearchResult] ) - List[SearchResult]: 语义去重——高相似度结果合并到同一组。 设计意图留下每组中 vector_score 最高的那条作为代表。 if len(results) 1: return results deduplicated: List[SearchResult] [] used_doc_ids: set set() for i, current in enumerate(results): if current.doc_id in used_doc_ids: continue best current for j, other in enumerate(results): if j i or other.doc_id in used_doc_ids: continue # 快速过滤如果两个结果的文档长度相差超过 30%不太可能是重复 len_ratio min( len(current.content), len(other.content) ) / max(len(current.content), len(other.content)) if len_ratio 0.7: continue # 通过向量相似度判断是否为重复内容 sim self._compute_similarity( current.vector_score, other.vector_score ) if sim self.similarity_threshold: used_doc_ids.add(other.doc_id) # 保留相似度更高的那条 if other.vector_score best.vector_score: best other deduplicated.append(best) used_doc_ids.add(current.doc_id) return deduplicated def rank( self, results: List[SearchResult], query: str ) - List[Tuple[SearchResult, float]]: 多因子融合排序。 返回 (结果, 最终得分) 的排序列表。 scored: List[Tuple[SearchResult, float]] [] for result in results: # 因子1向量相似度0-1——权重 0.40 similarity_score result.vector_score # 因子2关键词匹配度0-1——权重 0.25 # 计算查询词在文档中的覆盖率 keyword_score self._keyword_match_score(query, result.content) # 因子3时效性评分0-1——权重 0.15 # 最近 30 天更新的文档得满分90 天后线性衰减 freshness_score self._freshness_score(result.updated_at) # 因子4来源权威度0-1——权重 0.10 authority_score result.source_authority # 因子5文本质量0-1——权重 0.10 # 内容长度适中500-2000字、结构化程度高 quality_score self._quality_score(result.content) # 加权融合 final_score ( similarity_score * 0.40 keyword_score * 0.25 freshness_score * 0.15 authority_score * 0.10 quality_score * 0.10 ) scored.append((result, final_score)) # 按最终得分降序排列 scored.sort(keylambda x: x[1], reverseTrue) return scored def _keyword_match_score(self, query: str, content: str) - float: 计算查询关键词在文档中的覆盖比例 # 简单中文分词——取 2-4 字的连续子串作为关键词片段 keywords set() for length in [2, 3, 4]: for i in range(len(query) - length 1): keywords.add(query[i:i length]) if not keywords: return 0.0 content_lower content.lower() matched sum(1 for kw in keywords if kw.lower() in content_lower) return matched / len(keywords) def _freshness_score(self, updated_at: datetime) - float: 时效性评分——越新得分越高90天后归零 now datetime.now(timezone.utc) # 确保两个时间都有时区信息 if updated_at.tzinfo is None: updated_at updated_at.replace(tzinfotimezone.utc) days_ago (now - updated_at).days if days_ago 1: return 1.0 if days_ago 90: return 0.0 return 1.0 - (days_ago / 90) def _quality_score(self, content: str) - float: 文本质量评分——基于长度和结构特征 score 0.5 # 基础分 # 长度适中500-2000字得高分 length len(content) if 500 length 2000: score 0.3 elif 200 length 500: score 0.15 # 有结构化标记列表、标题格式 if re.search(r[\d][\.、], content): score 0.1 if # in content or ** in content: score 0.1 return min(score, 1.0) def _compute_similarity(self, score_a: float, score_b: float) - float: 通过向量相似度分值的接近程度估算内容相似度。 注生产环境应使用向量间的余弦相似度这里仅为简化示意。 return 1.0 - abs(score_a - score_b) * 0.5权重分配的说明相似度 40% 是最高权重——因为向量相似度仍然是相关性判断的最重要信号。关键词匹配 25% 用于修正向量检索可能忽略的精确匹配场景。时效性 15% 适用于食谱这类可能有过时信息的场景但对代码文档类搜索应将权重调低。权威度 10% 适合来源质量参差不齐的场景在企业内部知识库中可以忽略此项。四、权重调优与适用边界权重不是拍脑袋定的。在一个食谱场景中如果用户 80% 的点击落在排名第 3-5 位的结果上说明相似度权重过高而时效性权重过低。应该用点击率数据反向校准权重。最简单的做法是提取点击率 Top 3 的结果特征与当前排序对比调整权重使排名与点击率更一致。不适合多因子排序的场景企业内部知识库所有文档权威度相同、实时新闻检索时效性是唯一维度这些场景下的多因子反而增加了不必要的计算复杂度。去重阈值的敏感度similarity_threshold设为 0.92 在食谱场景中有效但在新闻场景中可能需要降至 0.85因为同一事件的多个报道不应被视为重复。五、总结RAG 检索结果的质量优化需要两方面的工作语义去重——在精排前剔除内容高度相似的冗余结果多因子融合——将相似度、关键词、时效性、权威度、文本质量五个维度加权融合权重通过点击率数据反向校准而非一次性固定。落地建议先上线基本的语义去重解决最明显的重复问题逐步加入关键词和时效性因子观察搜索结果的变化部署后采集用户点击数据每月校准一次权重对不同领域食谱 vs 代码文档维护独立的权重配置。