RAG 检索增强生成

5.1 为什么需要 RAG

RAG（Retrieval-Augmented Generation，检索增强生成）解决了大语言模型的几个核心问题：

5.2 RAG 工作流程

RAG 系统分为两个阶段：索引阶段（离线）和查询阶段（在线）。

5.3 文档加载

LangChain 提供了丰富的文档加载器，支持各种格式：

5.3.1 基础加载示例

# 安装依赖
# pip install langchain langchain-community pypdf python-docx

# PDF 加载
from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader("document.pdf")
pages = loader.load()

# pages 是 Document 对象列表
for page in pages:
    print(page.page_content)  # 页面文本
    print(page.metadata)       # 元数据（页码等）

# Word 文档
from langchain_community.document_loaders import Docx2txtLoader

loader = Docx2txtLoader("document.docx")
docs = loader.load()

# 网页内容
from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader("https://python.langchain.com/docs/")
docs = loader.load()

# 批量加载目录
from langchain_community.document_loaders import DirectoryLoader

loader = DirectoryLoader(
    "./documents/",
    glob="**/*.pdf",  # 只加载 PDF
    loader_cls=PyPDFLoader
)
docs = loader.load()

5.4 文本分割

将长文档分割成小的文本块（chunks），这是 RAG 的关键步骤。

5.4.1 基础分割器

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每个块的最大字符数
    chunk_overlap=50,    # 块之间的重叠字符数
    length_function=len, # 计算长度的函数
    separators=["\n\n", "\n", "。", "！", "？", " ", ""],  # 分隔符优先级
)

# 分割文档
chunks = text_splitter.split_documents(docs)

print(f"生成了 {len(chunks)} 个文本块")
for i, chunk in enumerate(chunks[:3]):
    print(f"块 {i+1}: {chunk.page_content[:100]}...")

5.4.2 Token 感知的分割

根据 token 数量分割（更准确）：

from langchain.text_splitter import TokenTextSplitter

# 按 token 分割
text_splitter = TokenTextSplitter(
    chunk_size=500,      # 最大 token 数
    chunk_overlap=50,    # 重叠 token 数
)

chunks = text_splitter.split_documents(docs)

5.4.3 Markdown 分割

保持 Markdown 结构：

from langchain.text_splitter import MarkdownTextSplitter

splitter = MarkdownTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)

# 分割 Markdown
chunks = splitter.split_text(markdown_content)

# 或者直接分割文档
splitter = MarkdownTextSplitter()
chunks = splitter.split_documents(markdown_docs)

5.5 向量化与嵌入

将文本转换为向量（embeddings），用于语义搜索。

5.5.1 使用 OpenAI Embeddings

from langchain_openai import OpenAIEmbeddings

# 创建嵌入模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"  # 或 "text-embedding-3-large"
)

# 单个文本嵌入
vector = embeddings.embed_query("你好，世界")
print(f"向量维度: {len(vector)}")

# 批量文本嵌入
texts = ["第一个文本", "第二个文本", "第三个文本"]
vectors = embeddings.embed_documents(texts)
print(f"生成了 {len(vectors)} 个向量")

5.5.2 其他嵌入提供商

# 使用本地嵌入模型
from langchain_community.embeddings import OllamaEmbeddings

embeddings = OllamaEmbeddings(
    model="nomic-embed-text"  # 或 "mxbai-embed-large"
)

# 使用 Cohere
from langchain_cohere import CohereEmbeddings

embeddings = CohereEmbeddings(
    model="embed-multilingual-v3.0",
    cohere_api_key="your-api-key"
)

5.6 向量存储

向量数据库用于存储和检索嵌入向量。

5.6.1 FAISS 本地存储

# pip install faiss-cpu  (或 faiss-gpu)

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 创建向量存储
vectorstore = FAISS.from_documents(
    documents=chunks,           # 文本块
    embedding=OpenAIEmbeddings()  # 嵌入模型
)

# 保存到本地
vectorstore.save_local("faiss_index")

# 加载
loaded_vectorstore = FAISS.load_local(
    "faiss_index",
    OpenAIEmbeddings(),
    allow_dangerous_deserialization=True  # 仅在信任的来源时使用
)

5.6.2 Chroma

# pip install chromadb

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"  # 持久化目录
)

# 保存
vectorstore.persist()

# 加载
client = chromadb.PersistentClient(path="./chroma_db")
vectorstore = Chroma(
    client=client,
    embedding_function=OpenAIEmbeddings()
)

5.7 检索与问答

5.7.1 基础相似性检索

# 相似性搜索
results = vectorstore.similarity_search(
    query="LangChain的Agent是什么？",
    k=3  # 返回最相似的 3 个文档块
)

for doc in results:
    print(doc.page_content)
    print("---")

# 带分数的搜索
results_with_scores = vectorstore.similarity_search_with_score(
    query="什么是LangChain？",
    k=3
)

for doc, score in results_with_scores:
    print(f"分数: {score:.4f}")
    print(f"内容: {doc.page_content[:200]}")
    print("---")

5.7.2 构建 RAG 问答链

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import ChatPromptTemplate

# 初始化模型
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 创建检索器
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 3}  # 检索 3 个文档块
)

# 创建 QA 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 将检索结果拼接到 prompt
    retriever=retriever,
    return_source_documents=True  # 返回源文档
)

# 问答
result = qa_chain.invoke({"query": "LangChain的Agent是什么？"})
print(result["result"])
print("\n来源文档：")
for doc in result["source_documents"]:
    print(f"- {doc.page_content[:100]}...")

5.7.3 LCEL 方式构建 RAG

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

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

# RAG 提示模板
RAG_PROMPT = ChatPromptTemplate.from_template("""根据以下上下文回答问题。如果上下文中没有相关信息，请说明不知道。

上下文：
{context}

问题：{question}

回答：""")

# 构建 RAG 链
rag_chain = (
    {"context": retriever | (lambda docs: "\n\n".join([d.page_content for d in docs])), "question": RunnablePassthrough()}
    | RAG_PROMPT
    | llm
    | StrOutputParser()
)

# 调用
result = rag_chain.invoke("LangChain的Agent是什么？")
print(result)

5.8 进阶优化

5.8.1 查询压缩（Query Compression）

将对话历史压缩成独立的查询：

from langchain.retrievers.multi_query import MultiQueryRetriever

# 多查询检索器 - 生成多个查询角度
retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm,
    include_original=True
)

# 会生成多个查询，如：
# - 原始查询: "Agent是什么？"
# - 衍生查询1: "LangChain中Agent的作用"
# - 衍生查询2: "如何使用Agent"
# 最终综合多个查询的结果

5.8.2 上下文压缩（Contextual Compression）

压缩检索结果，只保留相关内容：

from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.document_transformers import EmbeddingsFilter

# 嵌入过滤器 - 基于相似度过滤
embeddings_filter = EmbeddingsFilter(
    embeddings=embeddings,
    similarity_threshold=0.8  # 只保留相似度 > 0.8 的内容
)

# 上下文压缩检索器
compression_retriever = ContextualCompressionRetriever(
    base_retriever=vectorstore.as_retriever(),
    document_compressor=embeddings_filter
)

# 使用压缩检索器
results = compression_retriever.invoke("LangChain的Agent")
print(results)

5.8.3 混合搜索

结合向量搜索和关键词搜索：

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

# 关键词搜索
bm25_retriever = BM25Retriever.from_texts(
    texts=[doc.page_content for doc in chunks]
)
bm25_retriever.k = 2

# 向量搜索
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 组合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.5, 0.5]  # 各占一半
)

# 使用组合检索器
results = ensemble_retriever.invoke("LangChain Agent")
print(results)

5.8.4 重排序（Re-ranking）

对检索结果进行重排序，提高相关性：

# 使用 Cohere 进行重排序
from langchain_community.document_compressors import CohereRerank

compressor = CohereRerank(
    cohere_api_key="your-key",
    top_n=5  # 返回前 5 个
)

from langchain.retrievers import ContextualCompressionRetriever

compression_retriever = ContextualCompressionRetriever(
    base_retriever=vectorstore.as_retriever(),
    document_compressor=compressor
)

5.9 本章小结