LangChain：检索器类型与使用方法

在 LangChain 中，检索器是一个遵循统一接口的组件：输入一个查询（字符串），输出一个文档（Document）列表。

这个简单的接口极大地简化了信息检索的复杂性。你可以将任何能够返回相关文档的系统（如搜索引擎、数据库、向量存储）包装成一个检索器，然后在你的 LangChain 应用中以统一的方式调用它。

什么是检索器

在 LangChain 中，检索器是一个遵循统一接口的组件：输入一个查询（字符串），输出一个文档（Document）列表。

这个简单的接口极大地简化了信息检索的复杂性。你可以将任何能够返回相关文档的系统（如搜索引擎、数据库、向量存储）包装成一个检索器，然后在你的 LangChain 应用中以统一的方式调用它。

LangChain 的检索器是 RAG（检索增强生成）系统的核心组件，它提供了一套统一的接口来连接各种不同的数据源。无论是向量数据库、搜索引擎，还是传统的数据库，都可以通过这个接口被 LLM 应用轻松调用。

常见检索器类型及使用方法

LangChain 支持多种检索器，以满足不同的应用场景。以下是一些最常用和高级的类型：

检索器类型	核心概念	适用场景
向量存储检索器	基于语义相似度进行检索，将文本转换为向量后计算距离。	语义搜索、问答系统、知识库检索。
父文档检索器	索引小块（如句子）以获得高精度，但返回大块（段落）以提供完整上下文。	文档问答、法律合同分析、技术手册查询。
多查询检索器	使用 LLM 对用户问题生成多个不同视角的变体，分别检索后融合结果。	处理复杂、模糊的问题，提高召回率。
自查询检索器	让 LLM 从自然语言中提取出语义部分用于向量搜索，并提取出过滤条件用于元数据过滤。	数据分类清晰、需结构化筛选的场景，如电商、内容库。
集成检索器	结合多个检索器（如 BM25 + 向量搜索）的结果，使用加权或 RRF 算法重新排名。	需要兼顾关键词匹配和语义理解的搜索系统。
API/搜索引擎检索器	直接包装外部API（如搜索引擎、第三方数据服务）作为检索器，返回实时信息。	实时信息查询、联网搜索。

调用代码示例

向量存储检索器：

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

父文档检索器

retriever = ParentDocumentRetriever(vectorstore=..., docstore=..., child_splitter=..., parent_splitter=...)

多查询检索器

retriever = MultiQueryRetriever.from_llm(retriever=..., llm=...)

自查询检索器

retriever = SelfQueryRetriever.from_llm(llm, vectorstore, document_content_description, metadata_field_info)

集成检索器

retriever = EnsembleRetriever(retrievers=[bm25_retriever, vector_retriever], weights=[0.4, 0.6])

API/搜索引擎检索器

retriever = LinkupSearchRetriever(depth="deep")

详细使用指南与示例

向量存储检索器

这是最基础的检索器类型，几乎所有的向量数据库都支持将它转换为检索器。

使用步骤：

1. 加载和分割文档。
2. 生成嵌入向量并存入向量库。
3. 通过 as_retriever() 方法创建检索器。

示例代码：

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 1. 加载文档
loader = TextLoader("state_of_the_union.txt")
documents = loader.load()

# 2. 分割文本
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
docs = text_splitter.split_documents(documents)

# 3. 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(docs, embeddings)

# 4. 创建检索器 (指定返回前3个最相关的文档)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 5. 使用检索器
query = "What did the president say about Ketanji Brown Jackson?"
relevant_docs = retriever.invoke(query)

for doc in relevant_docs:
    print(doc.page_content[:200]) # 打印部分内容

父文档检索器

这种方法解决了检索精度与上下文完整性之间的矛盾。

工作原理：

• 将大文档分割成较大的“父文档”（如段落）。
• 再将父文档分割成很小的“子文档”（如句子）。
• 检索时，用子文档去匹配查询（精度高），但返回时返回其对应的父文档（上下文完整）。

示例代码：

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 假设我们已经有了一个 vectorstore
# vectorstore = FAISS.from_documents(...)

# 定义分割器
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=0)

# 初始化文档存储（用于存放父文档）
docstore = InMemoryStore()

# 创建父文档检索器
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# 添加文档（内部会自动完成分割和关联）
retriever.add_documents(docs)

# 检索时，返回的是父文档
retrieved_parent_docs = retriever.invoke("your query here")

自查询检索器

它让你可以用自然语言实现复杂的、带筛选条件的查询。

工作原理：

• 你定义好元数据字段的信息（如 year、genre）。
• 用户输入："Find movies made before 1960 that are rated higher than 8"。
• LLM 将这句话解析成一个结构化查询，包含 query 部分（”movies”）和 filter 部分（year < 1960 and rating > 8）。

示例代码（基于 MongoDB）：

from langchain_mongodb.retrievers import MongoDBAtlasSelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo

# 定义元数据字段信息
metadata_field_info = [
    AttributeInfo(name="year", description="The year the movie was released", type="integer"),
    AttributeInfo(name="rating", description="A 1-10 rating for the movie", type="float"),
]

# 创建检索器
retriever = MongoDBAtlasSelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore,
    metadata_field_info=metadata_field_info,
    document_contents="Descriptions of movies"
)

# 使用自然语言查询
result = retriever.invoke("Movies made before 1960 that are rated higher than 8")

集成检索器

混合检索是目前提升召回率的主流方法，通常将基于关键词（如BM25）和基于语义（向量）的检索器结合。

示例代码：

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# 假设已经创建了 bm25_retriever 和 vector_retriever
# bm25_retriever = BM25Retriever.from_documents(docs)
# vector_retriever = vectorstore.as_retriever()

# 创建集成检索器，并设置权重
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6] # 语义检索权重更高
)

docs = ensemble_retriever.invoke("What did the president say about Justice Breyer?")

完整的问答系统示例

最后，让我们看一个完整的例子，它使用向量存储检索器构建一个可以直接使用的问答链：

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 1. 假设我们已经有了一个向量库 (vectorstore)
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(docs, embeddings) # docs 是之前加载并分割好的文档

# 2. 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 3. 创建 LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 4. 创建 RetrievalQA 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff", # 常用类型，将所有文档一次性放入提示词
    retriever=retriever,
    return_source_documents=True # 返回源文档，方便溯源
)

# 5. 发起提问
query = "What did the president say about Ketanji Brown Jackson?"
result = qa_chain.invoke({"query": query})

# 6. 输出结果
print("Answer:", result["result"])
print("\nSources:")
for doc in result["source_documents"]:
    print(f"- {doc.metadata.get('source', 'Unknown')}")