AI 专题 DEC 05, 2025
RAG 检索优化：混合搜索、Reranking、多路召回
#RAG#检索优化#Reranking#混合搜索
RAG 检索优化：混合搜索、Reranking、多路召回
本文是【AI 专题精讲】系列第 06 篇。 上一篇：向量数据库选型：ChromaDB / pgvector / Pinecone / Milvus | 下一篇：意图识别：从关键词匹配到语义路由
这篇文章你会得到什么
前五篇我们搭好了 RAG 的基础设施：文件解析 → 切片 → 上传 → Embedding → 向量库。现在用户问一句话，能从知识库里检索出相关内容了。
但你很快会发现——基础的向量检索不够准。
用户问”2024年Q3财报营收多少”，返回的是一段讲公司愿景的内容。用户问”张三的工号”，明明知识库里有，就是搜不出来。
纯向量检索有天然的短板，今天我们用三板斧把召回率从 80% 拉到 95%：
- 混合搜索：向量检索 + 关键词检索，两条路同时走
- Reranking：用更精准的模型对候选结果二次排序
- 多路召回：不同策略并行检索，合并去重
再加上 Query 改写 和 评测体系，给你一套完整的检索优化 Pipeline。
纯向量检索的局限
先搞清楚为什么”只用向量搜索”不够。
问题 1：关键词精确匹配差
向量检索理解语义，但对精确关键词不敏感：
用户查询："合同编号 HT-2024-0815"
向量检索返回：一段讲"合同签署流程"的内容（语义相关但不是想要的）
期望返回：包含 "HT-2024-0815" 这个具体编号的内容编号、人名、专有名词——这些需要精确匹配的场景，向量检索天然弱势。
问题 2：短查询信息不足
用户查询："年假"
向量检索返回：可能命中"假期"、"休息"、"放假"等语义相近但不相关的内容查询太短，向量的语义空间很模糊，检索结果就容易飘。
问题 3：新词和低频词
Embedding 模型的训练数据不一定覆盖你的领域术语。“人天工时""BOM 物料清单”这些行业术语，向量表示可能不准。
方案一：混合搜索（Hybrid Search）
核心思路：向量检索找语义相关的，关键词检索找精确匹配的，两个结果融合。
BM25：经典关键词检索
BM25 是搜索引擎的经典算法，本质是”关键词出现频率 + 文档长度归一化”：
from rank_bm25 import BM25Okapi
import jieba
class BM25Retriever:
def __init__(self, chunks: list[dict]):
self.chunks = chunks
tokenized = [list(jieba.cut(c["content"])) for c in chunks]
self.bm25 = BM25Okapi(tokenized)
def search(self, query: str, top_k: int = 10) -> list[dict]:
tokens = list(jieba.cut(query))
scores = self.bm25.get_scores(tokens)
top_indices = scores.argsort()[::-1][:top_k]
return [
{
"id": self.chunks[i]["id"],
"content": self.chunks[i]["content"],
"score": float(scores[i]),
}
for i in top_indices
if scores[i] > 0
]BM25 对中文需要先分词（jieba），英文用空格分就行。
RRF：融合排序
两路检索各返回一个排序列表，怎么合并？RRF（Reciprocal Rank Fusion） 是最简单有效的方案：
def reciprocal_rank_fusion(
results_list: list[list[dict]],
k: int = 60,
top_k: int = 10,
) -> list[dict]:
"""
RRF 融合多路检索结果。
对每个文档，RRF 分数 = sum(1 / (k + rank_i))
"""
scores = {}
content_map = {}
for results in results_list:
for rank, item in enumerate(results, 1):
doc_id = item["id"]
scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank)
content_map[doc_id] = item
sorted_ids = sorted(scores, key=scores.get, reverse=True)[:top_k]
return [
{**content_map[doc_id], "rrf_score": scores[doc_id]}
for doc_id in sorted_ids
]RRF 的优势：不需要对齐两路的分数尺度。向量检索返回余弦相似度（0~1），BM25 返回的是另一种分数，直接加权不靠谱，但 RRF 只看排名位置，天然兼容。
完整混合搜索 Pipeline
class HybridSearcher:
def __init__(
self,
vector_store: VectorStore,
embedding_provider: EmbeddingProvider,
chunks: list[dict],
):
self.vector_store = vector_store
self.embedding_provider = embedding_provider
self.bm25 = BM25Retriever(chunks)
async def search(
self,
query: str,
top_k: int = 5,
vector_weight: float = 0.7,
) -> list[dict]:
# 1. 向量检索
query_result = self.embedding_provider.embed([query])
vector_results = await self.vector_store.search(
query_embedding=query_result.embeddings[0],
top_k=top_k * 2,
)
vector_list = [
{"id": r.id, "content": r.content, "score": r.score}
for r in vector_results
]
# 2. BM25 检索
bm25_list = self.bm25.search(query, top_k=top_k * 2)
# 3. RRF 融合
fused = reciprocal_rank_fusion(
[vector_list, bm25_list],
top_k=top_k,
)
return fused混合搜索的效果提升
以我的实际项目数据（500 chunks，50 条测试 query）：
| 方案 | Hit@5 | MRR |
|---|---|---|
| 纯向量检索 | 85% | 0.78 |
| 纯 BM25 | 72% | 0.65 |
| 混合搜索（RRF） | 92% | 0.86 |
两者互补效果明显：向量搜索找到 BM25 搜不到的语义匹配，BM25 补上向量搜索漏掉的关键词匹配。
方案二：Reranking（精排）
混合搜索召回了更多候选，但排序可能不够精准。Reranking 用更强的模型对候选结果二次排序。
为什么需要 Reranking
Embedding 模型是双编码器（Bi-Encoder）——query 和 document 分别编码，只能通过向量距离粗略比较。
Reranker 是交叉编码器（Cross-Encoder）——把 query 和 document 拼在一起输入模型，能做更深层的语义交互，排序更准确。
代价是速度慢——不能预计算，每次查询都要跑模型。所以 Reranker 只用在已召回的少量候选上（通常 10~30 条）。
用 BGE Reranker
from sentence_transformers import CrossEncoder
class BGEReranker:
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model = CrossEncoder(model_name)
def rerank(
self,
query: str,
candidates: list[dict],
top_k: int = 5,
) -> list[dict]:
pairs = [(query, c["content"]) for c in candidates]
scores = self.model.predict(pairs)
scored = [
{**candidates[i], "rerank_score": float(scores[i])}
for i in range(len(candidates))
]
scored.sort(key=lambda x: x["rerank_score"], reverse=True)
return scored[:top_k]用 Cohere Rerank API
不想本地部署，可以用 Cohere 的 Rerank API：
import httpx
class CohereReranker:
def __init__(self, api_key: str):
self.api_key = api_key
def rerank(
self,
query: str,
candidates: list[dict],
top_k: int = 5,
) -> list[dict]:
response = httpx.post(
"https://api.cohere.com/v2/rerank",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "rerank-v3.5",
"query": query,
"documents": [c["content"] for c in candidates],
"top_n": top_k,
},
)
data = response.json()
return [
{
**candidates[r["index"]],
"rerank_score": r["relevance_score"],
}
for r in data["results"]
]Jina Reranker
class JinaReranker:
def __init__(self, api_key: str):
self.api_key = api_key
def rerank(
self,
query: str,
candidates: list[dict],
top_k: int = 5,
) -> list[dict]:
response = httpx.post(
"https://api.jina.ai/v1/rerank",
headers={"Authorization": f"Bearer {self.api_key}"},
json={
"model": "jina-reranker-v2-base-multilingual",
"query": query,
"documents": [c["content"] for c in candidates],
"top_n": top_k,
},
)
data = response.json()
return [
{
**candidates[r["index"]],
"rerank_score": r["relevance_score"],
}
for r in data["results"]
]Reranking 效果对比
| 方案 | Hit@5 | MRR | 平均延迟 |
|---|---|---|---|
| 混合搜索（无 Rerank） | 92% | 0.86 | 50ms |
| + BGE Reranker（本地） | 96% | 0.92 | 120ms |
| + Cohere Rerank（API） | 95% | 0.91 | 300ms |
Reranking 在 MRR 上的提升尤其明显——正确答案被排到更前面了。
方案三：多路召回
不同的检索策略各有所长，并行跑多路再融合，是进一步提升召回率的方案。
策略 1：不同 chunk_size 的多路召回
async def multi_chunk_recall(
query: str,
stores: dict[str, VectorStore],
embedding_provider: EmbeddingProvider,
top_k: int = 10,
) -> list[dict]:
"""
用不同 chunk_size 切出的向量库并行召回。
小 chunk 精准，大 chunk 上下文完整。
"""
query_result = embedding_provider.embed([query])
query_vec = query_result.embeddings[0]
all_results = []
for label, store in stores.items():
results = await store.search(query_vec, top_k=top_k)
all_results.append([
{"id": r.id, "content": r.content, "score": r.score, "source": label}
for r in results
])
return reciprocal_rank_fusion(all_results, top_k=top_k)策略 2：Query 扩展多路召回
用 LLM 把用户的一句查询改写成多个版本，每个版本独立检索：
async def expand_query(query: str, client) -> list[str]:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"你是一个搜索查询扩展助手。"
"给定用户的查询，生成 3 个语义相关但措辞不同的查询变体。"
"每行一个，不要编号。"
),
},
{"role": "user", "content": query},
],
temperature=0.7,
)
variants = response.choices[0].message.content.strip().split("\n")
return [query] + [v.strip() for v in variants if v.strip()]
async def multi_query_recall(
query: str,
store: VectorStore,
embedding_provider: EmbeddingProvider,
client,
top_k: int = 10,
) -> list[dict]:
queries = await expand_query(query, client)
all_results = []
for q in queries:
q_result = embedding_provider.embed([q])
results = await store.search(q_result.embeddings[0], top_k=top_k)
all_results.append([
{"id": r.id, "content": r.content, "score": r.score}
for r in results
])
return reciprocal_rank_fusion(all_results, top_k=top_k)举例：
原始查询："年假怎么算"
LLM 扩展后：
- "年假怎么算"
- "员工年假天数计算规则"
- "工龄对应的年假标准"
- "带薪休假的计算方式"四个查询各自检索，合并后的召回率比单查询高很多。
策略 3：HyDE（假设性文档）
让 LLM 生成一段”假设性答案”，用答案去检索，而不是用问题：
async def hyde_retrieval(
query: str,
store: VectorStore,
embedding_provider: EmbeddingProvider,
client,
top_k: int = 5,
) -> list[dict]:
# LLM 生成假设性答案
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "请直接回答以下问题，不需要说明来源。如果不确定，给出你最好的猜测。",
},
{"role": "user", "content": query},
],
temperature=0.3,
)
hypothetical_doc = response.choices[0].message.content
# 用假设性答案做检索（答案和文档的语义更接近）
result = embedding_provider.embed([hypothetical_doc])
return await store.search(result.embeddings[0], top_k=top_k)原理：问题和答案的向量空间分布不同。“年假怎么算”（问题）和”员工入职满一年享有5天年假”（答案/文档）的向量距离，不如”员工年假标准是…”（假设性答案）和实际文档的距离近。
完整 Pipeline：检索 → 召回 → 精排
把上面的方案串起来：
class RAGRetriever:
def __init__(
self,
vector_store: VectorStore,
embedding_provider: EmbeddingProvider,
chunks: list[dict],
reranker: BGEReranker | None = None,
):
self.vector_store = vector_store
self.embedding_provider = embedding_provider
self.bm25 = BM25Retriever(chunks)
self.reranker = reranker
async def retrieve(
self,
query: str,
top_k: int = 5,
use_rerank: bool = True,
) -> list[dict]:
# 1. 粗召回：混合搜索，多拿一些候选
recall_k = top_k * 4 if use_rerank else top_k
query_result = self.embedding_provider.embed([query])
vector_results = await self.vector_store.search(
query_embedding=query_result.embeddings[0],
top_k=recall_k,
)
vector_list = [
{"id": r.id, "content": r.content, "score": r.score}
for r in vector_results
]
bm25_list = self.bm25.search(query, top_k=recall_k)
# 2. RRF 融合
candidates = reciprocal_rank_fusion(
[vector_list, bm25_list],
top_k=recall_k,
)
# 3. 精排（可选）
if use_rerank and self.reranker and len(candidates) > 0:
candidates = self.reranker.rerank(query, candidates, top_k=top_k)
else:
candidates = candidates[:top_k]
return candidates使用：
retriever = RAGRetriever(
vector_store=store,
embedding_provider=create_embedding_provider("bge"),
chunks=all_chunks,
reranker=BGEReranker(),
)
results = await retriever.retrieve("2024年Q3营收是多少", top_k=5)
for r in results:
print(f"[{r.get('rerank_score', r.get('rrf_score', 0)):.3f}] {r['content'][:80]}")评测体系
优化没有度量就是盲优化。RAG 检索的核心评测指标：
指标定义
| 指标 | 含义 | 计算 |
|---|---|---|
| Hit Rate@K | Top-K 结果中包含正确答案的比例 | 命中数 / 总查询数 |
| MRR | 正确答案的平均排名倒数 | mean(1/rank) |
| NDCG@K | 考虑位置权重的排序质量 | 排在前面的正确结果加分更多 |
评测脚本
import numpy as np
def evaluate_retriever(
retriever: RAGRetriever,
test_cases: list[dict],
top_k: int = 10,
) -> dict:
hits_at = {3: 0, 5: 0, 10: 0}
reciprocal_ranks = []
for case in test_cases:
results = retriever.retrieve(case["query"], top_k=top_k)
result_ids = [r["id"] for r in results]
relevant = set(case["relevant_ids"])
for k in hits_at:
if relevant & set(result_ids[:k]):
hits_at[k] += 1
for rank, rid in enumerate(result_ids, 1):
if rid in relevant:
reciprocal_ranks.append(1.0 / rank)
break
else:
reciprocal_ranks.append(0.0)
n = len(test_cases)
return {
"hit_rate@3": hits_at[3] / n,
"hit_rate@5": hits_at[5] / n,
"hit_rate@10": hits_at[10] / n,
"mrr": np.mean(reciprocal_ranks),
}评测数据怎么来
- 人工标注：从真实用户问题中抽样，人工标记正确的 chunk_id。最准但最费时间。
- LLM 辅助标注：把 query 和每个 chunk 扔给 LLM 判断相关性。速度快但有噪音。
- 用户反馈：用户点”有用/没用”，积累真实反馈数据。长期最靠谱。
我的建议：先用 LLM 批量标注一批，人工抽检修正，快速建立起评测集。
调参实战经验
1. RRF 的 k 值
默认 60，但不同场景可以调：
k 越小 → 排名靠前的结果权重越大 → 更激进
k 越大 → 排名差距的权重越小 → 更平滑我的经验：中文知识库用 40~60 比较稳。
2. 向量检索和 BM25 的召回数量
Reranking 之前要多召回一些候选：
recall_k = final_top_k * 3 # 至少 3 倍
recall_k = final_top_k * 5 # 搭配 Reranker 建议 5 倍3. Reranker 的候选数量
Cross-Encoder 速度慢，候选太多延迟上去了：
| 候选数 | BGE Reranker 延迟 |
|---|---|
| 10 条 | ~50ms |
| 20 条 | ~90ms |
| 50 条 | ~200ms |
推荐 20~30 条候选做 Reranking，性价比最高。
4. 分数阈值
检索结果不是越多越好。低相关度的 chunk 塞进 Prompt 反而干扰 LLM：
def filter_by_threshold(results: list[dict], min_score: float = 0.3) -> list[dict]:
return [r for r in results if r.get("rerank_score", r.get("score", 0)) >= min_score]总结
- 纯向量检索不够用——关键词匹配差、短查询飘、领域词汇弱。
- 混合搜索是基本操作——向量 + BM25 + RRF 融合，召回率直接提升 7~10 个百分点。
- Reranking 提升排序质量——BGE Reranker 本地部署免费，MRR 提升显著。
- 多路召回锦上添花——Query 改写、HyDE、多 chunk_size 并行，进一步提高上限。
- 评测驱动优化——没有 Hit Rate 和 MRR 数据，一切优化都是猜测。
- 完整 Pipeline：粗召回（混合搜索 × 多路）→ RRF 融合 → 精排（Reranker）→ 阈值过滤。
下一篇我们跳出 RAG，聊一个更通用的 AI 应用话题：意图识别。用户输入五花八门，怎么判断他是想查知识库、还是闲聊、还是下指令？
下一篇预告：07 | 意图识别：从关键词匹配到语义路由
讨论话题：你的 RAG 系统检索准确率够用吗？用了 Reranking 吗？评论区聊聊你的优化经验。