从零搭建RAG系统：手把手实现私有知识库智能问答-开发者社区

这次我们来看一个关于RAG（检索增强生成）的实战教程。标题里提到的“吴恩达亲授”并非指他本人亲自录制的视频，而是指其教学理念和课程体系在RAG领域的延伸应用。这个教程的核心价值在于，它提供了一个从零开始、手把手搭建一个可运行的检索增强生成系统的完整路径，特别适合那些已经了解了大模型基础，但不知道如何让模型“读懂”并“利用”自己私有数据的开发者。

对于想在企业内部部署智能问答、构建个人知识库助手，或者单纯想深入理解RAG技术栈的工程师来说，最关心的不是概念，而是“我能不能在自己的电脑上跑起来”、“需要多少资源”以及“怎么验证效果”。本文将围绕一个典型的RAG系统搭建流程，拆解从环境准备、文档处理、向量检索到与大模型集成的每一步，并提供可操作的代码示例和效果验证方法。

1. 核心能力速览

在深入细节之前，我们先通过一个表格快速了解基于本教程思路搭建的RAG系统具备哪些核心能力，以及大致的资源门槛。

能力项	说明与评估
项目类型	检索增强生成（RAG）系统实战教程/代码实现
核心功能	1. 文档解析与文本切片（Chunking） 2. 文本向量化嵌入（Embedding） 3. 向量存储与相似性检索（Vector Search） 4. 与大模型（LLM）集成进行增强生成
硬件门槛	开发/测试环境：普通CPU或带GPU的电脑均可。向量模型推理和LLM调用是主要计算点。 -CPU模式：可运行，但嵌入和生成速度较慢。 -GPU加速：推荐使用，能显著提升嵌入模型和LLM的推理速度。显存要求取决于模型大小，轻量级嵌入模型（如`BAAI/bge-small-zh`）1-2GB显存足够，7B参数的LLM需要8GB以上显存。
启动方式	通常为命令行分步执行或集成在一个Web应用（如使用Gradio/Streamlit）中启动。本文将以分步脚本和简易API服务为例。
是否支持API	是。可以封装核心的“检索-生成”流程为RESTful API，供其他系统调用。
是否支持批量任务	是。文档入库（向量化存储）环节天然支持批量处理。问答环节也可通过API实现批量提问。
适合场景	1. 个人或团队私有知识库问答 2. 企业文档智能查询与摘要 3. 作为AI应用的后端知识引擎 4. 学习RAG技术栈的动手实践

2. 适用场景与使用边界

一个自己搭建的RAG系统，其价值在于可控性和定制化。它最适合以下几类场景：

企业内部知识管理：将公司内部的产品手册、技术文档、会议纪要等非结构化数据转化为可查询的知识库，新员工或技术支持人员可以快速获取准确信息。
个人学习与研究：整理个人的读书笔记、收藏的博客文章、研究论文，构建一个随时可问的“第二大脑”。
客服与智能问答：基于产品文档和历史问答记录，搭建一个能准确回答常见问题的客服机器人原型。
内容创作辅助：为写作或报告生成提供基于特定知识库的事实依据和素材。

然而，在投入应用前，必须明确其使用边界：

知识局限性：系统的答案质量完全依赖于灌入的文档质量。它无法回答文档中未包含的信息（除非LLM本身具备该知识），且可能因检索不准或LLM“幻觉”而产生错误答案。
性能与成本：本地部署的轻量级模型效果不如云端大型API，但成本可控、数据隐私有保障。需要在效果、成本与隐私间权衡。
数据安全与合规：处理企业或个人敏感数据时，务必在安全的内部网络环境部署。使用开源模型和本地向量数据库是保障数据不出域的关键。
版权与授权：仅为学习与研究目的使用公开或自有版权的文档。切勿将未经授权的受版权保护的内容用于商业用途。

3. 环境准备与前置条件

让我们开始从零搭建。首先，确保你的开发环境满足以下基础要求。

操作系统：Linux (Ubuntu 20.04+)、macOS 或 Windows (WSL2推荐) 均可。本文命令以Linux/macOS为例，Windows用户可在WSL或PowerShell中对应调整。Python版本：推荐 Python 3.9 或 3.10，这是多数AI库兼容性较好的版本。包管理工具：使用pip或conda。建议先创建一个独立的虚拟环境。

# 创建并激活虚拟环境 (以conda为例) conda create -n rag_tutorial python=3.10 conda activate rag_tutorial # 或者使用 venv python -m venv rag_env source rag_env/bin/activate # Linux/macOS # rag_env\Scripts\activate # Windows

核心依赖库：我们将使用以下主流开源库，它们构成了现代RAG技术栈的基石。

# requirements.txt 核心内容 langchain==0.1.0 # RAG应用框架，提供高层抽象 langchain-community==0.0.10 # 社区贡献的组件 chromadb==0.4.22 # 轻量级向量数据库 sentence-transformers==2.2.2 # 用于生成文本向量的嵌入模型库 pypdf==3.17.4 # 用于解析PDF文档 python-dotenv==1.0.0 # 管理环境变量（如API密钥） openai==1.12.0 # 如需调用OpenAI API # 如果使用本地LLM，例如Ollama ollama==0.1.6 # Ollama客户端库 # 如果使用国内大模型API，例如智谱、DeepSeek zhipuai==2.0.1 # 智谱AI

使用pip一键安装：

pip install -r requirements.txt

硬件资源检查：

磁盘空间：预留至少2-5GB空间用于存放模型文件（嵌入模型、可能的本地LLM）。
内存：建议8GB以上。处理大量文档时，内存占用会上升。
GPU（可选但推荐）：如需运行本地嵌入模型或LLM，检查CUDA环境。
```
# 检查PyTorch是否能识别CUDA python -c "import torch; print(torch.cuda.is_available())"
```
输出True则表示GPU可用。确保已安装与CUDA版本匹配的torch。

4. 搭建流程与核心代码拆解

一个完整的RAG系统流程可以概括为两个阶段：知识库构建（索引）和问答（检索与生成）。下面我们分步实现。

4.1 阶段一：知识库构建 - 文档加载、切分与向量化

第一步是处理你的原始文档，将其转化为向量数据库中的一条条记录。

# build_knowledge_base.py import os from langchain_community.document_loaders import PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma class KnowledgeBaseBuilder: def __init__(self, persist_directory="./chroma_db"): """ 初始化知识库构建器。 :param persist_directory: 向量数据库持久化目录 """ self.persist_directory = persist_directory # 1. 初始化嵌入模型（使用开源模型，本地运行） # 模型会自动从HuggingFace下载，首次使用需要时间 self.embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-small-zh-v1.5", # 中文小模型，效果不错 model_kwargs={'device': 'cuda'}, # 使用GPU，如果只有CPU改为'cpu' encode_kwargs={'normalize_embeddings': True} # 归一化，提升检索效果 ) # 2. 初始化文本分割器 self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个文本块的最大字符数 chunk_overlap=50, # 块之间的重叠字符数，保持上下文 separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) def load_and_split_documents(self, doc_path): """ 加载并分割单个文档。 :param doc_path: 文档路径 :return: 分割后的文档列表 """ if doc_path.endswith('.pdf'): loader = PyPDFLoader(doc_path) elif doc_path.endswith('.txt'): loader = TextLoader(doc_path, encoding='utf-8') else: raise ValueError(f"Unsupported document type: {doc_path}") documents = loader.load() # 进行文本分割 split_docs = self.text_splitter.split_documents(documents) print(f"Loaded {len(documents)} pages, split into {len(split_docs)} chunks.") return split_docs def build_from_directory(self, input_dir): """ 从一个目录批量构建知识库。 :param input_dir: 存放原始文档的目录 """ all_splits = [] for filename in os.listdir(input_dir): file_path = os.path.join(input_dir, filename) if os.path.isfile(file_path) and (filename.endswith('.pdf') or filename.endswith('.txt')): try: splits = self.load_and_split_documents(file_path) all_splits.extend(splits) print(f"Processed: {filename}") except Exception as e: print(f"Error processing {filename}: {e}") if not all_splits: print("No valid documents found.") return # 3. 创建向量存储（Chromadb），并进行持久化 print(f"Creating vectorstore with {len(all_splits)} chunks...") vectordb = Chroma.from_documents( documents=all_splits, embedding=self.embeddings, persist_directory=self.persist_directory ) vectordb.persist() # 持久化到磁盘 print(f"Knowledge base built and saved to '{self.persist_directory}'.") if __name__ == "__main__": builder = KnowledgeBaseBuilder() # 假设你的文档放在 ./documents 目录下 builder.build_from_directory("./documents")

关键操作与解释：

嵌入模型选择：BAAI/bge-small-zh-v1.5是一个优秀的中文文本表示模型，体积小（约300MB），在CPU上也能运行，GPU上更快。首次运行会自动下载。
文本分割（Chunking）：这是影响检索效果的关键步骤。chunk_size=500意味着每块文本约500字符，chunk_overlap=50让相邻块有部分重叠，防止上下文被硬切断。
向量数据库：Chroma是一个轻量级、易用的向量数据库，它将文本块、其向量表示以及元数据（如来源）存储在一起。persist_directory指定了数据保存的位置。

运行与验证：

# 准备一个 documents 文件夹，放入你的PDF或TXT文件 mkdir -p documents # 将你的文档复制进去，例如：cp ~/my_doc.pdf documents/ # 运行构建脚本 python build_knowledge_base.py

如果看到类似“Knowledge base built and saved to './chroma_db'.”的输出，并且./chroma_db目录下生成了一些文件，说明知识库构建成功。

4.2 阶段二：问答系统 - 检索器、提示工程与大模型集成

知识库准备好后，我们就可以实现问答功能了。其核心是“检索-增强-生成”链。

# rag_qa_system.py from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain_community.llms import Ollama # 使用本地Ollama模型 # 也可以使用其他LLM，例如OpenAI # from langchain_openai import ChatOpenAI import sys class RAGQASystem: def __init__(self, persist_directory="./chroma_db"): """ 初始化RAG问答系统。 """ # 1. 加载之前创建的向量数据库 from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma self.embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-small-zh-v1.5", model_kwargs={'device': 'cuda'}, encode_kwargs={'normalize_embeddings': True} ) self.vectordb = Chroma( persist_directory=persist_directory, embedding_function=self.embeddings ) # 2. 将向量数据库转换为检索器，可以设置返回的相似文本块数量 self.retriever = self.vectordb.as_retriever(search_kwargs={"k": 3}) # 返回最相关的3个块 # 3. 定义提示词模板，这是指导LLM如何利用检索结果的关键 self.prompt_template = """请根据以下上下文信息回答问题。如果上下文信息不足以回答问题，请直接说“根据提供的信息无法回答此问题”，不要编造答案。 上下文信息： {context} 问题：{question} 请根据上下文提供准确的答案：""" self.PROMPT = PromptTemplate( template=self.prompt_template, input_variables=["context", "question"] ) # 4. 选择大语言模型（LLM） # 方案A：使用本地Ollama模型（需先安装Ollama并拉取模型，如qwen2:7b） self.llm = Ollama(model="qwen2:7b", base_url="http://localhost:11434") # 方案B：使用OpenAI API（需设置环境变量OPENAI_API_KEY） # from langchain_openai import ChatOpenAI # self.llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0) # 方案C：使用国内大模型API，如智谱GLM（需安装zhipuai并设置ZHIPUAI_API_KEY） # from langchain_community.llms import ZhipuAI # self.llm = ZhipuAI(model="glm-4", temperature=0) # 5. 构建检索问答链 self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", # 将检索到的所有文本“塞”进提示词 retriever=self.retriever, return_source_documents=True, # 返回源文档，便于追溯 chain_type_kwargs={"prompt": self.PROMPT} ) def ask(self, question): """ 向RAG系统提问。 :param question: 用户问题 :return: 答案和来源 """ result = self.qa_chain.invoke({"query": question}) answer = result["result"] source_docs = result["source_documents"] return answer, source_docs def main(): qa_system = RAGQASystem() print("RAG QA系统已加载。输入‘退出’或‘quit’结束。") while True: user_input = input("\n请输入您的问题：") if user_input.lower() in ["退出", "quit", "exit"]: break if not user_input.strip(): continue answer, sources = qa_system.ask(user_input) print(f"\n【答案】\n{answer}") print(f"\n【来源参考】") for i, doc in enumerate(sources): print(f" [{i+1}] {doc.metadata.get('source', '未知')} - 第{doc.metadata.get('page', 'N/A')}页") # 打印来源文本片段（前200字符） print(f" 片段：{doc.page_content[:200]}...") if __name__ == "__main__": main()

关键操作与解释：

检索器（Retriever）：从向量数据库中根据问题语义搜索最相关的文本块。search_kwargs={"k": 3}表示返回相似度最高的3个块。这个数字可以调整，太少可能信息不全，太多可能引入噪声并增加LLM的上下文长度负担。
提示词工程（Prompt Engineering）：我们定义了一个模板，明确要求LLM“根据上下文信息回答问题”，并指示它在信息不足时拒绝回答。这是减少LLM“幻觉”的关键。
大语言模型（LLM）选择：
- 本地模型（Ollama）：数据隐私性最好，零API成本。需要先在本地安装Ollama并拉取模型（如ollama pull qwen2:7b）。启动Ollama服务后，通过本地API（默认11434端口）调用。
- 云端API（OpenAI/智谱等）：效果通常更稳定强大，但会产生费用，且数据需传输到外部服务器。务必通过环境变量设置API密钥。
检索问答链（RetrievalQA）：chain_type="stuff"是最简单直接的方式，将所有检索到的上下文文本拼接后一次性送给LLM。对于较长的上下文，可以考虑map_reduce或refine等更复杂但能处理更长文本的链类型。

运行与验证：

如果使用Ollama：

# 首先确保Ollama服务运行，并拉取了模型 ollama pull qwen2:7b # 在另一个终端运行服务（如果还没运行） # ollama serve # 然后运行问答脚本 python rag_qa_system.py

如果使用OpenAI API：

# 设置API密钥 export OPENAI_API_KEY='your-api-key-here' # 修改 rag_qa_system.py 中的LLM初始化部分，注释掉Ollama，取消注释ChatOpenAI # 然后运行脚本 python rag_qa_system.py

启动后，系统会加载向量数据库。尝试问一个你文档中明确包含的问题，例如“XX产品的保修期是多久？”。观察它是否能返回正确答案，并列出答案的来源文档和页码。

5. 功能测试与效果验证

搭建完成后，需要通过系统性的测试来验证RAG系统的效果和稳定性。

5.1 基础问答准确性测试

准备一组测试问题，覆盖文档中的明确事实、需要归纳总结的内容以及文档外的知识。

# test_qa.py from rag_qa_system import RAGQASystem qa_system = RAGQASystem() test_questions = [ "文档中提到的项目启动时间是什么时候？", # 事实性问题 "请总结一下安全操作规程的要点。", # 归纳性问题 "火星上现在有多少人口？", # 文档外问题（应无法回答） ] for q in test_questions: print(f"\n=== 测试问题：{q} ===") answer, sources = qa_system.ask(q) print(f"答案：{answer}") if "无法回答" in answer or "没有提供" in answer: print("状态：✅ 正确拒答") elif sources: print(f"状态：✅ 基于文档回答 (参考源：{len(sources)}个)") else: print("状态：⚠️ 回答了但未提供来源，需检查")

成功标准：

事实性问题答案准确，且来源正确。
归纳性问题回答连贯，要点覆盖全面。
文档外问题能被系统明确拒绝，或声明信息不足，而不是胡编乱造。

5.2 检索相关性验证

有时答案不准，问题可能出在检索环节。我们需要检查系统检索到的文本块是否真的与问题相关。

# 在RAGQASystem类中添加一个方法 def retrieve_only(self, question, k=3): """仅执行检索，不调用LLM生成，用于调试检索效果""" docs = self.retriever.get_relevant_documents(question) return docs # 测试检索 debug_question = "如何申请休假？" retrieved_docs = qa_system.retrieve_only(debug_question) print(f"问题：'{debug_question}'") for i, doc in enumerate(retrieved_docs): print(f"\n[检索结果 {i+1}] 相关性分数（近似）: N/A") print(f" 来源：{doc.metadata.get('source')}") print(f" 内容预览：{doc.page_content[:300]}...")

检查检索到的文本片段是否确实包含“休假”、“申请”、“流程”等关键词。如果不相关，可能需要调整文本分割策略（chunk_size、chunk_overlap）或尝试不同的嵌入模型。

5.3 长文本与多轮对话压力测试

长文本输入：询问一个需要综合多个文档块信息才能回答的复杂问题。
多轮对话：进行连续提问，例如“上一段提到的那个功能，具体怎么操作？”。（注：基础的RetrievalQA链是无状态的，要实现多轮对话需要引入记忆机制，如ConversationalRetrievalChain）。

6. 封装为API服务与批量任务

要让其他应用调用，或者处理大量问题，我们需要将其服务化。

6.1 使用FastAPI封装RESTful API

# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from rag_qa_system import RAGQASystem import uvicorn app = FastAPI(title="RAG QA API Server") qa_system = RAGQASystem() # 启动时加载模型和向量库，注意内存占用 class QuestionRequest(BaseModel): question: str top_k: int = 3 # 可动态调整检索数量 class AnswerResponse(BaseModel): answer: str sources: list @app.post("/ask", response_model=AnswerResponse) async def ask_question(req: QuestionRequest): try: # 临时修改检索数量 qa_system.retriever.search_kwargs["k"] = req.top_k answer, source_docs = qa_system.ask(req.question) source_list = [{"content": doc.page_content[:500], "metadata": doc.metadata} for doc in source_docs] return AnswerResponse(answer=answer, sources=source_list) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health_check(): return {"status": "healthy"} if __name__ == "__main__": # 启动服务，监听本地7860端口 uvicorn.run(app, host="0.0.0.0", port=7860)

启动与调用：

# 启动API服务 python api_server.py # 服务将在 http://127.0.0.1:7860 运行

使用curl或Python测试API：

curl -X POST "http://127.0.0.1:7860/ask" \ -H "Content-Type: application/json" \ -d '{"question": "公司的年假有多少天？", "top_k": 2}'

6.2 批量问答任务

对于需要处理问题列表的场景，可以编写批量脚本。

# batch_qa.py import csv import json from rag_qa_system import RAGQASystem qa_system = RAGQASystem() input_file = "questions.csv" # 假设CSV文件，第一列是问题 output_file = "answers.jsonl" results = [] with open(input_file, 'r', encoding='utf-8') as f: reader = csv.reader(f) for row in reader: if not row: continue question = row[0].strip() if question: print(f"处理: {question}") answer, sources = qa_system.ask(question) result = { "question": question, "answer": answer, "sources": [{"metadata": doc.metadata, "preview": doc.page_content[:200]} for doc in sources] } results.append(result) # 保存为JSON Lines格式，每行一个结果 with open(output_file, 'w', encoding='utf-8') as f: for res in results: f.write(json.dumps(res, ensure_ascii=False) + '\n') print(f"批量处理完成，结果已保存至 {output_file}")

7. 资源占用与性能观察

运行RAG系统时，需要关注以下资源点：

内存与显存占用：
- 向量数据库（Chroma）：常驻内存，占用与存储的向量数量成正比。百万级向量可能需要数GB内存。
- 嵌入模型：加载BAAI/bge-small-zh模型，GPU显存占用约1-1.5GB，CPU内存占用约500MB。
- 大语言模型（如本地7B模型）：这是最大的资源消耗者。以Qwen2-7B为例，使用4-bit量化加载，GPU显存占用约5-6GB，纯CPU推理需要14GB以上内存。务必根据硬件条件选择模型。
- 观察命令：在Linux下可使用nvidia-smi（GPU）和htop（CPU/内存）实时监控。
响应时间：
- 检索阶段：通常在几十到几百毫秒，取决于向量库规模和硬件。
- 生成阶段：取决于LLM。本地7B模型在GPU上生成一段话可能需要2-10秒。API调用则受网络影响。
- 优化方向：使用更小的嵌入模型、对向量索引进行量化、为LLM使用更高效的推理框架（如vLLM）或降低生成参数（如max_tokens）。
磁盘空间：
- 向量数据库目录（chroma_db）会随着文档增多而变大。
- 模型文件（嵌入模型、LLM）是主要占用，提前规划好存储位置。

8. 常见问题与排查方法

在搭建和运行过程中，你可能会遇到以下问题：

问题现象	可能原因	排查方式	解决方案
运行`build_knowledge_base.py`时下载模型失败或极慢	网络连接HuggingFace不畅	检查网络，观察下载进度	1. 配置国内镜像源。 2. 手动下载模型文件到本地，修改代码指定本地路径。
导入`langchain`或`chromadb`时报错	依赖版本冲突或未安装	检查`pip list`，确认已安装正确版本	1. 在干净的虚拟环境中重新安装。 2. 严格按照提供的`requirements.txt`版本安装。
问答时返回“根据提供的信息无法回答此问题”，但文档中明明有答案	1. 检索失败，没找到相关文本块。 2. 检索到了但LLM没理解或没遵从指令。	1. 使用`retrieve_only`方法检查检索结果。 2. 查看检索到的原文片段。	1. 调整文本分割参数（增大`chunk_size`或`chunk_overlap`）。 2. 尝试不同的嵌入模型。 3. 优化提示词模板，更明确地要求“必须基于上下文”。
答案看起来是胡编乱造的（幻觉）	1. 检索到的上下文不相关或太弱。 2. LLM本身“幻觉”倾向强。 3. 提示词约束力不够。	1. 同上一问题，检查检索。 2. 测试一个文档外问题，看LLM是否老实拒答。	1. 提高检索质量（见上）。 2. 在提示词中加入更强烈的约束，如“仅使用上下文中的事实”。 3. 尝试换一个“更听话”的LLM。
使用Ollama时连接失败	Ollama服务未启动或模型未加载	检查Ollama服务状态`ollama list`	1. 确保已运行`ollama serve`。 2. 确保已拉取所需模型`ollama pull qwen2:7b`。 3. 检查`rag_qa_system.py`中的`base_url`是否正确。
API服务启动后无法访问	端口被占用或防火墙限制	检查端口`7860`是否被其他进程占用`lsof -i:7860`	1. 在代码中更换端口号。 2. 关闭占用端口的进程。
处理大量文档时内存/显存不足	1. 同时加载的文档太大。 2. 模型未量化，占用过高。	监控系统资源使用情况。	1. 分批处理文档，而不是一次性全部加载。 2. 使用CPU模式或量化版本的模型。 3. 增加硬件资源或使用云服务。

9. 最佳实践与使用建议

为了让你的RAG系统更健壮、易用，遵循以下建议：

文档预处理是关键：垃圾进，垃圾出。在向量化之前，尽量清洗文档格式（去除页眉页脚、无关字符），进行必要的文本规范化。
分步调试：不要一次性处理所有文档。先用一两篇小文档测试整个流程，确保每个环节（加载、分割、嵌入、检索、生成）都工作正常。
版本化管理：对代码、配置文件、提示词模板进行版本控制（如Git）。当更换模型或调整参数时，能清晰地回溯和对比效果。
效果评估：建立一个小型测试集（Q&A对），定期运行，量化评估系统的准确率、召回率等指标。这是迭代优化的基础。
关注安全与合规：
- 数据隔离：为不同部门或项目使用独立的向量数据库。
- 访问控制：API服务应部署在内网，或增加API密钥认证。
- 审计日志：记录所有的问答请求和结果，便于追踪和审计。
探索进阶特性：当基础系统跑通后，可以研究：
- 重排序（Re-ranking）：在初步检索后，使用一个更精细的模型对结果进行重排，提升Top结果的准确性。
- 多检索器混合：结合关键词检索（如BM25）和向量检索，取长补短。
- Agentic RAG：让RAG系统具备调用工具、自主规划多步查询的能力。