RAG 进阶检索架构
从单路向量检索到多路混合检索:掌握混合检索融合、Rerank 精排、层级 RAG(父子块)、结构化与非结构化联合检索、检索路由策略的完整架构设计。
1. 朴素 RAG 的瓶颈与进阶方向
1.1 标准 RAG 架构回顾与瓶颈分析
标准 RAG 流程(面试必画):
用户提问 → Embedding → 向量检索 Top-K → 拼接 Prompt → LLM 生成
这个流程的三大瓶颈:
瓶颈 1:语义鸿沟(Semantic Gap)
→ 用户问"苹果股价",向量检索可能返回"苹果的营养价值"
→ 向量检索对精确关键词/专有名词不敏感
→ 解决:混合检索(向量 + 关键词 BM25)
瓶颈 2:排序粗糙(Ranking Quality)
→ 向量相似度 ≠ 相关性,Top-5 中可能 3 条无关
→ Embedding 模型的能力上限导致粗排不够精准
→ 解决:Rerank 重排序(粗检索 100 条 → 精排 5 条)
瓶颈 3:上下文断裂(Context Fragmentation)
→ 小块检索精准,但丢失上下文
→ 大块保留上下文,但检索不精准
→ 解决:层级 RAG(小块检索 → 返回父块)1.2 进阶 RAG 架构全景图
进阶 RAG 检索架构全景:
用户提问
│
▼
┌──────────────┐
│ ① 查询处理 │ 查询改写 / HyDE / Multi-Query
└──────┬───────┘
│
▼
┌──────────────┐
│ ② 检索路由 │ 根据问题类型选择检索策略
└──┬───┬───┬──┘
│ │ │
▼ ▼ ▼
┌────┐┌────┐┌────┐
│向量││BM25││SQL │ ③ 多路并行检索
│检索││检索││查询│
└─┬──┘└─┬──┘└─┬──┘
│ │ │
└─────┼─────┘
▼
┌──────────────┐
│ ④ 融合排序 │ RRF / 加权融合
└──────┬───────┘
▼
┌──────────────┐
│ ⑤ Rerank │ Cross-Encoder 精排
└──────┬───────┘
▼
┌──────────────┐
│ ⑥ 上下文扩展 │ 父块扩展 / 窗口扩展
└──────┬───────┘
▼
┌──────────────┐
│ ⑦ LLM 生成 │ 拼接 Prompt + 生成答案
└──────────────┘💡 面试答题框架:被问"RAG 怎么优化"时,按这 7 层回答——查询改写 → 多路检索 → 融合排序 → Rerank → 上下文扩展 → LLM 生成。
2. 多路混合检索架构
单靠向量检索不够——关键词搜不到、专有名词匹配不上。混合检索是生产环境的标配。
2.1 向量检索 vs 关键词检索 vs 全文检索
三种检索方式对比:
检索方式 原理 擅长 不擅长
─────────────────────────────────────────────────────────────
向量检索 Embedding 相似度 语义理解/同义词 精确关键词/专有名词
关键词 BM25 词频统计 精确匹配/专有名词 语义理解/同义词
全文检索 倒排索引+TF-IDF 长文本/模糊匹配 语义理解
举例:
→ 问"Python 怎么读取 Excel"
→ 向量检索可能返回"如何用编程语言处理表格"(语义对但模糊)
→ BM25 直接匹配"Python" + "Excel"(精确命中)
→ 混合检索两个都返回,效果最好2.2 混合检索融合策略:RRF 与加权融合
# ═══════════════════════════════════════
# RRF(Reciprocal Rank Fusion)融合算法
# ═══════════════════════════════════════
def reciprocal_rank_fusion(
results_lists: list[list[dict]],
k: int = 60,
) -> list[dict]:
"""
RRF 融合:多路检索结果合并排序
- 每条结果的得分 = Σ 1/(k + rank)
- k=60 是论文推荐的常数
"""
scores = {}
for results in results_lists:
for rank, doc in enumerate(results, 1):
doc_id = doc["id"]
if doc_id not in scores:
scores[doc_id] = {"doc": doc, "score": 0}
scores[doc_id]["score"] += 1 / (k + rank)
# 按融合分数排序
fused = sorted(scores.values(), key=lambda x: x["score"], reverse=True)
return [item["doc"] for item in fused]
# 用法:
# vector_results = vector_store.search(query_emb, top_k=50)
# bm25_results = bm25_search(query_text, top_k=50)
# fused = reciprocal_rank_fusion([vector_results, bm25_results])
# → 融合后取 Top-52.3 实战:Qdrant + BM25 混合检索
# ═══════════════════════════════════════
# 混合检索器
# ═══════════════════════════════════════
from rank_bm25 import BM25Okapi
import jieba
class HybridRetriever:
"""混合检索:向量 + BM25"""
def __init__(self, vector_store, documents: list[dict]):
self.vector_store = vector_store
# 构建 BM25 索引
tokenized = [list(jieba.cut(d["content"])) for d in documents]
self.bm25 = BM25Okapi(tokenized)
self.documents = documents
def search(self, query: str, query_embedding: list[float],
top_k: int = 5, vector_weight: float = 0.6) -> list[dict]:
"""混合检索"""
# 路线 1: 向量检索
vector_results = self.vector_store.search(query_embedding, top_k=50)
# 路线 2: BM25 关键词检索
query_tokens = list(jieba.cut(query))
bm25_scores = self.bm25.get_scores(query_tokens)
bm25_top = sorted(
enumerate(bm25_scores), key=lambda x: x[1], reverse=True
)[:50]
bm25_results = [self.documents[i] for i, _ in bm25_top]
# 融合
fused = reciprocal_rank_fusion([vector_results, bm25_results])
return fused[:top_k]💡 面试答题要点:"混合检索用 RRF 融合,向量检索解决语义匹配,BM25 解决精确关键词匹配,两路互补。"
3. Rerank 重排序架构
粗检索只是"海选",Rerank 才是"精挑"——把真正相关的结果排到最前面。
3.1 粗检索 + 精排序的两阶段架构
两阶段检索架构:
阶段 1: 粗检索(Retrieval)
→ 方式:向量 + BM25 混合
→ 范围:从 100 万条中召回 Top-100
→ 速度:快(毫秒级)
→ 精度:一般(可能混入无关结果)
阶段 2: 精排序(Rerank)
→ 方式:Cross-Encoder 逐对打分
→ 范围:对 Top-100 重新排序,取 Top-5
→ 速度:慢(百毫秒级,但只排 100 条)
→ 精度:高(直接建模 query-doc 相关性)
为什么不直接用 Rerank 检索全部?
→ Cross-Encoder 需要逐对比较,100 万条太慢
→ 所以先粗检索缩小范围,再精排序提升质量3.2 Rerank 模型选型与对比
主流 Rerank 模型:
模型 中文 部署方式 效果 成本
──────────────────────────────────────────────────────────
Cohere Rerank v3 ✅ 云端 API ⭐⭐⭐ $2/1000次
BGE-Reranker-v2-m3 ⭐⭐⭐ 本地部署 ⭐⭐⭐ 免费
Jina-Reranker-v2 ✅ 本地/API ⭐⭐ 免费
ms-marco-MiniLM (英文) ❌ 本地部署 ⭐⭐ 免费
推荐:
→ 中文场景 → BGE-Reranker-v2-m3(免费、中文效果最好)
→ 快速验证 → Cohere Rerank API(零部署)3.3 实战:Cross-Encoder Rerank
# ═══════════════════════════════════════
# Rerank 实现(Cross-Encoder)
# ═══════════════════════════════════════
# pip install sentence-transformers
from sentence_transformers import CrossEncoder
class Reranker:
"""Cross-Encoder 重排序"""
def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, documents: list[dict],
top_k: int = 5) -> list[dict]:
"""对检索结果重新排序"""
# 构建 query-doc 对
pairs = [(query, doc["content"]) for doc in documents]
# Cross-Encoder 打分
scores = self.model.predict(pairs)
# 按分数排序
scored_docs = list(zip(documents, scores))
scored_docs.sort(key=lambda x: x[1], reverse=True)
return [
{**doc, "rerank_score": float(score)}
for doc, score in scored_docs[:top_k]
]
# 用法:
# retriever = HybridRetriever(vector_store, documents)
# candidates = retriever.search(query, query_emb, top_k=50)
# reranker = Reranker()
# final = reranker.rerank(query, candidates, top_k=5)
# → 检索质量显著提升(Recall@5 通常提升 10-20%)💡 面试答题要点:"Rerank 是 RAG 质量提升最明显的优化手段——只加一个 Rerank 步骤,Recall@5 通常提升 10-20%,投入产出比最高。"
4. 层级 RAG 架构(父子块)
分块的两难困境——小块精准但缺上下文,大块完整但检索不准。层级 RAG 同时解决两个问题。
4.1 父子块架构:小块检索 + 大块生成
父子块架构核心思想:
传统方案(只有一种块):
→ chunk_size=500(检索精准,但上下文少)
→ chunk_size=2000(上下文多,但检索不精准)
→ 怎么都不完美
父子块方案(两层块):
→ 子块(Child):200 字,用于检索(精准匹配)
→ 父块(Parent):1000 字,用于生成(完整上下文)
→ 检索时用子块 → 命中后返回对应的父块给 LLM
┌─────────────────────────────────────┐
│ 父块(1000 字) │ ← 这个给 LLM
│ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │子块1 │ │子块2 │ │子块3 │ │ ← 这些用来检索
│ └─────┘ └─────┘ └─────┘ │
└─────────────────────────────────────┘4.2 摘要级粗检 → 细节块精检
两级检索架构:
Level 1: 文档摘要级检索
→ 每篇文档生成一段摘要(LLM 总结)
→ 先用摘要做粗检索,定位相关文档
→ 快速过滤不相关的文档
Level 2: 细节块级精检
→ 在定位到的文档内,搜索具体的内容块
→ 返回精确匹配的段落
效果:
→ 100 篇文档 × 50 块 = 5000 个块(传统方案全搜)
→ 摘要检索选出 5 篇 × 50 块 = 250 个块(缩小 20 倍)4.3 实战:ParentDocumentRetriever
# ═══════════════════════════════════════
# 父子块检索器
# ═══════════════════════════════════════
class ParentChildRetriever:
"""父子块检索:子块检索 + 父块返回"""
def __init__(self, vector_store, parent_store: dict):
self.vector_store = vector_store # 子块向量库
self.parent_store = parent_store # 父块存储 {parent_id: content}
def ingest(self, documents: list[dict],
child_size: int = 200, parent_size: int = 1000):
"""入库:同时生成父块和子块"""
for doc in documents:
# 生成父块
parent_chunks = chunk_by_size(doc["content"], parent_size, overlap=0)
for p_idx, parent_text in enumerate(parent_chunks):
parent_id = f"{doc['id']}_p{p_idx}"
self.parent_store[parent_id] = parent_text
# 在父块内生成子块
child_chunks = chunk_by_size(parent_text, child_size, overlap=20)
for c_idx, child_text in enumerate(child_chunks):
self.vector_store.upsert({
"content": child_text,
"parent_id": parent_id, # 关联到父块
"embedding": embed(child_text),
})
def search(self, query_embedding: list[float], top_k: int = 3) -> list[str]:
"""检索子块 → 返回父块"""
# 用子块检索
child_results = self.vector_store.search(query_embedding, top_k=top_k * 2)
# 去重:同一个父块只返回一次
seen_parents = set()
parent_texts = []
for child in child_results:
parent_id = child["metadata"]["parent_id"]
if parent_id not in seen_parents:
seen_parents.add(parent_id)
parent_texts.append(self.parent_store[parent_id])
if len(parent_texts) >= top_k:
break
return parent_texts # 返回父块给 LLM💡 面试答题要点:"层级 RAG 解决分块粒度的两难问题——小块保证检索精度,父块保证上下文完整性。"
5. 混合数据源联合检索
企业知识不只在文档里——数据库有结构化数据,知识图谱有关系数据。联合检索才能全覆盖。
5.1 结构化 + 非结构化联合检索
联合检索架构:
用户问:"上个月销售额最高的产品是什么?相关的产品手册在哪?"
路线 1(结构化): Text-to-SQL
→ 查 MySQL:"SELECT product, SUM(amount) ... ORDER BY ... LIMIT 1"
→ 结果:"产品 A,销售额 500 万"
路线 2(非结构化): RAG 向量检索
→ 搜向量库:"产品 A 手册"
→ 结果:"产品 A 技术规格书.pdf 第 3 页"
融合输出:
→ "上月销售额最高的是产品 A(500 万),这是它的技术手册..."5.2 Text-to-SQL + RAG 混合架构
# ═══════════════════════════════════════
# Text-to-SQL + RAG 混合检索
# ═══════════════════════════════════════
class HybridDataRetriever:
"""结构化 + 非结构化联合检索"""
def __init__(self, llm, db_engine, vector_store):
self.llm = llm
self.db = db_engine
self.vector_store = vector_store
async def retrieve(self, query: str, query_embedding: list[float]) -> dict:
"""根据问题同时查数据库和向量库"""
results = {}
# 尝试 Text-to-SQL(如果问题涉及数据查询)
sql = await self.llm.invoke(f"""
根据以下问题生成 SQL(如果不适合 SQL 查询则返回 NONE):
问题: {query}
表结构: sales(date, product, amount, region)
""")
if "NONE" not in sql.content:
try:
db_result = self.db.execute(sql.content)
results["structured"] = db_result
except Exception:
pass
# 向量检索
rag_results = self.vector_store.search(query_embedding, top_k=3)
results["unstructured"] = rag_results
return results5.3 GraphRAG:知识图谱增强检索
GraphRAG 架构思路:
传统 RAG:
→ 文本 → 分块 → 向量检索 → 返回文本片段
GraphRAG:
→ 文本 → 提取实体关系 → 构建知识图谱
→ 检索时:向量检索 + 图谱遍历
→ 返回:相关文本 + 关联实体 + 上下游关系
优势:
→ 能回答关系类问题:"谁是 CEO?他管理哪些部门?"
→ 能做多跳推理:"A 的合作伙伴 B 的产品是什么?"
→ 传统 RAG 对这类问题无能为力💡 面试答题要点:"企业级 RAG 不能只有向量检索——结构化数据用 Text-to-SQL,关系数据用知识图谱,文档数据用向量检索,三路结果融合生成。"
6. 检索路由与自适应策略
不同的问题需要不同的检索策略——技术问题查文档,数据问题查数据库,模糊问题先改写再检索。
6.1 检索路由器:为不同问题选择最优策略
# ═══════════════════════════════════════
# 检索路由器
# ═══════════════════════════════════════
class RetrievalRouter:
"""根据问题类型选择检索策略"""
def __init__(self, llm):
self.llm = llm
async def route(self, query: str) -> str:
"""LLM 路由:判断该用哪种检索"""
response = await self.llm.invoke(f"""
判断以下问题应该用什么检索方式:
问题: {query}
选项:
- "vector": 语义理解类(概念解释、方案推荐)
- "keyword": 精确查找类(报错信息、API 名称)
- "hybrid": 混合检索(不确定时)
- "sql": 数据查询类(统计数字、排名)
- "none": 不需要检索(闲聊、常识)
只输出选项名称。
""")
return response.content.strip()
# 用法:
# strategy = await router.route("去年Q4的营收是多少?")
# → "sql"
# strategy = await router.route("什么是 RAG?")
# → "vector"6.2 查询改写与扩展:HyDE / Multi-Query
# ═══════════════════════════════════════
# 查询改写策略
# ═══════════════════════════════════════
# ① HyDE(Hypothetical Document Embedding)
# 思路:让 LLM 先生成一个"假设的答案",用答案去检索
async def hyde_rewrite(llm, query: str) -> str:
"""HyDE: 用假设文档替代原始查询"""
response = await llm.invoke(f"""
请写一段简短的文本来回答这个问题(不需要准确,只需要包含相关术语):
{query}
""")
return response.content
# → 用这段"假设答案"的 Embedding 去检索
# → 比直接用问题检索效果好 20%
# ② Multi-Query(多查询扩展)
# 思路:把一个问题改写成多个角度,分别检索再合并
async def multi_query_rewrite(llm, query: str) -> list[str]:
"""生成多个角度的查询"""
response = await llm.invoke(f"""
将以下问题从 3 个不同角度改写,用于搜索:
原始问题: {query}
输出 3 个改写后的查询,每行一个。
""")
return response.content.strip().split("\n")
# → 3 个查询分别检索,结果合并去重6.3 实战:Agentic RAG 自适应检索
Agentic RAG 自适应流程:
用户提问
│
▼
① Agent 判断:需不需要检索?
├── 不需要 → 直接回答
└── 需要 → 进入检索
│
▼
② Agent 判断:用什么策略?
├── 模糊问题 → HyDE 改写后检索
├── 精确问题 → 关键词检索
├── 数据问题 → Text-to-SQL
└── 通用问题 → 混合检索
│
▼
③ 检索结果够不够?
├── 够了 → 生成答案
└── 不够 → 换策略重试 / 改写查询重试💡 面试答题要点:"Agentic RAG 的核心是把检索决策交给 Agent——Agent 自主判断是否需要检索、用什么策略、结果够不够,而不是固定的检索流程。"
7. 检索质量评测与调优
7.1 评测指标与 A/B 测试方法
核心评测指标:
指标 含义 目标值 计算方式
──────────────────────────────────────────────────────────
Recall@5 Top-5 中包含相关文档 > 80% 命中数/相关总数
MRR 第一个相关结果的排名 > 0.7 1/排名 的均值
Hit Rate 至少有一条命中 > 90% 命中查询/总查询
NDCG@5 排名加权的相关性 > 0.8 DCG/IDCG
A/B 测试方案:
→ 控制组:朴素向量检索
→ 实验组 A:混合检索(向量 + BM25)
→ 实验组 B:混合检索 + Rerank
→ 实验组 C:混合检索 + Rerank + 父子块
→ 用 50 条测试 Query 对比各组 Recall@57.2 检索失败诊断与调优指南
检索失败诊断表:
症状 根因 解决方案
─────────────────────────────────────────────────────────────
关键词搜不到 纯向量检索 加 BM25 混合检索
语义相关但排名低 粗排不够精准 加 Rerank
检索到但上下文不够 分块太小 父子块 / 加 overlap
完全搜不到相关内容 数据没入库 / 查询太模糊 查入库日志 / HyDE 改写
返回过时内容 无时间过滤 加元数据时间过滤
同一文档重复出现 去重不够 按 doc_id 去重
参数调优速查:
→ Top-K 太小(如 3)→ 召回率低 → 调到 5-10
→ Top-K 太大(如 50)→ 噪声多 → 加 Rerank 精排
→ 相似度阈值太高 → 漏掉相关结果 → 降低阈值
→ chunk_size 太大 → 检索不精准 → 减小 + 父子块7.3 RAG 检索架构面试高频题
面试高频题 TOP 8:
❓ Q1: 朴素 RAG 有什么问题?怎么优化?
→ 三大瓶颈(语义鸿沟/排序粗糙/上下文断裂)
→ 对应方案(混合检索/Rerank/父子块)
❓ Q2: 向量检索和 BM25 各有什么优缺点?
→ 向量擅长语义、BM25 擅长精确关键词,混合互补
❓ Q3: RRF 融合算法是什么?
→ 多路检索结果按排名倒数求和,score = Σ 1/(k+rank)
❓ Q4: Rerank 放在哪一步?为什么?
→ 粗检索之后、LLM 之前。因为 Cross-Encoder 慢但准
❓ Q5: 什么是父子块?解决什么问题?
→ 小块检索精准,大块保留上下文,两全其美
❓ Q6: HyDE 是什么?
→ 先让 LLM 生成假设答案,用答案的 Embedding 去检索
❓ Q7: Agentic RAG 和普通 RAG 的区别?
→ 普通 RAG 是固定管线,Agentic RAG 是 Agent 自主决策
❓ Q8: 怎么评估 RAG 检索质量?
→ Recall@K + MRR + NDCG + RAGAS 端到端评测🎉 全文完成。RAG 进阶检索的核心优化链路:混合检索 → RRF 融合 → Rerank 精排 → 父子块扩展 → 查询改写 → 自适应路由。每加一层优化,Recall 提升 5-20%。