第6章 RAG 检索增强 — LangChain 实战教程

RAG 管道全流程

RAG（Retrieval-Augmented Generation，检索增强生成）的核心思想是：在让 LLM 生成答案之前，先从外部知识库检索相关文档片段，将这些片段作为上下文传入 Prompt，使模型能够基于真实信息回答，而非依赖训练时的"记忆"。

一个完整的 RAG 系统分为两个阶段：

索引阶段（离线）

加载文档（PDF、网页、数据库等）
分割为小块（chunk）
计算每块的 Embedding 向量
存入向量数据库（Chroma、FAISS、Qdrant）

检索阶段（在线）

接收用户问题
计算问题的 Embedding
在向量数据库中相似度搜索
将检索结果注入 Prompt 生成答案

文档加载器（Document Loaders）

LangChain 提供数十种文档加载器，覆盖各种数据来源：

from langchain_community.document_loaders import (
    PyPDFLoader,
    WebBaseLoader,
    TextLoader,
    CSVLoader,
    UnstructuredMarkdownLoader,
)

# 加载 PDF
pdf_loader = PyPDFLoader("./docs/langchain-guide.pdf")
pdf_docs = pdf_loader.load()
print(len(pdf_docs))              # 每页一个 Document
print(pdf_docs[0].page_content[:200])
print(pdf_docs[0].metadata)       # {'source': './docs/...', 'page': 0}

# 加载网页
web_loader = WebBaseLoader(
    web_paths=["https://python.langchain.com/docs/introduction/"],
    bs_kwargs={"parse_only": BeautifulSoup4.SoupStrainer(class_="main-content")},
)
web_docs = web_loader.load()

# 加载本地文本
text_loader = TextLoader("./docs/knowledge.txt", encoding="utf-8")
text_docs = text_loader.load()

# 加载目录下所有文件
from langchain_community.document_loaders import DirectoryLoader
dir_loader = DirectoryLoader(
    "./docs/",
    glob="**/*.md",
    loader_cls=UnstructuredMarkdownLoader,
)
all_docs = dir_loader.load()

文本分割（Text Splitters）

加载的文档通常太长，需要分割成适合嵌入（embedding）和检索的小块（chunk）。分割策略直接影响 RAG 的质量。

chunk_size

每个文本块的最大字符数。建议 500~1500 字符（中文），过小则每块信息不足，过大则 Embedding 稀释、token 消耗增加。

chunk_overlap

相邻块之间的重叠字符数。重叠可以防止关键信息恰好被切断在两块之间。通常设置为 chunk_size 的 10~20%。

RecursiveCharacterTextSplitter

最常用的分割器。依次尝试 ["\\n\\n", "\\n", " ", ""] 等分隔符，在语义边界处分割，尽量保持段落完整性。

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,          # 每块最大 800 字符
    chunk_overlap=100,       # 相邻块重叠 100 字符
    separators=["\n\n", "\n", "。", "？", "！", " ", ""],
    length_function=len,
)

# 分割文档列表
chunks = splitter.split_documents(pdf_docs)
print(len(chunks))             # 分割后的块数量
print(chunks[0].page_content)  # 第一块内容
print(chunks[0].metadata)      # 继承原文档的 metadata

# 基于 Token 的分割（更精确控制模型输入长度）
from langchain_text_splitters import TokenTextSplitter
token_splitter = TokenTextSplitter(chunk_size=300, chunk_overlap=50)
token_chunks = token_splitter.split_documents(pdf_docs)

Embedding 模型

Embedding 是将文本转换为高维数值向量的过程，语义相似的文本在向量空间中距离相近。这是语义搜索的基础。

from langchain_openai import OpenAIEmbeddings

# OpenAI text-embedding-3-small（推荐，性价比高）
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1536,
)

# 嵌入单个字符串
vector = embeddings.embed_query("LangChain 是什么？")
print(len(vector))  # 1536 维向量

# 嵌入文档列表（批量，更高效）
vectors = embeddings.embed_documents([
    "LangChain 是 LLM 应用框架",
    "Python 是一种编程语言",
])

# 国内开源替代：智谱 Embedding
from langchain_community.embeddings import ZhipuAIEmbeddings
zhipu_embeddings = ZhipuAIEmbeddings(
    model="embedding-3",
    api_key="your-zhipu-api-key"
)

向量存储（Vector Stores）

Chroma：本地开发首选

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建向量库（首次运行）
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",  # 持久化到本地磁盘
    collection_name="langchain_docs",
)

# 加载已有向量库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="langchain_docs",
)

# 相似度搜索
results = vectorstore.similarity_search("LCEL 是什么？", k=4)
for doc in results:
    print(doc.page_content[:100], "...")

# 带相似度分数的搜索
results_with_scores = vectorstore.similarity_search_with_score(
    "LCEL 管道", k=3
)
for doc, score in results_with_scores:
    print(f"Score: {score:.4f} | {doc.page_content[:80]}")

将向量库转为检索器（Retriever）

向量库通过 .as_retriever() 转换为实现 Runnable 接口的 Retriever，可以直接嵌入 LCEL 管道：

# 基础检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",    # 或 "mmr"、"similarity_score_threshold"
    search_kwargs={"k": 4},
)

# MMR 检索：最大边际相关性，避免返回重复内容
mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5},
)

# 带分数阈值过滤
threshold_retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.7, "k": 4},
)

完整 RAG 链构建

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser

def format_docs(docs: list) -> str:
    """将文档列表格式化为字符串"""
    return "\n\n---\n\n".join(
        f"来源：{doc.metadata.get('source', '未知')}\n{doc.page_content}"
        for doc in docs
    )

rag_prompt = ChatPromptTemplate.from_template("""
你是一位知识库问答助手。请根据以下检索到的上下文回答问题。
如果上下文中没有相关信息，请诚实地说"我在知识库中没有找到相关信息"，不要编造答案。

上下文：
{context}

问题：{question}

请给出详细的回答，并在最后注明信息来源。
""")

rag_chain = (
    {
        "context": retriever | RunnableLambda(format_docs),
        "question": RunnablePassthrough(),
    }
    | rag_prompt
    | model
    | StrOutputParser()
)

# 调用
answer = rag_chain.invoke("如何在 LangChain 中使用 LCEL？")
print(answer)

高级检索策略

EnsembleRetriever：混合检索

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

# 稠密检索（语义相似度）
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 稀疏检索（关键词 BM25）
bm25_retriever = BM25Retriever.from_documents(chunks, k=5)

# 融合：0.6 语义 + 0.4 关键词
ensemble_retriever = EnsembleRetriever(
    retrievers=[dense_retriever, bm25_retriever],
    weights=[0.6, 0.4],
)

MultiQueryRetriever：多查询扩展

from langchain.retrievers.multi_query import MultiQueryRetriever

# 用 LLM 将原始查询扩展为多个不同角度的查询，增加召回率
multi_retriever = MultiQueryRetriever.from_llm(
    retriever=retriever,
    llm=model,
)

# 会自动生成3个相关查询，合并去重检索结果
docs = multi_retriever.invoke("怎么用 LangChain 做 RAG？")

RAG 质量优化清单

RAG 系统常见的质量问题及对应优化方向：
1. 检索不到相关内容：调整 chunk_size、尝试 MMR 或混合检索、改善 Embedding 模型
2. 答案不准确：检查 chunk 是否完整、增加上下文窗口、改进 Prompt 指令
3. 答案来源于错误文档：添加元数据过滤、使用相似度阈值过滤
4. 性能慢：使用更快的向量库（如 FAISS）、添加缓存层

本章小结

RAG 分为索引（离线）和检索（在线）两个阶段
文档加载器支持 PDF、网页、文本、CSV 等数十种格式
RecursiveCharacterTextSplitter 是最常用的分割器，注意调整 chunk_size 和 overlap
Embedding 将文本转为向量，OpenAI text-embedding-3-small 是性价比最高的选择
Chroma 适合本地开发，生产环境推荐 Qdrant 或 Pinecone
混合检索（语义 + BM25）通常比单一策略效果更好
RAG 链的核心模式：{context: retriever | format_docs, question: passthrough} | prompt | model | parser