RAG检索增强生成是2024年最实用的AI落地技术之一。这篇文章手把手教你搭建一个企业级的RAG系统。前言我们团队之前做了一个内部知识库问答系统一开始直接把文档丢给GPT效果惨不忍睹——它要么胡说八道幻觉要么说我没有相关信息。后来引入RAG架构回答准确率从40%飙升到92%。这篇文章是我完整的实战笔记。一、RAG是什么一张图看懂传统方式用户问题 → LLM凭记忆回答 → 可能产生幻觉 RAG方式 用户问题 → 检索相关文档片段 → 问题 文档片段一起给LLM → 基于证据回答核心思路不让LLM凭空回答给它开卷考试的机会。二、整体架构┌─────────────────────────────────┐ │ RAG Pipeline │ ├─────────────────────────────────┤ 离线阶段 │ │ (数据准备) │ 文档 → 分块 → 向量化 → 存储 │ │ │ 在线阶段 │ 查询 → 向量检索 → 重排 → │ (问答) │ Prompt组装 → LLM生成 │ └─────────────────────────────────┘三、离线阶段数据准备3.1 文档加载fromlangchain_community.document_loadersimport(PyPDFLoader,TextLoader,UnstructuredMarkdownLoader,Docx2txtLoader)frompathlibimportPathclassDocumentLoader:多格式文档加载器LOADER_MAP{.pdf:PyPDFLoader,.txt:TextLoader,.md:UnstructuredMarkdownLoader,.docx:Docx2txtLoader,}defload_directory(self,dir_path:str)-list:加载目录下所有文档documents[]forfile_pathinPath(dir_path).rglob(*):suffixfile_path.suffix.lower()ifsuffixinself.LOADER_MAP:try:loaderself.LOADER_MAP[suffix](str(file_path))docsloader.load()# 添加元数据fordocindocs:doc.metadata[source]str(file_path)doc.metadata[file_type]suffix documents.extend(docs)print(f✅ 加载成功:{file_path.name})exceptExceptionase:print(f❌ 加载失败:{file_path.name}, 错误:{e})print(f\n总计加载{len(documents)}个文档片段)returndocuments3.2 文本分块最关键的一步fromlangchain.text_splitterimportRecursiveCharacterTextSplitterclassSmartChunker:智能分块器def__init__(self,chunk_size500,chunk_overlap100):# 递归字符分割器优先按段落 句子 字符分割self.splitterRecursiveCharacterTextSplitter(chunk_sizechunk_size,chunk_overlapchunk_overlap,separators[\n\n,\n,。,,,.,!,?, ],length_functionlen)defchunk_documents(self,documents:list)-list:对文档进行分块chunksself.splitter.split_documents(documents)# 为每个chunk添加上下文标题提高检索质量fori,chunkinenumerate(chunks):# 从元数据中提取来源信息拼接到内容前面sourcechunk.metadata.get(source,未知)chunk.page_contentf[来源:{source}]\n{chunk.page_content}chunk.metadata[chunk_index]iprint(f分块完成:{len(documents)}个文档 →{len(chunks)}个块)returnchunks踩坑chunk_size设多大chunk_size优点缺点200-300检索精准上下文不完整500-800平衡推荐大多数场景1000-2000上下文完整检索可能不够精准我的经验技术文档500-800字符FAQ类200-400字符每个QA对作为一个chunk合同/法律文档800-1200字符3.3 向量化与存储fromlangchain_openaiimportOpenAIEmbeddingsfromlangchain_community.vectorstoresimportChroma# 国内替代方案# from langchain_community.embeddings import DashScopeEmbeddingsclassVectorStore:向量存储管理def__init__(self,persist_dir./vector_db):self.persist_dirpersist_dir# 选择Embedding模型# 方案1: OpenAI效果好要钱self.embeddingsOpenAIEmbeddings(modeltext-embedding-3-small)# 方案2: 通义千问Embedding国内推荐# self.embeddings DashScopeEmbeddings(modeltext-embedding-v2)# 方案3: 本地开源模型免费但需要GPU# from langchain_huggingface import HuggingFaceEmbeddings# self.embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-large-zh-v1.5)defbuild_index(self,chunks:list):构建向量索引self.vectorstoreChroma.from_documents(documentschunks,embeddingself.embeddings,persist_directoryself.persist_dir,collection_nameknowledge_base)print(f✅ 向量索引构建完成共{len(chunks)}条记录)defload_index(self):加载已有索引self.vectorstoreChroma(persist_directoryself.persist_dir,embedding_functionself.embeddings,collection_nameknowledge_base)defsearch(self,query:str,top_k:int5)-list:相似度检索resultsself.vectorstore.similarity_search_with_score(query,ktop_k)returnresults四、在线阶段检索 生成4.1 基础RAG Chainfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnablePassthroughclassRAGEngine:RAG问答引擎def__init__(self,vector_store:VectorStore):self.vector_storevector_store self.llmChatOpenAI(modelgpt-4,temperature0.1)# RAG专用Promptself.promptChatPromptTemplate.from_template( 你是一个企业知识库问答助手。请严格根据以下参考文档来回答用户的问题。 ## 规则 1. 只根据参考文档中的信息回答不要编造 2. 如果文档中没有相关信息请明确说根据现有文档我没有找到相关信息 3. 回答时引用来源格式为 [来源: 文件名] 4. 用简洁专业的中文回答 ## 参考文档 {context} ## 用户问题 {question} ## 回答 )# 构建RAG Chainself.chain({context:lambdax:self._retrieve(x[question]),question:lambdax:x[question]}|self.prompt|self.llm|StrOutputParser())def_retrieve(self,query:str)-str:检索相关文档resultsself.vector_store.search(query,top_k5)# 格式化检索结果context_parts[]fordoc,scoreinresults:ifscore0.8:# 相似度阈值过滤Chroma用的是距离越小越相似context_parts.append(doc.page_content)ifnotcontext_parts:return未找到相关文档return\n\n---\n\n.join(context_parts)defask(self,question:str)-dict:问答answerself.chain.invoke({question:question})# 同时返回检索到的源文档方便验证sourcesself.vector_store.search(question,top_k3)source_fileslist(set(doc.metadata.get(source,未知)fordoc,_insources))return{answer:answer,sources:source_files}4.2 进阶加入重排序Reranker直接用向量检索有时候不够准确加一个Reranker可以大幅提升质量。# pip install FlagEmbeddingfromFlagEmbeddingimportFlagRerankerclassRerankedRAGEngine(RAGEngine):带重排序的RAG引擎def__init__(self,vector_store:VectorStore):super().__init__(vector_store)# 使用BGE-Reranker开源效果好self.rerankerFlagReranker(BAAI/bge-reranker-v2-m3,use_fp16True)def_retrieve(self,query:str)-str:# 1. 粗召回先检索20条raw_resultsself.vector_store.search(query,top_k20)# 2. 重排序用Reranker精排pairs[[query,doc.page_content]fordoc,_inraw_results]scoresself.reranker.compute_score(pairs)# 3. 按重排分数排序取Top 5rankedsorted(zip(raw_results,scores),keylambdax:x[1],reverseTrue)[:5]context_parts[doc.page_contentfor(doc,_),scoreinrankedifscore0]return\n\n---\n\n.join(context_parts)4.3 进阶查询改写Query Rewriting用户的问题可能表述不清楚改写后检索效果更好。classQueryRewriter:查询改写def__init__(self):self.llmChatOpenAI(modelgpt-4,temperature0)self.promptChatPromptTemplate.from_template( 请将用户的问题改写为更适合在知识库中检索的形式。 要求 1. 保持原意不变 2. 补充可能的关键词 3. 如果问题模糊生成2-3个不同角度的检索查询 用户原始问题{question} 改写后的检索查询每行一个 )defrewrite(self,question:str)-list:改写查询chainself.prompt|self.llm|StrOutputParser()resultchain.invoke({question:question})queries[q.strip()forqinresult.strip().split(\n)ifq.strip()]returnqueries[:3]# 最多3个改写查询五、完整Pipeline代码# main.py - 一键构建RAG系统fromdocument_loaderimportDocumentLoaderfromchunkerimportSmartChunkerfromvector_storeimportVectorStorefromrag_engineimportRerankedRAGEnginedefbuild_rag_system(docs_dir:str):构建RAG系统# Step 1: 加载文档loaderDocumentLoader()documentsloader.load_directory(docs_dir)# Step 2: 分块chunkerSmartChunker(chunk_size600,chunk_overlap100)chunkschunker.chunk_documents(documents)# Step 3: 向量化存储storeVectorStore(persist_dir./knowledge_vector_db)store.build_index(chunks)# Step 4: 创建问答引擎engineRerankedRAGEngine(store)returnenginedefmain():# 构建首次运行enginebuild_rag_system(./docs)# 问答whileTrue:questioninput(\n请输入问题输入q退出: )ifquestion.lower()q:breakresultengine.ask(question)print(f\n 回答:{result[answer]})print(f 来源:{, .join(result[sources])})if__name____main__:main()六、效果优化我试过的方法优化效果对比方案准确率备注基础RAG向量检索LLM72%基准 优化分块调整chunk_size78%分块太大太小都不行 Query改写82%多角度检索有效 Reranker重排序88%精排提升明显 混合检索向量BM2592%语义关键词互补混合检索实现fromlangchain_community.retrieversimportBM25Retrieverfromlangchain.retrieversimportEnsembleRetriever# BM25检索器关键词匹配bm25_retrieverBM25Retriever.from_documents(chunks)bm25_retriever.k10# 向量检索器vector_retrievervector_store.vectorstore.as_retriever(search_kwargs{k:10})# 混合检索Ensemblehybrid_retrieverEnsembleRetriever(retrievers[bm25_retriever,vector_retriever],weights[0.3,0.7]# 向量权重更高)七、踩坑与教训坑1PDF解析质量差很多PDF用PyPDF解析后是乱码或者丢失格式。# 解决用更好的PDF解析库# 方案1: unstructured效果好但慢fromlangchain_community.document_loadersimportUnstructuredPDFLoader# 方案2: pdfplumber表格解析好importpdfplumber# 方案3: 对于扫描件PDF用OCR# pip install pdf2image pytesseract坑2长文档分块后丢失上下文一个段落被切成两半导致信息不完整。# 解决增加overlap 父文档检索# 检索时命中子chunk返回时带上父chunk更大的上下文fromlangchain.retrieversimportParentDocumentRetriever坑3LLM不遵守不要编造的指令即使Prompt里写了GPT有时还是会编造信息。# 解决后处理验证defvalidate_answer(answer:str,sources:list)-str:验证回答是否有据可依# 简单方案检查回答中的关键信息是否在源文档中出现# 高级方案用另一个LLM做事实核查pass坑4向量检索的语义漂移问公司年假几天检索到公司成立几年了都有公司和几。# 解决混合检索 重排序 提高检索数量后过滤# BM25能抓住年假这个关键词# Reranker能判断语义相关性八、生产部署建议# 生产环境技术栈建议向量数据库:Milvus / Qdrant / Weaviate (替代ChromaDB)Embedding:BAAI/bge-large-zh-v1.5 (本地部署) 或 通义EmbeddingLLM:通义千问 / GLM-4 / DeepSeek (国内合规)API框架:FastAPI容器化:Docker K8s监控:记录每次检索的query、检索结果、生成结果、用户反馈写在最后RAG不是万能的它的上限取决于你的文档质量和检索精度。但相比纯LLM回答RAG至少解决了胡说八道的问题。如果你正在做企业知识库、客服问答、文档搜索这类需求RAG是目前最靠谱的方案。文章比较长如果对你有帮助点赞收藏一下吧下篇写MCP协议开发实战~ RAG实战-让AI基于你的私有文档回答问题而不是胡说八道
RAG检索增强生成是2024年最实用的AI落地技术之一。这篇文章手把手教你搭建一个企业级的RAG系统。前言我们团队之前做了一个内部知识库问答系统一开始直接把文档丢给GPT效果惨不忍睹——它要么胡说八道幻觉要么说我没有相关信息。后来引入RAG架构回答准确率从40%飙升到92%。这篇文章是我完整的实战笔记。一、RAG是什么一张图看懂传统方式用户问题 → LLM凭记忆回答 → 可能产生幻觉 RAG方式 用户问题 → 检索相关文档片段 → 问题 文档片段一起给LLM → 基于证据回答核心思路不让LLM凭空回答给它开卷考试的机会。二、整体架构┌─────────────────────────────────┐ │ RAG Pipeline │ ├─────────────────────────────────┤ 离线阶段 │ │ (数据准备) │ 文档 → 分块 → 向量化 → 存储 │ │ │ 在线阶段 │ 查询 → 向量检索 → 重排 → │ (问答) │ Prompt组装 → LLM生成 │ └─────────────────────────────────┘三、离线阶段数据准备3.1 文档加载fromlangchain_community.document_loadersimport(PyPDFLoader,TextLoader,UnstructuredMarkdownLoader,Docx2txtLoader)frompathlibimportPathclassDocumentLoader:多格式文档加载器LOADER_MAP{.pdf:PyPDFLoader,.txt:TextLoader,.md:UnstructuredMarkdownLoader,.docx:Docx2txtLoader,}defload_directory(self,dir_path:str)-list:加载目录下所有文档documents[]forfile_pathinPath(dir_path).rglob(*):suffixfile_path.suffix.lower()ifsuffixinself.LOADER_MAP:try:loaderself.LOADER_MAP[suffix](str(file_path))docsloader.load()# 添加元数据fordocindocs:doc.metadata[source]str(file_path)doc.metadata[file_type]suffix documents.extend(docs)print(f✅ 加载成功:{file_path.name})exceptExceptionase:print(f❌ 加载失败:{file_path.name}, 错误:{e})print(f\n总计加载{len(documents)}个文档片段)returndocuments3.2 文本分块最关键的一步fromlangchain.text_splitterimportRecursiveCharacterTextSplitterclassSmartChunker:智能分块器def__init__(self,chunk_size500,chunk_overlap100):# 递归字符分割器优先按段落 句子 字符分割self.splitterRecursiveCharacterTextSplitter(chunk_sizechunk_size,chunk_overlapchunk_overlap,separators[\n\n,\n,。,,,.,!,?, ],length_functionlen)defchunk_documents(self,documents:list)-list:对文档进行分块chunksself.splitter.split_documents(documents)# 为每个chunk添加上下文标题提高检索质量fori,chunkinenumerate(chunks):# 从元数据中提取来源信息拼接到内容前面sourcechunk.metadata.get(source,未知)chunk.page_contentf[来源:{source}]\n{chunk.page_content}chunk.metadata[chunk_index]iprint(f分块完成:{len(documents)}个文档 →{len(chunks)}个块)returnchunks踩坑chunk_size设多大chunk_size优点缺点200-300检索精准上下文不完整500-800平衡推荐大多数场景1000-2000上下文完整检索可能不够精准我的经验技术文档500-800字符FAQ类200-400字符每个QA对作为一个chunk合同/法律文档800-1200字符3.3 向量化与存储fromlangchain_openaiimportOpenAIEmbeddingsfromlangchain_community.vectorstoresimportChroma# 国内替代方案# from langchain_community.embeddings import DashScopeEmbeddingsclassVectorStore:向量存储管理def__init__(self,persist_dir./vector_db):self.persist_dirpersist_dir# 选择Embedding模型# 方案1: OpenAI效果好要钱self.embeddingsOpenAIEmbeddings(modeltext-embedding-3-small)# 方案2: 通义千问Embedding国内推荐# self.embeddings DashScopeEmbeddings(modeltext-embedding-v2)# 方案3: 本地开源模型免费但需要GPU# from langchain_huggingface import HuggingFaceEmbeddings# self.embeddings HuggingFaceEmbeddings(model_nameBAAI/bge-large-zh-v1.5)defbuild_index(self,chunks:list):构建向量索引self.vectorstoreChroma.from_documents(documentschunks,embeddingself.embeddings,persist_directoryself.persist_dir,collection_nameknowledge_base)print(f✅ 向量索引构建完成共{len(chunks)}条记录)defload_index(self):加载已有索引self.vectorstoreChroma(persist_directoryself.persist_dir,embedding_functionself.embeddings,collection_nameknowledge_base)defsearch(self,query:str,top_k:int5)-list:相似度检索resultsself.vectorstore.similarity_search_with_score(query,ktop_k)returnresults四、在线阶段检索 生成4.1 基础RAG Chainfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnablePassthroughclassRAGEngine:RAG问答引擎def__init__(self,vector_store:VectorStore):self.vector_storevector_store self.llmChatOpenAI(modelgpt-4,temperature0.1)# RAG专用Promptself.promptChatPromptTemplate.from_template( 你是一个企业知识库问答助手。请严格根据以下参考文档来回答用户的问题。 ## 规则 1. 只根据参考文档中的信息回答不要编造 2. 如果文档中没有相关信息请明确说根据现有文档我没有找到相关信息 3. 回答时引用来源格式为 [来源: 文件名] 4. 用简洁专业的中文回答 ## 参考文档 {context} ## 用户问题 {question} ## 回答 )# 构建RAG Chainself.chain({context:lambdax:self._retrieve(x[question]),question:lambdax:x[question]}|self.prompt|self.llm|StrOutputParser())def_retrieve(self,query:str)-str:检索相关文档resultsself.vector_store.search(query,top_k5)# 格式化检索结果context_parts[]fordoc,scoreinresults:ifscore0.8:# 相似度阈值过滤Chroma用的是距离越小越相似context_parts.append(doc.page_content)ifnotcontext_parts:return未找到相关文档return\n\n---\n\n.join(context_parts)defask(self,question:str)-dict:问答answerself.chain.invoke({question:question})# 同时返回检索到的源文档方便验证sourcesself.vector_store.search(question,top_k3)source_fileslist(set(doc.metadata.get(source,未知)fordoc,_insources))return{answer:answer,sources:source_files}4.2 进阶加入重排序Reranker直接用向量检索有时候不够准确加一个Reranker可以大幅提升质量。# pip install FlagEmbeddingfromFlagEmbeddingimportFlagRerankerclassRerankedRAGEngine(RAGEngine):带重排序的RAG引擎def__init__(self,vector_store:VectorStore):super().__init__(vector_store)# 使用BGE-Reranker开源效果好self.rerankerFlagReranker(BAAI/bge-reranker-v2-m3,use_fp16True)def_retrieve(self,query:str)-str:# 1. 粗召回先检索20条raw_resultsself.vector_store.search(query,top_k20)# 2. 重排序用Reranker精排pairs[[query,doc.page_content]fordoc,_inraw_results]scoresself.reranker.compute_score(pairs)# 3. 按重排分数排序取Top 5rankedsorted(zip(raw_results,scores),keylambdax:x[1],reverseTrue)[:5]context_parts[doc.page_contentfor(doc,_),scoreinrankedifscore0]return\n\n---\n\n.join(context_parts)4.3 进阶查询改写Query Rewriting用户的问题可能表述不清楚改写后检索效果更好。classQueryRewriter:查询改写def__init__(self):self.llmChatOpenAI(modelgpt-4,temperature0)self.promptChatPromptTemplate.from_template( 请将用户的问题改写为更适合在知识库中检索的形式。 要求 1. 保持原意不变 2. 补充可能的关键词 3. 如果问题模糊生成2-3个不同角度的检索查询 用户原始问题{question} 改写后的检索查询每行一个 )defrewrite(self,question:str)-list:改写查询chainself.prompt|self.llm|StrOutputParser()resultchain.invoke({question:question})queries[q.strip()forqinresult.strip().split(\n)ifq.strip()]returnqueries[:3]# 最多3个改写查询五、完整Pipeline代码# main.py - 一键构建RAG系统fromdocument_loaderimportDocumentLoaderfromchunkerimportSmartChunkerfromvector_storeimportVectorStorefromrag_engineimportRerankedRAGEnginedefbuild_rag_system(docs_dir:str):构建RAG系统# Step 1: 加载文档loaderDocumentLoader()documentsloader.load_directory(docs_dir)# Step 2: 分块chunkerSmartChunker(chunk_size600,chunk_overlap100)chunkschunker.chunk_documents(documents)# Step 3: 向量化存储storeVectorStore(persist_dir./knowledge_vector_db)store.build_index(chunks)# Step 4: 创建问答引擎engineRerankedRAGEngine(store)returnenginedefmain():# 构建首次运行enginebuild_rag_system(./docs)# 问答whileTrue:questioninput(\n请输入问题输入q退出: )ifquestion.lower()q:breakresultengine.ask(question)print(f\n 回答:{result[answer]})print(f 来源:{, .join(result[sources])})if__name____main__:main()六、效果优化我试过的方法优化效果对比方案准确率备注基础RAG向量检索LLM72%基准 优化分块调整chunk_size78%分块太大太小都不行 Query改写82%多角度检索有效 Reranker重排序88%精排提升明显 混合检索向量BM2592%语义关键词互补混合检索实现fromlangchain_community.retrieversimportBM25Retrieverfromlangchain.retrieversimportEnsembleRetriever# BM25检索器关键词匹配bm25_retrieverBM25Retriever.from_documents(chunks)bm25_retriever.k10# 向量检索器vector_retrievervector_store.vectorstore.as_retriever(search_kwargs{k:10})# 混合检索Ensemblehybrid_retrieverEnsembleRetriever(retrievers[bm25_retriever,vector_retriever],weights[0.3,0.7]# 向量权重更高)七、踩坑与教训坑1PDF解析质量差很多PDF用PyPDF解析后是乱码或者丢失格式。# 解决用更好的PDF解析库# 方案1: unstructured效果好但慢fromlangchain_community.document_loadersimportUnstructuredPDFLoader# 方案2: pdfplumber表格解析好importpdfplumber# 方案3: 对于扫描件PDF用OCR# pip install pdf2image pytesseract坑2长文档分块后丢失上下文一个段落被切成两半导致信息不完整。# 解决增加overlap 父文档检索# 检索时命中子chunk返回时带上父chunk更大的上下文fromlangchain.retrieversimportParentDocumentRetriever坑3LLM不遵守不要编造的指令即使Prompt里写了GPT有时还是会编造信息。# 解决后处理验证defvalidate_answer(answer:str,sources:list)-str:验证回答是否有据可依# 简单方案检查回答中的关键信息是否在源文档中出现# 高级方案用另一个LLM做事实核查pass坑4向量检索的语义漂移问公司年假几天检索到公司成立几年了都有公司和几。# 解决混合检索 重排序 提高检索数量后过滤# BM25能抓住年假这个关键词# Reranker能判断语义相关性八、生产部署建议# 生产环境技术栈建议向量数据库:Milvus / Qdrant / Weaviate (替代ChromaDB)Embedding:BAAI/bge-large-zh-v1.5 (本地部署) 或 通义EmbeddingLLM:通义千问 / GLM-4 / DeepSeek (国内合规)API框架:FastAPI容器化:Docker K8s监控:记录每次检索的query、检索结果、生成结果、用户反馈写在最后RAG不是万能的它的上限取决于你的文档质量和检索精度。但相比纯LLM回答RAG至少解决了胡说八道的问题。如果你正在做企业知识库、客服问答、文档搜索这类需求RAG是目前最靠谱的方案。文章比较长如果对你有帮助点赞收藏一下吧下篇写MCP协议开发实战~