Retrievers 检索器
检索器（Retriever）是 LangChain 中负责从向量存储或其他数据源中检索相关文档的组件。它封装了检索逻辑，为 RAG 系统提供简洁的接口。
概述
检索器的核心功能：
查询 → 检索器 → 相关文档列表python
# 检索器的基本接口
retriever = vector_store.as_retriever()
docs = retriever.invoke("你的查询问题")从向量存储创建检索器
python
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
# 创建向量存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.from_texts(
texts=["LangChain 是一个 LLM 开发框架",
"文档加载器用于加载各种格式的文档",
"检索器从向量数据库中找到相关文档"],
embedding=embeddings
)
# 创建检索器（默认返回 4 个文档）
retriever = vector_store.as_retriever()
# 检索
docs = retriever.invoke("LangChain 是什么？")
for doc in docs:
print(doc.page_content)检索器参数配置
搜索类型
python
# 相似性搜索（默认）
retriever = vector_store.as_retriever(
search_type="similarity",
search_kwargs={"k": 5}
)
# MMR（最大边际相关性）：兼顾相关性和多样性
retriever = vector_store.as_retriever(
search_type="mmr",
search_kwargs={
"k": 5,
"fetch_k": 20, # 先取 20 个候选
"lambda_mult": 0.5 # 多样性系数（0-1）
}
)
# 相似度分数阈值
retriever = vector_store.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"score_threshold": 0.8, # 只返回分数 > 0.8 的结果
"k": 10
}
)搜索参数
python
retriever = vector_store.as_retriever(
search_kwargs={
"k": 3, # 返回文档数量
"fetch_k": 10, # MMR 候选数
"lambda_mult": 0.7, # MMR 多样性
"score_threshold": 0.5, # 分数阈值
"filter": {"source": "knowledge"}, # 元数据过滤
}
)高级检索器
MultiQueryRetriever（多查询检索器）
自动从用户查询生成多个相似查询，增加检索覆盖率：
python
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI
# 基础检索器
base_retriever = vector_store.as_retriever()
# 多查询检索器
retriever = MultiQueryRetriever.from_llm(
retriever=base_retriever,
llm=ChatOpenAI(model="gpt-4o-mini"),
include_original=True # 同时使用原始查询
)
docs = retriever.invoke("如何安装 LangChain？")ContextualCompressionRetriever（上下文压缩检索器）
对检索到的文档进行压缩，提取最相关的部分：
python
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=vector_store.as_retriever()
)
docs = compression_retriever.invoke("LangChain 的核心组件")EnsembleRetriever（集成检索器）
组合多个检索器的结果：
python
from langchain.retrievers import EnsembleRetriever
from langchain_community.vectorstores import FAISS
# 创建两个不同的检索器
retriever_1 = vector_store_1.as_retriever(search_kwargs={"k": 3})
retriever_2 = vector_store_2.as_retriever(search_kwargs={"k": 3})
# 集成检索
ensemble_retriever = EnsembleRetriever(
retrievers=[retriever_1, retriever_2],
weights=[0.5, 0.5] # 权重
)
docs = ensemble_retriever.invoke("查询内容")ParentDocumentRetriever（父子文档检索器）
检索时先找到相关的小块文档，再返回包含该块的完整父文档：
python
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 父文档分割器（大块）
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
# 子文档分割器（小块）
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
retriever = ParentDocumentRetriever(
vectorstore=vector_store,
docstore=InMemoryStore(),
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
# 添加文档
retriever.add_documents(docs)
# 检索（返回父文档）
docs = retriever.invoke("查询")TimeWeightedVectorStoreRetriever（时间权重检索器）
结合语义相似度和时间衰减进行检索：
python
from langchain.retrievers import TimeWeightedVectorStoreRetriever
retriever = TimeWeightedVectorStoreRetriever(
vectorstore=vector_store,
decay_rate=0.01, # 时间衰减率
k=5, # 返回文档数
other_score_keys=["importance"] # 其他评分因素
)自定义检索器
可以通过继承 BaseRetriever 创建自定义检索器：
python
from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from typing import List
class MyCustomRetriever(BaseRetriever):
"""自定义检索器"""
documents: List[Document]
def _get_relevant_documents(self, query: str) -> List[Document]:
# 实现你的检索逻辑
relevant_docs = [
doc for doc in self.documents
if query in doc.page_content
]
return relevant_docs[:5]
# 使用
retriever = MyCustomRetriever(documents=docs)
results = retriever.invoke("查询")完整示例：端到端检索
python
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
# 1. 加载和分割文档
loader = TextLoader("knowledge.txt")
docs = loader.load()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, chunk_overlap=50
)
chunks = text_splitter.split_documents(docs)
# 2. 构建向量存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.from_documents(chunks, embeddings)
# 3. 创建高级检索器
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
base_retriever = vector_store.as_retriever(search_kwargs={"k": 3})
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=base_retriever,
llm=llm
)
# 4. 检索
query = "LangChain 中的检索器有什么作用？"
docs = multi_query_retriever.invoke(query)
print(f"查询: {query}\n")
for i, doc in enumerate(docs):
print(f"结果 {i+1}:")
print(f" 内容: {doc.page_content[:150]}...")
print(f" 来源: {doc.metadata.get('source', 'unknown')}")
print()检索器选择指南
| 检索器 | 适用场景 | 优势 |
|---|---|---|
| 基础向量检索器 | 通用检索 | 实现简单，速度快 |
| MultiQueryRetriever | 复杂查询 | 提高检索覆盖率和召回率 |
| ContextualCompressionRetriever | 长文档 | 压缩文档，节省 token |
| EnsembleRetriever | 多数据源 | 融合多路检索结果 |
| ParentDocumentRetriever | 需要完整上下文 | 检索小块，返回完整文档 |
| TimeWeightedRetriever | 时序数据 | 结合时间衰减 |
最佳实践
- 选择合适的检索策略：根据场景选择基础检索或高级检索器
- 调整 k 值：检索数量不宜过大，通常 3-5 个即可
- 使用 MMR：当需要结果多样性时使用 MMR 搜索
- 组合检索器：复杂场景使用 MultiQuery 或 Ensemble 检索器
- 缓存检索结果：重复查询时缓存结果提高性能