最近在落地一个企业知识库项目时团队反复被几个问题困扰文档检索不准、回答经常“胡言乱语”、系统响应慢。经过一番折腾我们最终基于 RAG检索增强生成技术栈结合本地向量库和 Agentic 设计模式构建了一个稳定、准确且高效的系统。本文将完整复盘这套实战方案从核心概念到一行行代码手把手带你搭建一个属于你自己的 RAG 知识库。无论你是想入门 LLM 应用开发还是正在为现有项目寻找检索增强的优化方案这篇文章都能提供从零到一的闭环指南。1. 背景与核心概念为什么需要 RAG在深入代码之前我们必须先理清几个核心概念明白我们到底要解决什么问题。1.1 大模型的困境与 RAG 的诞生大型语言模型LLM如 GPT-4、ChatGLM 等拥有强大的理解和生成能力但它们存在两个固有缺陷知识滞后性模型的训练数据有截止日期无法获取最新信息。幻觉问题当模型遇到知识盲区时可能会“自信地”编造错误答案。RAGRetrieval-Augmented Generation检索增强生成正是为了解决这些问题而生。它的核心思想是“不知道就去查”。在回答用户问题前先从外部知识库如你的文档、数据库中检索出最相关的信息然后将这些信息作为上下文连同问题一起交给 LLM 生成最终答案。1.2 RAG 系统的核心组件一个典型的 RAG 系统包含以下关键环节文档加载与处理将 PDF、Word、TXT 等格式的原始文档读入系统。文本分割分块将长文档切割成大小适宜的片段Chunk以便后续嵌入和检索。嵌入Embeddings使用嵌入模型将文本块转换为高维向量一组数字。语义相似的文本其向量在空间中的距离也相近。向量存储向量库将上一步生成的向量及其对应的原始文本存储起来便于快速相似性搜索。检索Retrieval当用户提问时将问题同样转换为向量并在向量库中搜索与之最相似的几个文本块。增强生成Generation将检索到的相关文本块作为上下文与用户问题组合成提示Prompt发送给 LLM 生成最终回答。1.3 什么是 Agentic RAG传统的 RAG 是一个线性流程检索 - 生成。Agentic RAG在此基础上引入了智能体Agent思想使系统具备决策和迭代能力。例如判断检索结果是否相关如果第一次检索的结果质量不高Agent 可以决定改写查询词重新检索。多步检索与推理对于复杂问题Agent 可以将其拆解成多个子问题分别检索后再综合回答。结果验证与修正生成答案后Agent 可以调用工具验证答案中的关键事实。Agentic RAG 让系统从“被动检索”升级为“主动求解”显著提升了复杂问答的准确性和可靠性。2. 环境准备与版本说明我们将使用 Python 作为开发语言并选择当前2026年初稳定且流行的开源库来构建项目。核心环境与工具操作系统Ubuntu 22.04 LTS / macOS Monterey 或更高 / Windows 11 (WSL2 推荐)Python 版本3.10 或 3.11确保稳定性包管理pip 或 condaIDEVS Code 或 PyCharm项目依赖库我们将主要使用langchain框架来简化流程并搭配轻量级本地向量库Chroma。# requirements.txt langchain0.2.1 langchain-community0.2.1 langchain-chroma0.1.0 chromadb0.4.24 sentence-transformers2.7.0 pypdf4.2.0 python-dotenv1.0.1 openai1.30.0 # 如需使用 OpenAI 接口版本说明langchain及其生态库版本迭代较快以上版本为当前稳定组合能避免常见的兼容性问题。sentence-transformers用于运行本地的嵌入模型无需网络调用。chromadb是一个轻量、易用的本地向量数据库非常适合原型开发和中小规模知识库。实际项目中请根据pip install -r requirements.txt时的提示微调版本。项目结构预览rag_project/ ├── knowledge_base/ # 存放原始文档 │ ├── product_manual.pdf │ └── faq.txt ├── data/ # 处理后的向量库数据 ├── src/ │ ├── __init__.py │ ├── document_loader.py # 文档加载与分割 │ ├── embeddings.py # 嵌入模型管理 │ ├── vector_store.py # 向量库操作 │ ├── retriever.py # 检索器 │ ├── agentic_rag.py # Agentic RAG 核心逻辑 │ └── config.py # 配置管理 ├── .env # 环境变量如 API Keys ├── requirements.txt └── main.py # 主程序入口3. 核心原理与关键技术拆解3.1 文本分割Chunking的艺术与科学分块大小直接影响检索质量。块太大会包含无关信息稀释核心语义块太小可能丢失完整上下文。常见策略与参数固定大小分割最常用。通常设置chunk_size500-1000字符chunk_overlap100-200字符。重叠部分可以避免将完整句子或概念割裂。按语义分割利用句子边界、段落或自然语言处理NLP模型进行更智能的分割能更好地保持语义完整性。递归分割先按大分隔符如\n\n分如果块还是太大再按小分隔符如\n、空格继续分。# src/document_loader.py from langchain.text_splitter import RecursiveCharacterTextSplitter def get_text_splitter(chunk_size800, chunk_overlap150): 获取递归字符文本分割器。 Args: chunk_size: 每个文本块的最大字符数。 chunk_overlap: 相邻块之间的重叠字符数。 Returns: TextSplitter 实例。 # 分隔符优先级列表先尝试按双换行分再按单换行再按句号最后按空格 separators [\n\n, \n, 。, , , \., \?, \!, , ] text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separatorsseparators, length_functionlen, is_separator_regexFalse, ) return text_splitter # 使用示例 splitter get_text_splitter() documents [{page_content: 这是一个很长的文档内容...}] chunks splitter.split_documents(documents) print(f原始文档被分割成了 {len(chunks)} 个块。)最佳实践建议对于技术文档、手册chunk_size600, chunk_overlap100是一个不错的起点。需要通过实际问答效果进行微调。3.2 嵌入模型Embeddings选型嵌入模型负责将文本转换为向量。选择取决于对精度、速度和成本的要求。主流选择本地模型推荐用于开发/内网all-MiniLM-L6-v2轻量级速度快中英文效果均衡是入门首选。bge-large-zh-v1.5专为中文优化在中文语义匹配任务上表现优异。text2vec系列国产优秀模型中文场景效果好。云服务 API用于生产/高精度需求OpenAItext-embedding-3-small/3-largeCohere Embed百度文心、智谱AI等国内厂商的嵌入API。为什么优先考虑本地模型数据隐私文档内容无需出公网。零成本无 API 调用费用。稳定性不依赖网络和第三方服务可用性。延迟低本地推理响应快。# src/embeddings.py from langchain.embeddings import HuggingFaceEmbeddings import sentence_transformers def get_local_embeddings(model_nameall-MiniLM-L6-v2): 加载本地嵌入模型。 Args: model_name: 模型名称可从 https://huggingface.co/models 查找。 Returns: Embeddings 实例。 # 设置模型参数 model_kwargs {device: cpu} # 使用CPU如有GPU可改为 cuda # 设置编码参数如归一化输出向量有利于相似度计算 encode_kwargs {normalize_embeddings: True} embeddings HuggingFaceEmbeddings( model_namefsentence-transformers/{model_name}, model_kwargsmodel_kwargs, encode_kwargsencode_kwargs, # cache_folder 可以指定模型缓存路径 ) return embeddings # 测试嵌入模型 if __name__ __main__: emb_model get_local_embeddings() text RAG技术简介 vector emb_model.embed_query(text) print(f文本“{text}”的向量维度为{len(vector)})3.3 向量数据库Vector Store核心操作向量库负责存储和检索向量。我们使用ChromaDB它简单易用支持持久化。核心概念Collection集合相当于一个表用于存放同一类文档的向量和元数据。Metadata元数据与每个向量块一起存储的附加信息如来源文件名、页码等便于过滤和追溯。相似性搜索通常使用余弦相似度或内积来计算向量间的距离返回最相似的 K 个结果。# src/vector_store.py import chromadb from chromadb.config import Settings from langchain_chroma import Chroma from src.embeddings import get_local_embeddings import os def create_or_get_chroma_store(persist_directory./data/chroma_db, collection_nameknowledge_base): 创建或连接一个持久化的Chroma向量库。 Args: persist_directory: 向量库数据持久化目录。 collection_name: 集合名称。 Returns: Chroma 向量库实例。 # 确保目录存在 os.makedirs(persist_directory, exist_okTrue) # 获取嵌入函数 embedding_function get_local_embeddings() # 创建 Chroma 客户端并启用持久化 client_settings Settings( chroma_db_implduckdbparquet, # 后端实现 persist_directorypersist_directory, anonymized_telemetryFalse # 关闭匿名数据收集 ) # 使用 LangChain 的 Chroma 包装器 vector_store Chroma( collection_namecollection_name, embedding_functionembedding_function, persist_directorypersist_directory, client_settingsclient_settings, ) return vector_store def add_documents_to_store(vector_store, documents): 将文档块添加到向量库。 Args: vector_store: Chroma 实例。 documents: 由 Document 对象组成的列表。 # LangChain 的 Chroma 封装了 add_documents 方法会自动处理向量化 vector_store.add_documents(documentsdocuments) # 持久化到磁盘 vector_store.persist() print(f成功添加 {len(documents)} 个文档块到向量库。) if __name__ __main__: # 示例初始化并添加文档 from langchain.schema import Document docs [ Document(page_contentLangChain是一个用于开发LLM应用的框架。, metadata{source: intro.txt}), Document(page_content向量数据库用于存储和检索嵌入向量。, metadata{source: vec_db.txt}), ] store create_or_get_chroma_store() add_documents_to_store(store, docs)4. 完整实战构建本地 RAG 知识库系统现在我们将把各个模块串联起来构建一个完整的、可运行的 RAG 知识库系统。4.1 项目初始化与依赖安装首先创建项目目录并安装依赖。# 1. 创建项目目录 mkdir rag_project cd rag_project # 2. 创建虚拟环境可选但推荐 python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 3. 创建 requirements.txt 并安装 cat requirements.txt EOF langchain0.2.1 langchain-community0.2.1 langchain-chroma0.1.0 chromadb0.4.24 sentence-transformers2.7.0 pypdf4.2.0 python-dotenv1.0.1 EOF pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 4. 创建项目结构 mkdir -p knowledge_base src data touch src/__init__.py src/config.py .env main.py4.2 实现文档加载与处理模块创建一个模块支持加载 PDF、TXT 等格式并进行分块。# src/document_loader.py from langchain.document_loaders import PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document import os class DocumentProcessor: def __init__(self, chunk_size800, chunk_overlap150): self.text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , \., \?, \!, , ], length_functionlen, ) def load_and_split(self, file_path): 根据文件后缀名加载并分割文档。 Args: file_path: 文档路径。 Returns: 分割后的 Document 列表。 _, ext os.path.splitext(file_path) ext ext.lower() if ext .pdf: loader PyPDFLoader(file_path) elif ext .txt: loader TextLoader(file_path, encodingutf-8) else: raise ValueError(f暂不支持的文件格式: {ext}) raw_documents loader.load() # 为每个文档块添加来源信息 for doc in raw_documents: doc.metadata[source] os.path.basename(file_path) split_documents self.text_splitter.split_documents(raw_documents) print(f[INFO] 文件 {os.path.basename(file_path)} 被分割为 {len(split_documents)} 块。) return split_documents def process_directory(self, dir_path): 处理一个目录下的所有支持文档。 Args: dir_path: 目录路径。 Returns: 所有文档分割后的总列表。 all_docs [] supported_ext [.pdf, .txt] for root, _, files in os.walk(dir_path): for file in files: if any(file.endswith(ext) for ext in supported_ext): full_path os.path.join(root, file) try: docs self.load_and_split(full_path) all_docs.extend(docs) except Exception as e: print(f[ERROR] 处理文件 {file} 时出错: {e}) return all_docs # 测试代码 if __name__ __main__: processor DocumentProcessor(chunk_size600, chunk_overlap100) # 假设 knowledge_base 目录下有一个 test.pdf test_docs processor.process_directory(./knowledge_base) print(f总共加载并分割了 {len(test_docs)} 个文档块。)4.3 构建向量库与检索器集成前面的模块实现知识库的构建索引功能。# src/retriever.py from src.vector_store import create_or_get_chroma_store from src.document_loader import DocumentProcessor from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # 注意LLMChainExtractor需要LLM初期可不用先实现基础检索 class KnowledgeBaseBuilder: def __init__(self, persist_dir./data/chroma_db): self.persist_dir persist_dir self.vector_store None self.retriever None def build_from_directory(self, docs_dir_path): 从文档目录构建知识库向量索引。 Args: docs_dir_path: 存放原始文档的目录路径。 print(开始构建知识库索引...) # 1. 加载并分割文档 processor DocumentProcessor() all_documents processor.process_directory(docs_dir_path) if not all_documents: print(未找到任何可处理的文档。) return # 2. 初始化向量库 self.vector_store create_or_get_chroma_store( persist_directoryself.persist_dir, collection_namecorporate_knowledge ) # 3. 清空现有集合如果是重建 # self.vector_store.delete_collection() # 慎用会删除所有数据 # self.vector_store create_or_get_chroma_store(...) # 重新创建 # 4. 添加文档到向量库 self.vector_store.add_documents(documentsall_documents) self.vector_store.persist() print(f知识库索引构建完成共处理 {len(all_documents)} 个文本块。) def get_retriever(self, search_typesimilarity, k4): 获取检索器。 Args: search_type: 检索类型可选 similarity, mmr (最大边际相关性), similarity_score_threshold。 k: 返回的最相关文档数量。 Returns: Retriever 实例。 if self.vector_store is None: # 如果之前没构建则尝试加载已有的向量库 self.vector_store create_or_get_chroma_store( persist_directoryself.persist_dir, collection_namecorporate_knowledge ) # 从向量库创建检索器 self.retriever self.vector_store.as_retriever( search_typesearch_type, search_kwargs{k: k} # 返回 top K 个结果 ) return self.retriever # 主程序构建知识库 if __name__ __main__: builder KnowledgeBaseBuilder() # 首次运行构建索引 builder.build_from_directory(./knowledge_base) # 获取检索器 retriever builder.get_retriever(k4) # 测试检索 test_query 什么是RAG docs retriever.invoke(test_query) print(f查询: {test_query}) for i, doc in enumerate(docs): print(f[结果{i1}] {doc.page_content[:150]}... (来源: {doc.metadata.get(source, N/A)}))4.4 实现基础 RAG 问答链有了检索器我们就可以结合 LLM 构建一个简单的问答链。# src/basic_rag.py from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain.chat_models import ChatOpenAI # 示例用 OpenAI可替换 from src.retriever import KnowledgeBaseBuilder import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class BasicRAGQABot: def __init__(self, model_namegpt-3.5-turbo, temperature0.1): 初始化基础 RAG 问答机器人。 Args: model_name: 使用的 LLM 模型名称。 temperature: 生成温度越低越确定。 # 初始化 LLM (这里以 OpenAI 为例生产环境请替换为本地或国内模型) self.llm ChatOpenAI( model_namemodel_name, temperaturetemperature, openai_api_keyos.getenv(OPENAI_API_KEY), # 从环境变量读取 # 如需使用本地模型例如 ChatGLM可使用 langchain.llms 或自定义封装 # from langchain.llms import ChatGLM # self.llm ChatGLM(endpoint_urlhttp://localhost:8000) ) self.qa_chain None def create_chain(self, retriever): 创建检索问答链。 Args: retriever: 检索器实例。 # 定义提示模板指导 LLM 基于上下文回答 prompt_template 请严格根据以下上下文信息来回答问题。如果上下文没有提供足够的信息请直接说“根据现有资料我无法回答这个问题”不要编造信息。 上下文 {context} 问题{question} 请给出准确、简洁的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 创建 RetrievalQA 链 self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, # 将检索到的所有文档“塞”进上下文 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue, # 返回源文档用于追溯 ) def ask(self, question): 提问并获取答案。 Args: question: 用户问题。 Returns: 答案和源文档。 if self.qa_chain is None: raise ValueError(请先调用 create_chain 方法创建问答链。) result self.qa_chain.invoke({query: question}) return result[result], result[source_documents] # 使用示例 if __name__ __main__: # 0. 确保 .env 文件中有 OPENAI_API_KEYsk-...或配置其他LLM # 1. 加载知识库检索器 builder KnowledgeBaseBuilder() retriever builder.get_retriever(k4) # 2. 创建 RAG 问答机器人 bot BasicRAGQABot(model_namegpt-3.5-turbo) bot.create_chain(retriever) # 3. 进行问答 questions [ 我们公司产品的保修期是多久, 如何重置设备密码, 请总结一下RAG技术的优点。 ] for q in questions: answer, sources bot.ask(q) print(f\nQ: {q}) print(fA: {answer}) print(参考来源) for src in sources[:2]: # 显示前两个来源 print(f - {src.metadata.get(source)} (片段: {src.page_content[:80]}...))4.5 进阶实现 Agentic RAG 智能体让我们为系统加入一些“智能”实现一个能判断检索质量并决定是否重试的简单 Agent。# src/agentic_rag.py from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain import hub # 用于拉取预设的提示 from src.basic_rag import BasicRAGQABot from langchain.chat_models import ChatOpenAI import re class SelfReflectiveRAGAgent: def __init__(self, basic_rag_bot, llm_for_agentNone): 初始化具有自反思能力的 RAG Agent。 Args: basic_rag_bot: 基础 RAG 问答机器人实例。 llm_for_agent: 用于驱动 Agent 决策的 LLM可与问答 LLM 不同。 self.basic_bot basic_rag_bot self.llm_for_agent llm_for_agent or ChatOpenAI(temperature0, model_namegpt-3.5-turbo) self.agent_executor None self._setup_tools() self._setup_agent() def _setup_tools(self): 定义 Agent 可以使用的工具。 # 工具1基础检索问答 def rag_qa_tool(query: str) - str: 使用基础RAG系统回答问题。输入应是一个清晰的问题。 answer, sources self.basic_bot.ask(query) # 将答案和来源信息格式化返回 source_info \n.join([f- {s.metadata.get(source, Unknown)}: {s.page_content[:100]}... for s in sources[:2]]) return f答案{answer}\n\n参考来源\n{source_info} # 工具2查询改写工具当检索结果不佳时使用 def query_rewriter_tool(original_query: str, feedback: str) - str: 根据对之前答案的反馈改写原始查询以获取更好结果。反馈应指出原答案的不足。 prompt f 原始查询{original_query} 反馈之前的答案不令人满意因为{feedback}。 请生成一个更可能从知识库中检索到准确信息的改写后的查询。只输出改写后的查询语句。 response self.llm_for_agent.invoke(prompt) return response.content.strip() self.tools [ Tool(nameRAG_QA, funcrag_qa_tool, description用于回答基于知识库的问题。输入问题。), Tool(nameQuery_Rewriter, funcquery_rewriter_tool, description当RAG答案不佳时用于改写查询词。输入原始查询和对答案的反馈。), ] def _setup_agent(self): 设置 ReAct 智能体。 # 从 LangChain Hub 拉取一个适合的 ReAct 提示模板 prompt hub.pull(hwchase17/react) self.agent create_react_agent(self.llm_for_agent, self.tools, prompt) self.agent_executor AgentExecutor(agentself.agent, toolsself.tools, verboseTrue, handle_parsing_errorsTrue) def ask_with_reflection(self, question: str, max_iterations3) - dict: 使用智能体进行问答具备自我反思和重试能力。 Args: question: 原始问题。 max_iterations: 最大重试/反思次数。 Returns: 包含最终答案、过程和历史记录的字典。 initial_input f用户的问题是{question}\n请使用可用的工具来回答这个问题。如果你对得到的答案不确定可以尝试改写问题重新检索。 result self.agent_executor.invoke({input: initial_input}) return result # 使用示例 if __name__ __main__: # 1. 初始化基础 RAG 系统 from src.retriever import KnowledgeBaseBuilder builder KnowledgeBaseBuilder() retriever builder.get_retriever() basic_bot BasicRAGQABot() basic_bot.create_chain(retriever) # 2. 创建 Agentic RAG reflective_agent SelfReflectiveRAGAgent(basic_bot) # 3. 提问一个可能初次检索不佳的问题 complex_question 请详细说明去年发布的那个新功能的具体操作步骤和注意事项。 print(f用户问题{complex_question}) print(*50) final_result reflective_agent.ask_with_reflection(complex_question) print(*50) print(\n最终答案) print(final_result.get(output, 未获得答案))5. 常见问题与排查思路在搭建和运行 RAG 系统时你可能会遇到以下典型问题。问题现象可能原因排查步骤与解决方案导入langchain库时报错版本冲突或依赖缺失。1. 检查 Python 版本是否为 3.10。2. 使用pip list检查langchain,langchain-community等核心库版本是否匹配。3. 尝试创建全新的虚拟环境严格按照requirements.txt安装。加载嵌入模型时下载失败或速度慢网络问题或sentence-transformers默认从 HuggingFace 下载。1. 配置国内镜像源export HF_ENDPOINThttps://hf-mirror.com(Linux/Mac) 或设置环境变量。2. 手动下载模型文件到本地在代码中指定cache_folder或model_kwargs{local_files_only: True}。向量库检索结果不相关1. 分块策略不佳。2. 嵌入模型不匹配如用英文模型处理中文。3. 检索参数k不合适。1.调整分块尝试减小chunk_size或使用语义分割。2.更换嵌入模型中文场景换用bge-large-zh-v1.5。3.优化检索尝试search_typemmr以增加结果多样性或调整k值。4.检查元数据确保检索时没有错误的过滤条件。LLM 回答忽略上下文胡编乱造1. 提示Prompt设计不佳。2. 检索到的上下文质量太差。3. LLM 温度temperature过高。1.强化提示在 Prompt 中明确指令如“严格根据上下文回答”。2.改进检索先解决检索质量问题。3.降低温度将temperature设为 0.1 或 0减少随机性。4.使用 LLM 提取器在检索后增加一个步骤用一个小型 LLM 从检索结果中提取最相关的句子。系统响应速度慢1. 嵌入模型在 CPU 上运行。2. 向量库未使用索引或规模太大。3. LLM API 调用延迟高。1.硬件加速如有 GPU将嵌入模型设置为devicecuda。2.向量库优化对于 Chroma确保使用持久化对于大规模数据考虑使用带 HNSW 索引的向量库如FAISS,Weaviate。3.缓存对常见查询的嵌入向量或最终答案进行缓存。4.异步处理对 I/O 密集型操作使用异步。RuntimeError: generator raised StopIteration常见于langchain与某些依赖库如pydantic的版本冲突。1. 确保pydantic版本兼容。尝试pip install pydantic1.10.13。2. 升级langchain到最新版本或回退到一个已知稳定的旧版本组合。6. 最佳实践与工程建议将 RAG 系统从原型推向生产环境需要考虑更多工程化细节。6.1 知识库构建与维护增量更新设计支持增量添加文档的流程避免每次全量重建索引。Chroma 的add_documents是增量的但需注意去重。版本控制对知识库文档进行版本管理如 Git记录每次更新的内容和时间便于追踪和回滚。质量评估定期用一组标准问题测试系统回答的准确性建立评估指标如检索命中率、答案相关度。元数据丰富化除了文件名可以为文档块添加更多元数据如文档类型手册、合同、部门、生效日期等便于高级过滤。6.2 检索质量优化混合检索结合密集向量检索语义相似和稀疏检索如 BM25关键词匹配。langchain的EnsembleRetriever可以轻松实现。重排序初步检索出较多结果如 20 个后使用一个更精细的交叉编码器模型对结果进行重排序选出最相关的 3-5 个能显著提升精度。查询扩展在用户查询基础上自动生成同义词或相关问题并行检索后再合并结果。父文档检索存储时将大块父文档和小块子文档关联。检索时先找到相关子文档然后返回其对应的父文档作为上下文以获取更完整的背景信息。6.3 提示工程与答案生成上下文管理注意 LLM 的上下文长度限制。如果检索到的总文本过长需要进行截断或摘要。引用溯源在生成的答案中明确标注信息来源如“根据《XX手册》第3章...”增加可信度。拒绝回答当检索到的上下文置信度很低或为空时应训练 LLM 学会说“我不知道”而不是猜测。多轮对话维护对话历史将历史信息也纳入后续查询的上下文或查询改写中。6.4 生产环境部署服务化使用 FastAPI 或 Flask 将 RAG 系统封装成 RESTful API 服务。配置化将所有参数模型路径、分块大小、检索数量等抽取到配置文件如config.yaml或环境变量中。日志与监控记录每一次问答的查询、检索到的文档、生成的答案、耗时和用户反馈用于分析和优化。安全与权限如果知识库包含敏感信息必须实施严格的访问控制、用户认证和审计日志。备份与灾备定期备份向量数据库和原始文档。6.5 技术栈选型扩展向量数据库Chroma 适合轻量级应用。生产环境可考虑Weaviate功能全面、Qdrant性能优异、Milvus适用于超大规模或PGVector与 PostgreSQL 生态集成。LLM 选择根据数据隐私和成本选择云端 APIOpenAI, Anthropic, 国内大厂或本地部署模型ChatGLM3, Qwen, Llama 等。框架LangChain/LangGraph是快速原型的好工具。对于更定制化、高性能的场景可以考虑直接使用各组件如sentence-transformers, 向量库 SDK进行底层编排。从零搭建一个 RAG 系统涉及多个环节但核心脉络是清晰的文档处理 - 向量化 - 存储 - 检索 - 生成。本文提供的代码是一个完整的起点你可以在此基础上根据“最佳实践”部分的建议逐步优化检索效果、引入智能体逻辑、完善工程部署。 基于RAG与本地向量库的企业知识库实战：从原理到代码实现
最近在落地一个企业知识库项目时团队反复被几个问题困扰文档检索不准、回答经常“胡言乱语”、系统响应慢。经过一番折腾我们最终基于 RAG检索增强生成技术栈结合本地向量库和 Agentic 设计模式构建了一个稳定、准确且高效的系统。本文将完整复盘这套实战方案从核心概念到一行行代码手把手带你搭建一个属于你自己的 RAG 知识库。无论你是想入门 LLM 应用开发还是正在为现有项目寻找检索增强的优化方案这篇文章都能提供从零到一的闭环指南。1. 背景与核心概念为什么需要 RAG在深入代码之前我们必须先理清几个核心概念明白我们到底要解决什么问题。1.1 大模型的困境与 RAG 的诞生大型语言模型LLM如 GPT-4、ChatGLM 等拥有强大的理解和生成能力但它们存在两个固有缺陷知识滞后性模型的训练数据有截止日期无法获取最新信息。幻觉问题当模型遇到知识盲区时可能会“自信地”编造错误答案。RAGRetrieval-Augmented Generation检索增强生成正是为了解决这些问题而生。它的核心思想是“不知道就去查”。在回答用户问题前先从外部知识库如你的文档、数据库中检索出最相关的信息然后将这些信息作为上下文连同问题一起交给 LLM 生成最终答案。1.2 RAG 系统的核心组件一个典型的 RAG 系统包含以下关键环节文档加载与处理将 PDF、Word、TXT 等格式的原始文档读入系统。文本分割分块将长文档切割成大小适宜的片段Chunk以便后续嵌入和检索。嵌入Embeddings使用嵌入模型将文本块转换为高维向量一组数字。语义相似的文本其向量在空间中的距离也相近。向量存储向量库将上一步生成的向量及其对应的原始文本存储起来便于快速相似性搜索。检索Retrieval当用户提问时将问题同样转换为向量并在向量库中搜索与之最相似的几个文本块。增强生成Generation将检索到的相关文本块作为上下文与用户问题组合成提示Prompt发送给 LLM 生成最终回答。1.3 什么是 Agentic RAG传统的 RAG 是一个线性流程检索 - 生成。Agentic RAG在此基础上引入了智能体Agent思想使系统具备决策和迭代能力。例如判断检索结果是否相关如果第一次检索的结果质量不高Agent 可以决定改写查询词重新检索。多步检索与推理对于复杂问题Agent 可以将其拆解成多个子问题分别检索后再综合回答。结果验证与修正生成答案后Agent 可以调用工具验证答案中的关键事实。Agentic RAG 让系统从“被动检索”升级为“主动求解”显著提升了复杂问答的准确性和可靠性。2. 环境准备与版本说明我们将使用 Python 作为开发语言并选择当前2026年初稳定且流行的开源库来构建项目。核心环境与工具操作系统Ubuntu 22.04 LTS / macOS Monterey 或更高 / Windows 11 (WSL2 推荐)Python 版本3.10 或 3.11确保稳定性包管理pip 或 condaIDEVS Code 或 PyCharm项目依赖库我们将主要使用langchain框架来简化流程并搭配轻量级本地向量库Chroma。# requirements.txt langchain0.2.1 langchain-community0.2.1 langchain-chroma0.1.0 chromadb0.4.24 sentence-transformers2.7.0 pypdf4.2.0 python-dotenv1.0.1 openai1.30.0 # 如需使用 OpenAI 接口版本说明langchain及其生态库版本迭代较快以上版本为当前稳定组合能避免常见的兼容性问题。sentence-transformers用于运行本地的嵌入模型无需网络调用。chromadb是一个轻量、易用的本地向量数据库非常适合原型开发和中小规模知识库。实际项目中请根据pip install -r requirements.txt时的提示微调版本。项目结构预览rag_project/ ├── knowledge_base/ # 存放原始文档 │ ├── product_manual.pdf │ └── faq.txt ├── data/ # 处理后的向量库数据 ├── src/ │ ├── __init__.py │ ├── document_loader.py # 文档加载与分割 │ ├── embeddings.py # 嵌入模型管理 │ ├── vector_store.py # 向量库操作 │ ├── retriever.py # 检索器 │ ├── agentic_rag.py # Agentic RAG 核心逻辑 │ └── config.py # 配置管理 ├── .env # 环境变量如 API Keys ├── requirements.txt └── main.py # 主程序入口3. 核心原理与关键技术拆解3.1 文本分割Chunking的艺术与科学分块大小直接影响检索质量。块太大会包含无关信息稀释核心语义块太小可能丢失完整上下文。常见策略与参数固定大小分割最常用。通常设置chunk_size500-1000字符chunk_overlap100-200字符。重叠部分可以避免将完整句子或概念割裂。按语义分割利用句子边界、段落或自然语言处理NLP模型进行更智能的分割能更好地保持语义完整性。递归分割先按大分隔符如\n\n分如果块还是太大再按小分隔符如\n、空格继续分。# src/document_loader.py from langchain.text_splitter import RecursiveCharacterTextSplitter def get_text_splitter(chunk_size800, chunk_overlap150): 获取递归字符文本分割器。 Args: chunk_size: 每个文本块的最大字符数。 chunk_overlap: 相邻块之间的重叠字符数。 Returns: TextSplitter 实例。 # 分隔符优先级列表先尝试按双换行分再按单换行再按句号最后按空格 separators [\n\n, \n, 。, , , \., \?, \!, , ] text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separatorsseparators, length_functionlen, is_separator_regexFalse, ) return text_splitter # 使用示例 splitter get_text_splitter() documents [{page_content: 这是一个很长的文档内容...}] chunks splitter.split_documents(documents) print(f原始文档被分割成了 {len(chunks)} 个块。)最佳实践建议对于技术文档、手册chunk_size600, chunk_overlap100是一个不错的起点。需要通过实际问答效果进行微调。3.2 嵌入模型Embeddings选型嵌入模型负责将文本转换为向量。选择取决于对精度、速度和成本的要求。主流选择本地模型推荐用于开发/内网all-MiniLM-L6-v2轻量级速度快中英文效果均衡是入门首选。bge-large-zh-v1.5专为中文优化在中文语义匹配任务上表现优异。text2vec系列国产优秀模型中文场景效果好。云服务 API用于生产/高精度需求OpenAItext-embedding-3-small/3-largeCohere Embed百度文心、智谱AI等国内厂商的嵌入API。为什么优先考虑本地模型数据隐私文档内容无需出公网。零成本无 API 调用费用。稳定性不依赖网络和第三方服务可用性。延迟低本地推理响应快。# src/embeddings.py from langchain.embeddings import HuggingFaceEmbeddings import sentence_transformers def get_local_embeddings(model_nameall-MiniLM-L6-v2): 加载本地嵌入模型。 Args: model_name: 模型名称可从 https://huggingface.co/models 查找。 Returns: Embeddings 实例。 # 设置模型参数 model_kwargs {device: cpu} # 使用CPU如有GPU可改为 cuda # 设置编码参数如归一化输出向量有利于相似度计算 encode_kwargs {normalize_embeddings: True} embeddings HuggingFaceEmbeddings( model_namefsentence-transformers/{model_name}, model_kwargsmodel_kwargs, encode_kwargsencode_kwargs, # cache_folder 可以指定模型缓存路径 ) return embeddings # 测试嵌入模型 if __name__ __main__: emb_model get_local_embeddings() text RAG技术简介 vector emb_model.embed_query(text) print(f文本“{text}”的向量维度为{len(vector)})3.3 向量数据库Vector Store核心操作向量库负责存储和检索向量。我们使用ChromaDB它简单易用支持持久化。核心概念Collection集合相当于一个表用于存放同一类文档的向量和元数据。Metadata元数据与每个向量块一起存储的附加信息如来源文件名、页码等便于过滤和追溯。相似性搜索通常使用余弦相似度或内积来计算向量间的距离返回最相似的 K 个结果。# src/vector_store.py import chromadb from chromadb.config import Settings from langchain_chroma import Chroma from src.embeddings import get_local_embeddings import os def create_or_get_chroma_store(persist_directory./data/chroma_db, collection_nameknowledge_base): 创建或连接一个持久化的Chroma向量库。 Args: persist_directory: 向量库数据持久化目录。 collection_name: 集合名称。 Returns: Chroma 向量库实例。 # 确保目录存在 os.makedirs(persist_directory, exist_okTrue) # 获取嵌入函数 embedding_function get_local_embeddings() # 创建 Chroma 客户端并启用持久化 client_settings Settings( chroma_db_implduckdbparquet, # 后端实现 persist_directorypersist_directory, anonymized_telemetryFalse # 关闭匿名数据收集 ) # 使用 LangChain 的 Chroma 包装器 vector_store Chroma( collection_namecollection_name, embedding_functionembedding_function, persist_directorypersist_directory, client_settingsclient_settings, ) return vector_store def add_documents_to_store(vector_store, documents): 将文档块添加到向量库。 Args: vector_store: Chroma 实例。 documents: 由 Document 对象组成的列表。 # LangChain 的 Chroma 封装了 add_documents 方法会自动处理向量化 vector_store.add_documents(documentsdocuments) # 持久化到磁盘 vector_store.persist() print(f成功添加 {len(documents)} 个文档块到向量库。) if __name__ __main__: # 示例初始化并添加文档 from langchain.schema import Document docs [ Document(page_contentLangChain是一个用于开发LLM应用的框架。, metadata{source: intro.txt}), Document(page_content向量数据库用于存储和检索嵌入向量。, metadata{source: vec_db.txt}), ] store create_or_get_chroma_store() add_documents_to_store(store, docs)4. 完整实战构建本地 RAG 知识库系统现在我们将把各个模块串联起来构建一个完整的、可运行的 RAG 知识库系统。4.1 项目初始化与依赖安装首先创建项目目录并安装依赖。# 1. 创建项目目录 mkdir rag_project cd rag_project # 2. 创建虚拟环境可选但推荐 python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 3. 创建 requirements.txt 并安装 cat requirements.txt EOF langchain0.2.1 langchain-community0.2.1 langchain-chroma0.1.0 chromadb0.4.24 sentence-transformers2.7.0 pypdf4.2.0 python-dotenv1.0.1 EOF pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 4. 创建项目结构 mkdir -p knowledge_base src data touch src/__init__.py src/config.py .env main.py4.2 实现文档加载与处理模块创建一个模块支持加载 PDF、TXT 等格式并进行分块。# src/document_loader.py from langchain.document_loaders import PyPDFLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document import os class DocumentProcessor: def __init__(self, chunk_size800, chunk_overlap150): self.text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , \., \?, \!, , ], length_functionlen, ) def load_and_split(self, file_path): 根据文件后缀名加载并分割文档。 Args: file_path: 文档路径。 Returns: 分割后的 Document 列表。 _, ext os.path.splitext(file_path) ext ext.lower() if ext .pdf: loader PyPDFLoader(file_path) elif ext .txt: loader TextLoader(file_path, encodingutf-8) else: raise ValueError(f暂不支持的文件格式: {ext}) raw_documents loader.load() # 为每个文档块添加来源信息 for doc in raw_documents: doc.metadata[source] os.path.basename(file_path) split_documents self.text_splitter.split_documents(raw_documents) print(f[INFO] 文件 {os.path.basename(file_path)} 被分割为 {len(split_documents)} 块。) return split_documents def process_directory(self, dir_path): 处理一个目录下的所有支持文档。 Args: dir_path: 目录路径。 Returns: 所有文档分割后的总列表。 all_docs [] supported_ext [.pdf, .txt] for root, _, files in os.walk(dir_path): for file in files: if any(file.endswith(ext) for ext in supported_ext): full_path os.path.join(root, file) try: docs self.load_and_split(full_path) all_docs.extend(docs) except Exception as e: print(f[ERROR] 处理文件 {file} 时出错: {e}) return all_docs # 测试代码 if __name__ __main__: processor DocumentProcessor(chunk_size600, chunk_overlap100) # 假设 knowledge_base 目录下有一个 test.pdf test_docs processor.process_directory(./knowledge_base) print(f总共加载并分割了 {len(test_docs)} 个文档块。)4.3 构建向量库与检索器集成前面的模块实现知识库的构建索引功能。# src/retriever.py from src.vector_store import create_or_get_chroma_store from src.document_loader import DocumentProcessor from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # 注意LLMChainExtractor需要LLM初期可不用先实现基础检索 class KnowledgeBaseBuilder: def __init__(self, persist_dir./data/chroma_db): self.persist_dir persist_dir self.vector_store None self.retriever None def build_from_directory(self, docs_dir_path): 从文档目录构建知识库向量索引。 Args: docs_dir_path: 存放原始文档的目录路径。 print(开始构建知识库索引...) # 1. 加载并分割文档 processor DocumentProcessor() all_documents processor.process_directory(docs_dir_path) if not all_documents: print(未找到任何可处理的文档。) return # 2. 初始化向量库 self.vector_store create_or_get_chroma_store( persist_directoryself.persist_dir, collection_namecorporate_knowledge ) # 3. 清空现有集合如果是重建 # self.vector_store.delete_collection() # 慎用会删除所有数据 # self.vector_store create_or_get_chroma_store(...) # 重新创建 # 4. 添加文档到向量库 self.vector_store.add_documents(documentsall_documents) self.vector_store.persist() print(f知识库索引构建完成共处理 {len(all_documents)} 个文本块。) def get_retriever(self, search_typesimilarity, k4): 获取检索器。 Args: search_type: 检索类型可选 similarity, mmr (最大边际相关性), similarity_score_threshold。 k: 返回的最相关文档数量。 Returns: Retriever 实例。 if self.vector_store is None: # 如果之前没构建则尝试加载已有的向量库 self.vector_store create_or_get_chroma_store( persist_directoryself.persist_dir, collection_namecorporate_knowledge ) # 从向量库创建检索器 self.retriever self.vector_store.as_retriever( search_typesearch_type, search_kwargs{k: k} # 返回 top K 个结果 ) return self.retriever # 主程序构建知识库 if __name__ __main__: builder KnowledgeBaseBuilder() # 首次运行构建索引 builder.build_from_directory(./knowledge_base) # 获取检索器 retriever builder.get_retriever(k4) # 测试检索 test_query 什么是RAG docs retriever.invoke(test_query) print(f查询: {test_query}) for i, doc in enumerate(docs): print(f[结果{i1}] {doc.page_content[:150]}... (来源: {doc.metadata.get(source, N/A)}))4.4 实现基础 RAG 问答链有了检索器我们就可以结合 LLM 构建一个简单的问答链。# src/basic_rag.py from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain.chat_models import ChatOpenAI # 示例用 OpenAI可替换 from src.retriever import KnowledgeBaseBuilder import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class BasicRAGQABot: def __init__(self, model_namegpt-3.5-turbo, temperature0.1): 初始化基础 RAG 问答机器人。 Args: model_name: 使用的 LLM 模型名称。 temperature: 生成温度越低越确定。 # 初始化 LLM (这里以 OpenAI 为例生产环境请替换为本地或国内模型) self.llm ChatOpenAI( model_namemodel_name, temperaturetemperature, openai_api_keyos.getenv(OPENAI_API_KEY), # 从环境变量读取 # 如需使用本地模型例如 ChatGLM可使用 langchain.llms 或自定义封装 # from langchain.llms import ChatGLM # self.llm ChatGLM(endpoint_urlhttp://localhost:8000) ) self.qa_chain None def create_chain(self, retriever): 创建检索问答链。 Args: retriever: 检索器实例。 # 定义提示模板指导 LLM 基于上下文回答 prompt_template 请严格根据以下上下文信息来回答问题。如果上下文没有提供足够的信息请直接说“根据现有资料我无法回答这个问题”不要编造信息。 上下文 {context} 问题{question} 请给出准确、简洁的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 创建 RetrievalQA 链 self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, # 将检索到的所有文档“塞”进上下文 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue, # 返回源文档用于追溯 ) def ask(self, question): 提问并获取答案。 Args: question: 用户问题。 Returns: 答案和源文档。 if self.qa_chain is None: raise ValueError(请先调用 create_chain 方法创建问答链。) result self.qa_chain.invoke({query: question}) return result[result], result[source_documents] # 使用示例 if __name__ __main__: # 0. 确保 .env 文件中有 OPENAI_API_KEYsk-...或配置其他LLM # 1. 加载知识库检索器 builder KnowledgeBaseBuilder() retriever builder.get_retriever(k4) # 2. 创建 RAG 问答机器人 bot BasicRAGQABot(model_namegpt-3.5-turbo) bot.create_chain(retriever) # 3. 进行问答 questions [ 我们公司产品的保修期是多久, 如何重置设备密码, 请总结一下RAG技术的优点。 ] for q in questions: answer, sources bot.ask(q) print(f\nQ: {q}) print(fA: {answer}) print(参考来源) for src in sources[:2]: # 显示前两个来源 print(f - {src.metadata.get(source)} (片段: {src.page_content[:80]}...))4.5 进阶实现 Agentic RAG 智能体让我们为系统加入一些“智能”实现一个能判断检索质量并决定是否重试的简单 Agent。# src/agentic_rag.py from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain import hub # 用于拉取预设的提示 from src.basic_rag import BasicRAGQABot from langchain.chat_models import ChatOpenAI import re class SelfReflectiveRAGAgent: def __init__(self, basic_rag_bot, llm_for_agentNone): 初始化具有自反思能力的 RAG Agent。 Args: basic_rag_bot: 基础 RAG 问答机器人实例。 llm_for_agent: 用于驱动 Agent 决策的 LLM可与问答 LLM 不同。 self.basic_bot basic_rag_bot self.llm_for_agent llm_for_agent or ChatOpenAI(temperature0, model_namegpt-3.5-turbo) self.agent_executor None self._setup_tools() self._setup_agent() def _setup_tools(self): 定义 Agent 可以使用的工具。 # 工具1基础检索问答 def rag_qa_tool(query: str) - str: 使用基础RAG系统回答问题。输入应是一个清晰的问题。 answer, sources self.basic_bot.ask(query) # 将答案和来源信息格式化返回 source_info \n.join([f- {s.metadata.get(source, Unknown)}: {s.page_content[:100]}... for s in sources[:2]]) return f答案{answer}\n\n参考来源\n{source_info} # 工具2查询改写工具当检索结果不佳时使用 def query_rewriter_tool(original_query: str, feedback: str) - str: 根据对之前答案的反馈改写原始查询以获取更好结果。反馈应指出原答案的不足。 prompt f 原始查询{original_query} 反馈之前的答案不令人满意因为{feedback}。 请生成一个更可能从知识库中检索到准确信息的改写后的查询。只输出改写后的查询语句。 response self.llm_for_agent.invoke(prompt) return response.content.strip() self.tools [ Tool(nameRAG_QA, funcrag_qa_tool, description用于回答基于知识库的问题。输入问题。), Tool(nameQuery_Rewriter, funcquery_rewriter_tool, description当RAG答案不佳时用于改写查询词。输入原始查询和对答案的反馈。), ] def _setup_agent(self): 设置 ReAct 智能体。 # 从 LangChain Hub 拉取一个适合的 ReAct 提示模板 prompt hub.pull(hwchase17/react) self.agent create_react_agent(self.llm_for_agent, self.tools, prompt) self.agent_executor AgentExecutor(agentself.agent, toolsself.tools, verboseTrue, handle_parsing_errorsTrue) def ask_with_reflection(self, question: str, max_iterations3) - dict: 使用智能体进行问答具备自我反思和重试能力。 Args: question: 原始问题。 max_iterations: 最大重试/反思次数。 Returns: 包含最终答案、过程和历史记录的字典。 initial_input f用户的问题是{question}\n请使用可用的工具来回答这个问题。如果你对得到的答案不确定可以尝试改写问题重新检索。 result self.agent_executor.invoke({input: initial_input}) return result # 使用示例 if __name__ __main__: # 1. 初始化基础 RAG 系统 from src.retriever import KnowledgeBaseBuilder builder KnowledgeBaseBuilder() retriever builder.get_retriever() basic_bot BasicRAGQABot() basic_bot.create_chain(retriever) # 2. 创建 Agentic RAG reflective_agent SelfReflectiveRAGAgent(basic_bot) # 3. 提问一个可能初次检索不佳的问题 complex_question 请详细说明去年发布的那个新功能的具体操作步骤和注意事项。 print(f用户问题{complex_question}) print(*50) final_result reflective_agent.ask_with_reflection(complex_question) print(*50) print(\n最终答案) print(final_result.get(output, 未获得答案))5. 常见问题与排查思路在搭建和运行 RAG 系统时你可能会遇到以下典型问题。问题现象可能原因排查步骤与解决方案导入langchain库时报错版本冲突或依赖缺失。1. 检查 Python 版本是否为 3.10。2. 使用pip list检查langchain,langchain-community等核心库版本是否匹配。3. 尝试创建全新的虚拟环境严格按照requirements.txt安装。加载嵌入模型时下载失败或速度慢网络问题或sentence-transformers默认从 HuggingFace 下载。1. 配置国内镜像源export HF_ENDPOINThttps://hf-mirror.com(Linux/Mac) 或设置环境变量。2. 手动下载模型文件到本地在代码中指定cache_folder或model_kwargs{local_files_only: True}。向量库检索结果不相关1. 分块策略不佳。2. 嵌入模型不匹配如用英文模型处理中文。3. 检索参数k不合适。1.调整分块尝试减小chunk_size或使用语义分割。2.更换嵌入模型中文场景换用bge-large-zh-v1.5。3.优化检索尝试search_typemmr以增加结果多样性或调整k值。4.检查元数据确保检索时没有错误的过滤条件。LLM 回答忽略上下文胡编乱造1. 提示Prompt设计不佳。2. 检索到的上下文质量太差。3. LLM 温度temperature过高。1.强化提示在 Prompt 中明确指令如“严格根据上下文回答”。2.改进检索先解决检索质量问题。3.降低温度将temperature设为 0.1 或 0减少随机性。4.使用 LLM 提取器在检索后增加一个步骤用一个小型 LLM 从检索结果中提取最相关的句子。系统响应速度慢1. 嵌入模型在 CPU 上运行。2. 向量库未使用索引或规模太大。3. LLM API 调用延迟高。1.硬件加速如有 GPU将嵌入模型设置为devicecuda。2.向量库优化对于 Chroma确保使用持久化对于大规模数据考虑使用带 HNSW 索引的向量库如FAISS,Weaviate。3.缓存对常见查询的嵌入向量或最终答案进行缓存。4.异步处理对 I/O 密集型操作使用异步。RuntimeError: generator raised StopIteration常见于langchain与某些依赖库如pydantic的版本冲突。1. 确保pydantic版本兼容。尝试pip install pydantic1.10.13。2. 升级langchain到最新版本或回退到一个已知稳定的旧版本组合。6. 最佳实践与工程建议将 RAG 系统从原型推向生产环境需要考虑更多工程化细节。6.1 知识库构建与维护增量更新设计支持增量添加文档的流程避免每次全量重建索引。Chroma 的add_documents是增量的但需注意去重。版本控制对知识库文档进行版本管理如 Git记录每次更新的内容和时间便于追踪和回滚。质量评估定期用一组标准问题测试系统回答的准确性建立评估指标如检索命中率、答案相关度。元数据丰富化除了文件名可以为文档块添加更多元数据如文档类型手册、合同、部门、生效日期等便于高级过滤。6.2 检索质量优化混合检索结合密集向量检索语义相似和稀疏检索如 BM25关键词匹配。langchain的EnsembleRetriever可以轻松实现。重排序初步检索出较多结果如 20 个后使用一个更精细的交叉编码器模型对结果进行重排序选出最相关的 3-5 个能显著提升精度。查询扩展在用户查询基础上自动生成同义词或相关问题并行检索后再合并结果。父文档检索存储时将大块父文档和小块子文档关联。检索时先找到相关子文档然后返回其对应的父文档作为上下文以获取更完整的背景信息。6.3 提示工程与答案生成上下文管理注意 LLM 的上下文长度限制。如果检索到的总文本过长需要进行截断或摘要。引用溯源在生成的答案中明确标注信息来源如“根据《XX手册》第3章...”增加可信度。拒绝回答当检索到的上下文置信度很低或为空时应训练 LLM 学会说“我不知道”而不是猜测。多轮对话维护对话历史将历史信息也纳入后续查询的上下文或查询改写中。6.4 生产环境部署服务化使用 FastAPI 或 Flask 将 RAG 系统封装成 RESTful API 服务。配置化将所有参数模型路径、分块大小、检索数量等抽取到配置文件如config.yaml或环境变量中。日志与监控记录每一次问答的查询、检索到的文档、生成的答案、耗时和用户反馈用于分析和优化。安全与权限如果知识库包含敏感信息必须实施严格的访问控制、用户认证和审计日志。备份与灾备定期备份向量数据库和原始文档。6.5 技术栈选型扩展向量数据库Chroma 适合轻量级应用。生产环境可考虑Weaviate功能全面、Qdrant性能优异、Milvus适用于超大规模或PGVector与 PostgreSQL 生态集成。LLM 选择根据数据隐私和成本选择云端 APIOpenAI, Anthropic, 国内大厂或本地部署模型ChatGLM3, Qwen, Llama 等。框架LangChain/LangGraph是快速原型的好工具。对于更定制化、高性能的场景可以考虑直接使用各组件如sentence-transformers, 向量库 SDK进行底层编排。从零搭建一个 RAG 系统涉及多个环节但核心脉络是清晰的文档处理 - 向量化 - 存储 - 检索 - 生成。本文提供的代码是一个完整的起点你可以在此基础上根据“最佳实践”部分的建议逐步优化检索效果、引入智能体逻辑、完善工程部署。