RAG 应用 #
RAG（Retrieval-Augmented Generation，检索增强生成）是让 LLM 访问外部知识的关键技术。通过 RAG，你可以构建基于私有数据的问答系统、智能客服、知识库等应用。
RAG 概述 #
text
┌─────────────────────────────────────────────────────────────┐
│ 什么是 RAG？ │
├─────────────────────────────────────────────────────────────┤
│ │
│ 问题：LLM 的知识有限 │
│ - 训练数据有时效性 │
│ - 无法访问私有数据 │
│ - 可能产生幻觉 │
│ │
│ RAG 解决方案： │
│ 1. 将文档向量化存储 │
│ 2. 根据问题检索相关内容 │
│ 3. 将检索结果作为上下文 │
│ 4. LLM 基于上下文生成答案 │
│ │
└─────────────────────────────────────────────────────────────┘
RAG 流程 #
text
┌─────────────────────────────────────────────────────────────┐
│ RAG 工作流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 离线阶段（索引）： │
│ │
│ 文档 ──> 文本分割 ──> 向量化 ──> 存储到向量数据库 │
│ │
│ 在线阶段（检索）： │
│ │
│ 问题 ──> 向量化 ──> 相似度检索 ──> 获取相关文档 │
│ │
│ 生成阶段： │
│ │
│ 问题 + 相关文档 ──> 构建提示 ──> LLM 生成答案 │
│ │
└─────────────────────────────────────────────────────────────┘
文档加载 #
文档加载器 #
python
from langchain_community.document_loaders import (
TextLoader,
PyPDFLoader,
CSVLoader,
JSONLoader,
UnstructuredMarkdownLoader,
WebBaseLoader,
DirectoryLoader,
)
# 加载文本文件
loader = TextLoader("document.txt")
documents = loader.load()
# 加载 PDF
loader = PyPDFLoader("document.pdf")
pages = loader.load()
# 加载 CSV
loader = CSVLoader("data.csv")
rows = loader.load()
# 加载 JSON
loader = JSONLoader(
file_path="data.json",
jq_schema=".[]",
text_content=False
)
data = loader.load()
# 加载 Markdown
loader = UnstructuredMarkdownLoader("readme.md")
docs = loader.load()
# 加载网页
loader = WebBaseLoader("https://example.com/article")
web_docs = loader.load()
# 加载目录下所有文件
loader = DirectoryLoader(
"./documents",
glob="**/*.txt",
loader_cls=TextLoader
)
all_docs = loader.load()
常用文档加载器 #
python
# Word 文档
from langchain_community.document_loaders import Docx2txtLoader
loader = Docx2txtLoader("document.docx")
# Excel
from langchain_community.document_loaders import UnstructuredExcelLoader
loader = UnstructuredExcelLoader("data.xlsx")
# PowerPoint
from langchain_community.document_loaders import UnstructuredPowerPointLoader
loader = UnstructuredPowerPointLoader("presentation.pptx")
# HTML
from langchain_community.document_loaders import BSHTMLLoader
loader = BSHTMLLoader("page.html")
# 代码文件
from langchain_community.document_loaders import PythonLoader
loader = PythonLoader("script.py")
# Notion
from langchain_community.document_loaders import NotionDBLoader
loader = NotionDBLoader(
integration_token="token",
database_id="db-id"
)
文本分割 #
分割器类型 #
python
from langchain_text_splitters import (
CharacterTextSplitter,
RecursiveCharacterTextSplitter,
TokenTextSplitter,
MarkdownHeaderTextSplitter,
CodeTextSplitter,
)
# 递归字符分割器（推荐）
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每块最大字符数
chunk_overlap=200, # 块之间的重叠字符数
length_function=len, # 长度计算函数
separators=["\n\n", "\n", " ", ""] # 分隔符优先级
)
chunks = text_splitter.split_text(long_text)
# 字符分割器
char_splitter = CharacterTextSplitter(
separator="\n\n",
chunk_size=1000,
chunk_overlap=200
)
# Token 分割器
token_splitter = TokenTextSplitter(
chunk_size=100, # token 数
chunk_overlap=20
)
# Markdown 分割器
md_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=[
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3")
]
)
分割文档 #
python
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 创建分割器
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
# 分割文档
documents = loader.load()
splits = text_splitter.split_documents(documents)
# 查看分割结果
for i, split in enumerate(splits[:3]):
print(f"Chunk {i}:")
print(f"Content: {split.page_content[:100]}...")
print(f"Metadata: {split.metadata}")
print("-" * 50)
text
┌─────────────────────────────────────────────────────────────┐
│ 文本分割策略 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 原始文档： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 这是第一段内容... │ │
│ │ │ │
│ │ 这是第二段内容... │ │
│ │ │ │
│ │ 这是第三段内容... │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 分割后（chunk_size=100, overlap=20）： │
│ │
│ Chunk 1: [第一段 + 第二段开头] │
│ ┌──────────────────────────────────────┐ │
│ │ 这是第一段内容...这是第二段开头... │ │
│ └──────────────────────────────────────┘ │
│ ↓ overlap │
│ Chunk 2: [第二段开头 + 第二段 + 第三段开头] │
│ ┌──────────────────────────────────────┐ │
│ │ 这是第二段开头...这是第二段... │ │
│ └──────────────────────────────────────┘ │
│ │
│ 重叠确保上下文连续性 │
│ │
└─────────────────────────────────────────────────────────────┘
向量存储 #
创建向量存储 #
python
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
# 创建嵌入模型
embeddings = OpenAIEmbeddings()
# 从文档创建向量存储
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings,
persist_directory="./chroma_db" # 持久化目录
)
# 从文本列表创建
texts = ["文本1", "文本2", "文本3"]
vectorstore = Chroma.from_texts(
texts=texts,
embedding=embeddings
)
常用向量数据库 #
python
# FAISS（本地，快速）
from langchain_community.vectorstores import FAISS
vectorstore = FAISS.from_documents(documents, embeddings)
# Pinecone（云端）
from langchain_community.vectorstores import Pinecone
import pinecone
pinecone.init(api_key="key", environment="env")
vectorstore = Pinecone.from_documents(
documents, embeddings, index_name="my-index"
)
# Weaviate
from langchain_community.vectorstores import Weaviate
vectorstore = Weaviate.from_documents(
documents, embeddings, weaviate_url="http://localhost:8080"
)
# Milvus
from langchain_community.vectorstores import Milvus
vectorstore = Milvus.from_documents(
documents, embeddings, connection_args={"host": "localhost", "port": "19530"}
)
# Qdrant
from langchain_community.vectorstores import Qdrant
vectorstore = Qdrant.from_documents(
documents, embeddings, url="http://localhost:6333"
)
检索器 #
基础检索 #
python
# 相似度检索
results = vectorstore.similarity_search(
query="什么是机器学习？",
k=4 # 返回最相似的 4 个文档
)
# 带分数的检索
results_with_scores = vectorstore.similarity_search_with_score(
query="什么是机器学习？",
k=4
)
for doc, score in results_with_scores:
print(f"Score: {score}")
print(f"Content: {doc.page_content[:100]}...")
print("-" * 50)
创建检索器 #
python
# 从向量存储创建检索器
retriever = vectorstore.as_retriever(
search_type="similarity", # 检索类型
search_kwargs={"k": 4} # 检索参数
)
# 也可以使用其他搜索类型
retriever = vectorstore.as_retriever(
search_type="mmr", # 最大边际相关性
search_kwargs={"k": 4, "fetch_k": 20}
)
retriever = vectorstore.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"score_threshold": 0.8, "k": 4}
)
# 使用检索器
docs = retriever.invoke("什么是机器学习？")
高级检索器 #
python
from langchain.retrievers import (
ContextualCompressionRetriever,
EnsembleRetriever,
MultiQueryRetriever,
)
# 多查询检索器
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=retriever,
llm=model
)
# 集成检索器（混合检索）
from langchain_community.retrievers import BM25Retriever
bm25_retriever = BM25Retriever.from_documents(documents)
ensemble_retriever = EnsembleRetriever(
retrievers=[retriever, bm25_retriever],
weights=[0.5, 0.5]
)
# 上下文压缩检索器
from langchain.retrievers.document_compressors import LLMChainExtractor
compressor = LLMChainExtractor.from_llm(model)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=retriever
)
构建 RAG 链 #
基础 RAG 链 #
python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
model = ChatOpenAI(model="gpt-4o-mini")
# 定义提示模板
prompt = ChatPromptTemplate.from_template("""根据以下上下文回答问题。如果上下文中没有相关信息，请说"我不知道"。
上下文：
{context}
问题：{question}
答案：""")
# 构建 RAG 链
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)
# 使用
answer = rag_chain.invoke("什么是机器学习？")
print(answer)
带来源的 RAG 链 #
python
from langchain_core.runnables import RunnableParallel
# 并行获取答案和来源
rag_chain_with_source = RunnableParallel(
answer=(
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
),
sources=retriever
)
result = rag_chain_with_source.invoke("什么是机器学习？")
print("答案:", result["answer"])
print("\n来源:")
for doc in result["sources"]:
print(f"- {doc.metadata.get('source', '未知')}")
对话式 RAG #
python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
# 带历史的提示
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的助手。根据以下上下文回答问题。\n\n上下文：{context}"),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{question}")
])
# 简单的历史管理
chat_history = []
def chat(question: str) -> str:
# 获取相关文档
docs = retriever.invoke(question)
context = format_docs(docs)
# 调用链
chain = prompt | model | StrOutputParser()
answer = chain.invoke({
"context": context,
"chat_history": chat_history,
"question": question
})
# 更新历史
chat_history.append(HumanMessage(content=question))
chat_history.append(AIMessage(content=answer))
return answer
# 使用
print(chat("什么是机器学习？"))
print(chat("它有哪些应用？")) # 会记住之前的对话
完整示例 #
文档问答系统 #
python
from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
# 1. 加载文档
loader = DirectoryLoader(
"./documents",
glob="**/*.pdf",
loader_cls=PyPDFLoader
)
documents = loader.load()
# 2. 分割文本
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
splits = text_splitter.split_documents(documents)
# 3. 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings,
persist_directory="./chroma_db"
)
# 4. 创建检索器
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 4}
)
# 5. 构建 RAG 链
model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("""你是一个专业的问答助手。请根据以下上下文回答问题。
上下文：
{context}
问题：{question}
请提供准确、详细的回答。如果上下文中没有相关信息，请明确说明。
答案：""")
def format_docs(docs):
return "\n\n---\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)
# 6. 使用
def ask(question: str) -> str:
# 获取相关文档
docs = retriever.invoke(question)
# 生成答案
answer = rag_chain.invoke(question)
# 打印来源
print(f"问题: {question}")
print(f"\n答案: {answer}")
print(f"\n参考来源:")
for i, doc in enumerate(docs):
source = doc.metadata.get('source', '未知')
page = doc.metadata.get('page', '未知')
print(f" [{i+1}] {source} (第 {page} 页)")
return answer
# 测试
ask("文档中提到了哪些重要内容？")
RAG 优化技巧 #
1. 改进检索质量 #
python
# 使用更好的分割策略
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 较小的块
chunk_overlap=100, # 适当的重叠
separators=["\n\n", "\n", "。", "！", "？", " ", ""]
)
# 使用混合检索
ensemble_retriever = EnsembleRetriever(
retrievers=[
vectorstore.as_retriever(search_kwargs={"k": 4}),
BM25Retriever.from_documents(splits, k=4)
],
weights=[0.6, 0.4]
)
# 使用重排序
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain.retrievers import ContextualCompressionRetriever
reranker = CrossEncoderReranker(model_name="cross-encoder/ms-marco-MiniLM-L-6-v2")
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=retriever
)
2. 改进提示 #
python
prompt = ChatPromptTemplate.from_template("""你是一个专业的问答助手。请根据以下上下文回答问题。
回答要求：
1. 只使用上下文中的信息
2. 如果上下文不足，明确说明
3. 引用具体的上下文来源
4. 回答要准确、完整
上下文：
{context}
问题：{question}
答案：""")
3. 添加元数据过滤 #
python
# 存储时添加元数据
documents = [
Document(
page_content="内容...",
metadata={
"source": "doc1.pdf",
"category": "技术文档",
"date": "2024-01-01"
}
)
]
# 检索时过滤
from langchain_community.vectorstores import Chroma
results = vectorstore.similarity_search(
query="问题",
k=4,
filter={"category": "技术文档"}
)
下一步 #
最后更新：2026-03-30