如何为你的 RAG 流水线加重排序

实践 · 约 10 分钟阅读 · 更新于 2026 年 6 月 21 日

改善平庸 RAG 答案的办法，很少是换更大的 LLM —— 通常是更好的检索排序。reranker 位于向量搜索和 LLM 调用之间，对候选重新打分，确保只有最相关的段落进入 prompt。

RAG 中的排序问题

典型的 RAG 系统会嵌入你的文档并存储向量。查询时它取出与查询嵌入最近的 k 个向量，把这些片段塞进 LLM 的 prompt。问题在于：独立嵌入之间的余弦相似度是个粗糙的相关性信号。正确的片段可能在前 20 个结果里，却排在第 14 位 —— 落在你真正送进模型的那 5 个之外。

这是“lost in the middle（迷失在中间）”问题的反面：正确答案一开始就没排在最前。重排序通过对检索器已经找到的短名单施加一个更昂贵、更准确的相关性模型来修复它。

检索—重排序—生成 范式

检索器负责规模（毫秒级处理数百万文档）。reranker 负责质量（对几十个候选做精确排序）。LLM 负责综合。每个阶段只做自己擅长的事。

代码讲解

下面是一个使用本地 bge-reranker 的自包含 Python 示例。在生产中，如果你更喜欢托管 API，可以把 rank_documents 换成对 Cohere、Jina 或 Voyage 的调用。

使用本地 cross-encoder（sentence-transformers）

from sentence_transformers import CrossEncoder

# Load once at startup — reuse across requests
reranker = CrossEncoder("BAAI/bge-reranker-base", max_length=512)

def rerank(query: str, candidates: list[str], top_n: int = 5) -> list[str]:
    """Return top_n candidates reranked by relevance to query."""
    pairs = [(query, doc) for doc in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, candidates), reverse=True)
    return [doc for _, doc in ranked[:top_n]]

# --- In your RAG pipeline ---
raw_chunks = vector_db.search(query, top_k=50)   # retrieve wide
best_chunks = rerank(query, raw_chunks, top_n=5)  # rerank tight
answer = llm.complete(build_prompt(query, best_chunks))  # generate

使用 Cohere 托管 API

import cohere

co = cohere.Client("YOUR_API_KEY")

def rerank(query: str, candidates: list[str], top_n: int = 5) -> list[str]:
    result = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=candidates,
        top_n=top_n,
    )
    return [candidates[r.index] for r in result.results]

使用 Jina Reranker API

import requests

def rerank(query: str, candidates: list[str], top_n: int = 5) -> list[str]:
    resp = requests.post(
        "https://api.jina.ai/v1/rerank",
        headers={"Authorization": "Bearer YOUR_KEY"},
        json={
            "model": "jina-reranker-v2-base-multilingual",
            "query": query,
            "documents": candidates,
            "top_n": top_n,
        },
    ).json()
    indices = [r["index"] for r in resp["results"]]
    return [candidates[i] for i in indices]

在框架内（LangChain · LlamaIndex · Haystack）

如果你已经在用某个 RAG 框架，重排序通常就是一个即插即用的节点，把你现有的检索器包起来。模式完全一样 —— 宽召回、重排序、保留 top-n —— 只是换成了框架的术语来表达。

# LangChain — ContextualCompressionRetriever wraps any base retriever
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")
compressor = CrossEncoderReranker(model=model, top_n=5)
retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 50}),
)
docs = retriever.invoke(query)   # retrieve 50 → rerank → top 5

# LlamaIndex — a node postprocessor on the query engine
from llama_index.core.postprocessor import SentenceTransformerRerank

reranker = SentenceTransformerRerank(model="BAAI/bge-reranker-base", top_n=5)
query_engine = index.as_query_engine(
    similarity_top_k=50,                 # retrieve wide
    node_postprocessors=[reranker],       # rerank to top 5
)
response = query_engine.query(query)

# Haystack 2.x — a Ranker component in the pipeline
from haystack import Pipeline
from haystack.components.rankers import TransformersSimilarityRanker

ranker = TransformersSimilarityRanker(model="BAAI/bge-reranker-base", top_k=5)
pipe = Pipeline()
pipe.add_component("retriever", retriever)   # returns ~50 docs
pipe.add_component("ranker", ranker)
pipe.connect("retriever.documents", "ranker.documents")

这三者也都能通过各自的集成包接入托管 reranker（Cohere、Jina、Voyage）—— 换掉组件，流水线照旧。

如何选 top-k 取值

你有两个 k 值要调：召回多少个，以及重排序后保留多少个。

延迟取舍

重排序会给你的流水线增加一次模型调用。成本取决于具体方案：

对大多数 RAG 应用来说，整条流水线 100–300 ms 的延迟完全可以接受，而质量提升是值得的。如果你的 SLA 非常紧，可以部署到 GPU、改用微型模型，或把 retrieval_k 从 50 降到 20–30。

如何衡量它是否真的有帮助

不要凭感觉就加 reranker —— 要量化它。构建一个小型评测集，把查询与应当被检索到的片段配对，然后在同一集合上对比「仅检索」与「检索 + 重排序」。三个指标几乎覆盖所有场景：

MRR 最直观：正确片段排第 1 得 1.0，排第 2 得 0.5，排第 4 得 0.25。在整个查询集上取平均，就得到一个数字，刻画「答案离顶部有多近？」—— 这正是 reranker 应当改善的东西。

# Minimal MRR + Recall@k over a labelled eval set.
# Each example: a query, the candidate texts, and the indices that are relevant.
def reciprocal_rank(ranked_ids, relevant_ids):
    for i, doc_id in enumerate(ranked_ids, start=1):
        if doc_id in relevant_ids:
            return 1.0 / i
    return 0.0

def evaluate(examples, rank_fn, k=5):
    mrr = recall = 0.0
    for ex in examples:
        ranked = rank_fn(ex["query"], ex["candidates"])  # returns ids, best first
        mrr += reciprocal_rank(ranked, ex["relevant"])
        recall += 1.0 if set(ranked[:k]) & set(ex["relevant"]) else 0.0
    n = len(examples)
    return {"MRR": mrr / n, f"Recall@{k}": recall / n}

base = evaluate(eval_set, retrieve_only)        # baseline
reranked = evaluate(eval_set, retrieve_then_rerank)  # with the reranker
print(base, reranked)   # expect MRR and nDCG to rise after reranking

该选哪个 reranker

简短版：

• 想自建、仅英文、免费：bge-reranker-v2-m3 —— 强大且广泛部署。
• 想要托管 API、最佳多语言质量：Cohere Rerank v3.5。
• 想要开源权重 + 托管 API + 可在浏览器运行的微型模型：Jina Reranker v2。
• 专注检索专项质量：Voyage Rerank 2。
• Want Apache 2.0 open weights, browser-runnable xsmall: mxbai-rerank.

查看完整模型对比 →

Keep reading