导语
2026 年，RAG（Retrieval-Augmented Generation，检索增强生成）已成为开发者必备技能。
但大多数教程直接让你用 LangChain、LlamaIndex 等框架。框架确实方便，但也带来了问题：
"大多数 RAG 失败不是因为 LLM 不好，而是因为分块策略糟糕。" —— 技术社区共识
当你不理解 RAG 的底层原理时： - 不知道如何选择合适的分块策略 - 不知道嵌入模型的选择如何影响检索质量 - 不知道如何调试检索失败的问题 - 面试时被问到原理，只能背概念
本文将不用任何 RAG 框架，纯 Python 从零实现一个完整的 RAG 系统。你会理解每个环节的原理，真正掌握这项技能。
一、RAG 是什么？为什么需要它？
1.1 核心问题：LLM 的局限性
问题 1：知识截止
- LLM 的训练数据有截止时间（如 GPT-4 是 2023 年）
问题 2：幻觉（Hallucination）
"Without RAG, the model answers from memory alone — whatever patterns it absorbed during training." —— MongoEngine
问题 3：无法追溯来源
1.2 RAG 的解决方案
核心思想：
"When given a query, the RAG model first searches and retrieves relevant passages from its knowledge base, then uses them as contextual grounding to help generate an accurate response." —— DataForest
工作流程：
用户提问 ↓检索相关文档（从知识库） ↓将文档 + 问题一起发给 LLM ↓LLM 基于文档生成答案 ↓返回答案（附带来源）优势： - ✅ 可以回答训练数据之外的问题 - ✅ 减少幻觉（答案基于检索到的文档） - ✅ 可以追溯来源 - ✅ 可以更新知识库（无需重新训练）
二、RAG 系统架构详解
2.1 完整流程
一个完整的 RAG 系统包含 5 个核心环节：
环节说明：
| | |
|---|
| 文档摄入 | 读取各种格式的文档（PDF、Markdown、TXT） | |
| 分块 | | Recursive Character Text Splitter |
| 嵌入 | | |
| 存储 | | |
| 检索 | | |
| 生成 | | |
三、从零实现 RAG 系统
3.1 环境准备
# 只依赖最基础的库pip install numpy # 向量计算pip install sentence-transformers # 嵌入模型pip install faiss-cpu # 向量检索（CPU 版本）pip install PyPDF2 # PDF 处理为什么不用 LangChain？
LangChain 等框架封装了太多细节： - 你不知道分块策略是什么 - 你不知道嵌入模型如何选择 - 你不知道检索是如何优化的
本文目标：理解每个环节的原理，而不是调用 API。
3.2 第一步：文档摄入
import osfrom pathlib import Pathfrom typing import Listclass DocumentLoader: """文档加载器：支持多种格式""" def __init__(self, directory: str): self.directory = Path(directory) def load_all(self) -> List[dict]: """加载目录下所有文档""" documents = [] for file_path in self.directory.glob("*"): if file_path.suffix == ".txt": content = self._load_txt(file_path) elif file_path.suffix == ".pdf": content = self._load_pdf(file_path) elif file_path.suffix == ".md": content = self._load_markdown(file_path) else: continue documents.append({ "content": content, "metadata": { "source": str(file_path), "filename": file_path.name } }) return documents def _load_txt(self, path: Path) -> str: """加载 TXT 文件""" with open(path, 'r', encoding='utf-8') as f: return f.read() def _load_pdf(self, path: Path) -> str: """加载 PDF 文件""" try: from PyPDF2 import PdfReader reader = PdfReader(path) text = "" for page in reader.pages: text += page.extract_text() return text except Exception as e: print(f"PDF 加载失败：{path} - {e}") return "" def _load_markdown(self, path: Path) -> str: """加载 Markdown 文件（去除格式）""" with open(path, 'r', encoding='utf-8') as f: content = f.read() # 简单处理：去除 Markdown 语法 import re content = re.sub(r'#{1,6}\s*', '', content) # 标题 content = re.sub(r'\*\*(.*?)\*\*', r'\1', content) # 粗体 content = re.sub(r'\*(.*?)\*', r'\1', content) # 斜体 content = re.sub(r'\[(.*?)\]\(.*?\)', r'\1', content) # 链接 return content关键点： - 支持多种格式（TXT、PDF、Markdown） - 保留元数据（来源、文件名） - 错误处理（PDF 可能损坏）
3.3 第二步：分块（Chunking）
这是 RAG 系统最重要的环节。
"Chunking is often overlooked but it defines the model's 'view' of the data." —— LinkedIn
"One of the most common mistakes in RAG is treating chunking like a fixed setting instead of a retrieval design choice." —— Facebook
为什么需要分块？
- 嵌入模型有长度限制
- 检索精度
- 成本
分块策略对比
实现递归分块器
from typing import List, Optionalclass RecursiveTextSplitter: """ 递归文本分块器 核心思想： "Recursive splitting works by trying larger, more meaningful boundaries first, then falling back only when needed." —— [Medium](https://medium.com/@ThinkingLoop/rag-chunking-changes-more-than-recall-7fd71e7e1fcd) """ def __init__( self, chunk_size: int = 1000, chunk_overlap: int = 200, separators: Optional[List[str]] = None ): """ 参数： - chunk_size: 每块最大字符数 - chunk_overlap: 块之间重叠的字符数（保持上下文连贯） - separators: 分隔符优先级列表（从大到小） """ self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap # 分隔符优先级：先按大结构分，再按小结构分 # "separator arrays like newline-newline, period, and comma to respect document hierarchy" self.separators = separators or [ "\n\n", # 段落 "\n", # 换行 "。", # 句号（中文） ".", # 句号（英文） "，", # 逗号（中文） ",", # 逗号（英文） " ", # 空格 "" # 字符级别（最后手段） ] def split_text(self, text: str) -> List[str]: """将文本分成块""" chunks = [] start = 0 while start < len(text): # 从 start 位置找到合适的切分点 end = self._find_chunk_end(text, start) # 提取块 chunk = text[start:end].strip() if chunk: chunks.append(chunk) # 移动起始位置（考虑重叠） start = end - self.chunk_overlap if start < 0: start = end return chunks def _find_chunk_end(self, text: str, start: int) -> int: """找到块的结束位置""" # 如果剩余文本小于 chunk_size，直接返回末尾 if start + self.chunk_size >= len(text): return len(text) # 在 chunk_size 范围内找最佳切分点 chunk_end = start + self.chunk_size # 按优先级尝试找分隔符 for separator in self.separators: # 在 [start, chunk_end] 范围内找最后一个分隔符 last_sep = text.rfind(separator, start, chunk_end) if last_sep > start: return last_sep + len(separator) # 没找到分隔符，强制在 chunk_size 处切分 return chunk_end关键设计：
- 分隔符优先级
- 重叠（Overlap）
- 语义完整性
参数建议：
# 通用场景splitter = RecursiveTextSplitter(chunk_size=1000, chunk_overlap=200)# 技术文档（代码片段多）splitter = RecursiveTextSplitter(chunk_size=800, chunk_overlap=150)# 长文章/书籍splitter = RecursiveTextSplitter(chunk_size=1500, chunk_overlap=300)
3.4 第三步：嵌入（Embedding）
什么是嵌入？
嵌入（Embedding）是将文本转换为向量（一组数字）的过程。语义相似的文本，向量也相似。
为什么需要嵌入？
选择嵌入模型
"Compare LLM fine-tuning and Retrieval-Augmented Generation (RAG) for enterprise AI. Learn costs, latency, accuracy trade-offs and implementation patterns." —— Mortex Solutions
2026 年主流嵌入模型：
| | | | |
|---|
| bge-large-en-v1.5 | | | | |
| bge-large-zh-v1.5 | | | | |
| all-mpnet-base-v2 | | | | |
| all-MiniLM-L6-v2 | | | | |
| text-embedding-3-small | | | | |
推荐： - 中文场景：bge-large-zh-v1.5 - 英文场景：bge-large-en-v1.5 - 快速原型：all-MiniLM-L6-v2
实现嵌入器
from sentence_transformers import SentenceTransformerimport numpy as npclass Embedder: """文本嵌入器""" def __init__(self, model_name: str = "BAAI/bge-large-zh-v1.5"): """ 参数： - model_name: HuggingFace 模型名称 """ print(f"加载嵌入模型：{model_name}") self.model = SentenceTransformer(model_name) self.dimension = self.model.get_sentence_embedding_dimension() print(f"模型维度：{self.dimension}") def embed(self, texts: List[str]) -> np.ndarray: """ 将文本列表转换为向量 返回：numpy 数组，形状为 (n_texts, dimension) """ embeddings = self.model.encode( texts, batch_size=32, # 批处理，加快速度 show_progress_bar=True, convert_to_numpy=True ) return embeddings def embed_query(self, text: str) -> np.ndarray: """嵌入单个查询（问题）""" # 对于 BGE 模型，查询需要加前缀 query = f"为这个句子生成表示以用于检索：{text}" embedding = self.model.encode([query], convert_to_numpy=True) return embedding[0] # 返回一维数组关键点： - 使用 sentence-transformers 库（HuggingFace） - 批处理提高效率 - 查询嵌入可能需要特殊前缀（BGE 模型）
3.5 第四步：存储（Vector Store）
为什么需要向量数据库？
- 向量数据库使用 HNSW、FAISS 等算法，复杂度 O(log n)
向量数据库对比
| | | |
|---|
| FAISS | | | |
| Chroma | | | |
| LanceDB | | | |
| Qdrant | | | |
| Pinecone | | | |
本文选择 FAISS： - 最快（C++ 实现） - 无需额外服务 - 适合学习原理
实现向量存储
import faissimport picklefrom pathlib import Pathclass VectorStore: """FAISS 向量存储""" def __init__(self, dimension: int, index_path: Optional[str] = None): """ 参数： - dimension: 向量维度（与嵌入模型一致） - index_path: 索引保存路径（可选） """ self.dimension = dimension self.chunks = [] # 存储原始文本块 self.metadata = [] # 存储元数据 # 创建 FAISS 索引（L2 距离，即欧几里得距离） # FAISS 支持多种索引类型，这里用最简单的 IndexFlatL2 self.index = faiss.IndexFlatL2(dimension) # 如果提供了保存路径，加载已有索引 if index_path and Path(index_path).exists(): self.load(index_path) def add(self, embeddings: np.ndarray, chunks: List[str], metadata: List[dict]): """ 添加向量到索引 参数： - embeddings: 向量数组，形状 (n, dimension) - chunks: 原始文本块列表 - metadata: 元数据列表 """ # FAISS 要求向量是 float32 类型 embeddings = embeddings.astype('float32') # 添加到索引 self.index.add(embeddings) # 保存原始文本和元数据 self.chunks.extend(chunks) self.metadata.extend(metadata) print(f"已添加 {len(chunks)} 个向量，当前总数：{self.index.ntotal}") def search(self, query_embedding: np.ndarray, top_k: int = 5) -> List[dict]: """ 搜索最相似的向量 参数： - query_embedding: 查询向量 - top_k: 返回最相似的 k 个结果 返回：列表，每项包含文本、元数据、相似度分数 """ # FAISS 要求查询向量是二维数组 query_embedding = query_embedding.reshape(1, -1).astype('float32') # 搜索 distances, indices = self.index.search(query_embedding, top_k) # 整理结果 results = [] for i, idx in enumerate(indices[0]): if idx < 0: # FAISS 返回 -1 表示没有找到 continue results.append({ "chunk": self.chunks[idx], "metadata": self.metadata[idx], "distance": float(distances[0][i]), "similarity": 1 / (1 + distances[0][i]) # 转换为相似度（距离越小越相似） }) return results def save(self, path: str): """保存索引到磁盘""" Path(path).parent.mkdir(parents=True, exist_ok=True) # 保存 FAISS 索引 faiss.write_index(self.index, f"{path}.faiss") # 保存文本块和元数据 with open(f"{path}.pkl", 'wb') as f: pickle.dump({ 'chunks': self.chunks, 'metadata': self.metadata }, f) print(f"索引已保存到：{path}") def load(self, path: str): """从磁盘加载索引""" # 加载 FAISS 索引 self.index = faiss.read_index(f"{path}.faiss") # 加载文本块和元数据 with open(f"{path}.pkl", 'rb') as f: data = pickle.load(f) self.chunks = data['chunks'] self.metadata = data['metadata'] print(f"索引已加载：{path}，共 {self.index.ntotal} 个向量")关键点： - FAISS 使用 L2 距离（欧几里得距离） - 需要同时保存向量和原始文本 - 支持持久化（保存/加载）
3.6 第五步：检索与生成
实现检索器
class Retriever: """检索器：结合向量存储和 LLM""" def __init__(self, vector_store: VectorStore, embedder: Embedder): self.vector_store = vector_store self.embedder = embedder def retrieve(self, query: str, top_k: int = 5) -> List[dict]: """ 检索与查询最相关的文档块 参数： - query: 用户问题 - top_k: 返回最相关的 k 个结果 """ # 1. 嵌入查询 query_embedding = self.embedder.embed_query(query) # 2. 搜索向量 results = self.vector_store.search(query_embedding, top_k) return results实现生成器（调用 LLM）
import requestsimport osclass Generator: """LLM 生成器""" def __init__(self, api_key: Optional[str] = None, model: str = "gpt-3.5-turbo"): self.api_key = api_key or os.environ.get("OPENAI_API_KEY") self.model = model self.base_url = "https://api.openai.com/v1" def generate(self, query: str, context: List[str]) -> str: """ 基于上下文生成答案 参数： - query: 用户问题 - context: 检索到的相关文档块列表 """ # 构建提示词 context_text = "\n\n".join(context) prompt = f"""基于以下信息回答问题。如果信息不足，请说明。【相关信息】{context_text}【问题】{query}【答案】""" # 调用 OpenAI API response = requests.post( f"{self.base_url}/chat/completions", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" }, json={ "model": self.model, "messages": [ {"role": "user", "content": prompt} ], "temperature": 0.3 # 降低随机性，提高准确性 } ) response.raise_for_status() result = response.json() return result["choices"][0]["message"]["content"]
3.7 完整 RAG 系统
class SimpleRAG: """完整的 RAG 系统""" def __init__( self, document_directory: str, index_path: str, embedding_model: str = "BAAI/bge-large-zh-v1.5", chunk_size: int = 1000, chunk_overlap: int = 200 ): """ 初始化 RAG 系统 参数： - document_directory: 文档目录 - index_path: 向量索引保存路径 - embedding_model: 嵌入模型名称 - chunk_size: 分块大小 - chunk_overlap: 分块重叠 """ print("=" * 60) print("初始化 RAG 系统") print("=" * 60) # 1. 加载文档 print("\n[1/5] 加载文档...") loader = DocumentLoader(document_directory) documents = loader.load_all() print(f"加载了 {len(documents)} 个文档") # 2. 分块 print("\n[2/5] 分块...") splitter = RecursiveTextSplitter(chunk_size, chunk_overlap) chunks = [] chunk_metadata = [] for doc in documents: doc_chunks = splitter.split_text(doc["content"]) chunks.extend(doc_chunks) chunk_metadata.extend([doc["metadata"]] * len(doc_chunks)) print(f"分成 {len(chunks)} 个块") # 3. 嵌入 print("\n[3/5] 嵌入...") self.embedder = Embedder(embedding_model) embeddings = self.embedder.embed(chunks) # 4. 存储 print("\n[4/5] 存储...") self.vector_store = VectorStore( dimension=self.embedder.dimension, index_path=index_path ) # 如果索引为空，添加新数据 if self.vector_store.index.ntotal == 0: self.vector_store.add(embeddings, chunks, chunk_metadata) self.vector_store.save(index_path) # 5. 初始化检索器 print("\n[5/5] 初始化检索器...") self.retriever = Retriever(self.vector_store, self.embedder) print("\n" + "=" * 60) print("RAG 系统初始化完成！") print("=" * 60) def query(self, question: str, top_k: int = 5) -> dict: """ 查询 参数： - question: 用户问题 - top_k: 检索的文档块数量 返回：包含答案、相关文档、来源的字典 """ print(f"\n🔍 问题：{question}") # 1. 检索相关文档 print("检索相关文档...") results = self.retriever.retrieve(question, top_k) # 2. 提取上下文 context_chunks = [r["chunk"] for r in results] sources = [r["metadata"]["filename"] for r in results] # 3. 生成答案 print("生成答案...") generator = Generator() answer = generator.generate(question, context_chunks) # 4. 返回结果 return { "question": question, "answer": answer, "relevant_chunks": context_chunks, "sources": list(set(sources)), # 去重 "similarity_scores": [r["similarity"] for r in results] }
3.8 使用示例
# 初始化 RAG 系统rag = SimpleRAG( document_directory="./docs", # 文档目录 index_path="./index/rag_index", # 索引保存路径 embedding_model="BAAI/bge-large-zh-v1.5", chunk_size=1000, chunk_overlap=200)# 查询result = rag.query("RAG 的分块策略有哪些？")print("\n" + "=" * 60)print(f"答案：{result['answer']}")print(f"\n来源：{result['sources']}")print(f"相似度：{result['similarity_scores']}")print("=" * 60)
四、常见陷阱与优化技巧
4.1 分块陷阱
陷阱 1：块太大
解决：减小 chunk_size（800-1000）
陷阱 2：块太小
解决：增加 chunk_overlap（150-200）
陷阱 3：分隔符不当
解决：使用递归分块，优先在自然边界切分
"The trick is not just the list. It is the order. Recursive splitting works by trying larger, more meaningful boundaries first." —— Medium
4.2 检索优化
技巧 1：混合检索（Hybrid Search）
技巧 2：MMR（Maximum Marginal Relevance）
技巧 3：元数据过滤
4.3 评估指标
如何知道 RAG 系统好不好？
| | |
|---|
| Precision@K | | |
| Recall@K | | |
| Faithfulness | | |
| Answer Relevance | | |
简单评估方法：
def evaluate_rag(rag: SimpleRAG, test_questions: List[str], ground_truth: List[str]): """简单评估 RAG 系统""" from sklearn.metrics import precision_score, recall_score predictions = [] for q in test_questions: result = rag.query(q) # 简单判断：答案是否包含关键信息 predictions.append(result['answer']) # 这里需要人工标注或更复杂的评估 print(f"测试了 {len(test_questions)} 个问题") print("请人工评估答案质量")
五、总结
5.1 核心要点
- RAG 的价值：解决 LLM 知识截止、幻觉、无法追溯来源的问题
- 分块是关键
- 嵌入模型选择
- 向量数据库
- 评估很重要：用 Precision/Recall/Faithfulness 衡量效果
5.2 下一步学习
进阶主题： - GraphRAG（基于知识图谱的 RAG） - 多模态 RAG（图像 + 文本） - 自适应分块
生产环境： - 并发处理 - 缓存优化 - 监控告警
框架学习： - LangChain（功能最全） - LlamaIndex（RAG 专用） - Haystack（企业级）
参考资料
核心论文
- • Lewis et al. "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks" (2020)
- • Adaptive Chunking: Optimizing Chunking-Method Selection for RAG (arXiv 2026)
技术教程
- • RAG Tutorial: A Comprehensive Guide for Beginners [Updated 2026]
- • RAG Systems Complete Guide
- • Retrieval-Augmented Generation (RAG) Tutorial
- • RAG Chunking Changes More Than Recall
- • Chunking: The Secret to Effective RAG Systems
- • Best Embedding Models 2025: MTEB Scores & Leaderboard
- • Best Vector Databases in 2026 Compared
生产实践
- • RAG in Production: From Website Crawl to Vector Search
- • RAG in 2026: Smarter Retrieval and Real-Time Responses
- • LLM Fine-Tuning vs RAG in 2026
- • What is GraphRAG? Complete Guide to Graph-Based RAG in 2026
开源项目
- • Sentence Transformers - 嵌入模型库
10个月宝宝每天需要喝多少奶粉？
