为你的向量数据库加上重排序

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

向量数据库返回的是「邻居」，不是「答案」。重排序这一步位于数据库和 prompt 之间，而各家的接入方式其实是同样的三行：放大 limit、给候选打分、截断。数据库之间的差别，只在于怎么多要几行、以及怎么把 payload 带回来。

通用模式

不管用哪个存储，加重排序对检索调用的改动只有一处 —— 你要的行数比实际要用的多：

CANDIDATES = 50   # what you ask the database for
KEEP = 5          # what reaches the prompt

rows = store.search(query_vector, limit=CANDIDATES)
scores = reranker.predict([(query, r.text) for r in rows])
top = [r for _, r in sorted(zip(scores, rows), key=lambda p: -p[0])][:KEEP]

有两个细节比选哪个数据库更重要。第一，让原始行对象贯穿整个重排序过程，这样排完之后 id、元数据和权限都还在 —— 只对文本重排、之后再靠字符串匹配找回来，是丢主键的可靠方式。第二，reranker 只看得到你递给它的文本，所以如果你存的是「标题+正文」拼在一起的大块，被打分的就是那一整块。

pgvector（Postgres）

pgvector 按距离排序返回行。你只需放大 LIMIT、保留 id，然后在应用层排序：

import psycopg
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)

def search(conn, query: str, query_vec, keep: int = 5):
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT id, content
            FROM documents
            ORDER BY embedding <=> %s
            LIMIT 50
            """,
            (query_vec,),
        )
        rows = cur.fetchall()

    pairs = [(query, content) for _id, content in rows]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, rows), key=lambda p: -p[0])
    return [row for _score, row in ranked[:keep]]

<=> 是余弦距离；L2 用 <->，内积用 <#>，与建索引时用的保持一致。注意把 limit 从 5 提到 50 会和索引配置相互影响 —— 用 HNSW 时可能需要把 SET LOCAL hnsw.ef_search 调到默认值以上，索引才会真的考察足够多的邻居、返回 50 条像样的候选。limit 大于 ef_search 时，你拿到的会是凑数而非候选，而这看起来和「reranker 没起作用」一模一样。

Qdrant

结构一样，把 payload 一起带上，避免丢元数据：

from qdrant_client import QdrantClient

client = QdrantClient(url="http://localhost:6333")

def search(query: str, query_vec, keep: int = 5):
    hits = client.query_points(
        collection_name="documents",
        query=query_vec,
        limit=50,
        with_payload=True,
    ).points

    scores = reranker.predict([(query, h.payload["text"]) for h in hits])
    ranked = sorted(zip(scores, hits), key=lambda p: -p[0])
    return [h for _s, h in ranked[:keep]]

如果你按租户、权限或日期过滤，请把过滤放进 query_points 调用里，而不是重排之后再过滤。放在后面意味着你花钱给一批马上要丢掉的行打了分，而且 50 个候选可能塌缩到只剩 3 个。

Elasticsearch

Elasticsearch 上重排序的收益最直观，因为你可以喂给它一个混合候选集 —— BM25 和向量在不同查询上各自失手，而 cross-encoder 负责把两者的并集理顺：

resp = es.search(
    index="documents",
    size=50,
    query={"bool": {"should": [
        {"match": {"content": query}},
        {"knn": {"field": "embedding", "query_vector": query_vec, "k": 50}},
    ]}},
)

hits = resp["hits"]["hits"]
scores = reranker.predict([(query, h["_source"]["content"]) for h in hits])
ranked = sorted(zip(scores, hits), key=lambda p: -p[0])
top = [h for _s, h in ranked[:5]]

BM25 分数和 kNN 分数量纲不同、没有可比性，这正是通常要用 RRF（倒数排名融合）的理由。cross-encoder 则直接绕开了这个问题：它用自己的一套尺度给每个候选重新打分，于是候选是怎么进池子的就不再重要了。

当数据库自带重排序

现在有几家存储支持在查询内部完成重排序 —— 或是与托管模型的托管式集成，或是原生的第二阶段。用它是个合理的默认选择，能省掉一次网络往返和一堆胶水代码。

但在投入之前，值得知道你让渡了什么：

• 模型选择被限制在厂商已集成的范围内，而在你的领域表现最好的那个未必在列。
• 评测变难 —— 当其中一个 reranker 住在查询引擎内部时，做 A/B 对比要费更多力气。
• 你的 API 密钥和段落内容，现在除了模型厂商，还会流经数据库厂商。

上面这套应用层写法在各家之间都是可移植的，所以即便你之后打算把这一步挪进数据库，从它起步也很稳妥。具体支持情况请查阅你所用存储的最新文档 —— 这块变化很快，我们刻意不去维护一张自己没把握保持准确的厂商功能对照表。

常见坑

cross-encoder 的开销与候选数量线性相关，所以从 50 提到 200，重排延迟大致翻两番，用托管 API 的话账单也一样。后半句的具体数字可以用成本计算器算出来。

继续阅读