企业官网建设流程全解析

RAG 检索结果去重与排序：从向量相似度到多因子融合

一、单一相似度排序在生活场景中的失败模式

一个家庭食谱 RAG 系统的真实案例：用户搜索"西红柿炒鸡蛋的家常做法"，向量检索返回了 10 条结果，其中 6 条是同一篇食谱的不同版本（因为该食谱被作者修改过 6 次），另外 4 条是含有"西红柿"和"鸡蛋"作为食材的其他菜谱。用户期待的"最经典的西红柿炒鸡蛋食谱"排在第三位，前两位分别是"西红柿鸡蛋面"和"西红柿鸡蛋馅饼"。

这个场景暴露了纯向量相似度排序的三个缺陷：

1. 内容重复：同一信息的多个版本挤占了检索结果的位置；
2. 关键词匹配优先于语义："西红柿"+"鸡蛋"的高频共现让无关菜谱排名靠前；
3. 缺乏时效性考量：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(key=lambda x: x[1], reverse=True) 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(tzinfo=timezone.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 检索结果的质量优化需要两方面的工作：

1. 语义去重——在精排前剔除内容高度相似的冗余结果；
2. 多因子融合——将相似度、关键词、时效性、权威度、文本质量五个维度加权融合；
3. 权重通过点击率数据反向校准，而非一次性固定。

落地建议：

1. 先上线基本的语义去重，解决最明显的重复问题；
2. 逐步加入关键词和时效性因子，观察搜索结果的变化；
3. 部署后采集用户点击数据，每月校准一次权重；
4. 对不同领域（食谱 vs 代码文档）维护独立的权重配置。