RAG从入门到高阶(一):构建你的第一个Naive RAG
引言
嗨,朋友们!
RAG 技术想必大家都不陌生了,但RAG的世界远比你想象的要精彩!在这个系列中,我将带你一步步深入探索各种复杂和高阶的 RAG 技术,让你的智能问答系统变得更强大。
准备好了吗?咱们这就开始 RAG 的进阶之旅吧!
第一篇我们先介绍最基础的RAG:Naive RAG
1. RAG 是什么?为什么我们需要它?
想象一下,你有一个非常聪明的朋友,他能回答各种各样的问题。这就是我们的人工智能大语言模型(LLM),比如 ChatGPT。但是,这些模型在训练的时候,只是学习了海量的通用知识,对于一些非常专业、最新的信息,或者你私有的文档内容,它们就无能为力了。
检索增强生成 (Retrieval Augmented Generation,简称 RAG) 技术应运而生,它就像给你的聪明朋友配了一个超级图书馆和一位快速检索员。当你有问题时,它不再只依靠自己的“记忆”,而是会:
-
检索 (Retrieval):快速在图书馆里找到与你的问题最相关的几本书(文档片段)。
-
增强 (Augmented):把这些找到的书籍内容提供给你的聪明朋友。
-
生成 (Generation):你的朋友结合这些“新知识”和自己的通用知识,给你一个更准确、更全面的答案。
Naive RAG 就是最基础、最直观的 RAG 实现,它能帮助我们快速搭建一个具备知识检索能力的问答系统。
2. Naive RAG 的工作流程:三步走战略
Naive RAG 就像一条高效的流水线,分为三个主要阶段:
第一阶段:知识库准备(索引构建)
在回答问题之前,我们得先把“图书馆”准备好。这个阶段就是把你的原始文档(比如产品手册、公司 FAQ 等)整理成 RAG 系统能理解和检索的格式。
-
文档分块:想象一下,一本书太厚了,不好一次性看完。所以我们把长文档切分成一个个小片段,就像把书拆成一页页的纸。这样做的好处是,每次检索时,我们只需要找到相关的几页,而不是整本书,这大大提高了检索效率和准确性。
-
向量化:机器不懂文字,但它懂数字。所以,我们会把这些文档片段和我们的问题都转换成一串串数字,这些数字被称为“向量”。你可以把向量想象成是文字的“指纹”,相似的文字会拥有相似的指纹(向量)。
-
向量库存储:把这些“数字指纹”和对应的文档片段一起存到一个特殊的数据库里,这个数据库就叫做“向量库”。当我们需要检索时,可以直接在这个库里快速查找。
第二阶段:问题来了!快速检索(检索召回)
当用户提出问题时,系统会迅速行动起来:
-
问题向量化:同样,把用户的问题也转换成“数字指纹”(向量)。
-
相似度检索:拿着问题的“指纹”,去向量库里和所有文档片段的“指纹”进行比较,找出最相似的几个文档片段。这就像你拿着一个指纹去指纹库里查找匹配的指纹一样。
第三阶段:生成答案(生成输出)
现在,我们有了用户的问题,也找到了最相关的文档片段,是时候让大语言模型发挥作用了:
-
上下文拼接:把找到的相关文档片段和用户的问题拼接起来,形成一个完整的“提示语”(Prompt)。这个提示语会告诉大语言模型:“这是你的参考资料,这是用户的问题,请基于参考资料回答问题。”
-
LLM 生成:大语言模型接收到这个提示语后,会阅读参考资料,并生成一个条理清晰、内容准确的答案。
3. 动手实践:用 Python 实现 Naive RAG
现在,让我们通过代码来一步步实现这个 Naive RAG 系统。
3.1 核心组件概览
我们的 Naive RAG 系统主要由以下几个核心组件构成:
-
DocumentProcessor(文档处理器):负责读取各种格式的文档(如 PDF, TXT),并将其切分成适合处理的小块。 -
VectorStore(向量存储):负责存储文档块的向量,并提供高效的相似度检索功能。这里我们使用Faiss,它是一个非常流行的向量相似度搜索库。 -
NaiveRAG(主控制器):整合前两个组件,负责整个 RAG 流程的调度,包括构建索引和回答用户查询。
3.2 文档处理器 (DocumentProcessor)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader, PDFMinerLoader # 使用PDFMinerLoader,因为它通常更稳定
from typing import List
from langchain_core.documents import Document
class DocumentProcessor:
def __init__(self, chunk_size: int = 512, chunk_overlap: int = 50):
# 初始化文本分割器,chunk_size 是每块的大小,chunk_overlap 是块之间的重叠部分
# 重叠部分有助于保持上下文的连贯性
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
def process(self, file_paths: List[str]) -> List[Document]:
documents = []
for path in file_paths:
# 根据文件扩展名选择合适的加载器
if path.endswith('.pdf'):
loader = PDFMinerLoader(path) # 用于加载PDF文件
elif path.endswith('.txt'):
loader = TextLoader(path) # 用于加载TXT文件
else:
print(f"Warning: Unsupported file type for {path}. Skipping.")
continue
try:
documents.extend(loader.load()) # 加载文档内容
except Exception as e:
print(f"Error loading {path}: {e}")
# 将加载的文档切分成小块
return self.text_splitter.split_documents(documents)
代码解析:
-
RecursiveCharacterTextSplitter: 这是一个智能的文本分割器,它会尝试以多种字符(如换行符、空格等)进行分割,以确保文档块的完整性。 -
chunk_size: 每个文档块包含的字符数。 -
chunk_overlap: 相邻文档块之间重叠的字符数。适当的重叠可以帮助模型更好地理解跨越两个文档块的上下文。 -
TextLoader,PDFMinerLoader:LangChain提供的文档加载器,可以方便地读取不同格式的文件。
3.3 向量存储 (VectorStore)
import faiss
import numpy as np
from typing import List, Tuple
class VectorStore:
def __init__(self, embedding_dim: int = 1536):
# 初始化 Faiss 索引。Faiss 是一个高效的相似度搜索库。
# IndexFlatIP 表示使用内积(Inner Product)作为相似度度量。
# embedding_dim 是向量的维度,例如 OpenAI 的 text-embedding-ada-002 模型是 1536 维。
self.index = faiss.IndexFlatIP(embedding_dim)
self.documents = [] # 存储原始的文档文本
self.embedder = None # 嵌入器,用于将文本转换为向量,稍后会注入
def add_documents(self, documents: List[str], embeddings: List[List[float]]):
# 将 embeddings 列表转换为 NumPy 数组,并确保数据类型是 float32,Faiss 要求。
embeddings_np = np.array(embeddings).astype('float32')
self.index.add(embeddings_np) # 将向量添加到 Faiss 索引中
self.documents.extend(documents) # 存储对应的原始文档文本
def search(self, query_embedding: List[float], k: int = 5) -> List[Tuple[str, float]]:
# 将查询向量转换为 NumPy 数组
query_np = np.array([query_embedding]).astype('float32')
# 在 Faiss 索引中搜索最相似的 k 个向量
scores, indices = self.index.search(query_np, k)
results = []
for score, idx in zip(scores[0], indices[0]):
if idx != -1: # 确保索引有效
results.append((self.documents[idx], float(score))) # 返回文档文本和相似度分数
return results
代码解析:
-
faiss.IndexFlatIP: Faiss 中的一种索引类型,适用于内积相似度计算。内积越大,表示两个向量越相似。 -
embedding_dim: 向量的维度。不同的嵌入模型会生成不同维度的向量。在实际使用中,你需要根据你选择的嵌入模型来设置这个值。 -
add_documents: 将文档文本和它们对应的向量添加到向量库中。 -
search: 接收一个查询向量,返回最相似的k个文档片段及其相似度分数。
3.4 主控制器 (NaiveRAG)
from openai import OpenAI
from sentence_transformers import SentenceTransformer
from typing import List, Dict
# 为了兼容 LangChain Document,我们需要导入它
from langchain_core.documents import Document
class NaiveRAG:
def __init__(self, embedder, vector_store, llm_model: str = "gpt-3.5-turbo"):
self.embedder = embedder # 嵌入器,用于文本到向量的转换
self.vector_store = vector_store # 向量存储
self.client = OpenAI() # 初始化 OpenAI 客户端,用于调用大语言模型
self.llm_model = llm_model # 使用的大语言模型名称
def build_index(self, file_paths: List[str]):
print("开始构建索引...")
# 1. 处理文档:加载并切分文档
processor = DocumentProcessor()
documents = processor.process(file_paths)
print(f"文档切分完成,共 {len(documents)} 个文档块。")
# 2. 向量化:将文档块转换为向量
# 注意:sentence-transformers 的 embed_documents 方法可能不存在,通常是直接调用 __call__ 或 encode
# 假设 self.embedder 是 SentenceTransformer 模型实例,可以直接调用 encode 方法
texts = [doc.page_content for doc in documents]
print("开始生成文档块嵌入向量...")
embeddings = self.embedder.encode(texts).tolist() # 将 numpy 数组转换为列表
print("文档块嵌入向量生成完成。")
# 3. 存储到向量库
self.vector_store.add_documents(texts, embeddings)
print("索引构建完成!")
def query(self, question: str) -> Dict:
print(f"接收到查询: {question}")
# 1. 查询向量化:将用户问题转换为向量
# 同样,对于 SentenceTransformer,使用 encode 方法
query_embedding = self.embedder.encode(question).tolist() # 转换为列表
print("查询向量化完成。")
# 2. 检索相关文档
# k=5 是从配置中获取的 top_k,这里先硬编码
results = self.vector_store.search(query_embedding, k=5)
if not results:
print("未找到相关信息。")
return {"answer": "未找到相关信息", "confidence": 0.0, "sources": 0}
print(f"检索到 {len(results)} 条相关文档。")
# 3. 构建上下文
# 这里的 score 是相似度分数,可以用来评估“置信度”
contexts = [doc for doc, _ in results]
context_text = "\n\n".join(contexts)
# 4. 生成答案
prompt = f"""基于以下文档回答问题:
{context_text}
问题:{question}
答案:"""
print("正在调用大语言模型生成答案...")
try:
response = self.client.chat.completions.create(
model=self.llm_model,
messages=[{"role": "user", "content": prompt}],
temperature=0.1 # 较低的温度使模型回答更确定
)
answer = response.choices[0].message.content
# 置信度可以使用检索到的最高相似度分数来近似表示
confidence = max(score for _, score in results) if results else 0.0
print("答案生成完成。")
return {
"answer": answer,
"confidence": confidence,
"sources": len(results)
}
except Exception as e:
print(f"调用大语言模型出错: {e}")
return {"answer": "生成答案失败。", "confidence": 0.0, "sources": 0}
代码解析:
-
__init__: 初始化 NaiveRAG 类,需要传入一个嵌入器 (embedder) 和一个向量存储 (vector_store) 实例,以及你希望使用的大语言模型名称。 -
build_index: 这是索引构建阶段的核心方法。它调用DocumentProcessor来处理文档,然后使用embedder将文档文本转换为向量,最后将文本和向量添加到VectorStore中。 -
query: 这是检索和生成阶段的核心方法。它首先将用户问题向量化,然后从VectorStore中检索最相关的文档片段。接着,它将这些文档片段和原始问题组合成一个提示语,发送给大语言模型,并返回生成的答案。 -
OpenAI API 调用: 我们使用
openai库与 OpenAI 的大语言模型进行交互。你需要确保已经配置了 OpenAI API 密钥。
3.5 使用示例
from sentence_transformers import SentenceTransformer
import os
# 确保你有存放文档的目录,例如 'docs'
# 如果没有,可以手动创建或者通过代码创建
os.makedirs("docs", exist_ok=True)
# 创建一些示例文档
# doc/manual.pdf (需要自行创建PDF文件,或者用txt代替)
with open("docs/faq.txt", "w", encoding="utf-8") as f:
f.write("Q: 我们的产品如何安装?\nA: 产品安装请参考用户手册第10页。首先,请确保您的设备已连接电源。然后,将安装光盘插入光驱。\n\nQ: 产品的保修期是多久?\nA: 产品提供一年免费保修服务,购买日起生效。\n\nQ: 如何联系客服?\nA: 您可以通过拨打客服热线400-123-4567或发送邮件至support@example.com联系我们。")
# 假设 embedder 的维度是 384 (all-MiniLM-L6-v2 的维度)
# 初始化组件
embedder = SentenceTransformer('all-MiniLM-L6-v2')
# 确保 VectorStore 的 embedding_dim 与 embedder 的输出维度匹配
vector_store = VectorStore(embedding_dim=embedder.get_sentence_embedding_dimension())
# 创建RAG系统
# 注意:使用 OpenAI API 需要配置 OPENAI_API_KEY 环境变量
# os.environ["OPENAI_API_KEY"] = "你的OpenAI API Key"
rag = NaiveRAG(embedder, vector_store, llm_model="gpt-3.5-turbo")
# 构建索引
# 假设 docs 文件夹下有 manual.pdf 和 faq.txt 文件
rag.build_index(["docs/faq.txt"]) # 示例中只用 faq.txt,如果你有PDF文件,请确保PDFMinerLoader已正确安装
# 查询
print("\n--- 进行查询 ---")
result = rag.query("如何安装产品?")
print(f"答案: {result['answer']}")
print(f"置信度: {result['confidence']:.2f}")
print(f"来源文档数量: {result['sources']}")
print("\n--- 进行另一个查询 ---")
result2 = rag.query("保修期是多久?")
print(f"答案: {result2['answer']}")
print(f"置信度: {result2['confidence']:.2f}")
print(f"来源文档数量: {result2['sources']}")
运行前准备:
-
安装必要的库:
pip install langchain-community langchain-core faiss-cpu numpy openai sentence-transformers pypdf-
langchain-community: 包含各种加载器和分割器 -
langchain-core: LangChain 的核心库 -
faiss-cpu: Faiss 库的 CPU 版本 -
numpy: 数值计算库 -
openai: OpenAI API 的 Python 客户端 -
sentence-transformers: 用于生成文本嵌入向量 -
pypdf: 如果处理 PDF 文件,PDFMinerLoader依赖它
-
-
设置 OpenAI API Key: 你需要从 OpenAI 官网获取 API Key,并将其设置为环境变量
OPENAI_API_KEY。 例如:在 Linux/macOS 下:
export OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxx"
在 Windows 下:
set OPENAI_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxx"
创建文档文件: 在项目根目录下创建 docs 文件夹,并在其中放入 faq.txt 和 manual.pdf (如果需要)。
4. Naive RAG 的优缺点
4.1 优点
-
实现简单,部署快速: 核心逻辑直观,代码量较少,非常适合入门和快速原型验证。
-
固定线性处理流程: 流程清晰,容易理解和调试。
-
成本低廉: 相对于复杂的 RAG 架构,Naive RAG 在计算资源和开发成本上都更低。
-
适合简单问答场景: 对于只需要从现有文档中提取信息并直接回答的场景,表现良好。
4.2 局限性
就像任何基础模型一样,Naive RAG 也有其不足之处,这正是我们未来优化和学习更复杂 RAG 架构的原因:
-
检索质量依赖简单向量相似度:
-
它只看向量的相似度,不理解深层语义,可能会召回一些表面相似但内容不相关的文档。
-
如果问题需要多步推理或结合多个独立信息点才能回答,Naive RAG 很难做到。例如,“公司去年最畅销的产品是什么?今年是否有新款上市?”这需要检索两个不同时间段的信息。
-
-
无法处理复杂多跳推理: 对于需要从多个文档片段中提取信息并进行整合、推理才能得到答案的问题,Naive RAG 显得力不从心。它只是简单地把检索到的内容一股脑地给 LLM。
-
缺乏对检索结果的质量控制: 它不会判断检索到的文档片段是否真的高质量或与问题强相关,可能把低质量或不准确的信息也提供给 LLM。
-
生成答案可解释性差: LLM 生成的答案,我们很难知道它具体是依据了哪个文档片段,或者哪些信息是它“脑补”出来的。这对于一些需要高可信度答案的场景是不足的。
-
上下文长度限制: 大语言模型接受的上下文长度是有限的。如果检索到的文档片段过多,可能会超出模型的输入限制,导致部分信息被截断。
5. 配置文件和部署:让系统更灵活
为了让我们的 Naive RAG 系统更易于配置和部署,我们可以引入配置文件和简单的 Web 服务。
5.1 配置文件 (config.yaml)
# config.yaml
naive_rag:
document:
chunk_size: 512
chunk_overlap: 50
embedding:
model: "all-MiniLM-L6-v2"
provider: "huggingface" # 可以扩展为 openai, google 等
retrieval:
top_k: 5
similarity_threshold: 0.7 # 可以用来过滤低相关性结果
generation:
model: "gpt-3.5-turbo"
temperature: 0.1
max_tokens: 1000
5.2 部署脚本 (deploy.py)
# deploy.py
import yaml
from flask import Flask, request, jsonify
import os
# 从我们的 NaiveRAG.py 中导入相关类
from naive_rag_tutorial import NaiveRAG, DocumentProcessor, VectorStore # 假设上面代码保存为 naive_rag_tutorial.py
from sentence_transformers import SentenceTransformer
def load_config(config_path: str) -> dict:
"""加载 YAML 配置文件。"""
with open(config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
def create_embedder(embedding_config: dict):
"""根据配置创建嵌入器。"""
model_name = embedding_config.get("model", "all-MiniLM-L6-v2")
provider = embedding_config.get("provider", "huggingface")
if provider == "huggingface":
return SentenceTransformer(model_name)
# 未来可以扩展支持 OpenAI, Google 等
# elif provider == "openai":
# from langchain_openai import OpenAIEmbeddings
# return OpenAIEmbeddings(model=model_name)
else:
raise ValueError(f"Unsupported embedding provider: {provider}")
if __name__ == "__main__":
config = load_config("config.yaml")
rag_config = config['naive_rag']
# 创建嵌入器和向量存储
embedder = create_embedder(rag_config['embedding'])
# 确保向量存储的维度与嵌入器输出维度匹配
vector_store = VectorStore(embedding_dim=embedder.get_sentence_embedding_dimension())
# 创建 NaiveRAG 系统
rag = NaiveRAG(
embedder,
vector_store,
llm_model=rag_config['generation']['model']
)
# 假设文档位于 'docs' 目录
document_paths = [os.path.join("docs", f) for f in os.listdir("docs") if f.endswith(('.txt', '.pdf'))]
if document_paths:
rag.build_index(document_paths)
else:
print("警告: 'docs' 目录下没有找到支持的文档 (.txt, .pdf)。请确保有文档以构建索引。")
# 启动 Flask 服务
app = Flask(__name__)
@app.route('/query', methods=['POST'])
def handle_query():
data = request.json
if not data or 'question' not in data:
return jsonify({"error": "请求体中必须包含 'question' 字段。"}), 400
question = data['question']
result = rag.query(question)
return jsonify(result)
# 可以在这里添加一个索引构建的接口,或者在启动时自动构建
@app.route('/build_index', methods=['POST'])
def re_build_index():
data = request.json
file_paths = data.get('file_paths', [])
if not file_paths:
return jsonify({"error": "请提供需要构建索引的文件路径列表。"}), 400
try:
rag.build_index(file_paths)
return jsonify({"message": "索引构建成功!"}), 200
except Exception as e:
return jsonify({"error": f"索引构建失败: {e}"}), 500
print(f"RAG 服务已启动,监听在 http://0.0.0.0:8000")
app.run(host='0.0.0.0', port=8000)
运行前准备:
-
安装 Flask 和 PyYAML:
pip install flask pyyaml -
将上面的 NaiveRAG 代码保存为
naive_rag_tutorial.py文件(或者你喜欢的文件名),并确保deploy.py在同一目录下可以正确导入。 -
确保
config.yaml和docs目录存在。
如何运行和测试:
-
运行
python deploy.py -
使用
curl或 Postman 等工具发送 POST 请求到http://localhost:8000/query:curl -X POST -H "Content-Type: application/json" -d '{"question": "如何安装产品?"}' http://localhost:8000/query你将收到类似这样的 JSON 响应:
{ "answer": "根据用户手册第10页的指示,产品安装的第一步是确保设备已连接电源。然后,将安装光盘插入光驱。", "confidence": 0.85, "sources": 1 }
6. 性能指标
对于 Naive RAG,我们可以关注以下几个性能指标:
-
检索精度: 65-75% - 表示检索到的文档片段与问题相关的比例。这是最关键的指标之一。
-
响应延迟: 100-300ms - 从接收问题到返回答案所需的时间。
-
内存占用: 100MB-1GB - 向量库和模型加载所需的内存。
-
适用文档量: <10K篇 - Naive RAG 对于小型到中等规模的知识库表现良好,但文档量过大时,检索效率和精度可能会下降。
这些指标会根据你的数据、硬件和选择的模型有所不同。
7. 优化建议:让 Naive RAG 更好用
尽管 Naive RAG 有局限性,但我们可以通过一些简单的优化来提升它的表现:
-
文档分块优化 (Chunking Strategy):
-
根据文档类型调整
chunk_size: 对于代码、表格、结构化文本等不同类型的文档,可能需要不同的分块策略和大小。 -
语义分块: 尝试根据语义而非固定长度来分块,例如将一整个段落作为一个块。
-
-
相似度阈值 (Similarity Threshold):
-
在
VectorStore的search方法中,可以设置一个similarity_threshold。只返回相似度超过这个阈值的文档片段,过滤掉低质量或不相关的结果,减少 LLM 的“噪音”输入。
-
-
上下文长度管理 (Context Length Management):
-
在将检索到的文档片段拼接成上下文时,需要考虑 LLM 的最大输入长度。如果检索到的内容太多,可以只选择最相关的几个片段,或者对内容进行摘要。
-
-
缓存机制 (Caching):
-
对于常见的或重复的查询,可以缓存其结果。当下次再次收到相同的查询时,直接返回缓存的结果,避免重复计算,提高响应速度。
-
-
选择更好的嵌入模型 (Better Embedding Models):
-
all-MiniLM-L6-v2是一个轻量级且高效的模型,但对于某些特定领域的知识,可能有更专业的嵌入模型。 -
例如,OpenAI 的
text-embedding-ada-002通常能提供更高质量的嵌入向量,但成本也更高。
-
预告:RAG 进阶之旅,未完待续...
我们已经学习了基础的 Naive RAG 架构,它帮助我们理解了 RAG 的核心机制。然而,这种基础的 RAG 在面对复杂或多样化的实际应用场景时,存在一些局限性,例如可能无法捕获所有相关信息,或难以处理多源异构数据。
在接下来的教程中,我们将继续深入探索更高级、更智能的 RAG 架构,解锁其在复杂应用中的巨大潜力:
-
Retrieve-and-Rerank RAG(检索重排 RAG):在初步检索后,我们将引入一个更精细的模型对检索结果进行二次排序,确保选出与用户查询最相关的文档。
-
Multimodal RAG (多模态 RAG):探索如何让 RAG 不仅能处理文本信息,还能理解图片、音频、视频等多种模态的数据,实现更全面的信息检索。
-
Graph RAG (图 RAG):学习如何利用知识图谱强大的推理能力,让 RAG 能够进行复杂的多跳推理,回答更深层次、更具关联性的问题。
-
Hybrid RAG (混合 RAG):探讨如何更智能地融合多种检索策略,并有效管理来自不同数据源的信息,以提升检索的召回率和准确性。
-
Agentic RAG Router (智能体路由):引入由 LLM 驱动的智能代理,让 RAG 系统能根据用户意图,动态选择最佳的工具、数据源或处理流程。
-
Agentic RAG Multi-Agent (多智能体):我们将进一步探索如何让多个专业的智能体进行协作,共同解决极其复杂的问题,并通过“辩论”等机制达成共识,提供更全面、更可靠的答案。
这些进阶架构将帮助我们构建出更智能、更鲁棒的问答系统,以应对未来更加复杂的挑战。敬请期待!
持续学习,才能玩转AI!
这篇RAG入门教程助你启程。想获取:
-
最新AI架构趋势深度解读
-
RAG、MCP及LLM应用实战教程与代码
-
精选学习资源与高效工具包
-
技术答疑与同行交流
👉 欢迎关注 【AI架构笔记】!

扫码 / 搜一搜:AI架构笔记,一起进阶,驾驭AI!
更多推荐


所有评论(0)