LangChain RAG 检索增强生成深度解析
RAG(Retrieval-Augmented Generation,检索增强生成)是解决 LLM 知识局限性的核心技术。通过将外部知识库与 LLM 结合,RAG 可以让 AI 获取最新、准确的信息,同时减少幻觉问题。本文将深入解析 LangChain 的 RAG 架构、文档加载、向量存储、检索策略、以及生产级 RAG 系统的最佳实践。
·
关于作者
- 深耕领域:大语言模型开发 / RAG 知识库 / AI Agent 落地 / 模型微调
- 技术栈:Python | RAG (LangChain / Dify + Milvus) | FastAPI + Docker
- 工程能力:专注模型工程化部署、知识库构建与优化,擅长全流程解决方案
「让 AI 交互更智能,让技术落地更高效」
欢迎技术探讨与项目合作,解锁大模型与智能交互的无限可能!
LangChain RAG 检索增强生成深度解析
摘要
RAG(Retrieval-Augmented Generation,检索增强生成)是解决 LLM 知识局限性的核心技术。通过将外部知识库与 LLM 结合,RAG 可以让 AI 获取最新、准确的信息,同时减少幻觉问题。本文将深入解析 LangChain 的 RAG 架构、文档加载、向量存储、检索策略、以及生产级 RAG 系统的最佳实践。
一、RAG 概述
1.1 为什么需要 RAG
LLM 存在以下局限性:
| 局限性 | 说明 | RAG 解决方案 |
|---|---|---|
| 知识截止 | 训练数据有时间截止点 | 实时检索最新信息 |
| 幻觉问题 | 可能生成虚假信息 | 基于真实文档回答 |
| 领域知识 | 缺乏专业领域知识 | 注入领域文档库 |
| 上下文限制 | 上下文窗口有限 | 检索相关片段 |
1.2 RAG 架构
1.3 LangChain RAG 组件
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain.chains import RetrievalQA
# RAG 核心组件
components = {
"文档加载器": "Document Loaders - 加载各种格式文档",
"文本分割器": "Text Splitters - 将文档切分成小块",
"嵌入模型": "Embeddings - 将文本转换为向量",
"向量存储": "Vector Stores - 存储和检索向量",
"检索器": "Retrievers - 封装检索逻辑",
"LLM": "Chat Models - 生成最终答案",
}
二、文档加载
2.1 文档加载器概述
LangChain 提供了 100+ 文档加载器:
2.2 常用加载器
from langchain_community.document_loaders import (
TextLoader,
PyPDFLoader,
WebBaseLoader,
DirectoryLoader,
UnstructuredMarkdownLoader,
)
# 1. 文本文件
loader = TextLoader("./data.txt", encoding="utf-8")
docs = loader.load()
# 2. PDF 文档
loader = PyPDFLoader("./document.pdf")
docs = loader.load() # 每页一个 Document
# 3. 网页内容
loader = WebBaseLoader("https://example.com/article")
docs = loader.load()
# 4. 目录下所有文件
loader = DirectoryLoader(
"./documents",
glob="**/*.txt",
loader_cls=TextLoader,
)
docs = loader.load()
# 5. Markdown 文件
loader = UnstructuredMarkdownLoader("./README.md")
docs = loader.load()
2.3 Document 对象
from langchain_core.documents import Document
# Document 结构
doc = Document(
page_content="这是文档的文本内容...",
metadata={
"source": "./data.txt",
"page": 1,
"author": "张三",
"created_at": "2024-01-01",
}
)
print(doc.page_content) # 文本内容
print(doc.metadata) # 元数据字典
2.4 自定义加载器
from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
from typing import Iterator
class MyCustomLoader(BaseLoader):
"""自定义文档加载器。"""
def __init__(self, file_path: str):
self.file_path = file_path
def lazy_load(self) -> Iterator[Document]:
"""懒加载文档。"""
with open(self.file_path, "r", encoding="utf-8") as f:
content = f.read()
# 可以按需分割成多个 Document
for i, paragraph in enumerate(content.split("\n\n")):
if paragraph.strip():
yield Document(
page_content=paragraph,
metadata={
"source": self.file_path,
"paragraph": i,
}
)
def load(self) -> list[Document]:
"""加载所有文档。"""
return list(self.lazy_load())
# 使用
loader = MyCustomLoader("./data.txt")
docs = loader.load()
三、文本分割
3.1 为什么需要分割
| 原因 | 说明 |
|---|---|
| 上下文限制 | LLM 有最大 Token 限制 |
| 检索精度 | 小块更精准匹配查询 |
| 成本控制 | 减少不必要的 Token 消耗 |
3.2 分割策略
3.3 RecursiveCharacterTextSplitter
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 递归字符分割器(推荐)
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每块最大字符数
chunk_overlap=200, # 块之间的重叠字符数
length_function=len, # 长度计算函数
separators=[ # 分割优先级
"\n\n", # 段落
"\n", # 行
"。", # 中文句号
".", # 英文句号
" ", # 空格
"", # 字符
],
)
# 分割文档
texts = splitter.split_text("很长的文本内容...")
# 分割 Document 对象
from langchain_core.documents import Document
docs = [Document(page_content="很长的文本内容...")]
split_docs = splitter.split_documents(docs)
3.4 代码分割器
from langchain_text_splitters import (
RecursiveCharacterTextSplitter,
Language,
)
# Python 代码分割器
python_splitter = RecursiveCharacterTextSplitter.from_language(
language=Language.PYTHON,
chunk_size=500,
chunk_overlap=50,
)
code = """
def hello():
print("Hello, World!")
class MyClass:
def __init__(self):
self.value = 1
"""
chunks = python_splitter.split_text(code)
# 会按函数、类等语法结构分割
3.5 Markdown 分割器
from langchain_text_splitters import MarkdownHeaderTextSplitter
markdown_text = """
# 主标题
这是主标题下的内容。
## 二级标题
这是二级标题下的内容。
### 三级标题
这是三级标题下的内容。
"""
# 按标题层级分割
splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "header1"),
("##", "header2"),
("###", "header3"),
]
)
docs = splitter.split_text(markdown_text)
# 每个 Document 的 metadata 会包含标题信息
3.6 Token 分割器
from langchain_text_splitters import TokenTextSplitter
# 按 Token 分割(精确控制)
splitter = TokenTextSplitter(
encoding_name="cl100k_base", # GPT-4 的编码
chunk_size=100, # 每 100 个 Token
chunk_overlap=20, # 重叠 20 个 Token
)
chunks = splitter.split_text("很长的文本...")
四、向量嵌入
4.1 嵌入模型概述
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import (
HuggingFaceEmbeddings,
OllamaEmbeddings,
)
# OpenAI 嵌入(推荐)
embeddings = OpenAIEmbeddings(
model="text-embedding-3-small", # 或 text-embedding-3-large
)
# HuggingFace 本地嵌入
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2",
)
# Ollama 本地嵌入
embeddings = OllamaEmbeddings(
model="nomic-embed-text",
)
4.2 嵌入操作
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
# 1. 嵌入单个文本
vector = embeddings.embed_query("这是一段文本")
print(len(vector)) # 1536 维(text-embedding-3-small)
# 2. 嵌入多个文本
texts = ["文本1", "文本2", "文本3"]
vectors = embeddings.embed_documents(texts)
print(len(vectors)) # 3 个向量
# 3. 异步嵌入
import asyncio
async def async_embed():
vector = await embeddings.aembed_query("异步嵌入")
return vector
vector = asyncio.run(async_embed())
4.3 嵌入模型对比
| 模型 | 维度 | 特点 | 适用场景 |
|---|---|---|---|
| text-embedding-3-small | 1536 | 性价比高 | 通用场景 |
| text-embedding-3-large | 3072 | 效果最好 | 高精度需求 |
| all-MiniLM-L6-v2 | 384 | 轻量本地 | 资源受限 |
| bge-large-zh | 1024 | 中文优化 | 中文场景 |
| nomic-embed-text | 768 | 开源免费 | 本地部署 |
五、向量存储
5.1 向量数据库概述
5.2 Chroma 使用
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
# 创建向量存储
embeddings = OpenAIEmbeddings()
# 方式1:从文档创建
docs = [
Document(page_content="Python 是一种编程语言", metadata={"source": "doc1"}),
Document(page_content="机器学习是 AI 的分支", metadata={"source": "doc2"}),
]
vectorstore = Chroma.from_documents(
documents=docs,
embedding=embeddings,
persist_directory="./chroma_db", # 持久化目录
)
# 方式2:从文本创建
vectorstore = Chroma.from_texts(
texts=["文本1", "文本2", "文本3"],
embedding=embeddings,
metadatas=[{"source": "a"}, {"source": "b"}, {"source": "c"}],
)
# 加载已有向量存储
vectorstore = Chroma(
persist_directory="./chroma_db",
embedding_function=embeddings,
)
5.3 检索操作
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
vectorstore = Chroma.from_texts(
texts=["Python 是编程语言", "机器学习是 AI 分支", "深度学习使用神经网络"],
embedding=OpenAIEmbeddings(),
)
# 1. 相似度搜索
results = vectorstore.similarity_search("编程", k=2)
for doc in results:
print(doc.page_content)
# 2. 带分数的相似度搜索
results = vectorstore.similarity_search_with_score("AI", k=2)
for doc, score in results:
print(f"分数: {score:.4f}, 内容: {doc.page_content}")
# 3. 向量相似度搜索
query_vector = embeddings.embed_query("人工智能")
results = vectorstore.similarity_search_by_vector(query_vector, k=2)
# 4. 异步搜索
import asyncio
async def async_search():
results = await vectorstore.asimilarity_search("AI", k=2)
return results
results = asyncio.run(async_search())
5.4 FAISS 使用
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
# 创建 FAISS 向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
texts=["文本1", "文本2", "文本3"],
embedding=embeddings,
)
# 保存到本地
vectorstore.save_local("./faiss_index")
# 加载
vectorstore = FAISS.load_local(
"./faiss_index",
embeddings,
allow_dangerous_deserialization=True,
)
# 搜索
results = vectorstore.similarity_search("查询", k=2)
5.5 添加和删除文档
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
vectorstore = Chroma.from_texts(
texts=["初始文档"],
embedding=OpenAIEmbeddings(),
)
# 添加文档
vectorstore.add_texts(["新文档1", "新文档2"])
vectorstore.add_documents([
Document(page_content="新文档3", metadata={"source": "new"}),
])
# 删除文档(需要 ID)
ids = vectorstore.add_texts(["临时文档"])
vectorstore.delete(ids)
# 清空所有文档
# vectorstore.delete_collection()
六、检索器
6.1 检索器概述
检索器是对向量存储的封装,提供统一的检索接口:
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
vectorstore = Chroma.from_texts(
texts=["Python 是编程语言", "机器学习是 AI 分支"],
embedding=OpenAIEmbeddings(),
)
# 转换为检索器
retriever = vectorstore.as_retriever(
search_type="similarity", # 搜索类型
search_kwargs={"k": 4}, # 搜索参数
)
# 使用检索器
docs = retriever.invoke("什么是 AI?")
for doc in docs:
print(doc.page_content)
6.2 检索类型
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
vectorstore = Chroma.from_texts(
texts=["文档1", "文档2", "文档3", "文档4", "文档5"],
embedding=OpenAIEmbeddings(),
)
# 1. 相似度搜索(默认)
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3},
)
# 2. 相似度阈值搜索
retriever = vectorstore.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"k": 5,
"score_threshold": 0.8, # 只返回分数 >= 0.8 的结果
},
)
# 3. MMR(最大边际相关性)- 减少重复
retriever = vectorstore.as_retriever(
search_type="mmr",
search_kwargs={
"k": 4,
"fetch_k": 20, # 先检索 20 个
"lambda_mult": 0.5, # 多样性权重(0-1,越大越相似)
},
)
6.3 多查询检索器
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
vectorstore = Chroma.from_texts(
texts=["Python 是编程语言", "机器学习是 AI 分支"],
embedding=OpenAIEmbeddings(),
)
# 创建多查询检索器
llm = ChatOpenAI(model="gpt-4o")
retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(),
llm=llm,
)
# 会自动生成多个相关查询并合并结果
docs = retriever.invoke("什么是 AI?")
6.4 上下文压缩检索器
from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
vectorstore = Chroma.from_texts(
texts=["很长的文档内容..."],
embedding=OpenAIEmbeddings(),
)
# 创建压缩器
llm = ChatOpenAI(model="gpt-4o")
compressor = LLMChainExtractor.from_llm(llm)
# 创建压缩检索器
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=vectorstore.as_retriever(),
)
# 检索结果会被压缩,只保留与查询相关的部分
docs = compression_retriever.invoke("查询内容")
6.5 集成检索器
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
texts = ["Python 编程", "机器学习", "深度学习", "自然语言处理"]
# 向量检索器
vectorstore = Chroma.from_texts(texts, embedding=OpenAIEmbeddings())
vector_retriever = vectorstore.as_retriever()
# BM25 检索器(关键词匹配)
bm25_retriever = BM25Retriever.from_texts(texts)
# 集成检索器
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.5, 0.5], # 权重
)
# 混合检索
docs = ensemble_retriever.invoke("AI 编程")
七、RAG 链构建
7.1 基础 RAG 链
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
# 1. 准备向量存储
texts = [
"Python 是一种高级编程语言,由 Guido van Rossum 创建。",
"机器学习是人工智能的一个分支,使计算机能够从数据中学习。",
"深度学习使用多层神经网络来处理复杂模式。",
]
vectorstore = Chroma.from_texts(texts, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()
# 2. 定义提示词模板
template = """根据以下上下文回答问题。如果上下文中没有答案,请说"我不知道"。
上下文:
{context}
问题: {question}
答案:"""
prompt = ChatPromptTemplate.from_template(template)
# 3. 构建 RAG 链
llm = ChatOpenAI(model="gpt-4o")
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()
)
# 4. 调用
answer = rag_chain.invoke("什么是机器学习?")
print(answer)
7.2 带来源的 RAG 链
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
# 准备数据
vectorstore = Chroma.from_texts(
["文档内容1", "文档内容2"],
embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()
# 提示词
prompt = ChatPromptTemplate.from_template("""根据上下文回答问题。
上下文: {context}
问题: {question}""")
llm = ChatOpenAI(model="gpt-4o")
# 构建链(返回答案和来源)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
chain_with_source = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough(),
"sources": retriever,
}
| {
"answer": prompt | llm,
"sources": lambda x: x["sources"],
}
)
# 调用
result = chain_with_source.invoke("问题内容")
print("答案:", result["answer"].content)
print("来源:", [doc.page_content for doc in result["sources"]])
7.3 对话式 RAG
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import AIMessage, HumanMessage
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
# 准备向量存储
vectorstore = Chroma.from_texts(
["Python 是编程语言", "机器学习是 AI 分支"],
embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()
# 带历史的提示词
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有用的助手。根据以下上下文回答问题:\n\n{context}"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{question}"),
])
llm = ChatOpenAI(model="gpt-4o")
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# 基础链
chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough(),
"chat_history": lambda x: x["chat_history"],
}
| prompt
| llm
)
# 添加历史记录
memory = ChatMessageHistory()
chain_with_history = RunnableWithMessageHistory(
chain,
get_session_history=lambda session_id: memory,
input_messages_key="question",
history_messages_key="chat_history",
)
# 对话
response1 = chain_with_history.invoke(
{"question": "什么是 Python?"},
config={"configurable": {"session_id": "test"}},
)
print(response1.content)
response2 = chain_with_history.invoke(
{"question": "它有什么特点?"}, # 会记住上文
config={"configurable": {"session_id": "test"}},
)
print(response2.content)
八、高级 RAG 技术
8.1 查询重写
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
# 查询重写链
rewrite_prompt = ChatPromptTemplate.from_template(
"""将用户的问题改写为更适合检索的形式。
原始问题: {question}
改写后的问题:"""
)
llm = ChatOpenAI(model="gpt-4o")
rewrite_chain = rewrite_prompt | llm
# 使用
original_query = "那个语言怎么样?"
rewritten = rewrite_chain.invoke({"question": original_query})
print(rewritten.content)
8.2 重排序
from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
# 使用 Cross-Encoder 重排序
model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")
compressor = CrossEncoderReranker(model=model, top_n=3)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10}),
)
# 先检索 10 个,重排序后返回前 3 个
docs = compression_retriever.invoke("查询内容")
8.3 自查询检索器
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
from langchain_openai import ChatOpenAI
# 定义元数据字段
metadata_field_info = [
AttributeInfo(
name="genre",
description="电影的类型,如'动作'、'喜剧'、'科幻'",
type="string",
),
AttributeInfo(
name="year",
description="电影的发行年份",
type="integer",
),
]
document_content_description = "电影简介"
llm = ChatOpenAI(model="gpt-4o")
# 创建自查询检索器
retriever = SelfQueryRetriever.from_llm(
llm=llm,
vectorstore=vectorstore,
document_contents=document_content_description,
metadata_field_info=metadata_field_info,
)
# 自动解析查询中的过滤条件
docs = retriever.invoke("2023年的科幻电影有哪些?")
# 会自动生成类似 {"genre": "科幻", "year": 2023} 的过滤条件
九、最佳实践
9.1 分块策略
| 文档类型 | 推荐策略 | 参数建议 |
|---|---|---|
| 通用文本 | RecursiveCharacterTextSplitter | chunk_size=1000, overlap=200 |
| 代码 | Language 专用分割器 | chunk_size=500, overlap=50 |
| Markdown | MarkdownHeaderTextSplitter | 按标题层级分割 |
| 长文档 | 分层索引 | 先检索块,再检索父文档 |
9.2 检索优化
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
# 1. 混合检索(向量 + 关键词)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
bm25_retriever = BM25Retriever.from_texts(texts)
bm25_retriever.k = 5
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4],
)
# 2. 重排序优化
# 先粗检索,再精排序
# 3. 多查询扩展
# 生成多个相关查询,合并结果
9.3 评估指标
| 指标 | 说明 | 计算方式 |
|---|---|---|
| 召回率 | 检索到相关文档的比例 | 相关文档被检索数 / 总相关文档数 |
| 精确率 | 检索结果中相关文档的比例 | 相关文档数 / 检索结果总数 |
| MRR | 平均倒数排名 | 第一个正确答案排名的倒数 |
| NDCG | 归一化折损累积增益 | 考虑排序位置的评估 |
十、总结
10.1 RAG 架构图
10.2 关键组件总结
| 组件 | 作用 | 推荐选择 |
|---|---|---|
| 文档加载器 | 加载各种格式文档 | 按需选择 |
| 文本分割器 | 切分文档为小块 | RecursiveCharacterTextSplitter |
| 嵌入模型 | 文本向量化 | text-embedding-3-small |
| 向量存储 | 存储和检索向量 | Chroma(开发)/ Milvus(生产) |
| 检索器 | 封装检索逻辑 | EnsembleRetriever |
| LLM | 生成答案 | GPT-4o / Claude |
参考资料
更多推荐



所有评论(0)