最近在尝试将大语言模型LLM应用到企业内部文档问答、智能客服等场景时你是否遇到过这样的困境模型对训练数据之外的知识一问三不知或者回答得似是而非、胡编乱造这正是RAG技术要解决的核心问题。本文将带你从零到一手把手构建一个企业级可用的RAG知识库系统涵盖从核心概念、技术选型、环境搭建、代码实战到性能优化的全链路。无论你是想快速入门RAG的开发者还是需要将RAG落地到实际项目的工程师都能从这篇实战指南中获得可直接复用的代码和清晰的架构思路。1. RAG核心概念与为什么需要它在深入代码之前我们必须先理解RAG是什么以及它为何成为连接大模型与私有知识的关键桥梁。1.1 什么是RAGRAG全称Retrieval-Augmented Generation即检索增强生成。它是一种将信息检索技术与大语言模型生成能力相结合的架构。其核心思想可以概括为先检索后生成。当用户提出一个问题Query时RAG系统不会让大模型凭空想象而是会先从外部的知识库如企业文档、数据库、网页中检索出与问题最相关的文档片段。然后将这些检索到的片段作为“参考依据”或“上下文”连同原始问题一起提交给大语言模型让模型基于这些可靠的依据来生成最终答案。1.2 RAG解决了什么问题传统大模型应用存在两大痛点知识滞后与静态性大模型的训练数据有截止日期无法获取最新信息或特定领域的非公开知识。幻觉Hallucination模型可能会生成看似合理但实际错误或虚构的内容这在需要高准确性的企业场景中是致命的。RAG通过引入外部知识源有效地缓解了以上问题知识实时更新只需更新知识库模型就能获取最新信息无需重新训练。答案有据可依生成的答案基于检索到的文档提高了可信度和可解释性。你可以追溯答案来源于哪份文档的哪个部分。成本与效率相比于为每个特定领域微调一个大模型成本高昂RAG是一种更轻量、更灵活的解决方案。1.3 一个典型的RAG系统工作流程一个完整的RAG系统通常包含两个主要阶段索引Indexing和检索生成Retrieval Generation。索引阶段线下进行文档加载从各种来源PDF、Word、TXT、网页、数据库加载原始文档。文档分割将长文档切分成语义连贯的小块Chunks以适应模型的上下文窗口。向量化使用嵌入模型Embedding Model将每个文本块转换为一个高维向量Vector。向量存储将这些向量及其对应的原始文本存储到向量数据库Vector Database中。检索生成阶段线上响应用户提问用户输入一个问题。问题向量化使用相同的嵌入模型将问题转换为向量。向量检索在向量数据库中搜索与问题向量最相似的几个文本块向量通常使用余弦相似度等度量方法。上下文构建将检索到的Top-K个相关文本块组合成提示词Prompt的上下文部分。增强生成将“上下文 原始问题”构成的完整提示词发送给大语言模型生成最终答案。2. 技术选型与环境准备为了构建一个企业级项目我们需要选择稳定、高效、生态丰富的技术组件。2.1 核心组件选型开发框架LangChain。它是一个用于开发LLM应用的强大框架提供了RAG所需的文档加载、文本分割、链Chain编排等高级抽象极大提升了开发效率。向量数据库Chroma。它轻量、易用、可嵌入式运行非常适合快速原型开发和中小规模项目。生产环境也可考虑Weaviate,Qdrant,Milvus等。嵌入模型text-embedding-3-small。我们使用OpenAI的嵌入模型API它性能优异接口简单。对于内网或离线环境可以选择BGE,Sentence-Transformers等开源模型。大语言模型GPT-3.5-Turbo。我们使用OpenAI的Chat Completion API作为生成模型。同样可根据需要替换为 Claude、国产大模型或本地部署的Llama等。Python环境Python 3.9。2.2 项目初始化与依赖安装首先创建项目目录并初始化虚拟环境。# 创建项目目录 mkdir enterprise-rag-tutorial cd enterprise-rag-tutorial # 创建虚拟环境 (可选但强烈推荐) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install langchain langchain-community langchain-openai chromadb pypdf这里我们安装了langchain: LangChain核心库。langchain-community: 社区维护的第三方集成。langchain-openai: OpenAI模型的官方LangChain集成。chromadb: 向量数据库Chroma的客户端。pypdf: 用于处理PDF文档。2.3 环境变量配置为了安全地管理API密钥我们使用环境变量。创建一个.env文件在项目根目录下# .env OPENAI_API_KEY你的-openai-api-key在代码中我们可以使用python-dotenv来加载。先安装它pip install python-dotenv。3. 构建知识库索引Indexing Pipeline这是RAG的基石。我们将创建一个脚本把本地PDF文档处理并存入向量数据库。3.1 文档加载与分割创建一个名为ingest.py的文件。# ingest.py import os from dotenv import load_dotenv from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 1. 加载环境变量 load_dotenv() openai_api_key os.getenv(OPENAI_API_KEY) # 2. 指定文档路径 pdf_directory ./docs # 假设你的PDF文档放在项目根目录的docs文件夹下 documents [] # 3. 遍历并加载所有PDF文档 for filename in os.listdir(pdf_directory): if filename.endswith(.pdf): pdf_path os.path.join(pdf_directory, filename) print(f正在加载: {filename}) loader PyPDFLoader(pdf_path) # loader.load() 返回一个Document对象列表每个对象包含页面内容和元数据 loaded_docs loader.load() documents.extend(loaded_docs) print(f共加载了 {len(documents)} 个文档页面。) # 4. 文本分割将长文档切分成小块 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个块的最大字符数 chunk_overlap200, # 块之间的重叠字符数保持上下文连贯 separators[\n\n, \n, 。, , , , , , ] # 分割符优先级 ) chunks text_splitter.split_documents(documents) print(f分割后得到 {len(chunks)} 个文本块。) # 5. 初始化嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small, openai_api_keyopenai_api_key) # 6. 创建并持久化向量存储 # persist_directory 指定向量数据库存储的本地路径 persist_directory ./chroma_db vectordb Chroma.from_documents( documentschunks, embeddingembeddings, persist_directorypersist_directory ) # 显式持久化到磁盘 vectordb.persist() print(f向量索引已创建并保存至: {persist_directory})关键参数解析chunk_size这是最重要的参数之一。太小会丢失上下文太大会降低检索精度并增加LLM处理负担。通常设置在500-1500之间需根据文档特点调整。chunk_overlap重叠部分可以防止一个完整的句子或概念被割裂到两个块中有助于提升检索质量。persist_directoryChroma会将向量数据持久化到本地这个目录下次启动时可以直接加载无需重新计算嵌入。运行前准备在项目根目录创建docs文件夹。放入几个用于测试的PDF文档可以是产品手册、公司制度等。确保.env文件中的OPENAI_API_KEY已正确设置。运行脚本python ingest.py如果一切顺利你会看到加载、分割、存储的日志并在项目根目录下生成一个chroma_db文件夹。4. 实现检索与问答链Retrieval Generation索引构建好后我们就可以实现问答功能了。创建query.py文件。4.1 基础检索式问答# query.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA # 1. 加载环境变量 load_dotenv() openai_api_key os.getenv(OPENAI_API_KEY) # 2. 加载已存在的向量数据库 persist_directory ./chroma_db embeddings OpenAIEmbeddings(modeltext-embedding-3-small, openai_api_keyopenai_api_key) vectordb Chroma(persist_directorypersist_directory, embedding_functionembeddings) # 3. 将向量数据库转换为检索器 (Retriever) # search_kwargs 控制检索行为k4 表示返回最相似的4个文本块 retriever vectordb.as_retriever(search_kwargs{k: 4}) # 4. 初始化大语言模型 llm ChatOpenAI(model_namegpt-3.5-turbo, temperature0, openai_api_keyopenai_api_key) # temperature0 使输出更确定、更少随机性适合事实性问答。 # 5. 创建检索增强生成链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最常用的链类型将所有检索到的文档“塞”进上下文 retrieverretriever, return_source_documentsTrue, # 返回源文档便于追溯 verboseTrue # 打印链的详细执行过程调试时非常有用 ) # 6. 进行问答 if __name__ __main__: while True: query input(\n请输入您的问题 (输入 quit 退出): ) if query.lower() quit: break if query.strip() : continue # 调用链 result qa_chain.invoke({query: query}) print(f\n【问题】: {query}) print(f\n【答案】: {result[result]}) print(f\n【参考来源】:) for i, doc in enumerate(result[source_documents]): print(f 片段 {i1}: {doc.page_content[:200]}...) # 打印前200个字符 print(f 来源: {doc.metadata.get(source, N/A)}, 页码: {doc.metadata.get(page, N/A)}) print(- * 50)代码解析Retriever检索器是LangChain的核心抽象它封装了从向量数据库查询相似向量的逻辑。RetrievalQA这是一个高级链它内部自动完成了“检索 - 组合上下文 - 调用LLM生成答案”的流程。chain_typestuff是最简单直接的方式。return_source_documentsTrue这个参数至关重要它让我们能够看到答案是基于哪些文档片段生成的增强了系统的可解释性和可信度。运行脚本并测试python query.py尝试问一些你文档中明确包含的问题观察模型是否能给出准确答案并列出正确的来源。4.2 优化提示词工程基础的RetrievalQA使用的默认提示词可能不够精准。我们可以自定义提示词来提升回答质量。# query_with_custom_prompt.py from langchain.prompts import PromptTemplate from langchain.chains import RetrievalQA # ... (前面的加载向量库、检索器、LLM的代码与之前相同) ... # 自定义提示词模板 prompt_template 请根据以下上下文信息回答用户的问题。如果上下文信息不足以回答问题请直接说“根据提供的信息无法回答此问题”不要编造信息。 上下文信息 {context} 用户问题{question} 请用中文给出专业、清晰的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 使用自定义提示词创建QA链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverretriever, chain_type_kwargs{prompt: PROMPT}, # 传入自定义提示词 return_source_documentsTrue, verboseFalse ) # ... (后续的问答循环代码与之前相同) ...这个自定义提示词做了几件事明确指令要求模型基于上下文回答。处理未知指示模型在信息不足时承认而不是幻觉。格式要求指定用中文回答并要求专业清晰。5. 企业级项目实战构建一个简单的Web问答应用现在我们将上面的核心功能封装成一个简单的Flask Web应用使其更接近真实产品。5.1 项目结构enterprise-rag-tutorial/ ├── docs/ # 存放原始文档 ├── chroma_db/ # 向量数据库存储目录由ingest.py生成 ├── app.py # Flask主应用 ├── ingest.py # 文档处理脚本 ├── rag_core.py # RAG核心逻辑封装 ├── requirements.txt # 项目依赖 ├── .env # 环境变量 └── templates/ # HTML模板 └── index.html5.2 封装RAG核心逻辑创建rag_core.py将加载向量库和生成答案的逻辑模块化。# rag_core.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate load_dotenv() class RAGSystem: def __init__(self, persist_directory./chroma_db): self.openai_api_key os.getenv(OPENAI_API_KEY) self.persist_directory persist_directory self.embeddings None self.vectordb None self.retriever None self.llm None self.qa_chain None self._initialize_components() def _initialize_components(self): 初始化所有必要组件 # 1. 嵌入模型 self.embeddings OpenAIEmbeddings( modeltext-embedding-3-small, openai_api_keyself.openai_api_key ) # 2. 加载向量数据库 self.vectordb Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) # 3. 创建检索器 self.retriever self.vectordb.as_retriever(search_kwargs{k: 4}) # 4. 初始化LLM self.llm ChatOpenAI( model_namegpt-3.5-turbo, temperature0, openai_api_keyself.openai_api_key ) # 5. 自定义提示词 prompt_template 你是一个专业的知识库助手。请严格根据以下上下文信息回答问题。如果上下文没有提供相关信息请如实告知用户你不知道不要编造答案。 上下文 {context} 问题{question} 请根据上下文提供准确、有用的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 6. 创建QA链 self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrieverself.retriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue ) def ask(self, question): 向RAG系统提问 if not question.strip(): return {answer: 问题不能为空。, sources: []} try: result self.qa_chain.invoke({query: question}) answer result[result] sources [] for doc in result[source_documents]: source_info { content_preview: doc.page_content[:150] ..., source: doc.metadata.get(source, 未知文档), page: doc.metadata.get(page, N/A) } sources.append(source_info) return {answer: answer, sources: sources} except Exception as e: return {answer: f系统处理问题时出现错误{str(e)}, sources: []} # 全局RAG系统实例 rag_system RAGSystem()5.3 创建Flask Web应用创建app.py。# app.py from flask import Flask, render_template, request, jsonify from rag_core import rag_system app Flask(__name__) app.route(/) def index(): 渲染首页 return render_template(index.html) app.route(/ask, methods[POST]) def ask_question(): 处理问答请求的API端点 data request.get_json() question data.get(question, ).strip() if not question: return jsonify({error: 问题内容为空}), 400 # 调用RAG系统 response rag_system.ask(question) return jsonify(response) if __name__ __main__: # 确保在运行前已经通过 ingest.py 创建了向量索引 print(启动企业级RAG知识库问答系统...) app.run(debugTrue, port5000)5.4 创建前端界面创建templates/index.html。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title企业级RAG知识库问答系统/title style body { font-family: sans-serif; max-width: 800px; margin: 40px auto; padding: 20px; } .container { border: 1px solid #ddd; border-radius: 8px; padding: 30px; } h1 { color: #333; text-align: center; } #question-input { width: 100%; padding: 12px; font-size: 16px; margin-bottom: 15px; border: 1px solid #ccc; border-radius: 4px; box-sizing: border-box; } #ask-btn { background-color: #007bff; color: white; padding: 12px 25px; border: none; border-radius: 4px; cursor: pointer; font-size: 16px; } #ask-btn:hover { background-color: #0056b3; } #answer-area, #sources-area { margin-top: 30px; padding: 20px; background-color: #f9f9f9; border-radius: 4px; white-space: pre-wrap; } .source-item { border-bottom: 1px dashed #ccc; padding: 10px 0; } .loading { display: none; color: #666; text-align: center; } /style /head body div classcontainer h1 企业知识库智能问答/h1 p基于RAG技术从上传的文档中获取精准答案。/p div textarea idquestion-input rows4 placeholder请输入您关于文档的问题.../textarea button idask-btn onclickaskQuestion()提问/button div idloading classloading正在思考中.../div /div div idanswer-area styledisplay:none; h3 答案/h3 p idanswer-text/p /div div idsources-area styledisplay:none; h3 参考来源/h3 div idsources-list/div /div /div script async function askQuestion() { const questionInput document.getElementById(question-input); const question questionInput.value.trim(); const answerArea document.getElementById(answer-area); const answerText document.getElementById(answer-text); const sourcesArea document.getElementById(sources-area); const sourcesList document.getElementById(sources-list); const loading document.getElementById(loading); if (!question) { alert(请输入问题); return; } // 显示加载中清空之前的结果 loading.style.display block; answerArea.style.display none; sourcesArea.style.display none; try { const response await fetch(/ask, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question: question }) }); const data await response.json(); loading.style.display none; if (response.ok) { // 显示答案 answerText.textContent data.answer; answerArea.style.display block; // 显示来源 if (data.sources data.sources.length 0) { sourcesList.innerHTML ; data.sources.forEach((source, idx) { const sourceDiv document.createElement(div); sourceDiv.className source-item; sourceDiv.innerHTML strong片段 ${idx 1}:/strong ${source.content_preview}br smallem文档: ${source.source} (页码: ${source.page})/em/small ; sourcesList.appendChild(sourceDiv); }); sourcesArea.style.display block; } else { sourcesArea.style.display none; } } else { alert(请求失败 (data.error || 未知错误)); } } catch (error) { loading.style.display none; alert(网络请求出错 error.message); } } // 支持按回车键提交 document.getElementById(question-input).addEventListener(keypress, function(e) { if (e.key Enter !e.shiftKey) { e.preventDefault(); askQuestion(); } }); /script /body /html5.5 运行完整应用确保索引已存在如果你还没运行过ingest.py请先运行它来创建向量数据库。python ingest.py安装Flaskpip install flask创建requirements.txtpip freeze requirements.txt方便他人部署启动Flask应用python app.py打开浏览器访问http://127.0.0.1:5000即可看到问答界面。输入问题系统会从你的docs/文件夹下的文档中检索并生成答案同时展示参考来源。6. 进阶优化与最佳实践一个基础RAG系统搭建完成了但要达到企业级可用还需要考虑以下优化点。6.1 检索优化策略多路召回与混合排序不要只依赖向量相似度。可以结合关键词检索如BM25从不同角度召回文档然后进行融合重排序提升召回率。元数据过滤在检索时可以附加过滤条件。例如只检索“技术部门”的文档或“2023年之后”的文档。这需要你在索引阶段为每个文本块添加丰富的元数据如部门、日期、文档类型。# 示例创建带元数据的检索器 retriever vectordb.as_retriever( search_kwargs{ k: 5, filter: {department: 技术部} # 根据元数据过滤 } )查询转换与扩展用户的原始问题可能不够精确。可以使用LLM对查询进行重写、扩展或分解。例如将“它怎么工作”在特定上下文中重写为“XX产品的工作原理是什么”。6.2 上下文处理与提示词工程超越“Stuff”链chain_typestuff简单但可能超出模型上下文长度。对于大量检索结果可以使用“map_reduce”、“refine”等更复杂的链类型它们能处理更长的上下文但调用LLM的次数更多成本更高。上下文压缩检索到的文档可能包含无关信息。可以使用ContextualCompressionRetriever在将文档送入LLM前先用一个更小的模型过滤掉不相关的句子。提示词迭代根据业务场景精心设计提示词。明确角色“你是一个专业的法律顾问”、规定格式“请用列表形式回答”、要求提供引用“在答案后注明来源页码”。6.3 性能与可观测性异步处理对于Web应用使用异步框架如FastAPI和异步的LangChain调用可以提高并发处理能力。缓存对频繁出现的相似查询结果进行缓存可以显著降低LLM API调用成本和延迟。日志与监控记录每一次问答的查询、检索到的文档、生成的答案、耗时和Token使用量。这对于分析系统表现、优化成本和排查问题至关重要。评估建立评估体系从答案相关性、事实准确性、信息完整性等维度评估RAG系统的输出质量。可以使用LLM作为裁判LLM-as-a-Judge进行自动评估。6.4 生产环境部署考量向量数据库选型Chroma适合原型和中小数据量。生产环境应考虑支持分布式、持久化、高可用的向量数据库如Qdrant、Weaviate、Milvus或PGVector如果已用PostgreSQL。嵌入模型部署如果涉及敏感数据或需要控制成本可以考虑在本地或内网部署开源的嵌入模型如BGE、E5替代OpenAI API。大模型选型根据数据敏感性、响应速度、成本预算选择适合的LLM。可以是云端APIGPT、Claude、文心一言等也可以是本地部署的模型Llama、Qwen、ChatGLM等。安全性确保知识库上传接口有严格的权限控制和文件类型、大小、病毒扫描。对用户输入进行必要的清洗和过滤防止提示词注入攻击。7. 常见问题与排查思路在开发和运行过程中你可能会遇到以下典型问题。问题现象可能原因排查与解决思路运行ingest.py时报错OpenAI API连接失败1. API密钥未设置或错误。2. 网络问题或代理配置。1. 检查.env文件格式和变量名是否正确。2. 在代码中打印os.getenv(“OPENAI_API_KEY”)的前几位确认。3. 检查网络连通性。问答时返回“根据提供的信息无法回答”但文档中明明有答案1. 检索到的文本块不相关。2. 文本块分割不合理太大或太小。3. 提示词不够明确。1. 检查检索到的源文档 (source_documents) 是否真的相关。2. 调整ingest.py中的chunk_size和chunk_overlap。3. 优化提示词明确要求“基于上下文”。4. 尝试增加检索数量 (k)。答案看起来是胡编乱造的幻觉1. 检索到的上下文质量差或不足。2. LLM的temperature参数过高。3. 提示词未限制模型编造。1. 确保检索环节有效检查源文档。2. 将LLM的temperature设为0或更低值。3. 在提示词中加入“如果上下文没有提供相关信息请说不知道”。处理长文档或大量文档时速度很慢1. 嵌入模型调用是主要瓶颈。2. 未使用批处理。3. 本地计算资源不足。1. 对于OpenAI API检查是否有速率限制考虑升级。2. 确保使用支持批处理的嵌入函数。3. 对于本地模型考虑使用GPU加速。Web应用运行正常但问答无结果或报错1. 向量数据库路径错误或未初始化。2. Flask应用与RAG核心模块路径或实例化问题。1. 确认persist_directory路径正确且chroma_db文件夹存在。2. 检查rag_core.py中RAGSystem初始化是否成功可在__init__中加入日志。3. 查看Flask应用的错误日志。构建一个健壮的企业级RAG系统是一个迭代过程。从本文介绍的最小可行产品MVP开始你可以逐步引入更复杂的检索策略、更精细的上下文管理、更强大的LLM以及完善的监控评估体系。记住RAG项目的成功不仅在于技术栈的选择更在于对业务需求的深刻理解、对数据质量的严格把控以及对系统效果的持续评估与优化。建议先从核心业务的一个小范围、高质量的知识库开始试点积累经验后再逐步扩大范围。 
长河君 · 2026/8/24 21:03:58 · 阅读 1.2k 从零构建企业级RAG知识库：基于LangChain与Chroma的实战指南
最近在尝试将大语言模型LLM应用到企业内部文档问答、智能客服等场景时你是否遇到过这样的困境模型对训练数据之外的知识一问三不知或者回答得似是而非、胡编乱造这正是RAG技术要解决的核心问题。本文将带你从零到一手把手构建一个企业级可用的RAG知识库系统涵盖从核心概念、技术选型、环境搭建、代码实战到性能优化的全链路。无论你是想快速入门RAG的开发者还是需要将RAG落地到实际项目的工程师都能从这篇实战指南中获得可直接复用的代码和清晰的架构思路。1. RAG核心概念与为什么需要它在深入代码之前我们必须先理解RAG是什么以及它为何成为连接大模型与私有知识的关键桥梁。1.1 什么是RAGRAG全称Retrieval-Augmented Generation即检索增强生成。它是一种将信息检索技术与大语言模型生成能力相结合的架构。其核心思想可以概括为先检索后生成。当用户提出一个问题Query时RAG系统不会让大模型凭空想象而是会先从外部的知识库如企业文档、数据库、网页中检索出与问题最相关的文档片段。然后将这些检索到的片段作为“参考依据”或“上下文”连同原始问题一起提交给大语言模型让模型基于这些可靠的依据来生成最终答案。1.2 RAG解决了什么问题传统大模型应用存在两大痛点知识滞后与静态性大模型的训练数据有截止日期无法获取最新信息或特定领域的非公开知识。幻觉Hallucination模型可能会生成看似合理但实际错误或虚构的内容这在需要高准确性的企业场景中是致命的。RAG通过引入外部知识源有效地缓解了以上问题知识实时更新只需更新知识库模型就能获取最新信息无需重新训练。答案有据可依生成的答案基于检索到的文档提高了可信度和可解释性。你可以追溯答案来源于哪份文档的哪个部分。成本与效率相比于为每个特定领域微调一个大模型成本高昂RAG是一种更轻量、更灵活的解决方案。1.3 一个典型的RAG系统工作流程一个完整的RAG系统通常包含两个主要阶段索引Indexing和检索生成Retrieval Generation。索引阶段线下进行文档加载从各种来源PDF、Word、TXT、网页、数据库加载原始文档。文档分割将长文档切分成语义连贯的小块Chunks以适应模型的上下文窗口。向量化使用嵌入模型Embedding Model将每个文本块转换为一个高维向量Vector。向量存储将这些向量及其对应的原始文本存储到向量数据库Vector Database中。检索生成阶段线上响应用户提问用户输入一个问题。问题向量化使用相同的嵌入模型将问题转换为向量。向量检索在向量数据库中搜索与问题向量最相似的几个文本块向量通常使用余弦相似度等度量方法。上下文构建将检索到的Top-K个相关文本块组合成提示词Prompt的上下文部分。增强生成将“上下文 原始问题”构成的完整提示词发送给大语言模型生成最终答案。2. 技术选型与环境准备为了构建一个企业级项目我们需要选择稳定、高效、生态丰富的技术组件。2.1 核心组件选型开发框架LangChain。它是一个用于开发LLM应用的强大框架提供了RAG所需的文档加载、文本分割、链Chain编排等高级抽象极大提升了开发效率。向量数据库Chroma。它轻量、易用、可嵌入式运行非常适合快速原型开发和中小规模项目。生产环境也可考虑Weaviate,Qdrant,Milvus等。嵌入模型text-embedding-3-small。我们使用OpenAI的嵌入模型API它性能优异接口简单。对于内网或离线环境可以选择BGE,Sentence-Transformers等开源模型。大语言模型GPT-3.5-Turbo。我们使用OpenAI的Chat Completion API作为生成模型。同样可根据需要替换为 Claude、国产大模型或本地部署的Llama等。Python环境Python 3.9。2.2 项目初始化与依赖安装首先创建项目目录并初始化虚拟环境。# 创建项目目录 mkdir enterprise-rag-tutorial cd enterprise-rag-tutorial # 创建虚拟环境 (可选但强烈推荐) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install langchain langchain-community langchain-openai chromadb pypdf这里我们安装了langchain: LangChain核心库。langchain-community: 社区维护的第三方集成。langchain-openai: OpenAI模型的官方LangChain集成。chromadb: 向量数据库Chroma的客户端。pypdf: 用于处理PDF文档。2.3 环境变量配置为了安全地管理API密钥我们使用环境变量。创建一个.env文件在项目根目录下# .env OPENAI_API_KEY你的-openai-api-key在代码中我们可以使用python-dotenv来加载。先安装它pip install python-dotenv。3. 构建知识库索引Indexing Pipeline这是RAG的基石。我们将创建一个脚本把本地PDF文档处理并存入向量数据库。3.1 文档加载与分割创建一个名为ingest.py的文件。# ingest.py import os from dotenv import load_dotenv from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 1. 加载环境变量 load_dotenv() openai_api_key os.getenv(OPENAI_API_KEY) # 2. 指定文档路径 pdf_directory ./docs # 假设你的PDF文档放在项目根目录的docs文件夹下 documents [] # 3. 遍历并加载所有PDF文档 for filename in os.listdir(pdf_directory): if filename.endswith(.pdf): pdf_path os.path.join(pdf_directory, filename) print(f正在加载: {filename}) loader PyPDFLoader(pdf_path) # loader.load() 返回一个Document对象列表每个对象包含页面内容和元数据 loaded_docs loader.load() documents.extend(loaded_docs) print(f共加载了 {len(documents)} 个文档页面。) # 4. 文本分割将长文档切分成小块 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个块的最大字符数 chunk_overlap200, # 块之间的重叠字符数保持上下文连贯 separators[\n\n, \n, 。, , , , , , ] # 分割符优先级 ) chunks text_splitter.split_documents(documents) print(f分割后得到 {len(chunks)} 个文本块。) # 5. 初始化嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small, openai_api_keyopenai_api_key) # 6. 创建并持久化向量存储 # persist_directory 指定向量数据库存储的本地路径 persist_directory ./chroma_db vectordb Chroma.from_documents( documentschunks, embeddingembeddings, persist_directorypersist_directory ) # 显式持久化到磁盘 vectordb.persist() print(f向量索引已创建并保存至: {persist_directory})关键参数解析chunk_size这是最重要的参数之一。太小会丢失上下文太大会降低检索精度并增加LLM处理负担。通常设置在500-1500之间需根据文档特点调整。chunk_overlap重叠部分可以防止一个完整的句子或概念被割裂到两个块中有助于提升检索质量。persist_directoryChroma会将向量数据持久化到本地这个目录下次启动时可以直接加载无需重新计算嵌入。运行前准备在项目根目录创建docs文件夹。放入几个用于测试的PDF文档可以是产品手册、公司制度等。确保.env文件中的OPENAI_API_KEY已正确设置。运行脚本python ingest.py如果一切顺利你会看到加载、分割、存储的日志并在项目根目录下生成一个chroma_db文件夹。4. 实现检索与问答链Retrieval Generation索引构建好后我们就可以实现问答功能了。创建query.py文件。4.1 基础检索式问答# query.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA # 1. 加载环境变量 load_dotenv() openai_api_key os.getenv(OPENAI_API_KEY) # 2. 加载已存在的向量数据库 persist_directory ./chroma_db embeddings OpenAIEmbeddings(modeltext-embedding-3-small, openai_api_keyopenai_api_key) vectordb Chroma(persist_directorypersist_directory, embedding_functionembeddings) # 3. 将向量数据库转换为检索器 (Retriever) # search_kwargs 控制检索行为k4 表示返回最相似的4个文本块 retriever vectordb.as_retriever(search_kwargs{k: 4}) # 4. 初始化大语言模型 llm ChatOpenAI(model_namegpt-3.5-turbo, temperature0, openai_api_keyopenai_api_key) # temperature0 使输出更确定、更少随机性适合事实性问答。 # 5. 创建检索增强生成链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最常用的链类型将所有检索到的文档“塞”进上下文 retrieverretriever, return_source_documentsTrue, # 返回源文档便于追溯 verboseTrue # 打印链的详细执行过程调试时非常有用 ) # 6. 进行问答 if __name__ __main__: while True: query input(\n请输入您的问题 (输入 quit 退出): ) if query.lower() quit: break if query.strip() : continue # 调用链 result qa_chain.invoke({query: query}) print(f\n【问题】: {query}) print(f\n【答案】: {result[result]}) print(f\n【参考来源】:) for i, doc in enumerate(result[source_documents]): print(f 片段 {i1}: {doc.page_content[:200]}...) # 打印前200个字符 print(f 来源: {doc.metadata.get(source, N/A)}, 页码: {doc.metadata.get(page, N/A)}) print(- * 50)代码解析Retriever检索器是LangChain的核心抽象它封装了从向量数据库查询相似向量的逻辑。RetrievalQA这是一个高级链它内部自动完成了“检索 - 组合上下文 - 调用LLM生成答案”的流程。chain_typestuff是最简单直接的方式。return_source_documentsTrue这个参数至关重要它让我们能够看到答案是基于哪些文档片段生成的增强了系统的可解释性和可信度。运行脚本并测试python query.py尝试问一些你文档中明确包含的问题观察模型是否能给出准确答案并列出正确的来源。4.2 优化提示词工程基础的RetrievalQA使用的默认提示词可能不够精准。我们可以自定义提示词来提升回答质量。# query_with_custom_prompt.py from langchain.prompts import PromptTemplate from langchain.chains import RetrievalQA # ... (前面的加载向量库、检索器、LLM的代码与之前相同) ... # 自定义提示词模板 prompt_template 请根据以下上下文信息回答用户的问题。如果上下文信息不足以回答问题请直接说“根据提供的信息无法回答此问题”不要编造信息。 上下文信息 {context} 用户问题{question} 请用中文给出专业、清晰的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 使用自定义提示词创建QA链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverretriever, chain_type_kwargs{prompt: PROMPT}, # 传入自定义提示词 return_source_documentsTrue, verboseFalse ) # ... (后续的问答循环代码与之前相同) ...这个自定义提示词做了几件事明确指令要求模型基于上下文回答。处理未知指示模型在信息不足时承认而不是幻觉。格式要求指定用中文回答并要求专业清晰。5. 企业级项目实战构建一个简单的Web问答应用现在我们将上面的核心功能封装成一个简单的Flask Web应用使其更接近真实产品。5.1 项目结构enterprise-rag-tutorial/ ├── docs/ # 存放原始文档 ├── chroma_db/ # 向量数据库存储目录由ingest.py生成 ├── app.py # Flask主应用 ├── ingest.py # 文档处理脚本 ├── rag_core.py # RAG核心逻辑封装 ├── requirements.txt # 项目依赖 ├── .env # 环境变量 └── templates/ # HTML模板 └── index.html5.2 封装RAG核心逻辑创建rag_core.py将加载向量库和生成答案的逻辑模块化。# rag_core.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate load_dotenv() class RAGSystem: def __init__(self, persist_directory./chroma_db): self.openai_api_key os.getenv(OPENAI_API_KEY) self.persist_directory persist_directory self.embeddings None self.vectordb None self.retriever None self.llm None self.qa_chain None self._initialize_components() def _initialize_components(self): 初始化所有必要组件 # 1. 嵌入模型 self.embeddings OpenAIEmbeddings( modeltext-embedding-3-small, openai_api_keyself.openai_api_key ) # 2. 加载向量数据库 self.vectordb Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) # 3. 创建检索器 self.retriever self.vectordb.as_retriever(search_kwargs{k: 4}) # 4. 初始化LLM self.llm ChatOpenAI( model_namegpt-3.5-turbo, temperature0, openai_api_keyself.openai_api_key ) # 5. 自定义提示词 prompt_template 你是一个专业的知识库助手。请严格根据以下上下文信息回答问题。如果上下文没有提供相关信息请如实告知用户你不知道不要编造答案。 上下文 {context} 问题{question} 请根据上下文提供准确、有用的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 6. 创建QA链 self.qa_chain RetrievalQA.from_chain_type( llmself.llm, chain_typestuff, retrieverself.retriever, chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue ) def ask(self, question): 向RAG系统提问 if not question.strip(): return {answer: 问题不能为空。, sources: []} try: result self.qa_chain.invoke({query: question}) answer result[result] sources [] for doc in result[source_documents]: source_info { content_preview: doc.page_content[:150] ..., source: doc.metadata.get(source, 未知文档), page: doc.metadata.get(page, N/A) } sources.append(source_info) return {answer: answer, sources: sources} except Exception as e: return {answer: f系统处理问题时出现错误{str(e)}, sources: []} # 全局RAG系统实例 rag_system RAGSystem()5.3 创建Flask Web应用创建app.py。# app.py from flask import Flask, render_template, request, jsonify from rag_core import rag_system app Flask(__name__) app.route(/) def index(): 渲染首页 return render_template(index.html) app.route(/ask, methods[POST]) def ask_question(): 处理问答请求的API端点 data request.get_json() question data.get(question, ).strip() if not question: return jsonify({error: 问题内容为空}), 400 # 调用RAG系统 response rag_system.ask(question) return jsonify(response) if __name__ __main__: # 确保在运行前已经通过 ingest.py 创建了向量索引 print(启动企业级RAG知识库问答系统...) app.run(debugTrue, port5000)5.4 创建前端界面创建templates/index.html。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title企业级RAG知识库问答系统/title style body { font-family: sans-serif; max-width: 800px; margin: 40px auto; padding: 20px; } .container { border: 1px solid #ddd; border-radius: 8px; padding: 30px; } h1 { color: #333; text-align: center; } #question-input { width: 100%; padding: 12px; font-size: 16px; margin-bottom: 15px; border: 1px solid #ccc; border-radius: 4px; box-sizing: border-box; } #ask-btn { background-color: #007bff; color: white; padding: 12px 25px; border: none; border-radius: 4px; cursor: pointer; font-size: 16px; } #ask-btn:hover { background-color: #0056b3; } #answer-area, #sources-area { margin-top: 30px; padding: 20px; background-color: #f9f9f9; border-radius: 4px; white-space: pre-wrap; } .source-item { border-bottom: 1px dashed #ccc; padding: 10px 0; } .loading { display: none; color: #666; text-align: center; } /style /head body div classcontainer h1 企业知识库智能问答/h1 p基于RAG技术从上传的文档中获取精准答案。/p div textarea idquestion-input rows4 placeholder请输入您关于文档的问题.../textarea button idask-btn onclickaskQuestion()提问/button div idloading classloading正在思考中.../div /div div idanswer-area styledisplay:none; h3 答案/h3 p idanswer-text/p /div div idsources-area styledisplay:none; h3 参考来源/h3 div idsources-list/div /div /div script async function askQuestion() { const questionInput document.getElementById(question-input); const question questionInput.value.trim(); const answerArea document.getElementById(answer-area); const answerText document.getElementById(answer-text); const sourcesArea document.getElementById(sources-area); const sourcesList document.getElementById(sources-list); const loading document.getElementById(loading); if (!question) { alert(请输入问题); return; } // 显示加载中清空之前的结果 loading.style.display block; answerArea.style.display none; sourcesArea.style.display none; try { const response await fetch(/ask, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question: question }) }); const data await response.json(); loading.style.display none; if (response.ok) { // 显示答案 answerText.textContent data.answer; answerArea.style.display block; // 显示来源 if (data.sources data.sources.length 0) { sourcesList.innerHTML ; data.sources.forEach((source, idx) { const sourceDiv document.createElement(div); sourceDiv.className source-item; sourceDiv.innerHTML strong片段 ${idx 1}:/strong ${source.content_preview}br smallem文档: ${source.source} (页码: ${source.page})/em/small ; sourcesList.appendChild(sourceDiv); }); sourcesArea.style.display block; } else { sourcesArea.style.display none; } } else { alert(请求失败 (data.error || 未知错误)); } } catch (error) { loading.style.display none; alert(网络请求出错 error.message); } } // 支持按回车键提交 document.getElementById(question-input).addEventListener(keypress, function(e) { if (e.key Enter !e.shiftKey) { e.preventDefault(); askQuestion(); } }); /script /body /html5.5 运行完整应用确保索引已存在如果你还没运行过ingest.py请先运行它来创建向量数据库。python ingest.py安装Flaskpip install flask创建requirements.txtpip freeze requirements.txt方便他人部署启动Flask应用python app.py打开浏览器访问http://127.0.0.1:5000即可看到问答界面。输入问题系统会从你的docs/文件夹下的文档中检索并生成答案同时展示参考来源。6. 进阶优化与最佳实践一个基础RAG系统搭建完成了但要达到企业级可用还需要考虑以下优化点。6.1 检索优化策略多路召回与混合排序不要只依赖向量相似度。可以结合关键词检索如BM25从不同角度召回文档然后进行融合重排序提升召回率。元数据过滤在检索时可以附加过滤条件。例如只检索“技术部门”的文档或“2023年之后”的文档。这需要你在索引阶段为每个文本块添加丰富的元数据如部门、日期、文档类型。# 示例创建带元数据的检索器 retriever vectordb.as_retriever( search_kwargs{ k: 5, filter: {department: 技术部} # 根据元数据过滤 } )查询转换与扩展用户的原始问题可能不够精确。可以使用LLM对查询进行重写、扩展或分解。例如将“它怎么工作”在特定上下文中重写为“XX产品的工作原理是什么”。6.2 上下文处理与提示词工程超越“Stuff”链chain_typestuff简单但可能超出模型上下文长度。对于大量检索结果可以使用“map_reduce”、“refine”等更复杂的链类型它们能处理更长的上下文但调用LLM的次数更多成本更高。上下文压缩检索到的文档可能包含无关信息。可以使用ContextualCompressionRetriever在将文档送入LLM前先用一个更小的模型过滤掉不相关的句子。提示词迭代根据业务场景精心设计提示词。明确角色“你是一个专业的法律顾问”、规定格式“请用列表形式回答”、要求提供引用“在答案后注明来源页码”。6.3 性能与可观测性异步处理对于Web应用使用异步框架如FastAPI和异步的LangChain调用可以提高并发处理能力。缓存对频繁出现的相似查询结果进行缓存可以显著降低LLM API调用成本和延迟。日志与监控记录每一次问答的查询、检索到的文档、生成的答案、耗时和Token使用量。这对于分析系统表现、优化成本和排查问题至关重要。评估建立评估体系从答案相关性、事实准确性、信息完整性等维度评估RAG系统的输出质量。可以使用LLM作为裁判LLM-as-a-Judge进行自动评估。6.4 生产环境部署考量向量数据库选型Chroma适合原型和中小数据量。生产环境应考虑支持分布式、持久化、高可用的向量数据库如Qdrant、Weaviate、Milvus或PGVector如果已用PostgreSQL。嵌入模型部署如果涉及敏感数据或需要控制成本可以考虑在本地或内网部署开源的嵌入模型如BGE、E5替代OpenAI API。大模型选型根据数据敏感性、响应速度、成本预算选择适合的LLM。可以是云端APIGPT、Claude、文心一言等也可以是本地部署的模型Llama、Qwen、ChatGLM等。安全性确保知识库上传接口有严格的权限控制和文件类型、大小、病毒扫描。对用户输入进行必要的清洗和过滤防止提示词注入攻击。7. 常见问题与排查思路在开发和运行过程中你可能会遇到以下典型问题。问题现象可能原因排查与解决思路运行ingest.py时报错OpenAI API连接失败1. API密钥未设置或错误。2. 网络问题或代理配置。1. 检查.env文件格式和变量名是否正确。2. 在代码中打印os.getenv(“OPENAI_API_KEY”)的前几位确认。3. 检查网络连通性。问答时返回“根据提供的信息无法回答”但文档中明明有答案1. 检索到的文本块不相关。2. 文本块分割不合理太大或太小。3. 提示词不够明确。1. 检查检索到的源文档 (source_documents) 是否真的相关。2. 调整ingest.py中的chunk_size和chunk_overlap。3. 优化提示词明确要求“基于上下文”。4. 尝试增加检索数量 (k)。答案看起来是胡编乱造的幻觉1. 检索到的上下文质量差或不足。2. LLM的temperature参数过高。3. 提示词未限制模型编造。1. 确保检索环节有效检查源文档。2. 将LLM的temperature设为0或更低值。3. 在提示词中加入“如果上下文没有提供相关信息请说不知道”。处理长文档或大量文档时速度很慢1. 嵌入模型调用是主要瓶颈。2. 未使用批处理。3. 本地计算资源不足。1. 对于OpenAI API检查是否有速率限制考虑升级。2. 确保使用支持批处理的嵌入函数。3. 对于本地模型考虑使用GPU加速。Web应用运行正常但问答无结果或报错1. 向量数据库路径错误或未初始化。2. Flask应用与RAG核心模块路径或实例化问题。1. 确认persist_directory路径正确且chroma_db文件夹存在。2. 检查rag_core.py中RAGSystem初始化是否成功可在__init__中加入日志。3. 查看Flask应用的错误日志。构建一个健壮的企业级RAG系统是一个迭代过程。从本文介绍的最小可行产品MVP开始你可以逐步引入更复杂的检索策略、更精细的上下文管理、更强大的LLM以及完善的监控评估体系。记住RAG项目的成功不仅在于技术栈的选择更在于对业务需求的深刻理解、对数据质量的严格把控以及对系统效果的持续评估与优化。建议先从核心业务的一个小范围、高质量的知识库开始试点积累经验后再逐步扩大范围。