RAG 架构原理
理解 RAG 的架构是构建高效系统的第一步。本章将深入分析 RAG 的工作流程、设计模式和核心概念，帮助你建立系统性的认知。
RAG 的完整工作流程
RAG 系统包含两个主要阶段：索引阶段（离线）和查询阶段（在线）。
索引阶段
索引阶段将原始文档转换为可检索的形式，通常在后台定期执行。
文档加载：从不同数据源获取文档。
from langchain_community.document_loaders import (
PyPDFLoader, # PDF 文件
WebBaseLoader, # 网页内容
TextLoader, # 纯文本
DirectoryLoader, # 目录批量加载
)
# 加载 PDF
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()
# 加载网页
web_loader = WebBaseLoader("https://example.com/article")
web_docs = web_loader.load()
文档解析：提取文本内容，保留结构信息（标题、段落、表格等）。
文档分块：将长文档切分为适当大小的片段，这是影响检索质量的关键步骤。
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每块最大字符数
chunk_overlap=200, # 块之间的重叠字符数
length_function=len,
separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
)
chunks = splitter.split_documents(docs)
向量化：使用嵌入模型将文本转换为向量。
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 单个文本
vector = embeddings.embed_query("这是一段文本")
# 批量文本
vectors = embeddings.embed_documents([chunk.page_content for chunk in chunks])
存储索引：将向量和元数据存入向量数据库。
from langchain_community.vectorstores import Chroma
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
查询阶段
查询阶段在用户提问时实时执行，需要尽可能快速。
问题向量化：将用户问题转换为向量。
question = "如何申请年假？"
question_vector = embeddings.embed_query(question)
向量检索：在向量数据库中查找最相似的文档。
# 基础检索
results = vectorstore.similarity_search(question, k=5)
# 带分数的检索
results_with_scores = vectorstore.similarity_search_with_score(question, k=5)
上下文组装：将检索结果与问题组合成提示词。
from langchain_core.prompts import ChatPromptTemplate
prompt_template = ChatPromptTemplate.from_messages([
("system", """你是一个智能助手。请根据以下参考文档回答问题。
如果参考文档中没有相关信息，请明确告知用户。
参考文档：
{context}"""),
("human", "{question}")
])
context = "\n\n".join([doc.page_content for doc in results])
prompt = prompt_template.invoke({
"context": context,
"question": question
})
LLM 生成：将组装好的提示词发送给大模型。
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke(prompt)
print(response.content)
核心概念详解
向量与嵌入
向量是文本在高维空间中的数值表示。语义相似的文本，其向量在高维空间中的距离也较近。
文本 向量表示
─────────────────────────────────────────────
"猫是一种宠物" → [0.2, -0.5, 0.8, ...]
"狗是常见的宠物" → [0.3, -0.4, 0.7, ...] (相似)
"汽车有四个轮子" → [-0.6, 0.9, -0.2, ...] (不同)
嵌入模型负责将文本转换为向量。好的嵌入模型能准确捕捉语义信息：
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
sentences = [
"猫是一种宠物",
"狗是常见的宠物",
"汽车有四个轮子"
]
embeddings = model.encode(sentences)
# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity
similarities = cosine_similarity(embeddings)
# 结果示例：
# sentences[0] 与 sentences[1] 相似度: 0.85 (高)
# sentences[0] 与 sentences[2] 相似度: 0.12 (低)
向量检索原理
向量检索的核心是计算查询向量与文档向量之间的相似度。最常用的是余弦相似度：
import numpy as np
def cosine_similarity(a, b):
"""计算两个向量的余弦相似度"""
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
# 相似度范围 [-1, 1]
# 1: 完全相同方向
# 0: 正交（无关）
# -1: 完全相反方向
当文档数量很大时，逐一计算相似度效率太低。向量数据库使用**近似最近邻搜索（ANN）**算法加速检索：
| 算法 | 原理 | 特点 |
|---|---|---|
| HNSW | 分层可导航小世界图 | 查询快、构建慢、内存占用高 |
| IVF | 倒排文件索引 | 构建快、查询中等、内存占用低 |
| PQ | 乘积量化 | 内存占用极低、精度略有损失 |
语义检索 vs 关键词检索
语义检索：基于向量相似度，能理解同义词、近义词。
查询: "怎么办理离职"
匹配: "员工辞职流程" ✓ (语义相似，无共同词)
关键词检索：基于词汇匹配，精确但缺乏语义理解。
查询: "怎么办理离职"
匹配: "离职办理窗口" ✓ (有共同词)
不匹配: "员工辞职流程" ✗ (无共同词)
混合检索：结合两种方法的优势，是生产环境的推荐方案。
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
# 向量检索器
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 关键词检索器
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5
# 混合检索
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.4, 0.6] # 关键词 40%，语义 60%
)
results = ensemble_retriever.invoke("怎么办理离职")
RAG 架构模式
简单 RAG
最基础的架构，适合快速原型和简单场景。
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
# 初始化组件
llm = ChatOpenAI(model="gpt-4o-mini")
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
# 创建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 4})
)
# 查询
answer = qa_chain.invoke({"query": "公司的报销流程是什么？"})
优点：实现简单、响应快速 缺点：上下文长度受限、无法处理复杂问题
对话式 RAG
支持多轮对话，需要维护对话历史。
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
# 问题重写提示词（基于历史生成独立问题）
contextualize_q_system_prompt = """给定聊天历史和用户最新问题，
将该问题重写为一个独立的问题，不需要聊天历史即可理解。"""
# 创建历史感知检索器
history_aware_retriever = create_history_aware_retriever(
llm, vectorstore.as_retriever(),
ChatPromptTemplate.from_messages([
("system", contextualize_q_system_prompt),
MessagesPlaceholder("chat_history"),
("human", "{input}")
])
)
# 创建 RAG 链
qa_chain = create_retrieval_chain(
history_aware_retriever,
create_stuff_documents_chain(llm, qa_prompt)
)
# 查询（带历史）
response = qa_chain.invoke({
"input": "那如果超时了呢？", # 指向上一轮对话
"chat_history": chat_history
})
多文档 RAG
处理大量文档或需要跨文档推理的场景。
from langchain.retrievers import MultiQueryRetriever
# 多查询扩展：从不同角度重写问题
retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(),
llm=llm
)
# 自动生成多个查询变体
# 原问题: "如何提高团队效率？"
# 变体1: "团队效率提升的方法有哪些？"
# 变体2: "如何改善团队协作？"
# 变体3: "提高团队生产力的策略"
results = retriever.invoke("如何提高团队效率？")
路由 RAG
根据问题类型路由到不同的知识库或处理流程。
from langchain.utils.math import cosine_similarity
from langchain_core.output_parsers import StrOutputParser
# 定义路由
route_templates = {
"技术文档": "关于代码、API、技术架构的问题",
"人力资源": "关于招聘、培训、员工福利的问题",
"财务": "关于报销、预算、财务流程的问题"
}
# 路由嵌入
route_embeddings = {
route: embeddings.embed_query(template)
for route, template in route_templates.items()
}
def route_query(question):
question_embedding = embeddings.embed_query(question)
scores = {
route: cosine_similarity([question_embedding], [emb])[0][0]
for route, emb in route_embeddings.items()
}
return max(scores, key=scores.get)
# 使用路由
route = route_query("Python API 怎么调用？")
# route = "技术文档"
自查询 RAG
自动从问题中提取过滤条件，结合元数据过滤。
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
# 定义元数据字段
metadata_field_info = [
AttributeInfo(
name="department",
description="文档所属部门",
type="string"
),
AttributeInfo(
name="year",
description="文档年份",
type="integer"
),
]
# 创建自查询检索器
retriever = SelfQueryRetriever.from_llm(
llm=llm,
vectorstore=vectorstore,
document_contents="公司内部文档",
metadata_field_info=metadata_field_info
)
# 自动提取过滤条件
# 问题: "2024年人力资源部发布的考勤制度"
# 自动生成: filter={"department": "人力资源", "year": 2024}
results = retriever.invoke("2024年人力资源部发布的考勤制度")
RAG 系统的关键指标
评估 RAG 系统的质量，需要关注两类指标：
检索指标
衡量检索结果的相关性和覆盖率。
| 指标 | 公式 | 含义 |
|---|---|---|
| Hit Rate | 命中数 / 总查询数 | 检索结果中包含正确答案的比例 |
| MRR | 正确答案的平均排名倒数 | |
| NDCG | 归一化折损累积增益 | 考虑排名位置的检索质量 |
| Recall@K | TopK 结果的召回率 |
生成指标
衡量生成回答的质量。
| 指标 | 说明 |
|---|---|
| 忠实度 | 回答是否基于检索文档，无幻觉 |
| 答案相关性 | 回答是否直接回答问题 |
| 上下文精确度 | 检索文档中相关内容的比例 |
| 上下文召回率 | 检索到的文档是否覆盖所有需要信息 |
# 使用 RAGAS 评估
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 忠实度
answer_relevancy, # 答案相关性
context_precision, # 上下文精确度
context_recall # 上下文召回率
)
results = evaluate(
dataset,
metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)
常见问题与解决方案
问题 1：检索不到相关文档
原因：分块过大或过小、嵌入模型不合适、查询与文档表述差异大
解决方案：
- 调整分块大小（通常 500-1000 字符）
- 尝试不同的嵌入模型（如中文场景使用 BGE）
- 使用查询扩展或重写
问题 2：检索结果太多噪音
原因：相似度阈值过低、缺少过滤条件
解决方案：
- 设置相似度阈值（如 0.7 以上）
- 添加元数据过滤
- 使用重排序模型
问题 3：回答不连贯或断章取义
原因：检索片段缺乏上下文、分块破坏语义完整性
解决方案：
- 增加分块重叠
- 检索时返回相邻片段
- 使用"父文档检索"策略
问题 4：响应太慢
原因：检索效率低、LLM 调用慢
解决方案：
- 使用高效的向量索引（HNSW）
- 减少检索数量（k 值）
- 使用更快的 LLM 或流式输出
小结
RAG 架构的核心是将检索与生成解耦，让大模型能够访问动态更新的知识。理解以下几点至关重要：
- 索引阶段决定知识的覆盖范围和检索质量
- 查询阶段决定用户体验和响应速度
- 向量和相似度是检索的数学基础
- 混合检索通常优于单一检索方法
- 评估指标帮助量化系统性能