10-使用Reranker提升RAG回答准确率

使用 Reranker（重排模型）提升 RAG 知识库回答准确率

系列：从零构建企业 RAG 知识库（第 10 篇）

1. 为什么召回后还要重排

第一阶段检索追求“别漏掉”，通常快速取 20—100 个候选；生成模型上下文却只适合少量高质量证据。

大范围召回（高 Recall） → Reranker 精排（高 Precision）
→ 去重与上下文预算 → 生成

Embedding 常将问题和文档分别编码，再比较向量；Cross-Encoder 类 Reranker 同时读取“问题 + 候选”，能进行更细致的相关性判断，但计算更慢。

2. 重排接口必须保留候选身份

from dataclasses import dataclass
from typing import Protocol

@dataclass(frozen=True)
class Candidate:
chunk_id: str
document_id: str
tenant_id: str
text: str
retrieval_score: float

@dataclass(frozen=True)
class RerankedCandidate:
candidate: Candidate
rerank_score: float
original_rank: int

class Reranker(Protocol):
def score(self, query: str, texts: list[str]) –> list[float]:
"""返回与 texts 一一对应的相关性分数。"""

3. 安全的批量重排

def rerank(
query: str,
candidates: list[Candidate],
reranker: Reranker,
top_n: int = 5,
) –> list[RerankedCandidate]:
if not query.strip():
raise ValueError("查询不能为空")
if not 1 <= top_n <= 20:
raise ValueError("top_n 必须在 1 到 20 之间")
if not candidates:
return []

scores = reranker.score(query, [item.text for item in candidates])
if len(scores) != len(candidates):
raise RuntimeError("Reranker 分数数量与候选数量不一致")
if any(not isinstance(score, (int, float)) for score in scores):
raise RuntimeError("Reranker 返回了非数值分数")

ranked = [
RerankedCandidate(candidate, float(score), original_rank)
for original_rank, (candidate, score) in enumerate(
zip(candidates, scores),
start=1,
)
]
ranked.sort(
key=lambda item: (
–item.rerank_score,
item.original_rank,
item.candidate.chunk_id,
)
)
return ranked[:top_n]

原始排名作为同分排序依据，让结果稳定且可诊断。

4. 用确定性 Fake Reranker 测流程

class KeywordReranker:
"""教学替身，只验证控制流，不代表真实重排模型。"""

def score(self, query: str, texts: list[str]) –> list[float]:
query_chars = set(query)
return [
float(len(query_chars & set(text))) / max(len(query_chars), 1)
for text in texts
]

def test_relevant_candidate_moves_forward() –> None:
candidates = [
Candidate("c1", "d1", "t1", "退款政策总览", 0.95),
Candidate("c2", "d2", "t1", "退款申请需要人工审核", 0.80),
]
result = rerank(
"退款申请如何审核",
candidates,
KeywordReranker(),
top_n=1,
)
assert result[0].candidate.chunk_id == "c2"
assert result[0].original_rank == 2

5. 真实 Cross-Encoder 的适配示意

class SentenceTransformersReranker:
"""示意适配器；模型名称和许可证需按当前官方资料选择。"""

def __init__(self, model) –> None:
self.model = model

def score(self, query: str, texts: list[str]) –> list[float]:
pairs = [(query, text) for text in texts]
raw_scores = self.model.predict(pairs)
return [float(value) for value in raw_scores]

# 示例初始化方式需要安装并锁定 sentence-transformers：
# from sentence_transformers import CrossEncoder
# model = CrossEncoder("经过团队评测并批准的模型路径")
# reranker = SentenceTransformersReranker(model)

本文没有下载或运行具体重排模型，不声称任何模型在企业数据上的准确率。

6. 重排之后还要处理上下文

def select_context(
ranked: list[RerankedCandidate],
max_chars: int,
min_score: float,
) –> list[Candidate]:
selected: list[Candidate] = []
used = 0
seen_documents: set[str] = set()

for item in ranked:
if item.rerank_score < min_score:
continue
size = len(item.candidate.text)
if used + size > max_chars:
continue
# 示例限制每份文档一个 Chunk，避免近重复结果占满上下文
if item.candidate.document_id in seen_documents:
continue
selected.append(item.candidate)
seen_documents.add(item.candidate.document_id)
used += size
return selected

真实预算应使用目标生成模型 Tokenizer，而不是字符数。

7. 可复验的失败测试

class BrokenReranker:
def score(self, query: str, texts: list[str]) –> list[float]:
return [1.0] # 故意返回错误数量

def test_broken_reranker_is_rejected() –> None:
candidates = [
Candidate("c1", "d1", "t1", "A", 0.9),
Candidate("c2", "d2", "t1", "B", 0.8),
]
try:
rerank("问题", candidates, BrokenReranker())
except RuntimeError:
pass
else:
raise AssertionError("错误的分数数量必须被拒绝")

8. 如何证明重排有效

固定第一阶段候选，比较重排前后：

• MRR、NDCG@K 和 Precision@K；
• 正确证据进入最终 Context 的比例；
• 最终答案正确性与忠实度；
• P95 延迟、吞吐和成本；
• 长文、编号、否定条件和多语言切片。

如果第一阶段没有召回正确证据，Reranker 无法凭空找回它。

9. 对抗性审查

• 权限过滤必须在重排前完成；
• 限制候选数量和单条长度，避免资源耗尽；
• 文档中的注入文本仍是不可信数据；
• 记录召回分数、重排分数和模型版本；
• 模型升级重新跑固定评测集；
• Reranker 故障可降级到原始检索排序，但要标记降级状态。

10. 总结

Reranker 的价值是把“可能相关”的大候选集压缩成“更适合进入上下文”的小集合。它优化排序而不是创造证据，必须建立在高召回、权限正确和可评测的第一阶段检索之上。