当前位置：首页 > news >正文

RAG 检索结果排序优化：召回对了，排在第二位也可能没用

news 2026/7/8 8:33:13

RAG 检索结果排序优化：召回对了，排在第二位也可能没用

一、向量召回对了，但大模型看不到

一个客服知识库 RAG 系统上线后，准确率只有 62%。
检查召回结果后发现，正确答案确实在召回列表中。
但它排在第 4 位。
大模型的上下文窗口有限，通常只读取前 3 条。
排在第 4 位的正确答案根本进不了 Prompt。

这就是"召回了但没利用"的问题。
学界称为检索增强生成中的排序偏差。
向量相似度排序是语义层面的。
但语义相似不等于答案有用。
一个 FAQ 条目可能语义高度匹配，却不能回答问题。

排序优化是整个 RAG 系统的关键一环。
召回决定上限，排序决定下限。

二、多因子排序：语义只是起点

单一语义相似度排序有天然缺陷。
好的排序需要综合多个信号。

flowchart TB A[用户 Query] --> B[向量召回 Top-50] B --> C[语义相似度打分] B --> D[关键词匹配打分 BM25] B --> E[文档质量信号] C --> F[多因子融合排序] D --> F E --> F F --> G[取 Top-5] G --> H[注入 Prompt 上下文]

关键排序因子：
语义相似度（向量余弦距离）；
关键词匹配度（BM25 或 TF-IDF）；
文档质量信号（创建时间、点击数、人工标记权重）；
问题-答案匹配度（专用排序模型）。

融合策略通常采用加权求和或 LambdaMART 等学习排序。
简单场景用加权求和已经足够。

三、重排序的 Python 实现

""" reranker.py - RAG 多因子重排序 """ import math import numpy as np from typing import List, Dict, Tuple from collections import Counter class SearchResult: """单条召回结果""" def __init__( self, doc_id: str, content: str, vector_score: float, metadata: Dict = None ): self.doc_id = doc_id self.content = content self.vector_score = vector_score # 向量相似度 [0,1] self.bm25_score = 0.0 # 关键词匹配得分 self.quality_score = 0.0 # 文档质量分 self.final_score = 0.0 # 融合后得分 self.metadata = metadata or {} class MultiFactorReranker: """多因子重排序器""" def __init__( self, weights: Dict[str, float] = None ): """ weights: 各因子权重 默认: 向量0.4, BM25 0.3, 质量0.2, 新鲜度0.1 """ self.weights = weights or { "vector": 0.4, "bm25": 0.3, "quality": 0.15, "freshness": 0.15, } def compute_bm25( self, query: str, results: List[SearchResult], k1: float = 1.5, b: float = 0.75, ): """计算 BM25 得分""" query_terms = query.lower().split() if not query_terms: return # 计算文档平均长度 doc_lengths = [len(r.content.split()) for r in results] avg_dl = sum(doc_lengths) / max(len(doc_lengths), 1) # 计算 IDF doc_count = len(results) df = Counter() for r in results: terms = set(r.content.lower().split()) for t in terms: df[t] += 1 for r in results: doc_terms = r.content.lower().split() doc_len = len(doc_terms) tf = Counter(doc_terms) score = 0.0 for term in query_terms: if term not in tf: continue # IDF idf = math.log( (doc_count - df.get(term, 0) + 0.5) / (df.get(term, 0) + 0.5) + 1.0 ) # BM25 公式 numerator = tf[term] * (k1 + 1) denominator = tf[term] + k1 * ( 1 - b + b * doc_len / avg_dl ) score += idf * numerator / max(denominator, 0.001) r.bm25_score = score def compute_quality(self, results: List[SearchResult]): """根据元数据计算文档质量分""" for r in results: score = 0.5 # 基础分 meta = r.metadata or {} # 点击数加分（归一化） if "clicks" in meta and meta["clicks"] > 0: score += min(math.log(meta["clicks"] + 1) / 10, 0.3) # 人工标记权重 if "manual_weight" in meta: score += meta["manual_weight"] * 0.2 r.quality_score = min(score, 1.0) def rank( self, query: str, results: List[SearchResult], top_k: int = 5, ) -> List[SearchResult]: """执行多因子融合排序""" if not results: return [] # 计算各因子得分 self.compute_bm25(query, results) self.compute_quality(results) # 归一化向量得分和 BM25 得分到 [0,1] max_bm25 = max(r.bm25_score for r in results) or 1.0 max_vector = max(r.vector_score for r in results) or 1.0 for r in results: norm_bm25 = r.bm25_score / max_bm25 norm_vector = r.vector_score / max_vector # 加权融合 r.final_score = ( self.weights["vector"] * norm_vector + self.weights["bm25"] * norm_bm25 + self.weights["quality"] * r.quality_score ) # 按最终得分降序排列 results.sort(key=lambda r: r.final_score, reverse=True) # 多样性重排：相邻文档去重 deduped = [] seen = set() for r in results: # 简单去重：检查前 50 字符是否重复 prefix = r.content[:50] if prefix not in seen: deduped.append(r) seen.add(prefix) if len(deduped) >= top_k: break return deduped # ---- 使用示例 ---- def demo(): raw_results = [ SearchResult("d1", "账户余额查询方法", 0.92), SearchResult("d2", "如何查看账户余额和交易明细", 0.88), SearchResult("d3", "账户余额不足时的处理方法", 0.75), SearchResult("d4", "充值后账户余额未到账怎么办", 0.71), SearchResult("d5", "企业账户余额管理指南", 0.65), ] # 设置文档质量元数据（点击量、人工权重） raw_results[3].metadata = {"clicks": 150, "manual_weight": 1.0} raw_results[0].metadata = {"clicks": 30} raw_results[4].metadata = {"clicks": 200} reranker = MultiFactorReranker() ranked = reranker.rank("账户余额查询", raw_results, top_k=3) for i, r in enumerate(ranked): print(f"[{i+1}] {r.doc_id}: {r.content} (score={r.final_score:.3f})") # 输出示例： # [1] d4: 充值后账户余额未到账怎么办 (score=0.xxx) ← BM25 + 高质量分提升 # [2] d1: 账户余额查询方法 (score=0.xxx) # [3] d2: 如何查看账户余额和交易明细 (score=0.xxx)