当前位置：首页>python>不用框架,纯 Python 手写 RAG 系统

不用框架,纯 Python 手写 RAG 系统

  • 2026-09-18 23:34:09
不用框架,纯 Python 手写 RAG 系统

导语

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

  • LLM 会"编造"看似合理但错误的答案
  • 尤其是涉及具体数据、引用时

问题 3：无法追溯来源

  • 传统 LLM 无法告诉你答案来自哪里
  • 企业场景需要可追溯、可验证

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）
PyPDF2、markdown
分块
将长文档切成小块，便于检索
Recursive Character Text Splitter
嵌入
将文本转换为向量（数值表示）
Sentence Transformers
存储
存储向量，支持快速检索
FAISS、Chroma、LanceDB
检索
根据问题找到最相关的文档块
向量相似度搜索
生成
LLM 基于检索到的内容生成答案
OpenAI、Claude、本地模型

三、从零实现 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

为什么需要分块？

  1. 嵌入模型有长度限制
     - 通常 512-8192 tokens
  2. 检索精度
     - 大段文本包含太多噪声，影响检索质量
  3. 成本
     - LLM 按 token 计费，检索小块更便宜

分块策略对比

策略
原理
优点
缺点
适用场景
固定长度
按字符数切分
简单快速
可能切断语义
快速原型
按段落
按段落/换行切分
保持语义完整
块大小不一
博客、文章
递归分块
按优先级切分（标题→段落→句子）
最佳语义保持
实现复杂
生产环境
自适应
根据内容类型选择策略
最优效果
需要分类器
高级应用

实现递归分块器

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

关键设计：

  1. 分隔符优先级
     - 先按段落分，再按句子分，最后按字符分
  2. 重叠（Overlap）
     - 保持上下文连贯，避免信息丢失
  3. 语义完整性
     - 尽量在自然边界（段落、句子）处切分

参数建议：

# 通用场景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）是将文本转换为向量（一组数字）的过程。语义相似的文本，向量也相似。

为什么需要嵌入？

  • LLM 只能处理数值，不能直接处理文本
  • 向量可以计算相似度（余弦相似度）
  • 向量数据库可以快速检索

选择嵌入模型

"Compare LLM fine-tuning and Retrieval-Augmented Generation (RAG) for enterprise AI. Learn costs, latency, accuracy trade-offs and implementation patterns." —— Mortex Solutions

2026 年主流嵌入模型：

模型
维度
MTEB 分数
速度
适用场景
bge-large-en-v1.5
1024
64.5
中等
英文通用
bge-large-zh-v1.5
1024
63.8
中等
中文通用
all-mpnet-base-v2
768
63.5
慢
高质量
all-MiniLM-L6-v2
384
61.2
快
快速原型
text-embedding-3-small
1536
62.0
快
OpenAI API

推荐： - 中文场景：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）

为什么需要向量数据库？

  • 朴素搜索（遍历所有向量）复杂度 O(n)，太慢
  • 向量数据库使用 HNSW、FAISS 等算法，复杂度 O(log n)
  • 支持过滤、元数据查询

向量数据库对比

数据库
类型
特点
适用场景
FAISS
库（非数据库）
Facebook 出品，最快
本地部署、单机
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：块太大

  • 包含太多噪声，检索精度下降
  • LLM 处理成本高

解决：减小 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
前 K 个结果中有多少是相关的
人工标注
Recall@K
所有相关结果中有多少被检索到
人工标注
Faithfulness
答案是否忠实于检索到的文档
LLM 评估
Answer Relevance
答案是否直接回答问题
LLM 评估

简单评估方法：

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 核心要点

  1. RAG 的价值
    ：解决 LLM 知识截止、幻觉、无法追溯来源的问题
  2. 分块是关键
    ：大多数 RAG 失败是因为分块策略糟糕
  3. 嵌入模型选择
    ：中文用 BGE-ZH，英文用 BGE-EN
  4. 向量数据库
    ：FAISS 最快，Chroma 最易用
  5. 评估很重要
    ：用 Precision/Recall/Faithfulness 衡量效果

5.2 下一步学习

  1. 进阶主题：    - GraphRAG（基于知识图谱的 RAG）    - 多模态 RAG（图像 + 文本）    - 自适应分块

  2. 生产环境：    - 并发处理    - 缓存优化    - 监控告警

  3. 框架学习：    - LangChain（功能最全）    - LlamaIndex（RAG 专用）    - Haystack（企业级）

参考资料

核心论文

  1. 1. RAG 原始论文
    • • Lewis et al. "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks" (2020)
  2. 2. 分块优化
    • • Adaptive Chunking: Optimizing Chunking-Method Selection for RAG (arXiv 2026)

技术教程

  1. 3. 综合教程
    • • RAG Tutorial: A Comprehensive Guide for Beginners [Updated 2026]
    • • RAG Systems Complete Guide
    • • Retrieval-Augmented Generation (RAG) Tutorial
  2. 4. 分块策略
    • • RAG Chunking Changes More Than Recall
    • • Chunking: The Secret to Effective RAG Systems
  3. 5. 嵌入模型
    • • Best Embedding Models 2025: MTEB Scores & Leaderboard
  4. 6. 向量数据库
    • • Best Vector Databases in 2026 Compared

生产实践

  1. 7. 生产部署
    • • RAG in Production: From Website Crawl to Vector Search
    • • RAG in 2026: Smarter Retrieval and Real-Time Responses
  2. 8. 对比分析
    • • LLM Fine-Tuning vs RAG in 2026
    • • What is GraphRAG? Complete Guide to Graph-Based RAG in 2026

开源项目

  1. 9. 代码参考
    • • FAISS - Facebook 向量检索库
    • • Sentence Transformers - 嵌入模型库
    • • Chroma - 向量数据库
    • • LanceDB - 磁盘向量数据库
本文来自网友投稿或网络内容，如有侵犯您的权益请联系我们删除，联系邮箱：596563534@qq.com 。

9 个自动化互联网的 Python 库,最后一个太离谱

Python Tkinter 界面控件自动生成器:用数据驱动 UI 的工程实践

最新文章

随机文章