RAG 检索结果怎么用 CrossEncoder 重新排序

RAG 的召回结果不理想时，通常不用立刻更换向量库。更实用的做法是保留第一阶段的高效召回，再把查询和每个候选片段一起交给 CrossEncoder 重新打分，按分数取最终 Top-K。它能看到查询与文档的交互关系，但需要为每一对文本做一次推理，所以只适合重排一个有限的候选池。

推荐的链路是“先召回 Top-N，再用 CrossEncoder 重排，最后把 Top-K 放进提示词”。N 要大于 K，但不能大到让重排延迟吞掉 RAG 的响应预算。

先把 RAG 的两阶段职责分开

在百万级语料中逐个计算查询与文档的 CrossEncoder 分数，成本会随候选数量线性增长。Sentence Transformers 的官方 Retrieve & Re-Rank 示例因此把流程拆成两段：双编码器或关键词检索先取出一小批候选，CrossEncoder 再给这些候选排序。第一段追求覆盖率，第二段追求相关性。

例如召回 50 段、最终放入 5 段，就只计算 50 个查询-文档对，而不是对整个知识库做配对。这里的 N 和 K 不是固定标准：N 太小会让重排器没有机会找回漏掉的答案，K 太大又会把相似但无关的片段塞回上下文。

用 CrossEncoder 给候选片段逐对打分

下面示例使用 Sentence Transformers 文档中的常见 MS MARCO CrossEncoder。模型名只是示例，生产环境应结合中文语料、领域和延迟预算选择经过验证的模型。候选文档保留 id、正文和来源，重排后只改变顺序，不丢掉这些元数据。

from sentence_transformers import CrossEncoder

# 用一个查询和多个候选片段组成成对输入。
query = "如何为 RAG 检索结果增加重排阶段"
candidates = [
    {"id": "doc-1", "text": "先用向量检索得到候选，再用 CrossEncoder 重新排序。", "source": "rag-guide"},
    {"id": "doc-2", "text": "BM25 适合按关键词匹配文档。", "source": "search-note"},
    {"id": "doc-3", "text": "CrossEncoder 同时读取查询和候选文本，输出相关性分数。", "source": "model-note"},
]

# 加载可用于句对相关性打分的模型；首次运行会下载模型文件。
model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L6-v2")
pairs = [[query, item["text"]] for item in candidates]
scores = model.predict(pairs, batch_size=16, show_progress_bar=False)

# 分数只用于同一查询下排序，元数据随候选一起保留。
ranked = sorted(zip(candidates, scores), key=lambda pair: float(pair[1]), reverse=True)
top_k = [{**item, "rerank_score": float(score)} for item, score in ranked[:2]]
for item in top_k:
    print(item["id"], item["rerank_score"], item["source"])

predict 接收的是句对列表，而不是一组孤立句子。示例里的 batch_size=16 只影响推理批次，不改变排序逻辑；显存不足时可以调小，吞吐允许时再逐步调大。scores 可能是 NumPy 数组或类似序列，转成 Python float 后再写入结果，便于序列化和日志记录。

把重排结果接回 RAG 上下文

重排结束后，不要只把一串文本直接拼进提示词。建议先按分数取 K，再做轻量去重和长度裁剪，最后保留来源标识。这样模型既能看到靠前的证据，也能在回答中指出证据来自哪一个片段。

# 只选重排后的前 K 个片段，并限制单片段长度。
def build_context(ranked_items, top_k=5, max_chars=1200):
    blocks = []
    seen = set()
    for item in ranked_items[:top_k]:
        # 用文档 ID 去重，避免同一片段的不同切片重复占用上下文。
        if item["id"] in seen:
            continue
        seen.add(item["id"])
        text = item["text"].strip()[:max_chars]
        blocks.append(f"[{item['source']}] {text}")
    return "\n\n".join(blocks)

context = build_context(top_k, top_k=2)
prompt = f"请只依据以下资料回答问题：\n{context}\n\n问题：{query}"
print(prompt)

最终的 K 应由提示词窗口、回答任务和离线评估共同决定。问答任务可以从较小的 K 开始；需要跨段落综合时，宁可扩大召回 N 并评估重排后的 Recall@K，也不要只凭一次示例把 K 写死。

用指标和边界控制重排成本

性能测试至少记录“每次查询的候选数 N、最终 K、重排耗时、端到端 P95 延迟、Recall@K 或 nDCG@K”。可以固定一组真实问题，分别比较不重排、N=20 和 N=50 的结果；如果 N 增大后 Recall@K 不再提升，却持续增加延迟，就没有必要继续扩大候选池。

CrossEncoder 的分数不是跨模型、跨查询都可直接比较的业务概率。它最可靠的用途是：在同一个查询的候选集合中排序。若业务要按阈值过滤，先用带标注的验证集校准阈值；若候选规模本身很小，也可以直接对全部候选打分，但不要把这种做法照搬到大语料全量扫描。

常见问题

CrossEncoder 能替代向量检索吗？

通常不能。它需要为查询和候选逐对计算，适合第二阶段重排；大规模语料仍应先用向量检索、BM25 或其他高效方法缩小候选集。

分数必须转换成 0 到 1 吗？

不必须。排序只需要比较相对大小。Sentence Transformers 文档说明部分 MS MARCO 模型输出 raw logits；若展示概率或按概率阈值决策，再使用 Sigmoid，并重新用验证集确认阈值。

N=100、K=5 是固定最佳值吗？

不是。N 影响召回覆盖和重排成本，K 影响上下文噪声与提示词长度。用真实查询集同时看 Recall@K、nDCG@K 和 P95 延迟，按业务预算选择。

落地时可以先从“召回 20 段、重排后取 5 段”做基线，再逐步增加候选数。只要把召回覆盖、重排排序和上下文组装分别记录，RAG 的问题就能从“感觉答得不准”变成可测量、可调参的检索链路。