基于RAG与本地向量库的企业知识库实战：从原理到代码实现

最近在落地一个企业知识库项目时，团队反复被几个问题困扰：文档检索不准、回答经常“胡言乱语”、系统响应慢。经过一番折腾，我们最终基于 RAG（检索增强生成）技术栈，结合本地向量库和 Agentic 设计模式，构建了一个稳定、准确且高效的系统。本文将完整复盘这套实战方案，从核心概念到一行行代码，手把手带你搭建一个属于你自己的 RAG 知识库。无论你是想入门 LLM 应用开发，还是正在为现有项目寻找检索增强的优化方案，这篇文章都能提供从零到一的闭环指南。

1. 背景与核心概念：为什么需要 RAG？

在深入代码之前，我们必须先理清几个核心概念，明白我们到底要解决什么问题。

1.1 大模型的困境与 RAG 的诞生

大型语言模型（LLM）如 GPT-4、ChatGLM 等，拥有强大的理解和生成能力，但它们存在两个固有缺陷：

1. 知识滞后性：模型的训练数据有截止日期，无法获取最新信息。
2. 幻觉问题：当模型遇到知识盲区时，可能会“自信地”编造错误答案。

RAG（Retrieval-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 或 conda
• IDE：VS Code 或 PyCharm

项目依赖库：我们将主要使用langchain框架来简化流程，并搭配轻量级本地向量库Chroma。

# requirements.txt langchain==0.2.1 langchain-community==0.2.1 langchain-chroma==0.1.0 chromadb==0.4.24 sentence-transformers==2.7.0 pypdf==4.2.0 python-dotenv==1.0.1 openai==1.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_size=500-1000字符，chunk_overlap=100-200字符。重叠部分可以避免将完整句子或概念割裂。
• 按语义分割：利用句子边界、段落或自然语言处理（NLP）模型进行更智能的分割，能更好地保持语义完整性。
• 递归分割：先按大分隔符（如\n\n）分，如果块还是太大，再按小分隔符（如\n、空格）继续分。

# src/document_loader.py from langchain.text_splitter import RecursiveCharacterTextSplitter def get_text_splitter(chunk_size=800, chunk_overlap=150): """ 获取递归字符文本分割器。 Args: chunk_size: 每个文本块的最大字符数。 chunk_overlap: 相邻块之间的重叠字符数。 Returns: TextSplitter 实例。 """ # 分隔符优先级列表：先尝试按双换行分，再按单换行，再按句号，最后按空格 separators = ["\n\n", "\n", "。", "？", "！", "\.", "\?", "\!", " ", ""] text_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=separators, length_function=len, is_separator_regex=False, ) return text_splitter # 使用示例 splitter = get_text_splitter() documents = [{"page_content": "这是一个很长的文档内容..."}] chunks = splitter.split_documents(documents) print(f"原始文档被分割成了 {len(chunks)} 个块。")

最佳实践建议：对于技术文档、手册，chunk_size=600, chunk_overlap=100是一个不错的起点。需要通过实际问答效果进行微调。

3.2 嵌入模型（Embeddings）选型

嵌入模型负责将文本转换为向量。选择取决于对精度、速度和成本的要求。

主流选择：

1. 本地模型（推荐用于开发/内网）：
   • all-MiniLM-L6-v2：轻量级，速度快，中英文效果均衡，是入门首选。
   • bge-large-zh-v1.5：专为中文优化，在中文语义匹配任务上表现优异。
   • text2vec系列：国产优秀模型，中文场景效果好。
2. 云服务 API（用于生产/高精度需求）：
   • OpenAItext-embedding-3-small/3-large
   • Cohere Embed
   • 百度文心、智谱AI等国内厂商的嵌入API。

为什么优先考虑本地模型？

• 数据隐私：文档内容无需出公网。
• 零成本：无 API 调用费用。
• 稳定性：不依赖网络和第三方服务可用性。
• 延迟低：本地推理，响应快。

# src/embeddings.py from langchain.embeddings import HuggingFaceEmbeddings import sentence_transformers def get_local_embeddings(model_name="all-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_name=f"sentence-transformers/{model_name}", model_kwargs=model_kwargs, encode_kwargs=encode_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_name="knowledge_base"): """ 创建或连接一个持久化的Chroma向量库。 Args: persist_directory: 向量库数据持久化目录。 collection_name: 集合名称。 Returns: Chroma 向量库实例。 """ # 确保目录存在 os.makedirs(persist_directory, exist_ok=True) # 获取嵌入函数 embedding_function = get_local_embeddings() # 创建 Chroma 客户端，并启用持久化 client_settings = Settings( chroma_db_impl="duckdb+parquet", # 后端实现 persist_directory=persist_directory, anonymized_telemetry=False # 关闭匿名数据收集 ) # 使用 LangChain 的 Chroma 包装器 vector_store = Chroma( collection_name=collection_name, embedding_function=embedding_function, persist_directory=persist_directory, client_settings=client_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(documents=documents) # 持久化到磁盘 vector_store.persist() print(f"成功添加 {len(documents)} 个文档块到向量库。") if __name__ == "__main__": # 示例：初始化并添加文档 from langchain.schema import Document docs = [ Document(page_content="LangChain是一个用于开发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' langchain==0.2.1 langchain-community==0.2.1 langchain-chroma==0.1.0 chromadb==0.4.24 sentence-transformers==2.7.0 pypdf==4.2.0 python-dotenv==1.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.py

4.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_size=800, chunk_overlap=150): self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", "。", "？", "！", "\.", "\?", "\!", " ", ""], length_function=len, ) 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, encoding='utf-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_size=600, chunk_overlap=100) # 假设 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_directory=self.persist_dir, collection_name="corporate_knowledge" ) # 3. 清空现有集合（如果是重建） # self.vector_store.delete_collection() # 慎用，会删除所有数据 # self.vector_store = create_or_get_chroma_store(...) # 重新创建 # 4. 添加文档到向量库 self.vector_store.add_documents(documents=all_documents) self.vector_store.persist() print(f"知识库索引构建完成！共处理 {len(all_documents)} 个文本块。") def get_retriever(self, search_type="similarity", k=4): """ 获取检索器。 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_directory=self.persist_dir, collection_name="corporate_knowledge" ) # 从向量库创建检索器 self.retriever = self.vector_store.as_retriever( search_type=search_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(k=4) # 测试检索 test_query = "什么是RAG？" docs = retriever.invoke(test_query) print(f"查询: '{test_query}'") for i, doc in enumerate(docs): print(f"[结果{i+1}] {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_name="gpt-3.5-turbo", temperature=0.1): """ 初始化基础 RAG 问答机器人。 Args: model_name: 使用的 LLM 模型名称。 temperature: 生成温度，越低越确定。 """ # 初始化 LLM (这里以 OpenAI 为例，生产环境请替换为本地或国内模型) self.llm = ChatOpenAI( model_name=model_name, temperature=temperature, openai_api_key=os.getenv("OPENAI_API_KEY"), # 从环境变量读取 # 如需使用本地模型，例如 ChatGLM，可使用 langchain.llms 或自定义封装 # from langchain.llms import ChatGLM # self.llm = ChatGLM(endpoint_url="http://localhost:8000") ) self.qa_chain = None def create_chain(self, retriever): """ 创建检索问答链。 Args: retriever: 检索器实例。 """ # 定义提示模板，指导 LLM 基于上下文回答 prompt_template = """请严格根据以下上下文信息来回答问题。如果上下文没有提供足够的信息，请直接说“根据现有资料，我无法回答这个问题”，不要编造信息。 上下文： {context} 问题：{question} 请给出准确、简洁的回答：""" PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) # 创建 RetrievalQA 链 self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", # 将检索到的所有文档“塞”进上下文 retriever=retriever, chain_type_kwargs={"prompt": PROMPT}, return_source_documents=True, # 返回源文档用于追溯 ) 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_KEY=sk-...，或配置其他LLM # 1. 加载知识库检索器 builder = KnowledgeBaseBuilder() retriever = builder.get_retriever(k=4) # 2. 创建 RAG 问答机器人 bot = BasicRAGQABot(model_name="gpt-3.5-turbo") bot.create_chain(retriever) # 3. 进行问答 questions = [ "我们公司产品的保修期是多久？", "如何重置设备密码？", "请总结一下RAG技术的优点。" ] for q in questions: answer, sources = bot.ask(q) print(f"\nQ: {q}") print(f"A: {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_agent=None): """ 初始化具有自反思能力的 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(temperature=0, model_name="gpt-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(name="RAG_QA", func=rag_qa_tool, description="用于回答基于知识库的问题。输入问题。"), Tool(name="Query_Rewriter", func=query_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(agent=self.agent, tools=self.tools, verbose=True, handle_parsing_errors=True) def ask_with_reflection(self, question: str, max_iterations=3) -> 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 系统时，你可能会遇到以下典型问题。

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 选择：根据数据隐私和成本，选择云端 API（OpenAI, Anthropic, 国内大厂）或本地部署模型（ChatGLM3, Qwen, Llama 等）。
• 框架：LangChain/LangGraph是快速原型的好工具。对于更定制化、高性能的场景，可以考虑直接使用各组件（如sentence-transformers, 向量库 SDK）进行底层编排。

从零搭建一个 RAG 系统涉及多个环节，但核心脉络是清晰的：文档处理 -> 向量化 -> 存储 -> 检索 -> 生成。本文提供的代码是一个完整的起点，你可以在此基础上，根据“最佳实践”部分的建议，逐步优化检索效果、引入智能体逻辑、完善工程部署。