RAG简介

检索增强生成（Retrieval-Augmented Generation，RAG）是目前最流行的LLM应用架构。

什么是RAG？

RAG通过检索外部知识库，增强LLM的生成能力。

为什么需要RAG？

LLM的局限性

RAG的优势

✅ 实时性 - 可访问最新信息 ✅ 准确性 - 减少幻觉，基于事实 ✅ 可控性 - 明确数据来源 ✅ 成本低 - 无需微调模型 ✅ 可更新 - 知识库易于维护 ✅ 隐私安全 - 数据本地化

RAG工作流程

完整流程

详细步骤

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
# 步骤1: 准备文档
documents = [
"Python是一种高级编程语言，由Guido van Rossum于1991年创建。",
"Python以其简洁清晰的语法而闻名，适合快速开发。",
"Python广泛应用于Web开发、数据科学、人工智能等领域。",
]
# 步骤2: 文本分块
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200,
chunk_overlap=50
)
chunks = text_splitter.split_documents(
[{"page_content": doc} for doc in documents]
)
# 步骤3: 创建向量数据库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings
)
# 步骤4: 定义检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
# 步骤5: 构建Prompt
prompt = ChatPromptTemplate.from_template("""
基于以下参考信息回答问题：
参考信息：
{context}
问题：{question}
如果参考信息不足，请明确说明。
""")
# 步骤6: 创建LLM
llm = ChatOpenAI(model="gpt-4o-mini")
# 步骤7: 构建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
| llm
)
# 步骤8: 执行查询
query = "Python是什么时候创建的？"
result = rag_chain.invoke(query)
print(result.content)

核心组件

1. 文档加载器

from langchain_community.document_loaders import (
TextLoader,
PyPDFLoader,
WebBaseLoader,
DirectoryLoader
)
# 加载文本文件
text_loader = TextLoader("document.txt")
docs = text_loader.load()
# 加载PDF
pdf_loader = PyPDFLoader("paper.pdf")
pdf_docs = pdf_loader.load()
# 加载网页
web_loader = WebBaseLoader("https://example.com")
web_docs = web_loader.load()
# 加载整个目录
dir_loader = DirectoryLoader(
"./docs/",
glob="**/*.txt",
loader_cls=TextLoader
)
all_docs = dir_loader.load()

2. 文本分割器

from langchain.text_splitter import (
RecursiveCharacterTextSplitter,
CharacterTextSplitter,
TokenTextSplitter
)
# 递归分割（推荐）
recursive_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""]
)
# 字符分割
character_splitter = CharacterTextSplitter(
chunk_size=1000,
chunk_overlap=0,
separator="\n\n"
)
# Token分割
token_splitter = TokenTextSplitter(
chunk_size=500,
chunk_overlap=50,
encoding_name="cl100k_base"  # GPT-4的tokenizer
)
# 使用
chunks = recursive_splitter.split_documents(docs)
print(f"分割后得到 {len(chunks)} 个文档块")

3. 嵌入模型

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
# OpenAI嵌入
openai_embeddings = OpenAIEmbeddings(
model="text-embedding-3-small",
openai_api_key="your_api_key"
)
# 开源嵌入（本地）
huggingface_embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
model_kwargs={'device': 'cuda'}  # 使用GPU
)
# 使用
text = "这是一个测试句子"
embedding_vector = openai_embeddings.embed_query(text)
print(f"嵌入向量维度: {len(embedding_vector)}")  # 1536 for text-embedding-3-small

4. 向量数据库

from langchain_community.vectorstores import Chroma, FAISS, Pinecone
import os
# Chroma（本地，推荐）
chroma_db = Chroma.from_documents(
documents=chunks,
embedding=openai_embeddings,
persist_directory="./chroma_db"
)
# FAISS（快速，本地）
faiss_db = FAISS.from_documents(
documents=chunks,
embedding=openai_embeddings
)
# Pinecone（云服务，推荐生产环境）
pinecone_db = Pinecone.from_documents(
documents=chunks,
embedding=openai_embeddings,
index_name="my-index",
namespace="test-namespace"
)
# 检索
query = "Python的特点"
results = chroma_db.similarity_search(query, k=3)
for i, doc in enumerate(results, 1):
print(f"结果 {i}: {doc.page_content[:100]}...")

5. 检索策略

# 相似度搜索
results = vectorstore.similarity_search(query, k=3)
# 带分数的相似度搜索
results_with_scores = vectorstore.similarity_search_with_score(query, k=3)
for doc, score in results_with_scores:
print(f"分数: {score:.4f} | 内容: {doc.page_content[:50]}...")
# 最大边界相关性搜索（MMR）
from langchain_community.vectorstores.utils import DistanceStrategy
mmr_results = vectorstore.max_marginal_relevance_search(
query,
k=3,
fetch_k=10  # 先获取10个候选，再选3个
)

RAG架构类型

1. Naive RAG（基础RAG）

最简单的架构，直接检索后生成。

2. Advanced RAG（高级RAG）

包含预处理、检索优化、后处理。

3. Modular RAG（模块化RAG）

灵活组合不同模块。

实践案例

案例1: 企业知识库问答

class EnterpriseKnowledgeBase:
"""企业知识库RAG系统"""
def __init__(self, docs_path: str):
self.docs_path = docs_path
self.embeddings = OpenAIEmbeddings()
self.vectorstore = None
self.llm = ChatOpenAI(model="gpt-4o")
def build_index(self):
"""构建索引"""
# 加载文档
loader = DirectoryLoader(self.docs_path, glob="**/*.md")
docs = loader.load()
# 分割
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = splitter.split_documents(docs)
# 创建向量数据库
self.vectorstore = Chroma.from_documents(
documents=chunks,
embedding=self.embeddings,
persist_directory="./chroma_kb"
)
print(f"✅ 索引构建完成，共 {len(chunks)} 个文档块")
def query(self, question: str) -> str:
"""查询"""
if not self.vectorstore:
return "请先构建索引"
# 检索
retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
docs = retriever.invoke(question)
# 构建Prompt
context = "\n\n".join(doc.page_content for doc in docs)
prompt = f"""
你是一个企业知识库助手。基于以下文档回答用户问题。
参考文档：
{context}
用户问题：{question}
请提供准确、有用的回答。如果文档中没有相关信息，请明确说明。
"""
# 生成
response = self.llm.invoke(prompt)
return response.content
# 使用
kb = EnterpriseKnowledgeBase("./company_docs/")
kb.build_index()
answer = kb.query("公司的请假政策是什么？")
print(answer)

案例2: 代码文档查询

class CodeDocSearcher:
"""代码文档查询系统"""
def __init__(self, repo_path: str):
self.repo_path = repo_path
self.embeddings = OpenAIEmbeddings()
self.vectorstore = None
def load_code(self):
"""加载代码文件"""
from langchain_community.document_loaders import TextLoader
code_files = []
for root, _, files in os.walk(self.repo_path):
for file in files:
if file.endswith(('.py', '.js', '.ts')):
filepath = os.path.join(root, file)
loader = TextLoader(filepath)
docs = loader.load()
code_files.extend(docs)
return code_files
def build_index(self):
"""构建代码索引"""
docs = self.load_code()
# 代码特定的分割
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\ndef ", "\nclass ", "\n# ", "\n"]
)
chunks = splitter.split_documents(docs)
self.vectorstore = Chroma.from_documents(
documents=chunks,
embedding=self.embeddings
)
print(f"✅ 代码索引构建完成，共 {len(chunks)} 个代码块")
def search_code(self, query: str):
"""搜索代码"""
if not self.vectorstore:
return []
results = self.vectorstore.similarity_search(query, k=5)
return results
# 使用
code_searcher = CodeDocSearcher("./my_project/")
code_searcher.build_index()
results = code_searcher.search_code("如何连接数据库")
for doc in results:
print(f"文件: {doc.metadata.get('source', 'unknown')}")
print(f"代码:\n{doc.page_content}\n")

学习要点

✅ RAG通过检索外部知识增强LLM ✅ 核心组件：文档加载、分割、嵌入、向量数据库、LLM ✅ 三种架构：Naive RAG、Advanced RAG、Modular RAG ✅ 可以用于企业知识库、代码查询等多种场景 ✅ 相比微调，RAG更灵活、成本更低

下一步: 学习 构建完整RAG系统 🏗️