从零构建Python RAG系统：超越玩具项目的实战指南

1. 从“玩具”到“工具”：为什么你需要一个真正的RAG项目

如果你最近在关注AI应用开发，尤其是大语言模型（LLM）的落地，那么“RAG”这个词一定在你眼前晃了无数次。RAG，检索增强生成，听起来像是一个解决LLM“幻觉”和知识过时问题的银弹。网上充斥着各种“5分钟搭建RAG”、“10行代码实现问答”的教程，它们大多基于LangChain或LlamaIndex这样的高级框架，用几行封装好的代码，就能把一个PDF喂给模型，然后得到一个看似能回答问题的系统。

我最初也是从这些“玩具”项目入门的。但当我真的想用它来解决一个实际问题——比如，让模型基于我们公司内部长达500页的技术文档来回答客户咨询时，问题接踵而至。那个“5分钟搭建”的系统，回答要么是胡言乱语，要么就是“根据提供的信息，我无法回答”。它成了一个精致的摆设。

这就是“玩具”和“工具”的区别。一个真正的、可用的RAG系统，远不止是调用几个API。它涉及到对原始文本的深度理解、对检索精度的苛刻要求、对生成结果的可控性设计。网上很多入门内容止步于“跑通Demo”，却很少告诉你，当你的文档超过10个，当你的问题变得复杂，当你的用户要求零错误时，你该怎么办。

这篇手册，我想和你分享的，就是如何跨过“玩具”的门槛，动手搭建一个扎实的、以Python为核心的RAG项目骨架。我们不追求最快，但追求每一步都知其所以然；我们不依赖“魔法”框架，而是从底层组件开始组装，让你在遇到问题时，有能力拆开它、调试它、优化它。你会发现，抛开那些高级抽象后，RAG的核心逻辑清晰而有力。

2. 拆解RAG：一个朴素的三段论工作流

在引入任何库之前，我们必须像设计一个普通软件模块一样，理解RAG到底在干什么。抛开所有华丽的术语，RAG的核心是一个“检索-增强-生成”的三段论管道。我们可以用一个图书馆管理员的比喻来理解它：

1. 建立索引（Indexing）：你有一屋子杂乱无章的书（你的文档库）。管理员（索引系统）需要把这些书的内容拆解成一个个有意义的章节或知识点（文本切分/分块），并为每个知识点制作一张精美的卡片（向量化），卡片上记录了知识点的核心摘要（向量嵌入）。最后，所有卡片被按照某种规律（向量相似度）整齐地排列在卡片柜里。
2. 检索（Retrieval）：当有读者（用户）来问一个问题时，管理员首先理解这个问题（将问题向量化），然后拿着这个“问题卡片”，快速地在卡片柜里寻找那些摘要最相似的几张卡片（计算余弦相似度，Top-K召回）。这几张卡片对应的书页，就是最相关的参考资料。
3. 生成（Generation）：管理员不会直接把这几页纸扔给读者。他会结合问题，仔细阅读这几页纸上的内容（将检索到的文本作为上下文），然后组织自己的语言，给读者一个准确、完整、流畅的答案（LLM生成）。

基于这个朴素模型，一个最小可用的RAG系统只需要三个核心组件：

• 文本加载与切分器（Loader & Splitter）：负责读取各种格式（TXT, PDF, MD, HTML）的文档，并将其切割成适合处理的片段。
• 向量模型与数据库（Embedding Model & Vector Store）：负责将文本片段转化为数学向量（嵌入），并存储、索引这些向量，支持快速相似度检索。
• 大语言模型（LLM）：负责理解“问题+检索到的上下文”，并生成最终答案。

我们的实战，就从亲手组装这三个组件开始。你会发现，即使只用最基础的库，你也能构建出比很多“玩具”项目更可控的系统。

3. 环境奠基：构建一个可复现的Python工作区

在开始写第一行业务代码前，一个隔离、干净、依赖明确的环境是专业项目的起点。我强烈建议你放弃直接使用系统Python或全局安装包的习惯。

3.1 为什么是Conda+Poetry？

虚拟环境工具很多，我选择Conda管理Python解释器本身（尤其是处理一些有C扩展依赖的包时更省心），用Poetry管理项目依赖。Poetry的pyproject.toml能清晰地声明依赖、分组（比如区分开发依赖和线上依赖）、并锁定精确版本，这比手写requirements.txt要优雅和可靠得多。

# 1. 创建并激活一个Conda环境（假设你已安装Miniconda） conda create -n rag_workshop python=3.10 -y conda activate rag_workshop # 2. 安装Poetry（如果你还没有） pip install poetry # 3. 在你的项目目录初始化Poetry poetry init

执行poetry init时，它会交互式地引导你创建pyproject.toml文件。对于依赖，我们暂时可以不填，后面用add命令来加。

3.2 核心依赖选型与安装

接下来，我们为RAG的三段论挑选具体的“武器”。这里的选择基于稳定性、社区活跃度和上手难度。

# 进入项目目录后，使用Poetry添加依赖 # 核心数据处理与向量计算 poetry add numpy pandas # 文本切分与处理 poetry add pypdf2 markdown beautifulsoup4 # 用于PDF、Markdown、HTML poetry add "langchain-text-splitters" # 使用LangChain的高质量切分器，但不引入其全量框架 # 向量模型（本地轻量首选） poetry add sentence-transformers # 向量数据库（本地轻量首选） poetry add chromadb # LLM调用（以OpenAI API为例，也可替换为其他） poetry add openai # 可选：用于更复杂的文本清理 poetry add tiktoken # OpenAI的Tokenizer，用于精确计算长度 # 添加开发依赖组，如代码格式化、类型检查等 poetry add --group dev black isort mypy pylint

注意：这里我们刻意只引入了langchain-text-splitters，而不是整个LangChain。这能让我们聚焦于核心流程，避免被框架的复杂性干扰初学时的理解。当你对底层了如指掌后，再使用框架来提升开发效率才是明智的。

执行poetry install后，所有依赖都会被安装在一个独立的虚拟环境中。你可以通过poetry shell进入该环境，或使用poetry run python your_script.py来运行脚本。

4. 第一步：文本加载与智能切分——质量决定上限

这是最容易被轻视，却对最终效果影响最大的环节。糟糕的切分会直接导致检索到无关信息，俗称“垃圾进，垃圾出”。

4.1 文档加载：处理多种格式

我们需要一个统一的接口来加载不同格式的文档。这里我们写一个简单的工具函数：

import PyPDF2 from bs4 import BeautifulSoup import markdown def load_document(file_path: str) -> str: """加载文本内容，支持 .txt, .pdf, .md, .html""" text = "" if file_path.endswith('.txt'): with open(file_path, 'r', encoding='utf-8') as f: text = f.read() elif file_path.endswith('.pdf'): with open(file_path, 'rb') as f: reader = PyPDF2.PdfReader(f) for page in reader.pages: text += page.extract_text() + "\n" elif file_path.endswith('.md'): with open(file_path, 'r', encoding='utf-8') as f: md_text = f.read() # 将markdown转为html，再提取纯文本 html = markdown.markdown(md_text) soup = BeautifulSoup(html, "html.parser") text = soup.get_text() elif file_path.endswith('.html') or file_path.endswith('.htm'): with open(file_path, 'r', encoding='utf-8') as f: soup = BeautifulSoup(f.read(), "html.parser") text = soup.get_text() else: raise ValueError(f"Unsupported file format: {file_path}") # 基础清理：合并多余空白字符 import re text = re.sub(r'\s+', ' ', text).strip() return text

4.2 文本切分：为什么不能简单按字数切？

很多新手会直接用text[i:i+chunk_size]来切片，这是灾难性的。它会粗暴地切断句子、甚至单词，破坏语义的完整性。

正确的做法是在自然的语义边界处进行切分，比如句子、段落，或者Markdown/HTML的标题。我们之前安装的langchain-text-splitters提供了非常优秀的RecursiveCharacterTextSplitter。

from langchain_text_splitters import RecursiveCharacterTextSplitter def split_text(text: str, chunk_size: int = 500, chunk_overlap: int = 50) -> list[str]: """ 使用递归字符切分器进行智能切分。 :param chunk_size: 每个文本块的目标最大字符数（并非严格相等）。 :param chunk_overlap: 块与块之间的重叠字符数，防止上下文断裂。 """ # 它默认按 ["\n\n", "\n", " ", ""] 的顺序尝试切分，优先保持段落、句子完整。 splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, length_function=len, # 使用简单的字符长度计算，也可以用tiktoken计算token数 separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] # 针对中文调整分隔符 ) chunks = splitter.split_text(text) return chunks

4.3 关键参数调优与实战心得

• chunk_size（块大小）：这是最重要的参数。太小（如100）会丢失上下文，检索到的信息碎片化；太大（如2000）可能包含过多无关信息，稀释核心内容，且可能超过LLM的上下文窗口限制。我的经验是，对于通用文档，500-800是一个不错的起点；对于代码或技术规范，可以更小一些（300-500）。
• chunk_overlap（重叠度）：重叠是为了避免一个完整的句子或概念被硬生生切成两半，导致任何一块都不完整。通常设置为chunk_size的10%-20%。重叠部分在后续向量化时会有重复计算，但这是保证召回率必要的代价。
• separators（分隔符）：对于中文文档，一定要调整默认分隔符。我上面的例子加入了中文标点。更高级的做法可以尝试用spaCy或jieba进行句子分割，但RecursiveCharacterTextSplitter在大多数场景下已经足够好。

一个常见的坑是，直接从PDF提取的文本可能包含大量的页眉、页脚、页码和换行符。在切分前，最好写一些正则表达式进行清洗。例如，移除形如“第 X 页”的字符串，或者将因为PDF换行而断开的单词重新连接起来。

5. 第二步：向量化与存储——将文本映射到数学空间

文本切分好后，我们需要把它们变成计算机能高效计算相似度的东西——向量。

5.1 嵌入模型选择：本地还是云端？

• OpenAItext-embedding-ada-002：效果稳定，简单易用，但需要API调用，有费用和延迟，且数据需出境。
• sentence-transformers：开源库，提供大量预训练模型，可在本地运行，数据隐私有保障，是入门和生产的首选。

我们选择sentence-transformers，并选用一个在中文上表现良好的模型，例如paraphrase-multilingual-MiniLM-L12-v2，它平衡了速度和效果。

from sentence_transformers import SentenceTransformer import numpy as np class LocalEmbedder: def __init__(self, model_name: str = 'paraphrase-multilingual-MiniLM-L12-v2'): # 首次运行会下载模型，请确保网络通畅 self.model = SentenceTransformer(model_name) print(f"Loaded embedding model: {model_name}") def embed_documents(self, texts: list[str]) -> np.ndarray: """将一批文本转换为向量。""" # 模型返回的是numpy数组 embeddings = self.model.encode(texts, convert_to_numpy=True, show_progress_bar=True, # 处理大量文本时显示进度 normalize_embeddings=True) # 归一化，方便余弦相似度计算 return embeddings def embed_query(self, query: str) -> np.ndarray: """将单个查询转换为向量。""" return self.model.encode([query], convert_to_numpy=True, normalize_embeddings=True)[0]

5.2 向量数据库：为什么需要它？

当你有几千、几万个文本块时，用embed_query得到问题向量后，难道要遍历计算和每一个块向量的相似度吗？这效率太低了。向量数据库（Vector Store）就是为解决这个问题而生的。它使用近似最近邻（ANN）算法，如HNSW、IVF，在精度损失很小的前提下，实现海量向量的毫秒级检索。

我们使用ChromaDB，因为它轻量、易用，且完全本地化。

import chromadb from chromadb.config import Settings class VectorStoreManager: def __init__(self, persist_directory: str = "./chroma_db"): # 配置ChromaDB，设置持久化目录 self.client = chromadb.PersistentClient( path=persist_directory, settings=Settings(anonymized_telemetry=False) # 禁用匿名数据收集 ) # 获取或创建一个集合（类似于数据库中的表） self.collection = self.client.get_or_create_collection( name="knowledge_base", metadata={"hnsw:space": "cosine"} # 使用余弦相似度作为距离度量 ) def add_documents(self, chunks: list[str], embeddings: np.ndarray, metadatas: list[dict] = None): """将文本块及其向量添加到集合中。""" # 生成唯一ID ids = [f"doc_{i}" for i in range(len(chunks))] # 将numpy数组转换为列表（ChromaDB接受的格式） embeddings_list = embeddings.tolist() # 如果没有提供元数据，则用空字典填充 if metadatas is None: metadatas = [{} for _ in chunks] self.collection.add( documents=chunks, embeddings=embeddings_list, metadatas=metadatas, ids=ids ) print(f"Added {len(chunks)} documents to vector store.") def search(self, query_embedding: np.ndarray, top_k: int = 5) -> list[tuple[str, float]]: """检索最相似的top_k个文本块，返回(文本, 相似度得分)。""" results = self.collection.query( query_embeddings=[query_embedding.tolist()], n_results=top_k, include=["documents", "distances"] # 返回文本和距离 ) # ChromaDB返回的距离是余弦距离（1 - 余弦相似度），值越小越相似 # 我们将其转换为相似度分数（越大越相似） retrieved_docs = [] for doc, distance in zip(results['documents'][0], results['distances'][0]): similarity_score = 1 - distance # 近似余弦相似度 retrieved_docs.append((doc, similarity_score)) return retrieved_docs

5.3 构建索引的完整流程

现在，我们可以将前两步串联起来，构建一个完整的索引管道：

def build_knowledge_base(doc_paths: list[str], vector_store_dir: str = "./chroma_db"): """从原始文档构建向量知识库。""" all_chunks = [] all_metadatas = [] # 1. 初始化组件 embedder = LocalEmbedder() vector_store = VectorStoreManager(persist_directory=vector_store_dir) for doc_path in doc_paths: print(f"Processing: {doc_path}") # 2. 加载文档 raw_text = load_document(doc_path) # 3. 智能切分 chunks = split_text(raw_text, chunk_size=600, chunk_overlap=80) # 4. 为每个块创建元数据（例如记录来源文件） metadatas = [{"source": doc_path, "chunk_index": i} for i in range(len(chunks))] all_chunks.extend(chunks) all_metadatas.extend(metadatas) print(f"Total chunks: {len(all_chunks)}") # 5. 批量生成向量（比逐条生成效率高很多） print("Generating embeddings...") embeddings = embedder.embed_documents(all_chunks) # 6. 存入向量数据库 print("Adding to vector store...") vector_store.add_documents(all_chunks, embeddings, all_metadatas) print("Knowledge base built successfully!") return embedder, vector_store

6. 第三步：检索与生成——组装最终答案

索引建好后，就进入了实时问答环节。

6.1 检索环节：不仅仅是相似度

基础的相似度检索（我们上面实现的）已经能解决大部分问题。但在复杂场景下，我们需要更智能的检索策略：

• 重排序（Re-ranking）：初步检索出Top-K（比如20个）相关文档后，使用一个更精细但更耗时的模型（如BGE-reranker）对它们进行重新打分和排序，只保留最相关的Top-N（比如5个）给LLM。这能显著提升精度。
• 混合检索（Hybrid Search）：结合稠密向量检索（我们正在做的）和稀疏向量检索（如BM25，基于关键词匹配）。前者语义理解好，后者对精确术语召回强。ChromaDB也支持集成BM25。

为了保持入门手册的简洁，我们先实现基础版本。但你需要知道，当效果遇到瓶颈时，“重排序”通常是第一个应该考虑的优化点。

6.2 提示工程：如何让LLM更好地利用上下文？

检索到的上下文不会自动变成答案。我们需要精心设计一个提示（Prompt）来引导LLM。一个健壮的提示模板通常包含以下几个部分：

def build_prompt(query: str, contexts: list[str]) -> str: """构建给LLM的提示。""" context_str = "\n\n---\n\n".join([f"[Context {i+1}]: {ctx}" for i, ctx in enumerate(contexts)]) prompt = f"""你是一个专业的助手，请严格根据以下提供的上下文信息来回答问题。如果上下文中的信息不足以回答问题，请直接说“根据已知信息无法回答该问题”，不要编造信息。 相关上下文信息： {context_str} 问题：{query} 请根据上述上下文回答。答案：""" return prompt

这个模板明确了角色、指令、上下文和问题。清晰的指令能极大减少LLM的“幻觉”。

6.3 调用LLM完成生成

我们以OpenAI API为例（你需要设置环境变量OPENAI_API_KEY）：

import os from openai import OpenAI class OpenAIGenerator: def __init__(self, model: str = "gpt-3.5-turbo"): self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) self.model = model def generate(self, prompt: str, temperature: float = 0.1) -> str: """调用LLM生成答案。""" try: response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "user", "content": prompt} ], temperature=temperature, # 低温度使输出更确定、更忠于上下文 max_tokens=1000 ) return response.choices[0].message.content.strip() except Exception as e: return f"Error generating answer: {e}"

6.4 串联成完整的RAG问答链

最后，我们把所有组件像流水线一样组装起来：

class SimpleRAGSystem: def __init__(self, embedder, vector_store, llm_generator): self.embedder = embedder self.vector_store = vector_store self.llm_generator = llm_generator def ask(self, question: str, top_k: int = 5) -> dict: """核心问答流程。""" # 1. 将问题向量化 print(f"Embedding question: {question[:50]}...") query_embedding = self.embedder.embed_query(question) # 2. 检索相关文档 print("Searching vector store...") retrieved_items = self.vector_store.search(query_embedding, top_k=top_k) contexts = [item[0] for item in retrieved_items] print(f"Retrieved {len(contexts)} contexts.") # 3. 构建提示 prompt = build_prompt(question, contexts) # （调试时可打印prompt） # print("--- Prompt ---\n", prompt[:500], "\n---") # 4. 生成答案 print("Generating answer with LLM...") answer = self.llm_generator.generate(prompt) # 5. 返回结果，包含答案和用于解释的引用来源 return { "question": question, "answer": answer, "source_documents": contexts, "similarity_scores": [item[1] for item in retrieved_items] } # 使用示例 if __name__ == "__main__": # 假设你已经运行过 build_knowledge_base，向量库已存在 embedder = LocalEmbedder() vector_store = VectorStoreManager() llm = OpenAIGenerator(model="gpt-3.5-turbo") # 或使用其他本地模型接口 rag = SimpleRAGSystem(embedder, vector_store, llm) while True: user_q = input("\n请输入你的问题 (输入 'quit' 退出): ") if user_q.lower() == 'quit': break result = rag.ask(user_q) print(f"\nAnswer: {result['answer']}") # 可选：显示来源 # for i, ctx in enumerate(result['source_documents']): # print(f"\n[Source {i+1}, Score: {result['similarity_scores'][i]:.3f}]: {ctx[:200]}...")

7. 从“能用”到“好用”：你必须面对的优化挑战

一个能返回答案的系统只是起点。要让它在实际中可靠，你必须关注以下问题：

7.1 效果评估：你的RAG系统真的准吗？

没有评估，优化就是盲人摸象。你需要一个评估集（一组问题+标准答案/相关文档）。可以从以下几个维度评估：

• 检索相关性：检索到的文档是否真的与问题相关？（可以人工打分，或使用LLM-as-a-judge自动评分）
• 答案忠实度：生成的答案是否严格基于检索到的上下文，没有胡编乱造？
• 答案准确性：基于上下文，答案本身是否正确？
• 答案流畅性：答案是否通顺、完整？

建立一个简单的评估脚本，定期跑分，是迭代优化的基础。

7.2 常见问题与调优方向

• 问题1：答案不相关或胡编乱造。

  • 检查检索：首先看检索到的source_documents是否相关。如果不相关，问题出在前端：调整chunk_size/overlap，尝试不同的embedding模型，或者引入重排序。
  • 检查提示：如果检索结果相关但答案胡扯，强化你的prompt指令，比如加上“如果信息不足，请明确说明”。
  • 降低LLM的temperature：将其设为0.1或0，减少随机性。

• 问题2：答案遗漏了关键信息。

  • 增加top_k：让LLM看到更多的上下文。
  • 优化分块策略：可能关键信息正好被切分在了两个块的边缘，尝试增加chunk_overlap。
  • 尝试不同的分块方法：对于结构化文档（如Markdown），可以尝试按标题（MarkdownHeaderTextSplitter）分块，能更好地保持语义单元完整。

• 问题3：处理长文档或大量文档时速度慢。

  • 批量嵌入：确保使用embed_documents进行批量处理，而非循环调用单条嵌入。
  • 向量数据库索引：ChromaDB默认使用HNSW，对于千万级以下的数据量性能很好。如果数据量极大，可以研究其持久化索引的配置参数。
  • 异步处理：对于构建索引的过程，可以考虑使用异步IO来并行处理多个文档。

7.3 引入路由与元数据过滤

更高级的RAG系统会引入“路由”概念。例如，你的向量库存储了公司“产品手册”、“技术博客”、“客服Q&A”等多种文档。当用户问“如何退款？”时，系统应该优先在“客服Q&A”中搜索。这可以通过在存储时为每个文本块添加metadata（如{"doc_type": "faq"}），并在检索时指定过滤条件来实现。

# 在检索时增加元数据过滤 results = self.collection.query( query_embeddings=[query_embedding.tolist()], n_results=top_k, where={"doc_type": {"$eq": "faq"}}, # 过滤条件 include=["documents", "distances"] )

8. 项目脚手架与后续演进

至此，你已经拥有了一个完全受控、可深度定制的RAG系统核心。我建议你将上述代码模块化，组织成一个真正的项目：

my_rag_project/ ├── pyproject.toml # Poetry依赖管理 ├── README.md ├── src/ │ ├── __init__.py │ ├── document_processor.py # 加载、切分模块 │ ├── embedding.py # 嵌入模型封装 │ ├── vector_store.py # 向量数据库操作 │ ├── llm_client.py # LLM调用封装 │ ├── prompt_templates.py # 提示词模板 │ └── rag_pipeline.py # 核心流水线组装 ├── scripts/ │ ├── build_kb.py # 构建知识库脚本 │ └── query_cli.py # 命令行问答脚本 ├── data/ # 存放原始文档 └── chroma_db/ # ChromaDB持久化数据

这个手工作坊式的项目，是你理解RAG每一寸肌肤的最佳方式。当你对数据流、瓶颈、调参点都有了切身感受后，你可以选择：

1. 引入LangChain/LlamaIndex：用它们来替换你手写的部分模块（如更复杂的文档加载器、链式调用），提升开发效率。这时你是在“驾驶”框架，而不是被框架“裹挟”。
2. 探索高级模式：如Agentic RAG（让LLM主动决定何时、如何检索）、Hypothetical Document Embeddings (HyDE)（先让LLM生成一个假设答案，再用这个答案去检索，效果奇佳）等。
3. 构建Web服务：使用FastAPI将你的RAG系统包装成API，供前端调用。
4. 持续迭代：根据评估结果，持续优化分块、嵌入模型、提示词，甚至微调一个本地的小型重排序模型。

RAG不是一个一蹴而就的框架调用，而是一个需要持续观察、分析和调优的系统工程。这份手册给你的不是一辆现成的汽车，而是一套完整的汽车零件和组装图纸。从拧第一个螺丝开始，你才能真正掌握驾驶它的能力。当你下次再看到“五分钟搭建RAG”的标题时，你心里会清楚，那只是旅程的起点，而真正的道路，现在才刚刚在你脚下展开。