RAG的五种重排技术：从快速检索到精准上下文匹配
为什么第一阶段检索会遗漏答案以及如何解决：跨编码器、RRF、Cohere Rerank、ColBERT和LLM作为评估工具，以及一种可从中开始的分层处理流程。
来源: **5 Reranking Techniques in RAG: From Fast Retrieval to Accurate Context**
想象一个构建良好的RAG流程：文档被分割成多个片段，选定了嵌入模型，并连接了向量数据库。当有查询出现时，检索器会返回50个片段，其中最相关的5个会被纳入提示词中。得到的答案还算不错，但并非完美，偶尔还会完全找不到正确答案。
查看被检索出的片段后，你常常会发现包含答案的片段位于第23位。嵌入模型认为它的相关性处于中等水平，未能进入前5名，因此大语言模型根本没有看到它。这种情况并非罕见，而是生产环境中的RAG系统最常出现的故障模式之一。
通常问题并不出在检索器本身，而在于检索之后的处理环节。
检索后的瓶颈
第一阶段的检索以速度为优化目标。无论是密集向量、稀疏BM25还是两者的结合，其目的都是在毫秒级时间内将百万份文档缩减为可处理的少量文档。但这样的做法会导致相关性判断过于快速且肤浅：模型会单独分析查询和每份文档，给出评分，然后返回评分最高的那个。
“评分最高”与“最有助于回答此问题”是两回事。初始检索会因以下四个原因而出现问题：
- 评分过于肤浅。嵌入模型衡量的是相似度而非实用性，因此某个文档虽然与查询很接近，却可能毫无有用信息。
- 缺乏查询与文档之间的交互。双编码器会分别对查询和文档进行编码，从未将它们放在一起对比，从而无法发现其中微妙的关系。
解决方案就是重新排序。
重新排序的作用
重新排序是第二轮处理。在快速、广泛的检索得到一批候选片段后，更强大的模型会根据这些片段与特定查询的匹配程度对它们进行重新排序。这可以视为上下文传递给大语言模型之前的质量管控。
整个流程包含三个步骤：
- 初始检索：快速、广泛的搜索会返回50到100个候选片段。
- 重新排序：更智能的模型会根据每个候选片段与查询的匹配度为其评分，并重新排列列表顺序。
这样既能获得近似搜索的速度，又能拥有深度评分的准确性。检索器并未被取代，只是得到了补充。
检索与重新排序并行处理
| Aspect | Initial Retrieval | Reranking |
| ------------------- | ------------------------ | ----------------------------- |
| Goal | Find candidates fast | Judge true relevance |
| Speed | Milliseconds | Tens to hundreds of milliseconds |
| Input | Query + index | Query + top-k candidates |
| Scoring depth | Shallow (embedding dot product) | Deep (cross-attention, token interaction) |
| Cost | Low (local compute) | Higher (model inference) |
| When to use | Every query | On top-k candidates only |
这样做的目的是避免在百万份文档上投入大量计算资源，而是将资源集中在最有可能有用的50份文档上。
五种重排序技术
重排序器的性能从低成本、快速处理到高成本、强功能不等。选择哪种取决于您的延迟预算、精度要求以及运营限制。这五种技术涵盖了不同的需求范围。
1. 交叉编码器重排序
交叉编码器是精度方面的标杆。双编码器会分别对查询和文档进行嵌入，而交叉编码器则将它们作为单个输入一起处理。这样一来，它就能捕捉到诸如代词指代、否定表达以及隐含关系等词元级交互，而这些是普通检索器完全无法识别的。
操作步骤如下：
- 检索器返回50个候选片段。
- 对于每个片段，将查询和该片段按
[CLS] 查询 [SEP] 片段 [SEP]的格式组合起来。
cross-encoder/ms-marco-MiniLM-L-6-v2。from sentence_transformers import CrossEncoder
# Load a cross-encoder reranker
# ms-marco-MiniLM-L-6-v2 is fast and accurate for general use
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def rerank_with_cross_encoder(query: str, retrieved_docs: list[str], top_k: int = 5):
"""
Rerank retrieved documents using a cross-encoder.
Args:
query: The user question
retrieved_docs: List of document chunks from initial retrieval
top_k: Number of documents to return after reranking
Returns:
List of (document, score) tuples, sorted by relevance
"""
# Create query-document pairs
pairs = [[query, doc] for doc in retrieved_docs]
# Get relevance scores
scores = cross_encoder.predict(pairs)
# Combine docs with scores and sort
scored_docs = list(zip(retrieved_docs, scores))
scored_docs.sort(key=lambda x: x[1], reverse=True)
return scored_docs[:top_k]
# Example usage
query = "What are the side effects of amoxicillin?"
retrieved = [
"Amoxicillin is a penicillin antibiotic used to treat bacterial infections.",
"Common side effects include nausea, vomiting, and diarrhea.",
"The drug was first discovered in 1958 by researchers at Beecham.",
"Patients with penicillin allergies should avoid amoxicillin.",
"Side effects may include rash, itching, and in rare cases, anaphylaxis.",
]
top_docs = rerank_with_cross_encoder(query, retrieved, top_k=3)
for doc, score in top_docs:
print(f"Score: {score:.4f} | {doc}")
当准确性是首要考虑因素，且可以接受额外增加50到200毫秒的延迟时，应选择交叉编码器。对于大多数生产环境中的RAG系统而言，这是最佳的起点。其局限性在于每个查询-文档对都需要单独处理，因此延迟会随着候选项数量的增加而上升。
2. 互反排名融合（RRF）
假设你完全不想运行神经网络模型。互反排名融合通过一种确定性公式将多个检索器的排序结果合并起来：无需训练，无需推理，仅需进行算术运算。
当已经采用混合检索方式时，RRF最为适用，例如在BM25关键词搜索之外再结合密集向量搜索。每个检索器都会提供自己的排序结果，RRF会将这些结果整合为更优的列表。
公式如下：
RRF_score(d) = 对所有检索器求和 1 / (k + rank_r(d))
此处，k 是一个常数，通常为60，而 rank_r(d) 表示文档 d 在检索器 r 中的排名。那些在多个检索器中都获得较高排名的文档会位居前列。
def reciprocal_rank_fusion(rankings: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
"""
Merge multiple document rankings using Reciprocal Rank Fusion.
Args:
rankings: List of rankings, where each ranking is a list of document IDs
ordered from most to least relevant
k: RRF constant (default 60, as recommended in the original paper)
Returns:
List of (document_id, rrf_score) tuples, sorted by fused score
"""
scores = {}
for ranking in rankings:
for rank, doc_id in enumerate(ranking, start=1):
if doc_id not in scores:
scores[doc_id] = 0.0
# RRF formula: 1 / (k + rank)
scores[doc_id] += 1.0 / (k + rank)
# Sort by score descending
return sorted(scores.items(), key=lambda x: x[1], reverse=True)
# Example: merging BM25 and vector search results
bm25_results = ["doc_5", "doc_2", "doc_8", "doc_1", "doc_9"]
vector_results = ["doc_1", "doc_5", "doc_3", "doc_8", "doc_7"]
fused = reciprocal_rank_fusion([bm25_results, vector_results])
print("Fused ranking:")
for doc_id, score in fused:
print(f" {doc_id}: {score:.4f}")
# Notice: doc_5 and doc_1 appear in both retrievers and get boosted to the top
RRF是此处最快的选择，几乎不会增加任何延迟。当您已经在使用混合检索方式，并希望在不进行模型推理的情况下以较低成本提升性能时，可选用它。虽然在原始准确率上无法超越交叉编码器，但考虑到其无需机器学习支持这一特点，实际表现却出奇地好。
3. Cohere 重排 API
有些团队不想自行托管和调优重排模型。如果您的重点在于产品功能而非模型运维，那么托管式的重排 API会很有吸引力，而Cohere的该服务在这一领域最为成熟。
该 API可接收查询语句及文档列表，并按相关性顺序返回结果。其背后运行着经过优化的现代重排模型，这类优化若由人工开发需要数周时间。它能处理多语言内容及长文档，并可根据需求扩展性能。
import cohere
from dotenv import load_dotenv
import os
load_dotenv()
# Initialize Cohere client
co = cohere.Client(os.getenv("COHERE_API_KEY"))
def rerank_with_cohere(query: str, documents: list[str], top_k: int = 5):
"""
Rerank documents using Cohere's managed Rerank API.
Args:
query: The user question
documents: List of document chunks from initial retrieval
top_k: Number of documents to return
Returns:
List of (document, relevance_score) tuples
"""
response = co.rerank(
model="rerank-v3.5",
query=query,
documents=documents,
top_n=top_k,
return_documents=True
)
results = []
for result in response.results:
results.append((
result.document.text,
result.relevance_score
))
return results
# Example usage
query = "How do I handle authentication in a FastAPI app?"
docs = [
"FastAPI is a modern web framework for building APIs with Python.",
"To add authentication, use OAuth2PasswordBearer and JWT tokens.",
"Pydantic models in FastAPI provide automatic request validation.",
"The OAuth2PasswordBearer class expects a token URL endpoint.",
"FastAPI was created by Sebastián Ramírez and released in 2018.",
]
ranked = rerank_with_cohere(query, docs, top_k=3)
for doc, score in ranked:
print(f"Score: {score:.4f} | {doc}")
当您需要在保证高准确度的同时避免运营负担时，可选择此方案。其代价在于每次调用的成本以及对外部依赖的需求，而在高流量场景下，网络调用的延迟也可能成为问题。对于大多数团队而言，节省在托管和优化上的时间远超过这些成本。
4. ColBERT
交叉编码器虽然准确但速度较慢，而双向编码器则快速但深度不足。ColBERT通过“延迟交互”找到了折中方案。
- 离线阶段：每份文档都会被逐个标记地编码为上下文嵌入向量，这些向量会被预先计算并存储在索引中。
- 在线阶段：当查询到来时，也会经过相同的标记级处理。
- 评分机制：每个查询标记都会与其最相似的文档标记进行匹配（即MaxSim步骤），随后将所有查询标记的最大相似度值相加，从而得出相关性得分。
由于文档嵌入是预先计算好的，因此其速度远快于交叉编码器。而且因为它保留了词元级的交互信息，在准确性方面也优于双编码器。
from colbert import Searcher
from colbert.infra import Run, RunConfig
def setup_colbert_searcher(index_path: str, checkpoint: str):
"""
Initialize a ColBERT searcher for late-interaction reranking.
Args:
index_path: Path to the pre-built ColBERT index
checkpoint: Path to the ColBERT model checkpoint
Returns:
Configured Searcher instance
"""
with Run().context(RunConfig(nranks=1, experiment="reranking")):
searcher = Searcher(
index=index_path,
checkpoint=checkpoint
)
return searcher
def rerank_with_colbert(searcher, query: str, doc_ids: list[str], top_k: int = 5):
"""
Rerank documents using ColBERT's late interaction.
Args:
searcher: Initialized ColBERT Searcher
query: The user question
doc_ids: List of document IDs from initial retrieval
top_k: Number of documents to return
Returns:
List of (doc_id, score) tuples
"""
# Search within the candidate set
results = searcher.search(
query,
k=top_k,
filter_fn=lambda pid: pid in doc_ids # Only rerank candidates
)
return list(zip(results[0], results[2])) # doc_ids, scores
# Note: ColBERT requires a pre-built index and model checkpoint.
# For production use, build the index once and load it at startup.
ColBERT适用于那些需要比双编码器更高的准确率、同时又比交叉编码器更低的延迟的大规模系统。其代价在于构建和维护索引的工作量。如果您熟悉向量数据库，并且希望在不会因候选项数量增加而出现线性延迟的情况下获得更优质的结果，那么ColBERT正是理想选择。
5. 以大语言模型作为评估器
在医疗诊断、法律研究以及金融合规等高风险领域，即便是交叉编码器也可能力不从心。这些场景中的查询十分复杂，上下文含义微妙，一旦给出错误答案会造成严重后果。在这种情况下，最精准的重新排序工具可以是大语言模型本身。
其思路是引导大型模型（如GPT-4、Claude或Gemini）对每个候选项与查询的相关性进行评分。凭借广泛的世界知识与强大的推理能力，这类模型能够以小型重新排序工具无法实现的方式来判断相关性。一个典型的提示语如下：
You are evaluating documents for a retrieval system.
Query: {query}
Document: {document}
Rate how relevant this document is for answering the query.
Respond with a single integer from 1 to 10, where 10 means perfectly relevant.
Relevance score:
对每个候选项运行该程序，解析得分后返回排名前k的选项。
from openai import OpenAI
import os
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def score_document_with_llm(query: str, document: str) -> int:
"""
Ask an LLM to score a document's relevance to a query.
Args:
query: The user question
document: A candidate document chunk
Returns:
Integer relevance score from 1-10
"""
prompt = f"""You are evaluating documents for a retrieval system.
Query: {query}
Document: {document}
Rate how relevant this document is for answering the query.
Respond with a single integer from 1 to 10, where 10 means perfectly relevant.
Be strict: only give high scores to documents that directly help answer the query.
Relevance score:"""
response = client.chat.completions.create(
model="gpt-4.1-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=5
)
try:
score = int(response.choices[0].message.content.strip())
return max(1, min(10, score)) # Clamp to 1-10
except ValueError:
return 5 # Default on parse failure
def rerank_with_llm_judge(query: str, documents: list[str], top_k: int = 3):
"""
Rerank documents using an LLM as a relevance judge.
Args:
query: The user question
documents: List of candidate document chunks
top_k: Number of documents to return
Returns:
List of (document, score) tuples, sorted by relevance
"""
scored = []
for doc in documents:
score = score_document_with_llm(query, doc)
scored.append((doc, score))
scored.sort(key=lambda x: x[1], reverse=True)
return scored[:top_k]
# Example usage
query = "What are the tax implications of RSU vesting for employees in California?"
docs = [
"RSUs are restricted stock units granted to employees as part of compensation.",
"In California, RSU income is taxed as ordinary income at vesting, not at grant.",
"Employers typically withhold federal and state taxes at vesting time.",
"Stock options and RSUs have different tax treatments under IRS rules.",
"California has one of the highest state income tax rates in the US.",
]
ranked = rerank_with_llm_judge(query, docs, top_k=3)
for doc, score in ranked:
print(f"Score: {score}/10 | {doc}")
这是目前最准确的方法，但同时也最为缓慢且成本最高。适用于那些答案错误会带来实际经济损失或严重风险的复杂、高价值查询场景。对于大多数应用而言，交叉编码器或托管型 API 即可达到约 90%的准确率，同时仅需 10%的成本（此为经验数值，并非基准数据）。只有在确实有必要时才使用大型语言模型评估工具。
如何选择
没有哪种重新排序方法在所有情况下都表现最佳。下表总结了各方法的局限性：
| Technique | Best For | Latency | Cost |
| --------------------- | ------------------------------------------------- | ------------ | -------------- |
| Cross-Encoder | Maximum quality on top-k candidates | 50-200ms | Local GPU/CPU |
| RRF | Hybrid retrieval without adding model inference | ~0ms | Free |
| Cohere Rerank API | Speed without operational overhead | 100-300ms | Per API call |
| ColBERT | Large-scale, low-latency use cases | 20-100ms | Index + GPU |
| LLM-as-a-Judge | Complex, high-value queries (medical, legal) | 1-5 seconds | Per API call |
一个合理的方案是从交叉编码器开始，因为它能在最低复杂度的情况下带来最大的准确率提升。如果延迟成为问题，可以考虑使用ColBERT或Cohere这类托管API。如果采用混合检索方式，则需加入RRF；而对于那些需要最高准确率来弥补成本的查询，再使用大语言模型作为判断器。
团队通常不必只选择其中一种方案。常见的生产环境架构如下：
- 结合RRF融合的混合检索（密集向量检索与BM25检索）。
- 对前50个候选结果进行交叉编码器重排序。
- 对于标记为高重要性的查询，以大语言模型作为最终判断器。
这样在普通情况下能保证速度，在关键场景下则能确保准确率。
总结
重排序或许是生产型RAG中最被低估的环节。团队们花费数月时间调整分块大小和嵌入模型，而最佳上下文却仍埋藏在第23位，而一个优秀的重排序器只需少量工程工作就能解决这个问题。
相关技术从免费且快速的（RRF）到成本高昂但功能强大的（以LLM作为评估器）不等，重排序也无需非得二选一。从简单方法开始，逐步叠加不同技术，只有在指标要求时再增加复杂性。
- 第一阶段检索能快速筛选出候选项，但不会直接确定答案。
- 重排序起到质量控制的作用，通过二次筛选来判断真正的相关性。
- 交叉编码器是最佳起点，具有高准确率且复杂度适中。
- 在已经采用混合检索方式的情况下，RRF能带来无需额外成本的优化效果。
如果在生产环境中运行 RAG 时不进行重排，答案质量就会大打折扣。请选择一种技术，在本周内实施并测量其效果差异。