纲要

• 检索器概述
  • Retriever 在 RAG 中的核心角色
  • 统一接口：invoke / ainvoke
• 基础检索器与词法检索器
  • 向量数据库一键封装：as_retriever()
  • 经典词法检索：BM25Retriever
• 查询重写：多重查询
  • 原理：将模糊问题改写为多个精准子问题
  • 实现：MultiQueryRetriever 与模拟模型
• 查询重构：自然语言转 SQL
  • 流程：自然语言 → SQL → 执行 → 答案
  • 使用 FakeListChatModel 模拟 SQL 生成与回答
• 检索策略对比与选型建议
• 完整可运行代码（无需 API Key）

引言

在 RAG 系统中，检索器（Retriever）负责将用户的自然语言问题与知识库中最相关的文档片段关联起来。LangChain 将检索器抽象为标准的 Runnable 组件，不仅支持基础的向量相似度搜索，还内置了查询重写、查询重构等高级策略，以显著提升召回率和答案质量。

本文将带你从最简单的 as_retriever() 起步，逐步深入到多重查询（MultiQuery）和文本到 SQL 的查询重构，并提供无需外部 API 即可运行的完整示例。

检索器基础

所有向量数据库都可以通过 .as_retriever() 方法一键转换为检索器，并直接使用 invoke 进行查询。以下示例使用 Chroma 内存向量库和 FakeEmbeddings 模拟嵌入过程。

from langchain_community.embeddings.fake import FakeEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document

# 准备文档
docs = [
    Document(page_content="LangChain 是一个强大的 LLM 应用框架。"),
    Document(page_content="RAG 结合检索与生成，提升回答准确性。"),
    Document(page_content="向量数据库用于存储文档的高维向量表示。"),
]

# 构建向量库
embeddings = FakeEmbeddings(size=128)
vectorstore = Chroma.from_documents(docs, embeddings, collection_name="intro")

# 转换为检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 查询
results = retriever.invoke("什么是 RAG？")
for doc in results:
    print(doc.page_content)

词法检索器：BM25

BM25 是一种基于词频和逆文档频率的经典排序算法，适合精准字面匹配。LangChain 提供了 BM25Retriever。

安装依赖：

pip install langchain-community rank_bm25

from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document

# 准备文档
corpus = [
    Document(page_content="LangChain 链式调用"),
    Document(page_content="RAG 检索增强生成"),
    Document(page_content="向量数据库 Chroma 和 FAISS"),
    Document(page_content="BM25 是一种词法检索算法"),
]

# 创建 BM25 检索器
bm25_retriever = BM25Retriever.from_documents(corpus)
bm25_retriever.k = 2  # 返回前 2 个

# 检索
results = bm25_retriever.invoke("检索算法")
for doc in results:
    print(doc.page_content)

查询重写：多重查询（MultiQueryRetriever）

用户的问题往往比较模糊，多重查询技术会先用大模型将原始问题改写为多个更精确的子问题，再对每个子问题分别检索，最后合并结果。这样可以大幅提高召回率和答案的相关性。

我们使用 FakeListChatModel 模拟大模型的改写能力，无需真实 API。

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings.fake import FakeEmbeddings
from langchain_core.documents import Document
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_community.chat_models.fake import FakeListChatModel

# 构建知识库
docs = [
    Document(page_content="LangChain 提供了丰富的组件用于构建 RAG 应用。"),
    Document(page_content="查询重写技术可以将一个笼统的问题拆解为多个子问题。"),
    Document(page_content="MultiQueryRetriever 利用 LLM 生成多个查询变体。"),
    Document(page_content="向量检索基于余弦相似度或欧氏距离。"),
]
embeddings = FakeEmbeddings(size=128)
vectorstore = Chroma.from_documents(docs, embeddings, collection_name="mq")

# 基础检索器
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 1})

# 模拟大模型：预设改写结果
# MultiQueryRetriever 期望模型返回一个包含 "queries" 的 JSON
fake_llm = FakeListChatModel(
    responses=[
        '{"queries": ["如何提高检索准确性？", "有哪些检索优化方法？", "MultiQueryRetriever 的工作原理"]}'
    ]
)

# 创建多重查询检索器
multi_retriever = MultiQueryRetriever.from_llm(
    retriever=base_retriever,
    llm=fake_llm
)

# 查询
results = multi_retriever.invoke("如何让搜索更准？")
for doc in results:
    print(doc.page_content)

运行后你会发现，原始问题被扩展为多个子问题，从而检索到更全面的相关文档。

查询重构：自然语言转 SQL

另一种常见的检索优化是将自然语言问题重构为精确的 SQL 语句，尤其适用于结构化数据查询。我们模拟一个简单场景：用户问“销售额最高的员工”，系统将其转为 SQL 并执行，再将结果喂给模型生成最终答案。

from langchain_community.chat_models.fake import FakeListChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda

# 模拟数据库查询函数
def execute_sql(sql: str) -> str:
    # 实际环境中连接数据库执行，这里返回模拟结果
    if "销售额" in sql:
        return "员工A: 100万, 员工B: 80万"
    return "无数据"

# 模拟大模型：第一次返回 SQL，第二次返回自然语言答案
fake_llm = FakeListChatModel(
    responses=[
        "SELECT employee, revenue FROM sales ORDER BY revenue DESC LIMIT 2;",
        "销售额最高的员工是员工A(100万)和员工B(80万)。"
    ]
)

# 第一步：生成 SQL
sql_prompt = ChatPromptTemplate.from_template(
    "将以下问题转换为 SQL 语句（仅返回 SQL）:\n{question}"
)
sql_chain = sql_prompt | fake_llm | StrOutputParser()

# 第二步：生成最终答案
answer_prompt = ChatPromptTemplate.from_template(
    "根据 SQL 查询结果回答问题:\n问题: {question}\nSQL 结果: {result}"
)
answer_chain = answer_prompt | fake_llm | StrOutputParser()

# 完整链：自然语言 -> SQL -> 执行 -> 答案
def full_chain(question: str):
    sql = sql_chain.invoke({"question": question})
    result = execute_sql(sql)
    answer = answer_chain.invoke({"question": question, "result": result})
    return {"sql": sql, "result": result, "answer": answer}

response = full_chain("哪两位员工的销售额最高？")
print("生成的 SQL:", response["sql"])
print("执行结果:", response["result"])
print("最终答案:", response["answer"])

这个流程清晰地展示了查询重构如何将自然语言映射到结构化查询语言，大大简化了数据访问。

检索策略对比

总结

检索器是 RAG 系统的灵魂，LangChain 不仅提供了统一的操作接口，还封装了多种检索优化策略。从基础的向量检索到智能的查询重写与重构，开发者可以根据业务需求灵活组合。

通过本文的模拟示例，你可以在没有外部 API 的情况下快速验证这些高级特性，为真实项目打下坚实基础。