跳至主要內容

RAG

zhangxj2026/6/20AILangChainRAGPython大约 15 分钟约 4481 字

什么是 RAG

RAG（Retrieval-Augmented Generation，检索增强生成）是一种将外部知识检索与大语言模型生成相结合的技术。核心思想是：在 LLM 生成回答之前，先从外部知识库中检索相关信息，然后将检索结果作为上下文提供给模型，从而生成更准确、更有依据的回答。

简单来说，RAG 就是让模型"先查资料，再回答问题"。

为什么需要 RAG

大语言模型存在一些固有的局限性，RAG 可以有效缓解这些问题：

问题说明RAG 如何解决
知识截止模型训练数据有时间截止点，无法获知最新信息通过检索实时获取最新知识
幻觉问题模型可能"编造"不存在的事实基于检索到的真实文档生成回答
领域知识不足通用模型对专业领域知识覆盖有限接入专业领域的知识库
上下文长度限制无法将所有相关资料一次性输入模型检索最相关的片段，精准投喂
无法引用来源回答缺乏可追溯性基于检索文档回答，可溯源

RAG 核心组件

RAG 系统由以下核心组件构成，每个组件都有对应的 LangChain 实现：

文档加载器（Document Loaders）

文档加载器负责从各种数据源读取数据，并将其转换为统一的 Document 对象。每个 Document 包含两个核心字段：

  • page_content：文档的文本内容
  • metadata：元数据（来源、页码等）
from langchain_core.documents import Document

# Document 的基本结构
doc = Document(
    page_content="这是文档的正文内容",
    metadata={"source": "example.txt", "page": 1}
)

常用文档加载器

LangChain 提供了丰富的文档加载器，覆盖文件、网页、数据库等多种数据源：

文件类加载器：

加载器用途安装包
TextLoader纯文本文件（.txt）langchain-community
CSVLoaderCSV 文件langchain-community
PyPDFLoaderPDF 文件pypdf
UnstructuredLoader多种格式（PDF/Word/HTML 等）unstructured
JSONLoaderJSON 文件langchain-community
DirectoryLoader批量加载目录下所有文件langchain-community
MarkdownLoaderMarkdown 文件unstructured

网页类加载器：

加载器用途安装包
WebBaseLoader网页内容beautifulsoup4
SitemapLoader网站 Sitemapbeautifulsoup4
FireCrawlLoader动态渲染网页firecrawl-py

其他常用加载器：

加载器用途安装包
NotionDirectoryLoaderNotion 导出langchain-community
GitLoaderGit 仓库文件langchain-community
WikipediaLoader维基百科langchain-community
ArxivLoaderarXiv 论文langchain-community

文档加载器使用示例

文本(txt)文件
from langchain_community.document_loaders import TextLoader

loader = TextLoader(
    file_path="example.txt",
    encoding="utf-8",        # 文件编码
    autodetect_encoding=False, # 是否自动检测编码
)
docs = loader.load()
print(docs[0].page_content)

参数说明：

参数类型默认值说明
file_pathstr必填文件路径
encodingstr"utf-8"文件编码，中文文件常用 "utf-8" 或 "gbk"
autodetect_encodingboolFalse是否自动检测文件编码（需要 chardet 包）

批量加载目录文件：

from langchain_community.document_loaders import DirectoryLoader, TextLoader

loader = DirectoryLoader(
    path="./docs",                    # 目录路径
    glob="**/*.md",                   # glob 模式匹配文件
    loader_cls=TextLoader,            # 指定文件加载器类
    loader_kwargs={"encoding": "utf-8"},  # 传递给加载器的参数
    show_progress=True,               # 显示加载进度条
    use_multithreading=True,          # 多线程加载
    max_concurrency=4,                # 最大并发数
    silent_errors=True,               # 跳过加载失败的文件（不抛异常）
)
docs = loader.load()
print(f"共加载 {len(docs)} 个文件")

DirectoryLoader 参数说明：

参数类型默认值说明
pathstr必填目录路径
globstr"**/[!.]*"glob 模式匹配文件，如 "**/*.pdf"、"*.txt"
loader_clsLoaderTextLoader指定文件加载器类，如 TextLoader、PyPDFLoader
loader_kwargsdict{}传递给加载器的额外参数
show_progressboolFalse是否显示进度条（需要 tqdm）
use_multithreadingboolFalse是否使用多线程并行加载
max_concurrencyint10最大并发加载数
silent_errorsboolFalse遇到加载错误时是否跳过（True 时不抛异常）
excludelist[str][]排除的文件模式列表，如 ["*.tmp"]

手动实现简易加载器：

如果不想安装额外依赖，也可以手动实现：

import pypdf
from langchain_core.documents import Document


def load_pdf_pages(file_path: str) -> list[Document]:
    reader = pypdf.PdfReader(file_path)
    return [
        Document(
            page_content=page.extract_text() or "",
            metadata={"source": file_path, "page": i},
        )
        for i, page in enumerate(reader.pages)
    ]

docs = load_pdf_pages("document.pdf")

文本切分器（Text Splitters）

文档加载后，往往需要切分成更小的片段（chunks）。原因：

  • 模型上下文窗口有限：过长的文本无法放入上下文
  • 检索精度：较小的片段更容易匹配到相关内容
  • 避免信息稀释：大段文本中关键信息可能被"淹没"

切分策略对比

切分器切分方式适用场景
RecursiveCharacterTextSplitter按分隔符递归切分（段落→句子→字符）通用文本（推荐默认选择）
CharacterTextSplitter按单个分隔符切分简单文本
TokenTextSplitter按 Token 数量切分需要精确控制 Token 时
MarkdownTextSplitter按 Markdown 标题切分Markdown 文档
HTMLSectionSplitter按 HTML 标签切分HTML 文档
RecursiveJsonSplitter按 JSON 结构切分JSON 数据

RecursiveCharacterTextSplitter（推荐）

这是 LangChain 最推荐的通用文本切分器。它会按照分隔符的层级递归切分，尽量保持较大的文本单元完整：

  1. 先尝试按 \n\n（段落）切分
  2. 如果段落仍然过大，按 \n（行）切分
  3. 如果仍然过大，按 （空格/句子）切分
  4. 最后按字符切分
from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,       # 每个片段的最大字符数
    chunk_overlap=200,     # 相邻片段的重叠字符数（保持上下文连贯）
    length_function=len,   # 长度计算函数（默认按字符数）
    add_start_index=True,  # 在元数据中记录片段在原文中的起始位置
)

# 切分 Document 对象
splits = text_splitter.split_documents(docs)
print(f"切分为 {len(splits)} 个片段")

基于 Token 的切分

如果你希望按 Token 数量精确控制（例如匹配模型的 Token 限制），可以使用基于 Token 的切分器：

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 使用 tiktoken 编码器（适配 OpenAI 模型）
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250,
    chunk_overlap=50,
)
splits = text_splitter.split_documents(docs)

Markdown 专用切分

from langchain_text_splitters import MarkdownTextSplitter

splitter = MarkdownTextSplitter(
    chunk_size=1000,
    chunk_overlap=100,
)
splits = splitter.split_documents(markdown_docs)

切分参数调优建议

参数建议值说明
chunk_size500 - 1500太小丢失上下文，太大降低检索精度
chunk_overlapchunk_size 的 10%-20%确保跨片段信息不丢失
add_start_indexTrue方便定位片段在原文中的位置

Embedding 模型（文本嵌入）

Embedding 模型将文本转换为向量（一组数字），使得语义相近的文本在向量空间中距离更近。这是实现语义检索的基础。

from langchain_openai import OpenAIEmbeddings

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

# 将文本转换为向量
vector = embeddings.embed_query("什么是 RAG？")
print(f"向量维度: {len(vector)}")  # 例如 3072

常用 Embedding 模型

模型提供商维度特点
text-embedding-3-largeOpenAI3072高精度，成本较高
text-embedding-3-smallOpenAI1536性价比高
text-embedding-ada-002OpenAI1536旧版，不推荐新项目
BGE-M3BAAI1024开源，支持多语言
nomic-embed-textNomic768开源，可本地部署

向量数据库（Vector Stores）

向量数据库用于存储 Embedding 向量，并提供高效的相似度检索能力。

常用向量数据库

数据库类型特点
FAISS内存Facebook 开源，高性能，适合本地开发
ChromaDB内存/持久化轻量级，开箱即用
InMemoryVectorStore内存LangChain 内置，无需额外安装
Pinecone云服务托管服务，无需运维
Weaviate自托管/云支持混合检索
Milvus自托管高性能，适合大规模数据
PGVectorPostgreSQL 扩展利用现有 PostgreSQL
Qdrant自托管/云Rust 实现，高性能

使用示例

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

# 创建向量库并存储文档
embeddings = OpenAIEmbeddings()
vectorstore = InMemoryVectorStore.from_documents(
    documents=splits,
    embedding=embeddings,
)

# 也可以分步操作
vectorstore = InMemoryVectorStore(embeddings)
vectorstore.add_documents(splits)

检索器（Retrievers）

检索器是 RAG 系统中负责根据用户问题查找相关文档的组件。VectorStore 可以直接转换为检索器：

# 将向量库转换为检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 检索类型
    search_kwargs={"k": 4},    # 返回前 4 个最相关的结果
)

# 使用检索器
docs = retriever.invoke("什么是 RAG？")
for doc in docs:
    print(doc.page_content[:100], "...")

检索类型

检索类型说明
similarity基于向量相似度检索（默认）
mmr最大边际相关性（平衡相关性和多样性）
similarity_score_threshold带相似度阈值的检索
# MMR 检索：避免返回过于相似的结果，提高多样性
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5},
)

完整 RAG 实战示例

传统 RAG 流程

下面是一个完整的 RAG 系统实现，包含文档加载、切分、存储、检索和生成：

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import bs4

# 1. 加载文档
loader = WebBaseLoader(
    web_paths=["https://example.com/article"],
    bs_kwargs={"parse_only": bs4.SoupStrainer(class_=("post-content",))},
)
docs = loader.load()

# 2. 切分文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    add_start_index=True,
)
splits = text_splitter.split_documents(docs)

# 3. 创建向量库并存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = InMemoryVectorStore.from_documents(
    documents=splits,
    embedding=embeddings,
)

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

# 5. 构建 RAG 链
llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题。如果上下文中没有相关信息，请说明你不确定。

上下文：
{context}

问题：{question}
""")


def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 6. 提问
answer = rag_chain.invoke("这篇文章的主要观点是什么？")
print(answer)

使用 LCEL 组装 RAG 链

LangChain 的 LCEL（LangChain Expression Language）提供了声明式的链式调用语法，用 | 操作符将各组件串联起来，非常直观：

# 上面的 RAG 链可以用更简洁的方式表达
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 流式输出
for chunk in rag_chain.stream("解释一下 RAG 的工作原理"):
    print(chunk, end="", flush=True)

进阶 RAG 技术

文档预处理

在切分前，可以对文档进行预处理来提升质量：

import bs4
from langchain_community.document_loaders import WebBaseLoader

# 使用 BeautifulSoup 过滤器只提取关键内容
loader = WebBaseLoader(
    web_paths=["https://example.com/article"],
    bs_kwargs={
        "parse_only": bs4.SoupStrainer(
            class_=("post-title", "post-content", "post-meta")
        )
    },
)
docs = loader.load()

多查询检索（Multi-Query）

同一个问题可以有不同的表述方式。多查询检索通过 LLM 生成多个查询变体，扩大检索覆盖面：

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

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

multi_query_prompt = ChatPromptTemplate.from_template(
    "你是一个 AI 助手。请为以下问题生成 3 个不同角度的检索查询，每行一个：\n"
    "问题：{question}"
)

# 生成多个查询
chain = multi_query_prompt | llm | StrOutputParser()
queries = chain.invoke({"question": "RAG 的优势是什么？"}).strip().split("\n")

# 用所有查询进行检索，合并去重
all_docs = []
seen = set()
for query in queries:
    for doc in retriever.invoke(query):
        if doc.page_content not in seen:
            seen.add(doc.page_content)
            all_docs.append(doc)

上下文压缩（Contextual Compression）

检索到的文档片段可能包含无关内容。上下文压缩通过 LLM 提取或压缩文档中的关键信息：

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

llm = ChatOpenAI(model="gpt-4o-mini")
compressor = LLMChainExtractor.from_llm(llm)

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

# 返回的文档只包含与问题相关的内容
compressed_docs = compression_retriever.invoke("RAG 的核心组件有哪些？")

Agentic RAG

传统 RAG 是"检索-生成"的线性流程。Agentic RAG 将 Agent 的推理能力引入 RAG，让模型自主决定何时检索、检索什么、以及检索结果是否足够。

传统 RAG vs Agentic RAG

实现示例

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

# 假设已有 vectorstore
retriever = vectorstore.as_retriever()


@tool
def search_knowledge_base(query: str) -> str:
    """搜索知识库，获取与查询相关的文档内容。当需要查找内部知识时使用此工具。"""
    docs = retriever.invoke(query)
    return "\n\n".join(doc.page_content for doc in docs)


# 创建带检索工具的 Agent
agent = create_agent(
    model="gpt-4o-mini",
    tools=[search_knowledge_base],
    system_prompt=(
        "你是一个专业的问答助手。你可以使用 search_knowledge_base 工具"
        "来搜索知识库获取信息。请基于检索到的信息回答问题。"
    ),
)

# Agent 会自主决定是否需要检索
response = agent.invoke({
    "messages": [("human", "RAG 和微调有什么区别？")]
})

Agentic RAG 的优势

特性传统 RAGAgentic RAG
检索时机固定在生成前Agent 自主决定
查询策略直接使用用户问题可改写、分解查询
结果判断无判断是否需要再次检索
多工具调用不支持支持多种检索工具
复杂问题效果一般可分步推理

RAG vs 微调（Fine-tuning）

RAG 和微调是两种让模型掌握特定知识的主要方式，各有优劣：

维度RAG微调
知识更新实时更新，修改知识库即可需要重新训练模型
成本无需训练，运行时成本较高训练成本高，推理成本低
可解释性可追溯到具体来源文档难以追溯知识来源
适用场景知识频繁更新、需要引用来源特定风格/格式的生成
幻觉控制较好，基于检索文档生成仍有幻觉风险
上下文长度受模型上下文窗口限制知识内化在模型参数中

简单决策：如果知识频繁更新或需要引用来源，选择 RAG；如果需要模型学习特定的输出风格或格式，选择微调。两者也可以结合使用。

学习资源

最近更新:
贡献者: zhangxj_a_sc
🍃你是故意找茬是不是?
Copyright © 2026 zhangxj