跳到主要内容

RAG 检索增强

RAG（Retrieval Augmented Generation，检索增强生成）是当前最主流的 LLM 应用架构之一。它将信息检索与文本生成相结合，让 LLM 能够基于私有数据或实时信息生成准确、可靠的回答。

什么是 RAG？​

RAG 的核心思想是：在生成回答之前，先从知识库中检索相关文档，将检索结果作为上下文提供给 LLM。这样，模型的回答不再是基于训练数据的"记忆"，而是基于实际检索到的"证据"。

RAG 解决的问题​

传统 LLM 存在几个根本性限制：

  • 知识截止：训练数据有截止日期，无法回答最新信息
  • 私有数据：无法访问企业内部文档、个人知识库
  • 幻觉问题：对不熟悉的主题可能编造错误信息
  • 可追溯性：无法说明答案的来源依据

RAG 通过检索外部知识库，有效缓解这些问题。模型的回答可以追溯到具体文档，且知识库可以实时更新。

RAG 架构概览​

RAG 系统分为两个主要阶段：

索引阶段（离线）：将文档处理成可检索的形式

  • 文档加载：从各种数据源读取文档
  • 文本分割：将长文档切分成适当大小的块
  • 向量嵌入：将文本转换为向量表示
  • 向量存储：将向量存入数据库，建立索引

查询阶段（在线）：处理用户问题并生成回答

  • 查询嵌入：将用户问题转换为向量
  • 相似度检索：从向量库中找到最相关的文档
  • 上下文构建：将检索结果与问题组合成提示词
  • 回答生成：LLM 基于上下文生成回答

完整 RAG 实现​

第一步：文档加载​

LangChain 提供了丰富的文档加载器，支持多种数据源：

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

# 加载纯文本
loader = TextLoader("document.txt", encoding="utf-8")
documents = loader.load()

# 加载 PDF
loader = PyPDFLoader("report.pdf")
documents = loader.load()  # 每页作为一个 Document

# 加载目录下所有 Markdown 文件
loader = DirectoryLoader(
    "docs/",
    glob="**/*.md",
    loader_cls=UnstructuredMarkdownLoader
)
documents = loader.load()

# 从网页加载
loader = WebBaseLoader("https://example.com/article")
documents = loader.load()

# 查看加载结果
for doc in documents[:2]:
    print(f"内容预览: {doc.page_content[:100]}...")
    print(f"元数据: {doc.metadata}")
    print("---")

Document 对象结构：

每个 Document 包含：

  • page_content：文档内容（字符串）
  • metadata：元数据（字典），如来源、页码、标题等

第二步：文本分割​

分割是影响 RAG 效果的关键环节。分割策略需要平衡两个因素：

  • 块太小：上下文不完整，信息碎片化
  • 块太大：检索不精确，包含过多噪音
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 目标块大小（字符数）
    chunk_overlap=50,      # 相邻块的重叠字符数
    length_function=len,   # 计算长度的函数
    separators=["\n\n", "\n", "。", ".", " ", ""]  # 分割符优先级
)

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

print(f"原文档数: {len(documents)}")
print(f"分割后块数: {len(splits)}")

# 查看分割效果
for i, split in enumerate(splits[:3]):
    print(f"\n块 {i+1} (长度: {len(split.page_content)}):")
    print(split.page_content[:200])

RecursiveCharacterTextSplitter 工作原理：

它会按照分隔符优先级依次尝试分割：

  1. 首先尝试按双换行（段落）分割
  2. 如果块仍然太大，按单换行（句子）分割
  3. 继续尝试句号、空格等

这种递归策略保证了分割后的文本尽可能保持语义完整性。

按 Token 分割：

某些嵌入模型对 Token 数量有限制，可以按 Token 分割：

from langchain.text_splitter import TokenTextSplitter

token_splitter = TokenTextSplitter(
    chunk_size=100,     # Token 数量
    chunk_overlap=20
)

splits = token_splitter.split_documents(documents)

第三步：向量嵌入与存储​

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

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

# 创建向量存储并持久化
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="my_knowledge_base"
)

# 验证存储
print(f"向量数量: {vectorstore._collection.count()}")

# 测试检索
results = vectorstore.similarity_search("测试查询", k=2)
for doc in results:
    print(f"内容: {doc.page_content[:100]}...")
    print(f"来源: {doc.metadata.get('source')}")

第四步：构建检索链​

使用 LCEL 构建标准的 RAG 检索链：

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain.output_parsers import StrOutputParser

# 初始化模型
model = init_chat_model("openai:gpt-4o-mini")

# 定义提示模板
qa_prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题。如果上下文中没有相关信息，请说明"根据提供的文档无法回答此问题"。

上下文：
{context}

问题：{question}

请给出详细、准确的回答：
""")

# 格式化文档函数
def format_docs(docs):
    """将文档列表格式化为字符串"""
    return "\n\n---\n\n".join(
        f"【文档 {i+1}】\n{doc.page_content}"
        for i, doc in enumerate(docs)
    )

# 构建 RAG 链
rag_chain = (
    {
        "context": vectorstore.as_retriever() | format_docs,
        "question": RunnablePassthrough()
    }
    | qa_prompt
    | model
    | StrOutputParser()
)

# 执行查询
answer = rag_chain.invoke("什么是机器学习？")
print(answer)

链的工作流程解析：

  1. RunnablePassthrough() 将用户问题原样传递
  2. vectorstore.as_retriever() 检索相关文档
  3. format_docs 将文档列表转为字符串
  4. qa_prompt 构建包含上下文的提示词
  5. model 调用 LLM 生成回答
  6. StrOutputParser() 提取文本内容

第五步：添加来源追踪​

让回答可追溯到具体文档：

from langchain_core.runnables import RunnableParallel

# 创建带来源的链
rag_chain_with_sources = RunnableParallel(
    # 生成回答
    answer=(
        {
            "context": vectorstore.as_retriever() | format_docs,
            "question": RunnablePassthrough()
        }
        | qa_prompt
        | model
        | StrOutputParser()
    ),
    # 保留来源文档
    sources=vectorstore.as_retriever()
)

# 执行
result = rag_chain_with_sources.invoke("什么是机器学习？")

print("回答:")
print(result["answer"])

print("\n来源文档:")
for i, doc in enumerate(result["sources"]):
    print(f"  {i+1}. {doc.metadata.get('source', '未知来源')}")
    print(f"     内容片段: {doc.page_content[:50]}...")

高级 RAG 模式​

1. 对话式 RAG​

支持多轮对话，模型能够理解上下文：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.messages import HumanMessage, AIMessage

# 历史感知的问题改写
contextualize_prompt = ChatPromptTemplate.from_template("""
根据对话历史，将当前问题改写为独立问题。不要回答问题，只改写。

对话历史：
{chat_history}

当前问题：{question}

独立问题：
""")

# 历史感知检索链
question_rewriter = contextualize_prompt | model | StrOutputParser()

def rag_with_history(question: str, chat_history: list = None):
    """带历史上下文的 RAG"""
    if chat_history is None:
        chat_history = []
    
    # 1. 将问题改写为独立问题
    if chat_history:
        standalone_question = question_rewriter.invoke({
            "question": question,
            "chat_history": chat_history
        })
    else:
        standalone_question = question
    
    print(f"改写后的问题: {standalone_question}")
    
    # 2. 检索并生成回答
    answer = rag_chain.invoke(standalone_question)
    return answer

# 使用示例
history = []

# 第一轮
q1 = "什么是机器学习？"
a1 = rag_with_history(q1, history)
history.extend([
    HumanMessage(content=q1),
    AIMessage(content=a1)
])

# 第二轮（引用上文）
q2 = "它有哪些主要应用？"
a2 = rag_with_history(q2, history)
# 模型会理解"它"指的是"机器学习"

2. RAG Agent​

使用 Agent 实现更灵活的检索策略：

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def search_knowledge_base(query: str) -> str:
    """
    搜索知识库，返回相关文档内容。
    
    参数:
        query: 搜索查询
    
    返回:
        相关文档内容
    """
    docs = vectorstore.similarity_search(query, k=3)
    return format_docs(docs)

# 创建 RAG Agent
agent = create_agent(
    model=model,
    tools=[search_knowledge_base],
    system_prompt="""你是一个知识问答助手。

当用户提问时：
1. 先分析问题需要哪些信息
2. 使用 search_knowledge_base 工具检索相关内容
3. 基于检索结果回答问题
4. 如果检索结果不足，如实说明
"""
)

# 执行
result = agent.invoke({
    "messages": [{"role": "user", "content": "介绍一下深度学习的发展历史"}]
})

for message in result["messages"]:
    if hasattr(message, "content"):
        print(message.content)

Agent 模式的优势：

  • 自主决定是否需要检索
  • 可以多次检索补充信息
  • 支持多工具协作

3. 多查询检索​

对于复杂问题，生成多个查询角度：

from langchain_core.prompts import ChatPromptTemplate

# 多查询生成
multi_query_prompt = ChatPromptTemplate.from_template("""
你是一个AI助手，帮助生成多个搜索查询。
用户问题：{question}

请生成3个相关的搜索查询，从不同角度检索信息。
每个查询一行，不要编号。
""")

def multi_query_retrieval(question: str, k: int = 3):
    """多查询检索"""
    # 生成多个查询
    multi_queries = (multi_query_prompt | model | StrOutputParser()).invoke(
        {"question": question}
    )
    
    queries = [q.strip() for q in multi_queries.strip().split("\n") if q.strip()]
    queries.append(question)  # 加上原问题
    
    print(f"生成的查询: {queries}")
    
    # 对每个查询检索
    all_docs = []
    seen = set()
    
    for query in queries:
        docs = vectorstore.similarity_search(query, k=k)
        for doc in docs:
            # 去重
            content_hash = hash(doc.page_content)
            if content_hash not in seen:
                seen.add(content_hash)
                all_docs.append(doc)
    
    return all_docs[:k*2]  # 返回最多 k*2 个不重复文档

# 使用
docs = multi_query_retrieval("如何优化机器学习模型的性能？")

检索优化策略​

1. 混合检索​

结合关键词和语义检索：

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

# BM25 关键词检索器
bm25_retriever = BM25Retriever.from_documents(splits)
bm25_retriever.k = 5

# 向量检索器
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 组合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # 向量检索权重更高
)

# 在链中使用
hybrid_chain = (
    {
        "context": ensemble_retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | qa_prompt
    | model
    | StrOutputParser()
)

2. 重排序​

使用重排序模型提升检索精度：

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 加载重排序模型
reranker_model = HuggingFaceCrossEncoder(
    model_name="BAAI/bge-reranker-base"
)

# 创建重排序压缩器
reranker = CrossEncoderReranker(
    model=reranker_model,
    top_n=5
)

# 创建重排序检索器
reranking_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

# 使用
docs = reranking_retriever.invoke("机器学习模型优化方法")

3. 元数据过滤​

利用文档元数据精确检索：

# 假设文档有 category 和 year 元数据
filtered_retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {
            "category": "技术文档",
            "year": {"$gte": 2023}
        }
    }
)

# 或在检索时动态过滤
results = vectorstore.similarity_search(
    "查询内容",
    k=5,
    filter={"source": "product_manual.pdf"}
)

实际应用示例​

企业知识库系统​

from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain.output_parsers import StrOutputParser

class KnowledgeBase:
    """企业知识库系统"""
    
    def __init__(self, docs_dir: str, db_dir: str = "./kb_db"):
        self.docs_dir = docs_dir
        self.db_dir = db_dir
        self.embeddings = OpenAIEmbeddings()
        self.model = init_chat_model("openai:gpt-4o-mini")
        
        # 初始化或加载数据库
        self._init_vectorstore()
        self._init_chain()
    
    def _init_vectorstore(self):
        """初始化向量存储"""
        try:
            # 尝试加载已有数据库
            self.vectorstore = Chroma(
                persist_directory=self.db_dir,
                embedding_function=self.embeddings
            )
            print(f"已加载知识库，共 {self.vectorstore._collection.count()} 条记录")
        except Exception:
            # 首次运行，构建索引
            print("构建知识库索引...")
            self._build_index()
    
    def _build_index(self):
        """构建向量索引"""
        # 加载文档
        loader = DirectoryLoader(
            self.docs_dir,
            glob="**/*.md",
            show_progress=True
        )
        documents = loader.load()
        
        # 分割
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
        splits = splitter.split_documents(documents)
        
        # 存储向量
        self.vectorstore = Chroma.from_documents(
            documents=splits,
            embedding=self.embeddings,
            persist_directory=self.db_dir
        )
        print(f"索引构建完成，共 {len(splits)} 个文档块")
    
    def _init_chain(self):
        """初始化 RAG 链"""
        prompt = ChatPromptTemplate.from_template("""
你是企业知识库助手。请基于以下文档回答问题。
如果文档中没有相关信息，请明确说明。

文档内容：
{context}

问题：{question}

回答：
""")
        
        def format_docs(docs):
            return "\n\n".join(doc.page_content for doc in docs)
        
        self.chain = (
            {
                "context": self.vectorstore.as_retriever() | format_docs,
                "question": RunnablePassthrough()
            }
            | prompt
            | self.model
            | StrOutputParser()
        )
    
    def query(self, question: str) -> str:
        """查询知识库"""
        return self.chain.invoke(question)
    
    def add_document(self, file_path: str):
        """添加新文档"""
        loader = TextLoader(file_path)
        documents = loader.load()
        
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
        splits = splitter.split_documents(documents)
        
        self.vectorstore.add_documents(splits)
        print(f"已添加 {len(splits)} 个文档块")

# 使用
kb = KnowledgeBase(docs_dir="./company_docs")

# 查询
answer = kb.query("公司的年假政策是什么？")
print(answer)

# 添加新文档
kb.add_document("./new_policy.md")

常见问题与解决​

1. 检索不到相关内容​

可能原因：

  • 文档分割过大或过小
  • 嵌入模型与文档语言不匹配
  • 查询表述与文档内容差异大

解决方案：

# 尝试不同的分割大小
splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=30)

# 使用多查询
# 使用混合检索

2. 回答不准确​

可能原因：

  • 检索到的文档不相关
  • 上下文窗口限制
  • 提示词设计不当

解决方案：

# 使用重排序提升检索质量
# 优化提示词，明确回答边界
# 使用更强的模型

3. 响应速度慢​

优化策略：

  • 减少检索数量 k
  • 使用更小的嵌入模型
  • 启用流式输出
  • 使用本地嵌入模型避免 API 调用

最佳实践总结​

  1. 文档分割：根据内容类型选择合适的 chunk_size，通常 300-800 字符
  2. 重叠设置：保持 10-20% 的重叠，保证上下文连续性
  3. 提示词设计：明确告知模型只能基于检索内容回答
  4. 来源追踪：始终返回文档来源，增强可信度
  5. 持续优化：记录查询日志，分析检索质量，迭代改进

下一步​

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