从零搭建RAG系统：用Python实现大模型私有知识库检索增强生成

1. 项目缘起：为什么你的大模型需要“外接大脑”？

最近跟不少做AI应用的朋友聊天，发现一个挺普遍的现象：大家兴致勃勃地部署好一个开源大模型，比如Qwen或者Llama，准备让它帮忙处理公司内部的文档、回答产品问题，结果第一轮测试就傻眼了。你问它“我们公司2024年第三季度的销售策略核心是什么？”，它要么开始一本正经地胡说八道，编造一些根本不存在的会议和数字；要么就礼貌地表示“作为一个AI模型，我无法获取贵公司的内部信息”。这感觉就像请了一位博古通今的大学教授，但他对你家书架上的书却一本都没看过。

这就是当前大模型面临的核心困境：它们拥有强大的通用知识和语言生成能力，但缺乏对特定、私有、最新信息的访问权限。模型训练时用的数据有截止日期，且不可能包含你公司的内部wiki、产品手册、客户邮件或者昨天的会议纪要。直接微调（Fine-tuning）模型来学习这些新知识呢？成本高、周期长，每次信息更新都得重新训练，完全不现实。

于是，RAG（Retrieval-Augmented Generation，检索增强生成）技术就成了解决这个问题的“银弹”。它的核心思想非常直观：不给模型换“脑”（不重新训练），而是给它配一个强大的“外部记忆库”和一位高效的“图书管理员”。当模型需要回答问题时，这位“图书管理员”会先快速地从你的私有知识库（比如一堆PDF、Word、网页）中检索出最相关的文档片段，然后把这些片段作为上下文，连同问题一起交给大模型，让它基于这些确凿的证据来生成答案。

这么做的好处是立竿见影的：

1. 答案准确性大幅提升：答案来源于你的真实文档，极大减少了模型“幻觉”（胡编乱造）。
2. 知识更新成本极低：只需要往“外部记忆库”里添加新文档，无需动模型本身。
3. 答案可追溯：你可以清楚地知道模型是基于哪份文档的哪段话得出的结论，增强了可信度。
4. 降低模型门槛：即使是一个参数量较小的“轻量级”模型，在优质上下文的辅助下，也能完成专业领域的问答。

所以，今天我们就抛开那些复杂的框架和云服务，从最底层开始，用Python亲手搭建一个属于你自己的RAG系统。你会清晰地看到，一个RAG系统是如何由数据准备、向量检索、提示工程这几个核心环节像搭积木一样组合起来的。我们将使用完全本地、可掌控的技术栈，让你彻底理解其工作原理，并能根据自身需求灵活定制。

2. 核心组件拆解：一个RAG系统由哪些部分构成？

在开始写代码之前，我们必须像建筑师看蓝图一样，先理解RAG系统的整体架构。一个典型的、可运行的RAG流程，主要包含以下五个核心环节，它们环环相扣：

2.1 文档加载与解析（Document Loading & Parsing）这是数据处理的入口。你的知识可能存在于各种格式的文件中：PDF报告、Word文档、Markdown笔记、HTML网页，甚至是数据库里的记录。这一步的任务就是把这些不同格式的“原材料”统一加载进来，并解析出其中纯文本和元数据（如标题、作者、页码）。例如，一个PDF文件，你需要能提取出它的文字内容，并知道某段文字来自第几页。

2.2 文本分割（Text Splitting / Chunking）这是至关重要且容易被忽视的一步。你不能把一整本100页的产品手册直接扔给模型。原因有二：第一，大模型有上下文长度限制（如4096、8192个token），装不下；第二，信息密度不均，检索时需要精准定位。因此，我们需要把长文档切割成大小合适的“文本块”（Chunks）。 分割策略很有讲究：

• 固定大小分割：简单，但可能把一个完整的句子或段落从中间切断。
• 基于分隔符分割：按段落、标题等自然分隔符切割，能更好地保持语义完整性。
• 智能重叠分割：在分割时，让相邻的块有一小部分重叠（例如100个字符）。这能防止关键信息恰好落在两个块的边界而被割裂，确保检索时上下文连贯。

2.3 文本嵌入与向量化（Text Embedding & Vectorization）这是实现“智能检索”的魔法所在。我们需要把上一步得到的文本块，转换成计算机能理解的“语义”。嵌入模型（Embedding Model）就是这个翻译官，它把一个句子或段落，转换成一个固定长度的、高维度的数字向量（比如384维或768维的一串数字）。这个向量的神奇之处在于：语义相近的文本，其向量在空间中的距离（通常用余弦相似度衡量）也会很近。例如，“如何更换轮胎”和“汽车轮胎拆卸步骤”这两个句子的向量就会非常接近。我们将所有文本块的向量存储起来，就构建了一个“向量数据库”的雏形。

2.4 向量存储与检索（Vector Storage & Retrieval）我们需要一个专门的地方来高效存储和查询这些向量，这就是向量数据库（Vector Database）。它不同于传统的关系型数据库（如MySQL），其核心能力是进行“近似最近邻搜索”（Approximate Nearest Neighbor, ANN），即快速从百万甚至千万级向量中，找到与问题向量最相似的那几个。 当用户提出一个问题时，我们先用同样的嵌入模型把问题也转换成向量，然后向向量数据库发起查询：“请找出和这个‘问题向量’最相似的Top K个文本块向量”。向量数据库会高效地返回最相关的几个文本块及其原始内容。

2.5 提示构建与生成（Prompt Construction & Generation）这是最后一步，也是点睛之笔。我们拿到了最相关的文本块（作为证据或上下文），以及用户的原始问题。现在需要构造一个清晰的指令（即提示词Prompt），交给大模型去生成最终答案。 一个典型的RAG提示词模板长这样：

请基于以下提供的上下文信息，回答用户的问题。如果上下文中的信息不足以回答问题，请直接说“根据已知信息无法回答该问题”，不要编造信息。 上下文： {context_text_1} {context_text_2} ... {context_text_k} 问题：{user_question} 请用中文给出专业、清晰的回答：

然后，我们将这个精心构造的提示词，发送给大模型（如通过OpenAI API、或本地部署的Ollama服务），模型就会基于我们提供的“证据”生成最终答案。

理解了这五个环节，我们就有了清晰的施工图。接下来，我们就用Python，从零开始，把这些环节一个一个实现出来。

3. 实战环境搭建与工具选型

工欲善其事，必先利其器。我们选择工具的核心原则是：轻量、本地化、可控、易于理解。这样能让我们聚焦于RAG流程本身，而不是陷入复杂的云服务配置中。

3.1 Python环境与包管理首先确保你有一个Python环境（3.8以上版本）。强烈建议使用虚拟环境来隔离项目依赖，避免包冲突。

# 创建并激活虚拟环境（以venv为例） python -m venv rag_env # Windows rag_env\Scripts\activate # macOS/Linux source rag_env/bin/activate

3.2 核心库安装与选型理由我们将通过pip安装一系列核心库，每个库都承担着RAG流水线中的一个特定角色。

pip install langchain langchain-community langchain-chroma pypdf python-dotenv sentence-transformers tiktoken

下面我来逐一解释为什么选它们：

• langchain&langchain-community：这不是必须的，但对于快速构建原型和串联流程非常有帮助。LangChain是一个框架，它把文档加载、分割、嵌入、检索、提示等环节抽象成了标准的“链”（Chain）。我们用它来快速搭建流程，理解各个环节的接口。但请注意，我们不会完全依赖它的“黑盒”，关键步骤我们会拆开看其原理。
• langchain-chroma：这是Chroma向量数据库的LangChain集成包。Chroma是一个轻量级、开源、可嵌入的向量数据库，非常适合本地开发和实验，无需单独部署服务。
• pypdf：一个纯Python的PDF解析库，用于从PDF文件中提取文本。相比某些依赖外部工具（如poppler）的库，它更干净。
• python-dotenv：用于管理环境变量。如果后续我们用到需要API Key的在线模型（如OpenAI的嵌入模型），可以用它来安全地加载密钥。
• sentence-transformers：这是本次项目的核心之一。它提供了数百种预训练的句子嵌入模型。我们将使用一个轻量级且效果不错的双语模型paraphrase-multilingual-MiniLM-L12-v2，它支持中文，并且可以在CPU上运行，无需GPU。
• tiktoken：OpenAI开源的快速BPE分词器。我们用它来精确计算文本的token长度，这对于控制文本分割和符合大模型上下文窗口至关重要。

3.3 备用方案与扩展思考我们的选择是基于“最小可行”和“完全本地”的原则。当你需要扩展时，可以有以下选择：

• 文档加载器：除了PDF，langchain-community还提供了对Word、Markdown、HTML、甚至Notion、Confluence的加载器支持。
• 嵌入模型：如果你有GPU且追求更高精度，可以选用更大的模型如bge-large-zh-v1.5。如果追求极致的本地化且不介意牺牲一些精度，可以用text2vec等纯本地模型。如果网络允许且不计成本，OpenAI的text-embedding-3-small是云端非常强大的选择。
• 向量数据库：除了Chroma，生产环境常考虑Milvus、Qdrant、Weaviate等，它们支持分布式、持久化、更丰富的过滤查询。但对于从零开始学习，Chroma的简单易用是无与伦比的。
• 大模型：生成部分，为了完全本地化，我们可以使用Ollama在本地运行Qwen2.5:7b或Llama3.2等模型。本文为了流程清晰，将先使用模拟的生成步骤，但会给出集成Ollama的真实代码示例。

环境准备好后，我们创建一个项目目录，比如my_rag_project，并在里面开始我们的代码之旅。

4. 第一步：将你的私有文档“喂”给系统

假设我们有一个名为knowledge_base的文件夹，里面存放着公司的几份PDF产品手册和一份Markdown格式的Q&A文档。我们的任务就是把这些文档“吃进去”，并处理好。

4.1 实现一个通用的文档加载器我们不希望为每种文件类型写不同的代码。利用LangChain的文档加载器抽象，我们可以轻松实现一个统一加载接口。

# file_loader.py import os from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader from langchain.schema import Document from typing import List def load_documents_from_directory(directory_path: str) -> List[Document]: """ 从指定目录加载所有支持格式的文档。 返回一个包含所有文档内容的Document对象列表。 """ documents = [] supported_extensions = { '.pdf': PyPDFLoader, '.txt': TextLoader, '.md': UnstructuredMarkdownLoader, # 可以继续扩展，如 '.docx': DocxLoader } for root, _, files in os.walk(directory_path): for file in files: file_ext = os.path.splitext(file)[1].lower() if file_ext in supported_extensions: file_path = os.path.join(root, file) print(f"正在加载文件: {file_path}") try: loader_class = supported_extensions[file_ext] # 注意：有些Loader可能需要额外参数，这里做简单处理 if loader_class == PyPDFLoader: loader = loader_class(file_path) else: loader = loader_class(file_path, mode="elements") # 对于Markdown等，可能需指定模式 loaded_docs = loader.load() # 为每个文档片段添加来源元数据 for doc in loaded_docs: doc.metadata["source"] = file_path doc.metadata["page"] = doc.metadata.get("page", 0) # PDF有页码，其他格式默认0 documents.extend(loaded_docs) except Exception as e: print(f"加载文件 {file_path} 时出错: {e}") print(f"共加载了 {len(documents)} 个文档片段。") return documents # 使用示例 if __name__ == "__main__": docs = load_documents_from_directory("./knowledge_base") for i, doc in enumerate(docs[:2]): # 打印前两个片段看看 print(f"片段 {i} 内容 (前200字符): {doc.page_content[:200]}...") print(f"元数据: {doc.metadata}\n")

注意：不同的文档解析库质量参差不齐。对于复杂的PDF（特别是扫描版或特殊排版），PyPDF可能提取效果不佳，可以考虑pdfplumber或pymupdf。Unstructured库功能强大但更重。在实际项目中，你可能需要根据文档质量混合使用多种加载器。

4.2 智能文本分割：为什么“怎么切”比“切多少”更重要加载上来的是原始文档流，现在我们需要把它切成块。前面提到，简单的按字符数切割会破坏语义。我们来实现一个更聪明的、带重叠的分割器。

# text_splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter import tiktoken # 用于精确计算token数 def create_text_splitter(chunk_size=500, chunk_overlap=50): """ 创建一个递归字符文本分割器。 参数： chunk_size: 每个文本块的目标大小（单位：字符）。这是一个软限制。 chunk_overlap: 相邻块之间的重叠字符数。用于保持上下文连贯。 返回：配置好的分割器对象。 """ # 首先，我们定义一个分隔符优先级列表。分割器会按这个顺序尝试分割。 separators = ["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] # 使用递归分割器，它会尽量按分隔符来保持语义完整。 text_splitter = RecursiveCharacterTextSplitter( separators=separators, chunk_size=chunk_size, chunk_overlap=chunk_overlap, length_function=len, # 这里用字符长度，简单。生产环境可用tiktoken计算token。 # 如果你想用token长度（更准确对应LLM上下文），可以这样： # length_function=lambda text: len(tiktoken.get_encoding("cl100k_base").encode(text)), # chunk_size和chunk_overlap也需要调整为token数，如 chunk_size=800（token） ) return text_splitter def split_documents(documents: List[Document], text_splitter) -> List[Document]: """ 将文档列表分割成更小的文本块。 """ print("开始分割文档...") all_splits = text_splitter.split_documents(documents) print(f"分割完成，共得到 {len(all_splits)} 个文本块。") # 打印一个示例块，检查分割效果 if all_splits: sample = all_splits[0] print(f"\n示例文本块 (前300字符):\n{sample.page_content[:300]}...") print(f"该块元数据: {sample.metadata}") print(f"该块长度（字符）: {len(sample.page_content)}") return all_splits # 在main.py中整合使用 if __name__ == "__main__": from file_loader import load_documents_from_directory raw_docs = load_documents_from_directory("./knowledge_base") splitter = create_text_splitter(chunk_size=500, chunk_overlap=50) chunked_docs = split_documents(raw_docs, splitter)

4.3 关键参数调优心得

• chunk_size（块大小）：这不是越大越好。它需要与你选用的大模型的上下文窗口以及你的问题复杂度匹配。如果你的问题通常需要综合多个段落的信息来回答，块可以稍大（如800-1000字符）。如果问题非常具体，指向性明确，较小的块（如300-500字符）能让检索更精准。一个经验法则是：块大小应能容纳一个完整的“思想单元”，比如一个概念解释、一个操作步骤。
• chunk_overlap（重叠大小）：通常设置为chunk_size的10%-20%。重叠部分就像一个“缓冲区”，确保关键信息不会因为恰好落在边界而被丢失。例如，一个重要的定义可能在一段的末尾开始，在下一段的开头结束，重叠就能把它保留在同一个检索上下文中。
• 分隔符顺序：我们定义的["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]是中文环境下比较合理的顺序。它优先按空行、换行、句号等语义强边界分割，最后才按空格和字符分割，最大程度保持句子和段落的完整性。

完成这一步后，你的原始文档就变成了一系列干净、大小适中、带有来源信息的文本块，为下一步的“向量化”做好了准备。

5. 第二步：构建语义搜索引擎——向量数据库

这是RAG系统的“记忆核心”。我们需要把上一步得到的文本块，转换成向量，并存储到一个能快速进行相似性搜索的数据库里。

5.1 选择与初始化嵌入模型我们使用sentence-transformers库中的轻量级双语模型。它能在CPU上良好运行，并且对中文有不错的支持。

# embedding_model.py from sentence_transformers import SentenceTransformer import numpy as np class LocalEmbeddingModel: def __init__(self, model_name='paraphrase-multilingual-MiniLM-L12-v2'): """ 初始化本地嵌入模型。 首次运行时会从Hugging Face Hub下载模型，请确保网络通畅。 """ print(f"正在加载嵌入模型: {model_name}， 首次加载可能需要几分钟...") self.model = SentenceTransformer(model_name) # 获取模型的向量维度，后续存入数据库需要知道 self.dimension = self.model.get_sentence_embedding_dimension() print(f"模型加载完成，向量维度: {self.dimension}") def embed_documents(self, texts: List[str]) -> np.ndarray: """将文本列表编码为向量矩阵。""" print(f"正在为 {len(texts)} 个文本块生成嵌入向量...") # model.encode 返回 numpy.ndarray, 形状为 (num_texts, dimension) embeddings = self.model.encode(texts, convert_to_numpy=True, normalize_embeddings=True, # 归一化，方便用余弦相似度 show_progress_bar=True) print("嵌入向量生成完毕。") return embeddings def embed_query(self, query: str) -> np.ndarray: """将单个查询文本编码为向量。""" # 注意：查询和文档应使用相同的模型和参数进行编码，以确保向量空间一致。 return self.model.encode([query], convert_to_numpy=True, normalize_embeddings=True)[0] # 取第一个也是唯一一个结果 # 使用示例 if __name__ == "__main__": embedder = LocalEmbeddingModel() sample_texts = ["如何更换汽车轮胎？", "汽车轮胎拆卸步骤详解。", "今天的天气真好。"] embeddings = embedder.embed_documents(sample_texts) print(f"向量形状: {embeddings.shape}") # 例如 (3, 384) # 计算前两个句子的余弦相似度 from numpy import dot from numpy.linalg import norm cos_sim = dot(embeddings[0], embeddings[1]) / (norm(embeddings[0]) * norm(embeddings[1])) print(f"句子1和2的余弦相似度: {cos_sim:.4f}") # 预期应该很高 cos_sim_2 = dot(embeddings[0], embeddings[2]) / (norm(embeddings[0]) * norm(embeddings[2])) print(f"句子1和3的余弦相似度: {cos_sim_2:.4f}") # 预期应该很低

实操心得：normalize_embeddings=True这个参数非常重要。它将向量归一化为单位长度，此时余弦相似度计算简化为向量点积，计算效率更高，并且是许多向量数据库默认的相似度计算方式。

5.2 初始化并持久化向量数据库（Chroma）我们将使用Chroma，它可以直接将数据持久化到本地磁盘，下次启动时无需重新计算嵌入。

# vector_store.py import chromadb from chromadb.config import Settings from langchain_chroma import Chroma from typing import List import hashlib import os def create_or_load_vector_store(chunked_documents: List[Document], embedding_model, persist_directory: str = "./chroma_db"): """ 创建或加载一个Chroma向量存储。 如果指定目录已存在数据，则加载；否则，创建新的存储并添加文档。 """ # 提取纯文本和元数据 texts = [doc.page_content for doc in chunked_documents] metadatas = [doc.metadata for doc in chunked_documents] # 为每个文本块生成一个唯一ID（基于内容哈希），避免重复插入 ids = [hashlib.md5((text + str(metadata.get("source", ""))).encode()).hexdigest()[:20] for text, metadata in zip(texts, metadatas)] # 配置Chroma客户端，设置持久化目录 client_settings = Settings( chroma_db_impl="duckdb+parquet", # 使用DuckDB引擎，Parquet格式存储，性能好 persist_directory=persist_directory, anonymized_telemetry=False # 禁用匿名遥测 ) # 使用LangChain的Chroma包装器，方便集成 vectorstore = Chroma.from_texts( texts=texts, embedding=embedding_model, # 这里需要传入一个LangChain兼容的Embedding对象，稍后适配 metadatas=metadatas, ids=ids, client_settings=client_settings, collection_name="my_knowledge_collection", persist_directory=persist_directory ) print(f"向量数据库已初始化/加载，存储路径: {os.path.abspath(persist_directory)}") return vectorstore # 我们需要创建一个适配器，让我们的LocalEmbeddingModel符合LangChain的Embedding接口 from langchain.embeddings.base import Embeddings class LocalEmbeddingsAdapter(Embeddings): """将我们的LocalEmbeddingModel适配成LangChain的Embeddings接口。""" def __init__(self, model: LocalEmbeddingModel): self.model = model def embed_documents(self, texts: List[str]) -> List[List[float]]: # 返回List[List[float]]，这是LangChain期望的格式 numpy_embeddings = self.model.embed_documents(texts) return numpy_embeddings.tolist() def embed_query(self, text: str) -> List[float]: numpy_embedding = self.model.embed_query(text) return numpy_embedding.tolist() # 在main.py中整合 if __name__ == "__main__": from file_loader import load_documents_from_directory from text_splitter import create_text_splitter, split_documents from embedding_model import LocalEmbeddingModel from vector_store import LocalEmbeddingsAdapter, create_or_load_vector_store # 1. 加载并分割文档 raw_docs = load_documents_from_directory("./knowledge_base") splitter = create_text_splitter(chunk_size=500, chunk_overlap=50) chunks = split_documents(raw_docs, splitter) # 2. 初始化嵌入模型和适配器 local_model = LocalEmbeddingModel() embeddings_adapter = LocalEmbeddingsAdapter(local_model) # 3. 创建/加载向量存储 vectorstore = create_or_load_vector_store(chunks, embeddings_adapter, "./chroma_db") # 4. 测试检索功能 test_query = "我们产品的主要优势是什么？" print(f"\n测试查询: '{test_query}'") # 使用 similarity_search_with_score 可以同时返回文档和相似度分数 results = vectorstore.similarity_search_with_score(test_query, k=3) # 返回最相关的3个结果 for i, (doc, score) in enumerate(results): print(f"\n--- 结果 {i+1} (相似度分数: {score:.4f}) ---") print(f"内容片段: {doc.page_content[:200]}...") print(f"来源: {doc.metadata.get('source', 'N/A')}")

运行这段代码，你会看到系统首先加载模型（第一次需要下载），然后为每个文本块计算向量，最后存入本地的chroma_db文件夹。下次再运行，如果文档没有变化，它会直接加载已有的数据库，速度飞快。

5.3 向量检索的幕后原理与调优当你调用similarity_search_with_score时，Chroma在背后做了什么？

1. 编码查询：用同样的嵌入模型把你的问题转换成向量。
2. 近似最近邻搜索：在数百万个向量中，快速找到与查询向量最相似的K个。它可能使用HNSW（Hierarchical Navigable Small World）等索引算法来加速，而不是暴力计算所有距离。
3. 返回结果：返回Top K的文本块及其相似度分数（通常是余弦相似度，值越接近1越相似）。

关键参数k：它决定了返回多少个相关文本块。k太小，可能遗漏关键信息；k太大，会给大模型带来无关噪音，增加成本并可能干扰判断。通常从3-5开始尝试，根据答案质量调整。对于复杂问题，可能需要更大的k（如8-10）。

至此，你的私有知识库的“搜索引擎”已经构建完毕。它已经能够理解你文档的语义，并根据问题找到最相关的段落。

6. 第三步：让大模型基于证据“开口说话”

现在，我们有了问题，也有了从知识库中检索到的相关证据（上下文）。最后一步，就是构造一个清晰的指令，让大模型扮演一个“基于给定材料回答问题”的专家角色。

6.1 设计一个高效的提示词模板提示词（Prompt）是引导大模型行为的关键。一个结构良好的RAG提示词应包含以下几个部分：

1. 角色指令：明确告诉模型它应该扮演什么角色。
2. 任务说明：清晰说明它需要完成什么任务。
3. 上下文提供：清晰地标注出我们提供的背景材料。
4. 问题：用户的具体问题。
5. 回答格式与约束：规定回答的格式、语言、以及当信息不足时的行为。

# prompt_builder.py def build_rag_prompt(context_texts: List[str], user_question: str) -> str: """ 构建RAG提示词。 参数： context_texts: 检索到的相关文本块列表。 user_question: 用户原始问题。 返回：构造好的完整提示字符串。 """ # 将多个上下文文本块合并，用明显的分隔符隔开 context_str = "\n\n---\n\n".join(context_texts) prompt_template = f"""你是一个专业的助手，严格根据我提供的背景资料来回答问题。 请仔细阅读以下资料，这些资料来自可信的内部文档： 【背景资料开始】 {context_str} 【背景资料结束】 现在，请基于以上背景资料，回答下面的问题。 如果背景资料中没有足够的信息来回答问题，请直接说“根据提供的资料，我无法回答这个问题”，不要编造任何信息。 问题：{user_question} 请用中文给出清晰、准确、专业的回答： """ return prompt_template # 示例 if __name__ == "__main__": sample_contexts = [ "本公司产品A采用了最新的纳米涂层技术，防水等级达到IP68。", "产品A的电池续航在标准模式下为72小时，节能模式下可达120小时。" ] sample_question = "产品A的防水等级和电池续航分别是多少？" prompt = build_rag_prompt(sample_contexts, sample_question) print("生成的提示词示例：\n") print(prompt)

这个模板有几个设计要点：

• 清晰的边界：用【背景资料开始/结束】明确标定了上下文的范围，防止模型混淆。
• 强硬的约束：明确要求模型“严格根据资料”，并在信息不足时“直接说无法回答”。这能有效抑制幻觉。
• 结构化输出：要求“用中文给出清晰、准确、专业的回答”，引导模型输出格式。

6.2 集成本地大模型（以Ollama为例）为了完全本地化，我们使用Ollama来运行一个开源大模型。假设你已经在本地安装并运行了Ollama，并且拉取了qwen2.5:7b模型。

# 在终端中启动Ollama服务（如果尚未运行），并拉取模型 # ollama pull qwen2.5:7b # Ollama服务默认运行在 http://localhost:11434

然后，我们在Python中通过HTTP请求调用它。

# llm_client.py import requests import json class OllamaClient: def __init__(self, base_url="http://localhost:11434", model="qwen2.5:7b"): self.base_url = base_url self.model = model self.generate_endpoint = f"{base_url}/api/generate" def generate(self, prompt: str, temperature=0.1, max_tokens=1024) -> str: """ 向Ollama发送生成请求。 参数： temperature: 温度参数，控制随机性。越低（接近0）答案越确定、保守，适合事实问答。 max_tokens: 生成的最大token数。 """ payload = { "model": self.model, "prompt": prompt, "stream": False, # 非流式响应，一次性返回 "options": { "temperature": temperature, "num_predict": max_tokens } } try: response = requests.post(self.generate_endpoint, json=payload, timeout=60) response.raise_for_status() # 检查HTTP错误 result = response.json() return result.get("response", "").strip() except requests.exceptions.RequestException as e: print(f"调用Ollama API失败: {e}") return f"模型调用出错: {e}" except json.JSONDecodeError as e: print(f"解析Ollama响应失败: {e}") return "模型响应解析失败。" # 使用示例 if __name__ == "__main__": from prompt_builder import build_rag_prompt client = OllamaClient(model="qwen2.5:7b") test_contexts = ["..."] # 你的上下文 test_question = "..." test_prompt = build_rag_prompt(test_contexts, test_question) print("正在生成回答...") answer = client.generate(test_prompt, temperature=0.1) print(f"\n模型回答：\n{answer}")

6.3 温度参数（Temperature）的实战意义在调用大模型时，temperature是一个至关重要的参数。它控制了模型生成文本的随机性。

• temperature=0.1：非常低，模型输出高度确定，倾向于选择概率最高的下一个词。这非常适合事实性问答、总结、基于明确上下文的生成，能最大程度保证答案的稳定性和准确性，减少“胡言乱语”。
• temperature=0.7~0.9：中等偏高，输出更有创造性、多样性。适合写故事、诗歌、头脑风暴等需要创意的任务。 在RAG场景下，我们强烈建议使用较低的temperature（如0.1-0.3），因为我们希望模型严格遵从我们提供的证据，而不是自由发挥。

7. 第四步：组装完整流程与效果优化

现在，我们把前三个步骤的代码像管道一样连接起来，形成一个完整的、端到端的RAG问答系统。

7.1 构建完整的RAG流水线创建一个主程序，串联所有模块。

# main.py import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from file_loader import load_documents_from_directory from text_splitter import create_text_splitter, split_documents from embedding_model import LocalEmbeddingModel from vector_store import LocalEmbeddingsAdapter, create_or_load_vector_store from prompt_builder import build_rag_prompt from llm_client import OllamaClient class RAGPipeline: def __init__(self, knowledge_dir="./knowledge_base", persist_dir="./chroma_db"): self.knowledge_dir = knowledge_dir self.persist_dir = persist_dir self.vectorstore = None self.llm_client = OllamaClient() self._initialize_pipeline() def _initialize_pipeline(self): """初始化流程：加载文档、分割、创建向量库。如果向量库已存在，则跳过前两步。""" # 检查向量库是否已存在 if os.path.exists(self.persist_dir) and os.listdir(self.persist_dir): print("检测到已存在的向量数据库，直接加载...") # 这里需要重新初始化嵌入模型适配器用于加载 local_model = LocalEmbeddingModel() embeddings_adapter = LocalEmbeddingsAdapter(local_model) # 注意：直接加载需要知道collection name和embedding function # 为了简化，我们使用一个假设已存在的加载函数。实际中，Chroma.from_persistent_directory更合适。 # 我们重构一下vector_store.py，提供一个加载函数。 from vector_store import load_existing_vector_store self.vectorstore = load_existing_vector_store(self.persist_dir, embeddings_adapter) else: print("未找到向量数据库，开始构建...") # 1. 加载文档 raw_docs = load_documents_from_directory(self.knowledge_dir) if not raw_docs: print("错误：未在指定目录下找到任何文档。") return # 2. 分割文档 splitter = create_text_splitter(chunk_size=500, chunk_overlap=50) chunks = split_documents(raw_docs, splitter) # 3. 初始化嵌入模型和适配器 local_model = LocalEmbeddingModel() embeddings_adapter = LocalEmbeddingsAdapter(local_model) # 4. 创建向量存储 self.vectorstore = create_or_load_vector_store(chunks, embeddings_adapter, self.persist_dir) print("RAG流水线初始化完成。") def ask(self, question: str, k=4) -> str: """核心问答函数。""" if self.vectorstore is None: return "流水线未正确初始化。" # 1. 检索 print(f"\n>>> 正在检索与问题相关的信息 (k={k})...") retrieved_docs_with_scores = self.vectorstore.similarity_search_with_score(question, k=k) if not retrieved_docs_with_scores: return "未在知识库中找到相关信息。" retrieved_texts = [doc.page_content for doc, _ in retrieved_docs_with_scores] print(f"检索到 {len(retrieved_texts)} 个相关片段。") # 2. 构建提示词 prompt = build_rag_prompt(retrieved_texts, question) # （可选）打印提示词用于调试 # print("\n--- 发送给模型的提示词（前500字符）---") # print(prompt[:500] + "...\n") # 3. 调用大模型生成 print(">>> 正在生成回答...") answer = self.llm_client.generate(prompt, temperature=0.1, max_tokens=1024) # 4. （可选）返回检索到的来源，用于引用和验证 sources = [{"content": doc.page_content[:150], "source": doc.metadata.get("source"), "score": score} for doc, score in retrieved_docs_with_scores] return answer, sources # 在vector_store.py中补充加载函数 def load_existing_vector_store(persist_directory: str, embedding_function): """加载已存在的Chroma向量存储。""" from langchain_chroma import Chroma vectorstore = Chroma( persist_directory=persist_directory, embedding_function=embedding_function, collection_name="my_knowledge_collection" ) print(f"从 {persist_directory} 加载了已有的向量数据库。") return vectorstore if __name__ == "__main__": # 初始化流水线 print("="*50) print("正在启动RAG问答系统...") print("="*50) pipeline = RAGPipeline(knowledge_dir="./my_docs", persist_dir="./chroma_db") # 交互式问答循环 print("\n系统已就绪！请输入你的问题（输入 'quit' 或 '退出' 结束）：") while True: user_input = input("\n你的问题: ").strip() if user_input.lower() in ['quit', '退出', 'exit']: print("再见！") break if not user_input: continue answer, sources = pipeline.ask(user_input) print(f"\n--- 回答 ---\n{answer}\n") print("--- 参考来源 (前3个) ---") for i, src in enumerate(sources[:3]): print(f"[{i+1}] 相似度: {src['score']:.3f} | 来源: {os.path.basename(src['source'])}") print(f" 片段: {src['content']}...\n")

运行这个main.py，你就拥有了一个本地的、基于私有知识库的智能问答系统！第一次运行会构建向量库，之后每次启动都是秒级加载。

7.2 效果优化与高级技巧一个基础的RAG系统跑通了，但答案质量可能还不尽如人意。以下是几个立竿见影的优化方向：

1. 检索优化：超越简单的相似度搜索

• 重排序（Re-ranking）：初步检索返回的Top K个结果，可能不是最相关的。可以引入一个更精细但更耗时的“交叉编码器”模型，对候选结果进行重新打分和排序，把最相关的排到最前面。例如，使用cross-encoder/ms-marco-MiniLM-L-6-v2模型。
• 元数据过滤：在检索时加入过滤条件。例如，只检索来自“产品手册.pdf”的文档，或者只检索“2024年”的文档。这需要你在分割文档时，把更多的元信息（如文档类型、年份、部门）存入向量数据库。
• 混合搜索（Hybrid Search）：结合关键词搜索（如BM25）和向量搜索。有些问题用关键词匹配更准（如产品型号“ABC-123”），有些用语义搜索更准（如“如何解决开机慢的问题”）。将两者的结果融合，能提高召回率。

2. 提示词工程优化

• 少样本示例（Few-Shot）：在提示词中给模型一两个“问题-答案”的例子，教它如何利用上下文。例如：

  示例1： 背景资料：...（资料1）... 问题：...（问题1）... 答案：...（基于资料1的答案）... 示例2： 背景资料：...（资料2）... 问题：...（问题2）... 答案：...（基于资料2的答案）... 现在请根据以下背景资料回答新问题： 【背景资料开始】...【背景资料结束】 问题：{user_question} 答案：

• 分步思考（Chain-of-Thought）：对于复杂问题，要求模型先推理再回答。例如：“请先列出背景资料中提到的所有相关要点，然后综合这些要点给出最终答案。”
• 指定回答格式：如果需要结构化输出，可以明确要求：“请以要点列表的形式回答。”

3. 后处理与评估

• 答案验证：对于关键事实，可以让模型同时输出引用的原文片段在上下文中的位置，方便人工核对。
• 设置置信度阈值：如果检索到的所有片段与问题的相似度分数都低于某个阈值（如0.7），可以直接返回“信息不足”，而不是让模型基于弱相关上下文强行生成，这能减少幻觉。

7.3 常见问题排查（踩坑实录）在搭建过程中，你可能会遇到以下问题：

问题1：检索到的内容完全不相关。

• 检查嵌入模型：你用的嵌入模型是否支持中文？对专业术语的编码效果如何？可以先用embedding_model.embed_query计算几个相关和不相关句子的相似度，看模型本身的分辨能力。
• 检查文本分割：块是不是太大了？导致一个块里包含多个不相关主题，稀释了核心语义。尝试减小chunk_size。
• 检查检索参数k：k是否太小？可能最相关的答案排在第5位，但你只取了前3个。适当增大k。

问题2：模型回答“根据资料无法回答”，但明明资料里有。

• 检查上下文是否完整传入：打印出构建的提示词，看看检索到的文本块是否真的包含了答案信息。可能分割时把答案切碎了。
• 检查提示词指令：是否明确要求模型“必须基于资料回答”？指令是否足够强硬？尝试在提示词开头加上“你必须使用以下背景资料来回答问题，你的答案必须完全来源于这些资料。”
• 调整模型温度：确保temperature设置得足够低（如0.1），避免模型过于“创造性”而忽略资料。

问题3：回答包含幻觉，即编造了资料中没有的内容。

• 这是RAG中最棘手的问题。除了优化上述两点，还可以：
• 在提示词中加强警告：“如果资料中没有提到，绝对不要猜测或编造。直接说不知道。”
• 使用更强大的模型：某些更大的模型（如Qwen2.5-14B, Llama3-70B）遵循指令和抑制幻觉的能力更强。
• 实施后处理校验：让模型在生成答案后，再基于同样的资料判断“答案中的每一个关键事实是否都能在资料中找到支撑”。这需要多轮调用，成本较高。

搭建一个可用的RAG系统是第一步，而持续优化它以达到生产级可靠性和准确性，则是一个需要不断迭代和调试的过程。希望这个从零开始的指南，能为你打下坚实的基础，让你有能力去探索更高级的RAG技术，如Agentic RAG、多跳检索（Multi-hop Retrieval）等，真正让你的大模型成为精通你私有知识的专家。