检索策略
检索策略决定了如何从知识库中找到最相关的文档。好的检索策略能显著提升 RAG 系统的效果。本章将深入分析各种检索方法的原理、优缺点和适用场景。
检索策略概览
┌─────────────────────────────────────────────────────────────────┐
│ 检索策略分类 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 向量检索 关键词检索 混合检索 │
│ ────────── ────────── ────────── │
│ 语义理解 精确匹配 综合优势 │
│ 同义词匹配 专业术语 更高召回 │
│ 多语言支持 编号/代码 生产推荐 │
│ │
└─────────────────────────────────────────────────────────────────┘
向量检索（语义检索）
向量检索基于文本的语义相似度，能理解同义词和近义表达。
工作原理
查询: "如何提高团队效率？"
↓ 嵌入模型
查询向量: [0.23, -0.15, 0.89, ...]
↓ 相似度计算
文档向量数据库
↓ 返回最相似
"团队协作优化的方法" (0.92)
"提升团队生产力的策略" (0.88)
"管理高绩效团队的技巧" (0.85)
实现
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
# 初始化
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents, embeddings)
# 基础向量检索
results = vectorstore.similarity_search(
query="如何提高团队效率？",
k=5 # 返回前 5 个结果
)
# 带分数的检索
results_with_scores = vectorstore.similarity_search_with_score(
query="如何提高团队效率？",
k=5
)
for doc, score in results_with_scores:
print(f"相似度: {score:.3f} - {doc.page_content[:50]}...")
优缺点
| 优点 | 缺点 |
|---|---|
| 理解语义、同义词 | 对专业术语可能不准确 |
| 多语言支持好 | 编号、代码匹配较差 |
| 无需精确用词 | 可能有语义漂移 |
关键词检索（词汇检索）
关键词检索基于词汇匹配，能精确查找特定术语和编号。
工作原理
查询: "ISO 27001 认证流程"
↓ 分词
关键词: ["ISO", "27001", "认证", "流程"]
↓ 倒排索引查找
匹配文档（包含这些词的文档）
↓ 排序
返回结果
BM25 算法
BM25 是最常用的关键词检索算法，考虑词频和文档长度：
from langchain_community.retrievers import BM25Retriever
# 创建 BM25 检索器
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5
# 检索
results = bm25_retriever.invoke("ISO 27001 认证流程")
优缺点
| 优点 | 缺点 |
|---|---|
| 精确匹配术语 | 不理解同义词 |
| 编号、代码匹配好 | 需要精确用词 |
| 可解释性强 | 多语言支持差 |
混合检索
结合向量检索和关键词检索的优势，是生产环境的推荐方案。
实现方式
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
# 向量检索器
vector_retriever = vectorstore.as_retriever(
search_kwargs={"k": 5}
)
# 关键词检索器
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5
# 混合检索器
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.4, 0.6] # 关键词 40%, 向量 60%
)
# 检索
results = ensemble_retriever.invoke("ISO 27001 信息安全认证")
权重调优
权重需要根据实际数据调整：
# 测试不同权重
weight_combinations = [
(0.2, 0.8), # 偏向语义
(0.4, 0.6), # 平衡
(0.5, 0.5), # 均等
(0.6, 0.4), # 偏向关键词
]
for bm25_weight, vector_weight in weight_combinations:
retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[bm25_weight, vector_weight]
)
# 评估效果...
去重策略
混合检索可能返回重复文档，需要去重：
def deduplicate_results(results, keep_first=True):
"""去重并保留分数最高的"""
seen = {}
for doc in results:
doc_id = doc.metadata.get("id", doc.page_content[:100])
if doc_id not in seen:
seen[doc_id] = doc
return list(seen.values())
# 使用
unique_results = deduplicate_results(results)
多查询检索
从不同角度重写问题，提高召回率。
实现
from langchain.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
# 多查询检索器
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
llm=llm
)
# 检索
results = multi_query_retriever.invoke("如何提高团队效率？")
# 内部流程：
# 原问题: "如何提高团队效率？"
# LLM 生成变体:
# 1. "团队效率提升的方法有哪些？"
# 2. "如何改善团队协作？"
# 3. "提高团队生产力的策略？"
# 分别检索并合并结果
自定义提示词
from langchain_core.prompts import PromptTemplate
QUERY_PROMPT = PromptTemplate(
input_variables=["question"],
template="""你是一个AI助手，帮助生成搜索查询。
请为以下问题生成3个不同版本的查询，以帮助找到相关文档。
每个查询应该从不同角度表达相同的意图。
原问题: {question}
生成的查询（每行一个）:"""
)
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(),
llm=llm,
prompt=QUERY_PROMPT
)
自查询检索
从问题中提取过滤条件，结合元数据过滤。
实现
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
# 定义元数据字段
metadata_field_info = [
AttributeInfo(
name="category",
description="文档类别，如：技术、管理、人事",
type="string"
),
AttributeInfo(
name="year",
description="文档发布的年份",
type="integer"
),
AttributeInfo(
name="department",
description="发布部门",
type="string"
),
]
# 创建自查询检索器
retriever = SelfQueryRetriever.from_llm(
llm=llm,
vectorstore=vectorstore,
document_contents="公司内部文档",
metadata_field_info=metadata_field_info,
verbose=True # 打印生成的过滤条件
)
# 查询
results = retriever.invoke("2024年技术部发布的安全规范")
# 内部解析：
# 过滤条件: {"year": 2024, "department": "技术部", "category": "安全"}
# 先过滤元数据，再进行向量检索
上下文压缩检索
压缩检索结果，只保留与问题相关的内容。
实现
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
# 创建压缩器
compressor = LLMChainExtractor.from_llm(llm)
# 创建压缩检索器
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)
# 检索
results = compression_retriever.invoke("年假制度")
# 原始文档（可能很长）:
# "公司制度汇编：第一章 考勤管理...第二章 年假制度：员工入职满一年..."
# 压缩后（只保留相关部分）:
# "年假制度：员工入职满一年后可享受年假，天数根据工龄计算..."
父文档检索
检索小块以精确匹配，返回包含上下文的大块。
实现
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 小块分割器（用于索引）
child_splitter = RecursiveCharacterTextSplitter(
chunk_size=400,
chunk_overlap=50
)
# 大块分割器（用于返回）
parent_splitter = RecursiveCharacterTextSplitter(
chunk_size=2000,
chunk_overlap=400
)
# 创建检索器
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=InMemoryStore(), # 存储父文档
child_splitter=child_splitter,
parent_splitter=parent_splitter
)
# 索引
retriever.add_documents(documents)
# 检索
results = retriever.invoke("年假如何计算？")
# 流程：
# 1. 检索匹配的小块（精确）
# 2. 返回包含小块的大块（完整上下文）
检索策略选择指南
| 场景 | 推荐策略 | 原因 |
|---|---|---|
| 通用问答 | 混合检索 | 综合优势 |
| 精确术语查询 | 关键词检索 | 精确匹配 |
| 同义词丰富领域 | 向量检索 | 语义理解 |
| 复杂问题 | 多查询检索 | 多角度召回 |
| 元数据过滤需求 | 自查询检索 | 自动解析 |
| 需要完整上下文 | 父文档检索 | 上下文完整 |
检索参数调优
Top-K 选择
# 简单问题：较少结果
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# 复杂问题：更多结果
retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
# 权衡：
# - K 太小：可能遗漏信息
# - K 太大：噪音增加，成本上升
相似度阈值
from langchain.retrievers import VectorStoreRetriever
# 设置相似度阈值
retriever = vectorstore.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"k": 10,
"score_threshold": 0.7 # 只返回相似度 > 0.7 的结果
}
)
results = retriever.invoke("查询内容")
检索效果评估
def evaluate_retrieval(test_cases, retriever):
"""评估检索效果"""
hit_rates = []
mrrs = []
for case in test_cases:
query = case["query"]
relevant_ids = set(case["relevant_ids"])
results = retriever.invoke(query)
retrieved_ids = [r.metadata["id"] for r in results]
# Hit Rate
hit = 1 if set(retrieved_ids) & relevant_ids else 0
hit_rates.append(hit)
# MRR (Mean Reciprocal Rank)
for i, rid in enumerate(retrieved_ids):
if rid in relevant_ids:
mrrs.append(1 / (i + 1))
break
else:
mrrs.append(0)
return {
"hit_rate": sum(hit_rates) / len(hit_rates),
"mrr": sum(mrrs) / len(mrrs)
}
小结
检索策略是 RAG 系统效果的关键：
- 混合检索是生产推荐方案
- 多查询检索提高复杂问题的召回
- 自查询检索适合有丰富元数据的场景
- 父文档检索平衡精度和上下文
- 持续评估和调优是必要的