在实际的大语言模型LLM应用开发中如何让模型理解并回答超出其训练数据范围或需要最新信息的问题是一个核心挑战。简单地将用户问题直接抛给模型往往得到的是“幻觉”或过时的答案。检索增强生成RAG技术通过引入外部知识库有效地解决了这一问题。一个典型的 RAG 系统包含三个关键环节将知识转化为机器可理解的向量Embeddings高效存储和检索这些向量的本地向量库以及决定如何与知识库交互的智能体Agent。掌握这三者是构建可靠、可控、可解释的 LLM 应用的基础。本文将以一个从零开始的 RAG 知识库搭建项目为线索深入讲解 Embeddings 模型的选择与使用、本地向量数据库的部署与操作并探讨如何引入 Agentic RAG 思想来优化检索流程。无论你是希望为内部文档构建一个智能问答助手还是想为你的产品增加基于私有知识的对话能力这篇文章都将提供一条清晰的实践路径。我们将从核心概念入手逐步完成环境准备、代码实现、系统验证并最终讨论生产环境中的常见问题与优化策略。1. 理解 RAG 系统的核心组件与工作流在动手编码之前必须清晰地理解 RAG 系统是如何工作的。它不是一个单一的技术而是一个由多个组件协同工作的架构。1.1 RAG 的基本流程检索与生成的结合RAG 的核心思想可以概括为“先检索后生成”。当用户提出一个问题Query时系统不会直接让 LLM 凭空回答。而是执行以下步骤知识库预处理离线将你的原始文档如 PDF、Word、Markdown、网页进行文本提取、分割Chunking然后通过 Embeddings 模型转换为高维向量最后存入向量数据库。检索在线将用户的 Query 同样通过 Embeddings 模型转换为向量然后在向量数据库中进行相似度搜索如余弦相似度找出与 Query 最相关的几个文本片段Chunks。增强提示在线将检索到的相关文本片段作为“上下文”Context与用户的原始 Query 一起组合成一个新的、信息更丰富的提示Prompt提交给 LLM。生成在线LLM 基于这个包含了相关上下文的 Prompt 生成最终答案。由于答案的依据来自提供的上下文其准确性和可信度大大提高同时也能有效减少模型“幻觉”。这个流程的关键在于LLM 的答案能力与外部知识库的检索能力被解耦了。你可以独立优化检索部分如使用更好的 Embeddings 模型、更优的分块策略和生成部分如更换更强大的 LLM而无需重新训练模型。1.2 关键组件深度解析Embeddings向量化模型这是将文本语义映射到数学空间的核心。一个好的 Embeddings 模型应该能将语义相似的句子映射到向量空间中相近的点。例如“如何配置数据库连接”和“设置数据库的链接参数”这两个句子的向量应该非常接近。常见的开源模型包括text-embedding-ada-002的替代品如BAAI/bge-small-zh-v1.5用于中文、all-MiniLM-L6-v2等。选择时需考虑语言中/英、性能、向量维度影响存储和计算效率和本地部署能力。向量数据库专门为高维向量的存储和快速相似性搜索而设计的数据库。它需要高效地处理成千上万个向量并在毫秒级内返回最相似的 Top-K 个结果。本地部署的常见选择有ChromaDB轻量级、易于使用特别适合原型开发和中小规模项目。FAISSFacebook 开源的库专注于向量相似性搜索性能极高通常作为库集成到应用中。Milvus/Qdrant功能更全面的专业向量数据库支持分布式、持久化、动态数据管理适合大规模生产环境。Agentic RAG这是对传统 RAG 的智能升级。在传统 RAG 中检索是一次性的、被动的。而 Agentic RAG 引入了“智能体”的概念让检索过程变得主动和迭代。例如智能体可以判断是否需要检索如果问题是一般性对话可能不需要。改写用户 Query 以提升检索效果如将口语化问题改写成更正式的文档用语。多路检索尝试不同的分块大小或检索策略然后对结果进行重排序Rerank。迭代检索根据 LLM 的初步回答生成新的、更聚焦的 Query 进行二次检索以补充或验证信息。Agentic RAG 的核心是让 LLM 本身或一个更小的规划模型来协调和控制 RAG 流程中的决策点使其成为一个闭环的、自适应的系统。2. 项目环境准备与核心依赖配置我们将构建一个基于 Python 的本地 RAG 系统。为了兼顾学习成本和功能完整性技术栈选择如下Embeddings 模型BAAI/bge-small-zh-v1.5优秀的中文开源模型向量数据库ChromaDB内存/磁盘模式易于上手LLM通过 OpenAI 兼容的 API 调用例如使用开源模型部署的本地 API或云服务商 API。我们将使用openai库的标准方式调用便于切换后端。框架langchain和langchain-community。它们提供了连接各组件的高层抽象能极大减少样板代码让我们聚焦于流程设计。对于 Agentic 部分我们会结合langchain的智能体工具。2.1 创建项目与虚拟环境首先创建一个干净的项目目录并设置独立的 Python 环境这是管理依赖的最佳实践。# 创建项目目录 mkdir rag-project-2026 cd rag-project-2026 # 创建并激活虚拟环境 (以 conda 为例也可使用 venv) conda create -n rag-env python3.10 -y conda activate rag-env2.2 安装项目依赖创建requirements.txt文件并填入以下核心依赖# 核心框架 langchain0.1.0 langchain-community0.0.10 langchain-openai0.0.5 # 向量数据库与Embeddings chromadb0.4.22 sentence-transformers2.2.2 # 用于运行本地Embeddings模型 # 文档加载与处理 pypdf3.17.4 # 读取PDF unstructured0.10.30 # 解析多种格式文档 markdown3.5.2 # OpenAI兼容客户端 (用于调用LLM) openai1.12.0 # 其他工具 tiktoken0.5.2 # 用于Token计数 python-dotenv1.0.0 # 管理环境变量然后安装它们pip install -r requirements.txt注意langchain版本迭代较快上述版本号旨在保证示例代码的稳定性。在实际项目中请根据官方文档和兼容性说明选择合适的版本。2.3 准备模型与API配置对于 Embeddings 模型sentence-transformers库会在首次使用时自动从 Hugging Face 下载BAAI/bge-small-zh-v1.5模型。确保你的开发环境可以正常访问相关资源。对于 LLM API我们需要一个提供 OpenAI 兼容接口的 LLM 服务。这可以是云服务如 OpenAI API、Azure OpenAI、DeepSeek API 等。本地部署使用ollama、vLLM、LM Studio等工具部署开源模型如 Qwen、Llama、ChatGLM并开启其兼容 OpenAI 的 API 服务。本文假设你已有一个可用的 OpenAI 兼容 API 端点。将 API 基础地址和密钥配置在环境变量中。创建一个.env文件# .env 文件 OPENAI_API_KEYyour_api_key_here # 如果是云服务商填写其密钥 OPENAI_API_BASEhttp://localhost:11434/v1 # 如果是本地 ollama默认地址如此 # 如果使用官方OpenAI则注释掉 OPENAI_API_BASE # OPENAI_API_BASEhttps://api.openai.com/v1在代码中使用python-dotenv加载配置from dotenv import load_dotenv import os load_dotenv() # 加载 .env 文件中的环境变量 api_key os.getenv(OPENAI_API_KEY) api_base os.getenv(OPENAI_API_BASE, https://api.openai.com/v1) # 提供默认值3. 构建基础 RAG 流水线从文档到答案现在我们开始实现一个最基础的、端到端的 RAG 系统。这个系统能够读取本地 PDF 文档构建向量库并回答用户问题。3.1 文档加载与文本分割原始文档需要被处理成适合检索的文本块。块的大小和重叠度是关键参数。# file_processor.py from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document def load_and_split_pdfs(pdf_paths, chunk_size500, chunk_overlap50): 加载PDF文档并分割成文本块。 参数: pdf_paths: PDF文件路径列表。 chunk_size: 每个文本块的最大字符数。 chunk_overlap: 相邻块之间的重叠字符数用于保持上下文连贯。 返回: 包含所有文本块的列表。 all_docs [] text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , , ] # 中文友好的分隔符 ) for pdf_path in pdf_paths: try: loader PyPDFLoader(pdf_path) pages loader.load() # 每个页面是一个Document对象 # 将页面内容合并然后由分割器处理 chunks text_splitter.split_documents(pages) all_docs.extend(chunks) print(f已加载并分割 {pdf_path}, 得到 {len(chunks)} 个文本块。) except Exception as e: print(f处理文件 {pdf_path} 时出错: {e}) print(f总计加载 {len(all_docs)} 个文本块。) return all_docs # 使用示例 if __name__ __main__: pdfs [./docs/产品手册.pdf, ./docs/技术白皮书.pdf] # 假设的文档路径 documents load_and_split_pdfs(pdfs, chunk_size500, chunk_overlap50) # 查看第一个块的内容和元数据 if documents: print(示例块内容:, documents[0].page_content[:200]) print(元数据:, documents[0].metadata)关键参数解释chunk_size决定每个向量代表多少文本。太小可能丢失全局语义太大可能包含无关信息降低检索精度。通常 256-1024 个字符是常见范围需要根据文档类型技术文档、小说、对话调整。chunk_overlap防止一个完整的句子或概念被硬生生切断。重叠部分可以确保边界信息不会丢失。3.2 向量化与向量库构建接下来我们将分割好的文本块转换为向量并存储到 ChromaDB 中。# vector_store_builder.py from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma import os def create_vector_store(documents, persist_directory./chroma_db): 创建并持久化向量存储。 参数: documents: 分割后的Document对象列表。 persist_directory: 向量数据库持久化目录。 返回: vector_store: 向量存储对象可用于检索。 # 1. 初始化Embeddings模型 # 使用开源的中文Embedding模型 model_name BAAI/bge-small-zh-v1.5 model_kwargs {device: cpu} # 如果GPU可用可改为 cuda encode_kwargs {normalize_embeddings: True} # 归一化便于余弦相似度计算 embeddings HuggingFaceEmbeddings( model_namemodel_name, model_kwargsmodel_kwargs, encode_kwargsencode_kwargs ) # 2. 创建向量存储并持久化 # Chroma.from_documents 会完成向量化并存入本地目录 vector_store Chroma.from_documents( documentsdocuments, embeddingembeddings, persist_directorypersist_directory ) # 显式持久化虽然from_documents通常会自动保存但显式调用更安全 vector_store.persist() print(f向量库已创建并保存至 {persist_directory}) return vector_store def load_existing_vector_store(persist_directory./chroma_db): 加载已存在的向量存储。 embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-small-zh-v1.5) vector_store Chroma( persist_directorypersist_directory, embedding_functionembeddings ) print(f已从 {persist_directory} 加载现有向量库。) return vector_store # 使用示例构建新库 if __name__ __main__: from file_processor import load_and_split_pdfs # 假设文档已处理 # documents load_and_split_pdfs([...]) # vs create_vector_store(documents) # 使用示例加载已有库 # vs load_existing_vector_store()3.3 检索与生成链路的集成有了向量库我们就可以实现检索和回答的完整链路了。这里使用langchain的RetrievalQA链。# rag_chain.py from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from dotenv import load_dotenv import os load_dotenv() def create_rag_chain(vector_store): 创建RAG问答链。 参数: vector_store: 已加载的向量存储对象。 返回: qa_chain: 可用于问答的链对象。 # 1. 初始化LLM (使用OpenAI兼容接口) llm ChatOpenAI( modelgpt-3.5-turbo, # 模型名根据你的API后端调整例如 qwen-turbo, llama3 openai_api_keyos.getenv(OPENAI_API_KEY), openai_api_baseos.getenv(OPENAI_API_BASE), temperature0.1, # 低温度使输出更确定更依赖上下文 streamingFalse, # 如需流式输出可设为True ) # 2. 定义自定义提示模板明确要求模型基于上下文回答 prompt_template 请严格根据以下提供的上下文信息来回答问题。如果上下文信息中没有明确答案请直接说“根据已知信息无法回答该问题”不要编造信息。 上下文 {context} 问题{question} 基于上下文的答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 3. 创建检索问答链 # chain_type 可选 stuff默认所有上下文塞进prompt, map_reduce, refine, map_rerank qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 简单场景下足够 retrievervector_store.as_retriever( search_typesimilarity, # 相似度搜索 search_kwargs{k: 4} # 检索返回的最相关文档数量 ), chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue, # 返回源文档便于追溯 ) return qa_chain def ask_question(qa_chain, question): 使用RAG链回答问题。 result qa_chain.invoke({query: question}) answer result[result] source_docs result[source_documents] print(f\n问题{question}) print(f答案{answer}) print(\n--- 参考来源 ---) for i, doc in enumerate(source_docs): print(f[来源{i1}] {doc.page_content[:150]}...) # 打印片段 print(f 元数据{doc.metadata}\n) return answer, source_docs # 主程序入口 if __name__ __main__: from vector_store_builder import load_existing_vector_store # 加载向量库 print(正在加载向量库...) vector_store load_existing_vector_store() # 创建RAG链 print(正在创建RAG问答链...) qa_chain create_rag_chain(vector_store) # 交互式问答 print(\n RAG 问答系统已就绪 ) print(输入 quit 或 退出 结束程序。) while True: user_input input(\n请输入您的问题) if user_input.lower() in [quit, 退出, exit]: break if user_input.strip(): ask_question(qa_chain, user_input)现在一个最基础的 RAG 系统就完成了。运行python rag_chain.py系统会加载向量库然后等待你输入问题。它会从向量库中检索相关文档片段并生成基于上下文的答案。4. 进阶实现 Agentic RAG 以优化检索流程基础 RAG 是线性的“检索-生成”。Agentic RAG 则引入决策和迭代。我们将实现一个简单的智能体它能够判断问题是否需要检索并在需要时执行检索。4.1 设计智能体工具与决策逻辑我们将使用langchain的智能体框架。首先我们需要定义两个关键工具一个“检索工具”和一个“直接回答工具”。智能体将根据对问题的分析决定调用哪个工具。# agentic_rag.py from langchain.agents import Tool, AgentExecutor, create_react_agent from langchain import hub from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.memory import ConversationBufferMemory from dotenv import load_dotenv import os load_dotenv() class AgenticRAGSystem: def __init__(self, vector_store): self.vector_store vector_store self.llm ChatOpenAI( modelgpt-3.5-turbo, openai_api_keyos.getenv(OPENAI_API_KEY), openai_api_baseos.getenv(OPENAI_API_BASE), temperature0, streamingFalse, ) self.memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) self._setup_tools() self._setup_agent() def _setup_tools(self): 定义智能体可用的工具。 # 工具1基于知识库检索并回答 def rag_qa_tool(input_text): from rag_chain import create_rag_chain qa_chain create_rag_chain(self.vector_store) result qa_chain.invoke({query: input_text}) return f答案{result[result]}\n\n参考来源{result[source_documents]} # 工具2直接对话用于通用问候、无需检索的问题 def direct_chat_tool(input_text): # 这里可以是一个简单的对话链或者直接让LLM回答 # 为简化我们直接调用LLM进行通用回复 from langchain_core.messages import HumanMessage messages [HumanMessage(contentinput_text)] response self.llm.invoke(messages) return response.content self.tools [ Tool( nameKnowledgeBaseQA, funcrag_qa_tool, description当用户的问题涉及到具体的产品信息、技术细节、文档内容、历史记录等需要从知识库中查找答案时使用此工具。输入应该是清晰的问题。 ), Tool( nameGeneralChat, funcdirect_chat_tool, description当用户进行一般性问候、闲聊、询问你的能力或者问题与知识库内容完全无关时使用此工具。例如‘你好’、‘你是谁’、‘今天天气怎么样’。 ), ] def _setup_agent(self): 创建智能体执行器。 # 从LangChain Hub拉取一个ReAct风格的提示词模板 prompt hub.pull(hwchase17/react-chat) # 创建智能体 agent create_react_agent(self.llm, self.tools, prompt) self.agent_executor AgentExecutor( agentagent, toolsself.tools, memoryself.memory, verboseTrue, # 设置为True可以看到智能体的思考过程 handle_parsing_errorsTrue, max_iterations3, # 限制最大迭代次数防止死循环 ) def ask(self, question): 向智能体提问。 try: result self.agent_executor.invoke({input: question}) return result[output] except Exception as e: return f处理问题时出现错误{e} # 使用示例 if __name__ __main__: from vector_store_builder import load_existing_vector_store print(正在加载向量库和初始化Agentic RAG系统...) vs load_existing_vector_store() agent_system AgenticRAGSystem(vs) print(\n Agentic RAG 系统已就绪 ) # 测试不同类型的问题 test_questions [ 你好介绍一下你自己。, # 应触发 GeneralChat 我们产品的主要特性是什么, # 应触发 KnowledgeBaseQA 今天的日期是什么, # 应触发 GeneralChat (知识库无此信息) 根据文档如何安装这个软件, # 应触发 KnowledgeBaseQA ] for q in test_questions: print(f\n用户{q}) answer agent_system.ask(q) print(f系统{answer}) print(- * 50)运行这个脚本你会看到智能体的“思考”过程因为verboseTrue。它会先判断问题类型然后决定调用哪个工具。例如对于“你好”它会调用GeneralChat对于“产品特性”它会调用KnowledgeBaseQA并从向量库中检索答案。4.2 实现查询重写与多轮检索更高级的 Agentic RAG 还可以在检索前对查询进行优化。例如如果用户问“它怎么用”智能体可以根据对话历史将其重写为更具体的“如何安装XX软件”。这可以通过在工具调用链中增加一个“查询重写”步骤来实现。# 在 AgenticRAGSystem 类中添加一个内部方法 def _rewrite_query_for_retrieval(self, original_query, chat_history): 根据对话历史重写查询以优化检索效果。 rewrite_prompt PromptTemplate.from_template( 你是一个查询优化助手。请根据对话历史和当前问题生成一个更清晰、更具体、更适合从知识库中检索文档的问题。 如果历史对话为空或与当前问题无关则直接优化当前问题即可。 对话历史 {history} 当前问题{question} 优化后的问题 ) from langchain_core.messages import HumanMessage, SystemMessage # 简化处理直接调用LLM messages [ SystemMessage(content你负责优化用户问题以便更好地检索知识库。), HumanMessage(contentrewrite_prompt.format(historychat_history, questionoriginal_query)) ] rewritten self.llm.invoke(messages).content.strip() print(f查询重写: {original_query} - {rewritten}) return rewritten # 然后修改 rag_qa_tool 函数在检索前先调用重写 def rag_qa_tool(input_text): # 获取当前对话历史简化表示 chat_history str(self.memory.chat_memory.messages[-4:]) if self.memory.chat_memory.messages else rewritten_query self._rewrite_query_for_retrieval(input_text, chat_history) # 使用重写后的查询进行检索 from rag_chain import create_rag_chain qa_chain create_rag_chain(self.vector_store) result qa_chain.invoke({query: rewritten_query}) return f基于优化查询‘{rewritten_query}’检索\n答案{result[result]}5. 系统验证、常见问题排查与生产优化构建完系统后必须进行系统的验证和测试并了解如何排查生产环境中可能出现的问题。5.1 如何验证你的 RAG 系统是否有效不要只问几个简单问题就认为系统成功了。需要设计系统的测试方案检索准确性测试准备一组标准问题Q和对应的标准文档片段A。运行系统检查检索到的 Top-K 个片段中是否包含 A以及排名是否靠前。答案忠实性测试检查 LLM 生成的答案是否严格基于检索到的上下文有没有“幻觉”出上下文不存在的信息。可以人工审核或使用“答案-上下文一致性”评估模型。边界测试无关问题问“今天天气如何”系统应礼貌拒绝或调用通用对话工具而不是从知识库胡编乱造。模糊问题问“那个功能”系统是否能结合对话历史如果有进行澄清或检索。复杂/多跳问题问“文档A中提到的XX方法和文档B中的YY方法有什么区别”系统是否能检索到多篇相关文档并进行综合比较。性能测试测量从提问到返回答案的端到端延迟特别是检索阶段的耗时。对于大量文档检索速度是关键。5.2 常见问题与排查路径问题现象可能原因检查点与解决方案答案与文档内容不符幻觉1. 检索到的上下文不相关。2. Prompt 未强制要求基于上下文回答。3. LLM 温度Temperature过高。1. 检查检索结果 (source_documents)看相关性。2. 强化 Prompt如添加“如果上下文未提及请说不知道”。3. 将 LLM 的temperature参数调低如 0.1。检索不到任何相关内容1. Embeddings 模型与文档语言/领域不匹配。2. 文本分块策略不合理太大或太小。3. 查询词太短或太模糊。4. 向量库未正确构建或加载。1. 尝试不同的 Embeddings 模型。2. 调整chunk_size和chunk_overlap尝试语义分块。3. 实现查询扩展或重写如前文 Agentic 部分。4. 检查向量库持久化路径确认文档已成功嵌入。回答“根据已知信息无法回答”过于频繁1. 检索数量k太小。2. 相似度阈值设置过高如果设置了。3. 知识库本身缺乏该信息。1. 适当增加search_kwargs{“k”: 4}中的k值。2. 在as_retriever()中调整score_threshold如果支持。3. 扩充知识库文档。系统响应速度慢1. Embeddings 模型推理慢特别是首次加载。2. 向量库规模大检索慢。3. LLM API 调用延迟高。1. 考虑使用更轻量级的 Embeddings 模型或启用 GPU。2. 对向量库建立索引如 HNSW或使用性能更高的向量数据库如 Qdrant。3. 检查网络或考虑使用延迟更低的 LLM 服务。无法处理长文档或复杂格式1. 文档加载器不支持该格式。2. 分块时破坏了表格、代码等结构。1. 使用unstructured库它支持更多格式。2. 尝试使用专门的分割器如MarkdownHeaderTextSplitter用于 Markdown。5.3 面向生产环境的优化建议向量数据库选型对于生产环境ChromaDB 的简单持久化可能不够。应考虑Milvus、Qdrant或PgVector如果已在用 PostgreSQL。它们支持分布式、高可用、更丰富的索引和过滤功能。Embeddings 模型部署将 Embeddings 模型部署为独立的推理服务如使用FastAPI封装而不是每次启动都加载。这可以提高并发处理能力并方便模型更新。检索优化混合检索结合稠密向量检索和传统的稀疏检索如 BM25兼顾语义匹配和关键词匹配提高召回率。重排序Rerank使用一个更精细但较慢的交叉编码器模型如BGE-reranker对初步检索到的结果进行重排提升 Top1 的准确率。元数据过滤在检索时加入过滤器如“只检索某年某部门的文档”这需要你在存储时保留并索引文档的元数据。链路可观测性记录每一次问答的日志包括原始问题、检索到的片段及得分、发送给 LLM 的完整 Prompt、LLM 的原始回复。这对于调试和优化至关重要。缓存策略对常见的、不变的查询结果进行缓存可以极大降低 LLM 调用成本和响应延迟。版本管理与回滚知识库更新时应有版本控制。当新文档导致回答质量下降时能快速回滚到上一个版本的向量库。从基础 RAG 到 Agentic RAG本质上是为系统增加了“思考”和“决策”的能力使其从静态的管道升级为动态的、自适应的智能体。这通常通过引入一个“规划器”LLM 来协调多个工具调用检索、重写、重排、生成等来实现。本文实现的智能体决策只是一个起点更复杂的架构如LangGraph可以用于描述包含循环、分支的复杂工作流。下一步你可以尝试将向量数据库更换为 Milvus实现混合检索和重排序或者用 LangGraph 构建一个支持多轮追问、自我验证的更强健的 Agentic RAG 系统。记住评估始终是核心在添加任何复杂功能前后都要用你的测试集去量化地衡量效果提升。