RAG 检索增强

RAG（Retrieval Augmented Generation，检索增强生成）是当前最流行的 LLM 应用架构。本章将详细介绍如何在 LangChain 中实现 RAG 系统。

什么是 RAG？

RAG 是一种结合检索和生成的架构，让 LLM 能够基于私有数据生成更准确、更可靠的答案。

RAG 工作原理

核心流程：

索引阶段：将文档分割、嵌入、存入向量数据库
查询阶段：用户提问 → 检索相关文档 → 生成回答

为什么需要 RAG？

知识更新：LLM 训练数据有截止日期，RAG 可以访问最新信息
私有数据：让 LLM 能够使用企业内部文档
减少幻觉：基于真实文档生成，减少编造答案
成本效益：无需重新训练模型即可扩展知识

RAG 架构

完整 RAG 流程

实现 RAG

1. 文档加载

from langchain_community.document_loaders import TextLoader, PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 加载文本文档
loader = TextLoader("document.txt")
documents = loader.load()

# 加载 PDF
loader = PyPDFLoader("document.pdf")
documents = loader.load()

# 加载目录中的所有文档
loader = DirectoryLoader("docs/", glob="*.md")
documents = loader.load()

# 查看加载的文档
for doc in documents:
    print(f"内容: {doc.page_content[:100]}...")
    print(f"元数据: {doc.metadata}\n")

2. 文本分割

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每个块的最大字符数
    chunk_overlap=200,    # 块之间的重叠字符数
    separators=["\n\n", "\n", "。", ".", " "],  # 分割符
)

# 分割文档
splits = text_splitter.split_documents(documents)

print(f"分割成 {len(splits)} 个块")
for i, split in enumerate(splits[:3]):
    print(f"块 {i+1}: {split.page_content[:100]}...")

3. 嵌入和向量化

from langchain_openai import OpenAIEmbeddings

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

# 对文本进行嵌入
text = "Python 是一种高级编程语言"
vector = embeddings.embed_query(text)

print(f"嵌入维度: {len(vector)}")
print(f"嵌入向量前5个值: {vector[:5]}")

4. 向量存储

from langchain_community.vectorstores import Chroma

# 创建向量存储
vectorstore = Chroma.from_documents(
    documents=splits,        # 分割后的文档块
    embedding=embeddings,   # 嵌入模型
    collection_name="my_docs"  # 集合名称
)

# 保存到磁盘
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 持久化目录
)

5. 检索

# 相似度检索
results = vectorstore.similarity_search(
    query="什么是 Python 装饰器?",
    k=3  # 返回前3个最相关的文档块
)

for i, doc in enumerate(results):
    print(f"结果 {i+1}:")
    print(doc.page_content[:200])
    print(f"来源: {doc.metadata}\n")

6. 生成回答

from langchain.chat_models import init_chat_model
from langchain.prompts import ChatPromptTemplate

model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

# 构建 RAG prompt
template = """基于以下参考文档回答问题。如果文档中没有相关信息，请如实说明。

参考文档：
{context}

问题：{question}

回答："""

prompt = ChatPromptTemplate.from_template(template)

# 创建 RAG Chain
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": lambda x: format_docs(x["docs"]), "question": lambda x: x["question"]}
    | prompt
    | model
)

# 执行 RAG
docs = vectorstore.similarity_search("什么是 Python 装饰器?", k=3)
result = rag_chain.invoke({"docs": docs, "question": "什么是 Python 装饰器?"})
print(result.content)

使用 LangChain RAG 组件

create_retrieval_chain

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

# 问答链
qa_prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题：

{context}

问题：{input}
""")

# 创建文档处理链
qa_chain = create_stuff_documents_chain(llm=model, prompt=qa_prompt)

# 创建检索链
retrieval_chain = create_retrieval_chain(
    retriever=vectorstore.as_retriever(),
    return_source_documents=True,
    combine_docs_chain=qa_chain
)

# 执行
result = retrieval_chain.invoke({"input": "Python装饰器是什么?"})
print(f"回答: {result['answer']}")
print(f"来源: {result['source_documents']}")

create_history_aware_retriever

带对话历史的检索器：

from langchain.chains import create_retrieval_chain
from langchain.chains.retrieval import create_history_aware_retriever

# 历史感知检索器
history_prompt = ChatPromptTemplate.from_template("""
给定对话历史，将当前问题改写为一个独立的问题。

对话历史：
{chat_history}

当前问题：{input}

独立问题：""")

history_retriever = create_history_aware_retriever(
    llm=model,
    retriever=vectorstore.as_retriever(),
    prompt=history_prompt
)

# 使用
result = history_retriever.invoke({
    "input": "它的主要特点是什么?",
    "chat_history": [
        ("user", "Python 是什么?"),
        ("assistant", "Python 是一种高级编程语言")
    ]
})

检索优化

1. 检索器配置

# 创建可配置的检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",  # similarity, similarity_threshold, mmr, similarity_score_threshold
    search_kwargs={
        "k": 5,                    # 返回文档数量
        "score_threshold": 0.8,    # 相似度阈值
        "filter": {"source": "doc"} # 过滤条件
    }
)

2. MMR 检索

# MMR (最大边际相关性) - 多样性检索
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 5,              # 召回数量
        "fetch_k": 20,       # 候选数量
        "lambda_mult": 0.5   # 0=最相似, 1=最多样
    }
)

3. 混合检索

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 使用 LLM 压缩检索结果
compressor = LLMChainExtractor.from_llm(model)

compression_retriever = ContextualCompressionRetriever(
    base_retriever=retriever,
    document_compressor=compressor
)

高级 RAG 技术

1. 父子文档检索

# 检索小块，返回关联的大块
from langchain.retrievers import ParentDocumentRetriever
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 子块分割器
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=400,
    chunk_overlap=50
)

# 父块分割器
parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=2000,
    chunk_overlap=200
)

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,  # 文档存储
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

2. 句法感知检索

from langchain.retrievers import SentenceTransformersRetriever

# 使用句子级嵌入
retriever = SentenceTransformersRetriever(
    model_name="all-MiniLM-L6-v2"
)

3. 知识图谱增强

# 使用知识图谱增强检索
from langchain.graphs import Neo4jGraph

graph = Neo4jGraph()
graph.query("CREATE (a:Article {title: 'Python'})")

# 在 RAG 中使用图谱
# ... 结合图谱关系进行检索

实际应用示例

1. 企业知识库

from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

# 1. 加载文档
loader = DirectoryLoader("company_docs/", glob="**/*.md")
docs = loader.load()

# 2. 分割
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = splitter.split_documents(docs)

# 3. 嵌入并存储
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings()
)

# 4. 创建检索链
retrieval_chain = create_retrieval_chain(
    vectorstore.as_retriever(),
    create_stuff_documents_chain(model, qa_prompt)
)

# 5. 查询
def query_knowledgebase(question):
    result = retrieval_chain.invoke({"input": question})
    return result["answer"], result["source_documents"]

# 使用
answer, sources = query_knowledgebase("公司年假制度是什么?")
print(answer)

2. PDF 问答

from langchain_community.document_loaders import PyPDFLoader
from langchain.chains import create_retrieval_chain

# 加载 PDF
loader = PyPDFLoader("annual_report.pdf")
pages = loader.load()

# 分割
splits = text_splitter.split_documents(pages)

# 向量存储
vectorstore = Chroma.from_documents(splits, embeddings)

# 创建问答系统
qa_chain = create_retrieval_chain(vectorstore.as_retriever(), qa_chain)

# 询问关于 PDF 的问题
result = qa_chain.invoke({"input": "公司去年的收入是多少?"})

3. 多文档问答

# 从多个数据源构建 RAG
from langchain_community.document_loaders import TextLoader, CSVLoader

# 加载多种类型文档
loaders = [
    TextLoader("policy.txt"),
    TextLoader("faq.md"),
    # CSVLoader("products.csv"),
]

docs = []
for loader in loaders:
    docs.extend(loader.load())

# 统一处理
splits = text_splitter.split_documents(docs)
vectorstore = Chroma.from_documents(splits, embeddings)

# 问答
result = qa_chain.invoke({"input": "退货政策是什么?"})

RAG 评估

使用 LangSmith

import os
os.environ["LANGSMITH_TRACING"] = "true"

# 评估 RAG 系统
from langsmith.evaluation import evaluate

def evaluate_rag(dataset):
    # 评估逻辑
    pass

evaluate(
    "my-rag-system",
    evaluate_rag,
    data=dataset
)

常见问题

1. 检索不到相关内容

检查文档是否正确加载
调整 chunk_size 和 overlap
尝试不同的嵌入模型

2. 生成回答不准确

增加检索的文档数量
优化 prompt 模板
使用更强大的 LLM

3. 处理长文档

使用父子文档检索
使用摘要式记忆
分批处理

下一步

现在你已经掌握了 RAG 的核心概念，接下来学习：

向量数据库 - 深入了解向量存储
Tools 工具 - 为代理添加工具能力

什么是 RAG？​

RAG 工作原理​

为什么需要 RAG？​

RAG 架构​

完整 RAG 流程​

实现 RAG​

1. 文档加载​

2. 文本分割​

3. 嵌入和向量化​

4. 向量存储​

5. 检索​

6. 生成回答​

使用 LangChain RAG 组件​

create_retrieval_chain​

create_history_aware_retriever​

检索优化​

1. 检索器配置​

2. MMR 检索​

3. 混合检索​

高级 RAG 技术​

1. 父子文档检索​

2. 句法感知检索​

3. 知识图谱增强​

实际应用示例​

1. 企业知识库​

2. PDF 问答​

3. 多文档问答​

RAG 评估​

使用 LangSmith​

常见问题​

1. 检索不到相关内容​

2. 生成回答不准确​

3. 处理长文档​

下一步​

什么是 RAG？

RAG 工作原理

为什么需要 RAG？

RAG 架构

完整 RAG 流程

实现 RAG

1. 文档加载

2. 文本分割

3. 嵌入和向量化

4. 向量存储

5. 检索

6. 生成回答

使用 LangChain RAG 组件

create_retrieval_chain

create_history_aware_retriever

检索优化

1. 检索器配置

2. MMR 检索

3. 混合检索

高级 RAG 技术

1. 父子文档检索

2. 句法感知检索

3. 知识图谱增强

实际应用示例

1. 企业知识库

2. PDF 问答

3. 多文档问答

RAG 评估

使用 LangSmith

常见问题

1. 检索不到相关内容

2. 生成回答不准确

3. 处理长文档

下一步