交叉编码器重排:在向量召回后用小模型精排 TopK,提升 RAG 命中率。
RAG 通常分两阶段:先用 bi-encoder 向量召回 50–100 个候选,再用 cross-encoder 把 query 和每个文档拼在一起打分,保留 Top 5–10。交叉编码器能建模词间交互,精度通常高于向量相似度,但每个候选都要一次前向计算,延迟和成本随候选数线性增长。
常见选择是本地 BAAI/bge-reranker-v2-m3(0.6B、多语言)或 Cohere Rerank API。本地模型数据不出域、无按次费用,但要 GPU 和运维;API 省运维、支持多语言,但按搜索单元计费且有网络延迟。选型要看延迟预算、数据合规和候选规模。
Reranker 不能弥补召回缺失:如果正确文档不在候选集里,重排也找不到。先用 recall@K 评估召回,再调 reranker 的 TopK 和阈值;换模型后分数分布会变,阈值必须重新校准。
安装本地重排库和 Cohere SDK:
pip install -U FlagEmbedding sentence-transformers cohere
用 FlagEmbedding 加载 bge-reranker,normalize=True 把分数压到 0–1:
from FlagEmbedding import FlagReranker
reranker = FlagReranker("BAAI/bge-reranker-v2-m3", use_fp16=True)
pairs = [("OOMKilled 怎么排查", "先看 memory limit,再查容器内存曲线"), ("OOMKilled 怎么排查", "今天天气不错")]
print(reranker.compute_score(pairs, normalize=True))
用 sentence-transformers 的 CrossEncoder 接口,适合已有 embedding 管线:
from sentence_transformers import CrossEncoder
model = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)
query = "Pod 一直重启怎么办"
docs = ["检查 restartCount 和 events", "调整副本数", "查看 OOMKilled 事件"]
print(model.predict([(query, d) for d in docs]))
调用 Cohere Rerank,index 指向原始 documents 的下标:
import cohere
co = cohere.ClientV2(api_key="...")
docs = ["检查 restartCount 和 events", "调整副本数", "查看 OOMKilled 事件"]
res = co.rerank(model="rerank-v3.5", query="Pod 一直重启怎么办", documents=docs, top_n=2)
for item in res.results:
print(item.index, item.relevance_score, docs[item.index])
把重排接进召回结果,按分数过滤后再截断:
def rerank(query, candidates, top_k=5, threshold=0.3):
pairs = [(query, c["text"]) for c in candidates]
scores = reranker.compute_score(pairs, normalize=True)
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
return [(c, s) for c, s in ranked if s >= threshold][:top_k]
用固定候选集压测延迟,分别记录 P50 和 P95:
import time
start = time.perf_counter()
for _ in range(20):
reranker.compute_score([(query, c["text"]) for c in candidates], normalize=True)
print("avg_ms", (time.perf_counter() - start) / 20 * 1000)
检查 GPU 显存和利用率,批量过大时先降 batch size:
nvidia-smi --query-gpu=memory.used,memory.total,utilization.gpu --format=csv
对比“只用向量”和“向量+重排”的 recall@K,确认重排确实有收益:
def recall_at_k(ranked_ids, gold_id, k):
return int(gold_id in ranked_ids[:k])
print(recall_at_k([d["id"] for d in candidates], "doc-42", 5))
normalize=True 后的 0–1 也不是概率;换模型必须重新校准阈值。use_fp16=True、减小 batch,必要时换更小模型或 API。documents 必须是字符串,结构化文档先序列化;top_n 不能超过文档数,返回的是原数组下标。