1. 项目缘起为什么你的大模型需要“外接大脑”最近跟不少做AI应用的朋友聊天发现一个挺普遍的现象大家兴致勃勃地部署好一个开源大模型比如Qwen或者Llama准备让它帮忙处理公司内部的文档、回答产品问题结果第一轮测试就傻眼了。你问它“我们公司2024年第三季度的销售策略核心是什么”它要么开始一本正经地胡说八道编造一些根本不存在的会议和数字要么就礼貌地表示“作为一个AI模型我无法获取贵公司的内部信息”。这感觉就像请了一位博古通今的大学教授但他对你家书架上的书却一本都没看过。这就是当前大模型面临的核心困境它们拥有强大的通用知识和语言生成能力但缺乏对特定、私有、最新信息的访问权限。模型训练时用的数据有截止日期且不可能包含你公司的内部wiki、产品手册、客户邮件或者昨天的会议纪要。直接微调Fine-tuning模型来学习这些新知识呢成本高、周期长每次信息更新都得重新训练完全不现实。于是RAGRetrieval-Augmented Generation检索增强生成技术就成了解决这个问题的“银弹”。它的核心思想非常直观不给模型换“脑”不重新训练而是给它配一个强大的“外部记忆库”和一位高效的“图书管理员”。当模型需要回答问题时这位“图书管理员”会先快速地从你的私有知识库比如一堆PDF、Word、网页中检索出最相关的文档片段然后把这些片段作为上下文连同问题一起交给大模型让它基于这些确凿的证据来生成答案。这么做的好处是立竿见影的答案准确性大幅提升答案来源于你的真实文档极大减少了模型“幻觉”胡编乱造。知识更新成本极低只需要往“外部记忆库”里添加新文档无需动模型本身。答案可追溯你可以清楚地知道模型是基于哪份文档的哪段话得出的结论增强了可信度。降低模型门槛即使是一个参数量较小的“轻量级”模型在优质上下文的辅助下也能完成专业领域的问答。所以今天我们就抛开那些复杂的框架和云服务从最底层开始用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/activate3.2 核心库安装与选型理由我们将通过pip安装一系列核心库每个库都承担着RAG流水线中的一个特定角色。pip install langchain langchain-community langchain-chroma pypdf python-dotenv sentence-transformers tiktoken下面我来逐一解释为什么选它们langchainlangchain-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。tiktokenOpenAI开源的快速BPE分词器。我们用它来精确计算文本的token长度这对于控制文本分割和符合大模型上下文窗口至关重要。3.3 备用方案与扩展思考我们的选择是基于“最小可行”和“完全本地”的原则。当你需要扩展时可以有以下选择文档加载器除了PDFlangchain-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格式的QA文档。我们的任务就是把这些文档“吃进去”并处理好。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, modeelements) # 对于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_size500, chunk_overlap50): 创建一个递归字符文本分割器。 参数 chunk_size: 每个文本块的目标大小单位字符。这是一个软限制。 chunk_overlap: 相邻块之间的重叠字符数。用于保持上下文连贯。 返回配置好的分割器对象。 # 首先我们定义一个分隔符优先级列表。分割器会按这个顺序尝试分割。 separators [\n\n, \n, 。, , , , , , ] # 使用递归分割器它会尽量按分隔符来保持语义完整。 text_splitter RecursiveCharacterTextSplitter( separatorsseparators, chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, # 这里用字符长度简单。生产环境可用tiktoken计算token。 # 如果你想用token长度更准确对应LLM上下文可以这样 # length_functionlambda text: len(tiktoken.get_encoding(cl100k_base).encode(text)), # chunk_size和chunk_overlap也需要调整为token数如 chunk_size800token ) 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_size500, chunk_overlap50) 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_nameparaphrase-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_numpyTrue, normalize_embeddingsTrue, # 归一化方便用余弦相似度 show_progress_barTrue) print(嵌入向量生成完毕。) return embeddings def embed_query(self, query: str) - np.ndarray: 将单个查询文本编码为向量。 # 注意查询和文档应使用相同的模型和参数进行编码以确保向量空间一致。 return self.model.encode([query], convert_to_numpyTrue, normalize_embeddingsTrue)[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_embeddingsTrue这个参数非常重要。它将向量归一化为单位长度此时余弦相似度计算简化为向量点积计算效率更高并且是许多向量数据库默认的相似度计算方式。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_implduckdbparquet, # 使用DuckDB引擎Parquet格式存储性能好 persist_directorypersist_directory, anonymized_telemetryFalse # 禁用匿名遥测 ) # 使用LangChain的Chroma包装器方便集成 vectorstore Chroma.from_texts( textstexts, embeddingembedding_model, # 这里需要传入一个LangChain兼容的Embedding对象稍后适配 metadatasmetadatas, idsids, client_settingsclient_settings, collection_namemy_knowledge_collection, persist_directorypersist_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_size500, chunk_overlap50) 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, k3) # 返回最相关的3个结果 for i, (doc, score) in enumerate(results): print(f\n--- 结果 {i1} (相似度分数: {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在背后做了什么编码查询用同样的嵌入模型把你的问题转换成向量。近似最近邻搜索在数百万个向量中快速找到与查询向量最相似的K个。它可能使用HNSWHierarchical Navigable Small World等索引算法来加速而不是暴力计算所有距离。返回结果返回Top K的文本块及其相似度分数通常是余弦相似度值越接近1越相似。关键参数k它决定了返回多少个相关文本块。k太小可能遗漏关键信息k太大会给大模型带来无关噪音增加成本并可能干扰判断。通常从3-5开始尝试根据答案质量调整。对于复杂问题可能需要更大的k如8-10。至此你的私有知识库的“搜索引擎”已经构建完毕。它已经能够理解你文档的语义并根据问题找到最相关的段落。6. 第三步让大模型基于证据“开口说话”现在我们有了问题也有了从知识库中检索到的相关证据上下文。最后一步就是构造一个清晰的指令让大模型扮演一个“基于给定材料回答问题”的专家角色。6.1 设计一个高效的提示词模板提示词Prompt是引导大模型行为的关键。一个结构良好的RAG提示词应包含以下几个部分角色指令明确告诉模型它应该扮演什么角色。任务说明清晰说明它需要完成什么任务。上下文提供清晰地标注出我们提供的背景材料。问题用户的具体问题。回答格式与约束规定回答的格式、语言、以及当信息不足时的行为。# 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_urlhttp://localhost:11434, modelqwen2.5:7b): self.base_url base_url self.model model self.generate_endpoint f{base_url}/api/generate def generate(self, prompt: str, temperature0.1, max_tokens1024) - 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, jsonpayload, timeout60) 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(modelqwen2.5:7b) test_contexts [...] # 你的上下文 test_question ... test_prompt build_rag_prompt(test_contexts, test_question) print(正在生成回答...) answer client.generate(test_prompt, temperature0.1) print(f\n模型回答\n{answer})6.3 温度参数Temperature的实战意义在调用大模型时temperature是一个至关重要的参数。它控制了模型生成文本的随机性。temperature0.1非常低模型输出高度确定倾向于选择概率最高的下一个词。这非常适合事实性问答、总结、基于明确上下文的生成能最大程度保证答案的稳定性和准确性减少“胡言乱语”。temperature0.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_size500, chunk_overlap50) 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, k4) - str: 核心问答函数。 if self.vectorstore is None: return 流水线未正确初始化。 # 1. 检索 print(f\n 正在检索与问题相关的信息 (k{k})...) retrieved_docs_with_scores self.vectorstore.similarity_search_with_score(question, kk) 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, temperature0.1, max_tokens1024) # 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_directorypersist_directory, embedding_functionembedding_function, collection_namemy_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[{i1}] 相似度: {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。检查检索参数kk是否太小可能最相关的答案排在第5位但你只取了前3个。适当增大k。问题2模型回答“根据资料无法回答”但明明资料里有。检查上下文是否完整传入打印出构建的提示词看看检索到的文本块是否真的包含了答案信息。可能分割时把答案切碎了。检查提示词指令是否明确要求模型“必须基于资料回答”指令是否足够强硬尝试在提示词开头加上“你必须使用以下背景资料来回答问题你的答案必须完全来源于这些资料。”调整模型温度确保temperature设置得足够低如0.1避免模型过于“创造性”而忽略资料。问题3回答包含幻觉即编造了资料中没有的内容。这是RAG中最棘手的问题。除了优化上述两点还可以在提示词中加强警告“如果资料中没有提到绝对不要猜测或编造。直接说不知道。”使用更强大的模型某些更大的模型如Qwen2.5-14B, Llama3-70B遵循指令和抑制幻觉的能力更强。实施后处理校验让模型在生成答案后再基于同样的资料判断“答案中的每一个关键事实是否都能在资料中找到支撑”。这需要多轮调用成本较高。搭建一个可用的RAG系统是第一步而持续优化它以达到生产级可靠性和准确性则是一个需要不断迭代和调试的过程。希望这个从零开始的指南能为你打下坚实的基础让你有能力去探索更高级的RAG技术如Agentic RAG、多跳检索Multi-hop Retrieval等真正让你的大模型成为精通你私有知识的专家。 从零搭建RAG系统：用Python实现大模型私有知识库检索增强生成
1. 项目缘起为什么你的大模型需要“外接大脑”最近跟不少做AI应用的朋友聊天发现一个挺普遍的现象大家兴致勃勃地部署好一个开源大模型比如Qwen或者Llama准备让它帮忙处理公司内部的文档、回答产品问题结果第一轮测试就傻眼了。你问它“我们公司2024年第三季度的销售策略核心是什么”它要么开始一本正经地胡说八道编造一些根本不存在的会议和数字要么就礼貌地表示“作为一个AI模型我无法获取贵公司的内部信息”。这感觉就像请了一位博古通今的大学教授但他对你家书架上的书却一本都没看过。这就是当前大模型面临的核心困境它们拥有强大的通用知识和语言生成能力但缺乏对特定、私有、最新信息的访问权限。模型训练时用的数据有截止日期且不可能包含你公司的内部wiki、产品手册、客户邮件或者昨天的会议纪要。直接微调Fine-tuning模型来学习这些新知识呢成本高、周期长每次信息更新都得重新训练完全不现实。于是RAGRetrieval-Augmented Generation检索增强生成技术就成了解决这个问题的“银弹”。它的核心思想非常直观不给模型换“脑”不重新训练而是给它配一个强大的“外部记忆库”和一位高效的“图书管理员”。当模型需要回答问题时这位“图书管理员”会先快速地从你的私有知识库比如一堆PDF、Word、网页中检索出最相关的文档片段然后把这些片段作为上下文连同问题一起交给大模型让它基于这些确凿的证据来生成答案。这么做的好处是立竿见影的答案准确性大幅提升答案来源于你的真实文档极大减少了模型“幻觉”胡编乱造。知识更新成本极低只需要往“外部记忆库”里添加新文档无需动模型本身。答案可追溯你可以清楚地知道模型是基于哪份文档的哪段话得出的结论增强了可信度。降低模型门槛即使是一个参数量较小的“轻量级”模型在优质上下文的辅助下也能完成专业领域的问答。所以今天我们就抛开那些复杂的框架和云服务从最底层开始用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/activate3.2 核心库安装与选型理由我们将通过pip安装一系列核心库每个库都承担着RAG流水线中的一个特定角色。pip install langchain langchain-community langchain-chroma pypdf python-dotenv sentence-transformers tiktoken下面我来逐一解释为什么选它们langchainlangchain-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。tiktokenOpenAI开源的快速BPE分词器。我们用它来精确计算文本的token长度这对于控制文本分割和符合大模型上下文窗口至关重要。3.3 备用方案与扩展思考我们的选择是基于“最小可行”和“完全本地”的原则。当你需要扩展时可以有以下选择文档加载器除了PDFlangchain-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格式的QA文档。我们的任务就是把这些文档“吃进去”并处理好。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, modeelements) # 对于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_size500, chunk_overlap50): 创建一个递归字符文本分割器。 参数 chunk_size: 每个文本块的目标大小单位字符。这是一个软限制。 chunk_overlap: 相邻块之间的重叠字符数。用于保持上下文连贯。 返回配置好的分割器对象。 # 首先我们定义一个分隔符优先级列表。分割器会按这个顺序尝试分割。 separators [\n\n, \n, 。, , , , , , ] # 使用递归分割器它会尽量按分隔符来保持语义完整。 text_splitter RecursiveCharacterTextSplitter( separatorsseparators, chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, # 这里用字符长度简单。生产环境可用tiktoken计算token。 # 如果你想用token长度更准确对应LLM上下文可以这样 # length_functionlambda text: len(tiktoken.get_encoding(cl100k_base).encode(text)), # chunk_size和chunk_overlap也需要调整为token数如 chunk_size800token ) 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_size500, chunk_overlap50) 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_nameparaphrase-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_numpyTrue, normalize_embeddingsTrue, # 归一化方便用余弦相似度 show_progress_barTrue) print(嵌入向量生成完毕。) return embeddings def embed_query(self, query: str) - np.ndarray: 将单个查询文本编码为向量。 # 注意查询和文档应使用相同的模型和参数进行编码以确保向量空间一致。 return self.model.encode([query], convert_to_numpyTrue, normalize_embeddingsTrue)[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_embeddingsTrue这个参数非常重要。它将向量归一化为单位长度此时余弦相似度计算简化为向量点积计算效率更高并且是许多向量数据库默认的相似度计算方式。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_implduckdbparquet, # 使用DuckDB引擎Parquet格式存储性能好 persist_directorypersist_directory, anonymized_telemetryFalse # 禁用匿名遥测 ) # 使用LangChain的Chroma包装器方便集成 vectorstore Chroma.from_texts( textstexts, embeddingembedding_model, # 这里需要传入一个LangChain兼容的Embedding对象稍后适配 metadatasmetadatas, idsids, client_settingsclient_settings, collection_namemy_knowledge_collection, persist_directorypersist_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_size500, chunk_overlap50) 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, k3) # 返回最相关的3个结果 for i, (doc, score) in enumerate(results): print(f\n--- 结果 {i1} (相似度分数: {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在背后做了什么编码查询用同样的嵌入模型把你的问题转换成向量。近似最近邻搜索在数百万个向量中快速找到与查询向量最相似的K个。它可能使用HNSWHierarchical Navigable Small World等索引算法来加速而不是暴力计算所有距离。返回结果返回Top K的文本块及其相似度分数通常是余弦相似度值越接近1越相似。关键参数k它决定了返回多少个相关文本块。k太小可能遗漏关键信息k太大会给大模型带来无关噪音增加成本并可能干扰判断。通常从3-5开始尝试根据答案质量调整。对于复杂问题可能需要更大的k如8-10。至此你的私有知识库的“搜索引擎”已经构建完毕。它已经能够理解你文档的语义并根据问题找到最相关的段落。6. 第三步让大模型基于证据“开口说话”现在我们有了问题也有了从知识库中检索到的相关证据上下文。最后一步就是构造一个清晰的指令让大模型扮演一个“基于给定材料回答问题”的专家角色。6.1 设计一个高效的提示词模板提示词Prompt是引导大模型行为的关键。一个结构良好的RAG提示词应包含以下几个部分角色指令明确告诉模型它应该扮演什么角色。任务说明清晰说明它需要完成什么任务。上下文提供清晰地标注出我们提供的背景材料。问题用户的具体问题。回答格式与约束规定回答的格式、语言、以及当信息不足时的行为。# 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_urlhttp://localhost:11434, modelqwen2.5:7b): self.base_url base_url self.model model self.generate_endpoint f{base_url}/api/generate def generate(self, prompt: str, temperature0.1, max_tokens1024) - 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, jsonpayload, timeout60) 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(modelqwen2.5:7b) test_contexts [...] # 你的上下文 test_question ... test_prompt build_rag_prompt(test_contexts, test_question) print(正在生成回答...) answer client.generate(test_prompt, temperature0.1) print(f\n模型回答\n{answer})6.3 温度参数Temperature的实战意义在调用大模型时temperature是一个至关重要的参数。它控制了模型生成文本的随机性。temperature0.1非常低模型输出高度确定倾向于选择概率最高的下一个词。这非常适合事实性问答、总结、基于明确上下文的生成能最大程度保证答案的稳定性和准确性减少“胡言乱语”。temperature0.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_size500, chunk_overlap50) 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, k4) - str: 核心问答函数。 if self.vectorstore is None: return 流水线未正确初始化。 # 1. 检索 print(f\n 正在检索与问题相关的信息 (k{k})...) retrieved_docs_with_scores self.vectorstore.similarity_search_with_score(question, kk) 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, temperature0.1, max_tokens1024) # 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_directorypersist_directory, embedding_functionembedding_function, collection_namemy_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[{i1}] 相似度: {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。检查检索参数kk是否太小可能最相关的答案排在第5位但你只取了前3个。适当增大k。问题2模型回答“根据资料无法回答”但明明资料里有。检查上下文是否完整传入打印出构建的提示词看看检索到的文本块是否真的包含了答案信息。可能分割时把答案切碎了。检查提示词指令是否明确要求模型“必须基于资料回答”指令是否足够强硬尝试在提示词开头加上“你必须使用以下背景资料来回答问题你的答案必须完全来源于这些资料。”调整模型温度确保temperature设置得足够低如0.1避免模型过于“创造性”而忽略资料。问题3回答包含幻觉即编造了资料中没有的内容。这是RAG中最棘手的问题。除了优化上述两点还可以在提示词中加强警告“如果资料中没有提到绝对不要猜测或编造。直接说不知道。”使用更强大的模型某些更大的模型如Qwen2.5-14B, Llama3-70B遵循指令和抑制幻觉的能力更强。实施后处理校验让模型在生成答案后再基于同样的资料判断“答案中的每一个关键事实是否都能在资料中找到支撑”。这需要多轮调用成本较高。搭建一个可用的RAG系统是第一步而持续优化它以达到生产级可靠性和准确性则是一个需要不断迭代和调试的过程。希望这个从零开始的指南能为你打下坚实的基础让你有能力去探索更高级的RAG技术如Agentic RAG、多跳检索Multi-hop Retrieval等真正让你的大模型成为精通你私有知识的专家。