手把手带你从零搭建一个可用的 RAG 知识库问答系统支持上传 PDF/Word 文档基于文档内容精准回答问题。什么是 RAGRAGRetrieval-Augmented Generation检索增强生成是目前企业落地大模型最主流的方案用户问题 → 向量检索找到相关文档片段→ 大模型基于检索结果生成答案解决的核心问题大模型不了解你的私有数据避免模型幻觉编造不存在的信息答案可溯源有据可查系统架构┌─────────────┐ ┌──────────────┐ ┌─────────────┐ │ 文档上传 │ → │ 文本分割 │ → │ 向量化存储 │ │ PDF/Word/MD │ │ Chunk Split │ │ ChromaDB │ └─────────────┘ └──────────────┘ └─────────────┘ ↓ ┌─────────────┐ ┌──────────────┐ ┌─────────────┐ │ 最终回答 │ ← │ DeepSeek │ ← │ 向量检索 │ │ 含来源引用 │ │ 生成回答 │ │ Top-K 结果 │ └─────────────┘ └──────────────┘ └─────────────┘环境准备pip install langchain langchain-community langchain-openai pip install chromadb pip install pypdf python-docx pip install sentence-transformers一、文档加载与分割from langchain.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter import os def load_documents(file_path: str): 根据文件类型加载文档 ext os.path.splitext(file_path)[1].lower() if ext .pdf: loader PyPDFLoader(file_path) elif ext in [.docx, .doc]: loader Docx2txtLoader(file_path) elif ext .txt or ext .md: loader TextLoader(file_path, encodingutf-8) else: raise ValueError(f不支持的文件类型: {ext}) return loader.load() def split_documents(documents, chunk_size500, chunk_overlap50): 将文档切割成小块 splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, # 每块最大字符数 chunk_overlapchunk_overlap, # 块之间的重叠字符数 separators[\n\n, \n, 。, , , , ], length_functionlen ) return splitter.split_documents(documents) # 使用示例 docs load_documents(company_manual.pdf) chunks split_documents(docs) print(f共切割为 {len(chunks)} 个文本块) # 输出共切割为 156 个文本块二、向量化与存储from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings def create_vector_store(chunks, persist_dir./chroma_db): 创建向量数据库 # 使用本地嵌入模型免费不调用API embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, # 中文效果好的小模型 model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) # 创建并持久化向量库 vector_store Chroma.from_documents( documentschunks, embeddingembeddings, persist_directorypersist_dir ) print(f✅ 向量库创建完成共 {vector_store._collection.count()} 条记录) return vector_store def load_vector_store(persist_dir./chroma_db): 加载已有向量库 embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) return Chroma( persist_directorypersist_dir, embedding_functionembeddings )三、接入 DeepSeek 生成回答from langchain_openai import ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate def create_qa_chain(vector_store): 创建问答链 # 使用 DeepSeek兼容 OpenAI API llm ChatOpenAI( modeldeepseek-chat, api_keyYOUR_DEEPSEEK_API_KEY, base_urlhttps://api.deepseek.com, temperature0.1, # 低温度回答更准确 ) # 自定义提示词要求基于文档回答 prompt_template 你是一个专业的知识库助手。请严格基于以下检索到的文档内容回答用户问题。 如果文档中没有相关信息请明确说明根据现有文档无法找到相关信息不要编造答案。 检索到的文档内容 {context} 用户问题{question} 请给出准确、简洁的回答并在回答末尾注明信息来源文档名称和页码 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 创建检索问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervector_store.as_retriever( search_typesimilarity, search_kwargs{k: 4} # 检索最相关的4个文档块 ), chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue # 返回来源文档 ) return qa_chain def ask(qa_chain, question: str): 提问并获取带来源的回答 result qa_chain({query: question}) answer result[result] sources result[source_documents] print(f\n❓ 问题{question}) print(f\n 回答{answer}) print(f\n 参考来源) seen set() for doc in sources: source doc.metadata.get(source, 未知来源) page doc.metadata.get(page, ) key f{source}-{page} if key not in seen: seen.add(key) print(f - {source} (f 第{page1}页 if page ! else )) return answer四、完整系统整合import os class KnowledgeBase: def __init__(self, persist_dir./chroma_db): self.persist_dir persist_dir self.vector_store None self.qa_chain None def build(self, file_paths: list): 从文件列表构建知识库 all_chunks [] for path in file_paths: print(f 加载文档{path}) docs load_documents(path) chunks split_documents(docs) all_chunks.extend(chunks) print(f 切割为 {len(chunks)} 块) print(f\n⏳ 正在向量化 {len(all_chunks)} 个文本块...) self.vector_store create_vector_store(all_chunks, self.persist_dir) self.qa_chain create_qa_chain(self.vector_store) print(✅ 知识库构建完成) def load(self): 加载已有知识库 self.vector_store load_vector_store(self.persist_dir) self.qa_chain create_qa_chain(self.vector_store) print(✅ 知识库加载完成) def chat(self, question: str) - str: 提问 if not self.qa_chain: raise RuntimeError(知识库未初始化请先调用 build() 或 load()) return ask(self.qa_chain, question) def add_document(self, file_path: str): 向已有知识库添加文档 docs load_documents(file_path) chunks split_documents(docs) self.vector_store.add_documents(chunks) print(f✅ 已添加 {len(chunks)} 个文本块) # 使用示例 # 第一次构建知识库 kb KnowledgeBase() kb.build([ company_manual.pdf, # 公司手册 product_docs.docx, # 产品文档 faq.md # 常见问题 ]) # 后续使用直接加载 kb KnowledgeBase() kb.load() # 提问 kb.chat(公司的年假政策是什么) kb.chat(产品支持哪些操作系统) kb.chat(如何申请报销)五、实际效果展示❓ 问题年假是几天 回答根据公司员工手册年假天数与工龄挂钩 - 工龄1-3年5天 - 工龄3-10年10天 - 工龄10年以上15天 年假需提前3个工作日申请经直属领导审批后方可使用。 参考来源 - company_manual.pdf 第12页六、性能优化建议# 1. 批量处理嵌入减少 API 调用 embeddings HuggingFaceEmbeddings( encode_kwargs{ normalize_embeddings: True, batch_size: 64 # 批量处理 } ) # 2. 混合检索相似度 关键词 from langchain.retrievers import BM25Retriever, EnsembleRetriever bm25_retriever BM25Retriever.from_documents(chunks) bm25_retriever.k 4 vector_retriever vector_store.as_retriever(search_kwargs{k: 4}) # 混合检索效果更好 ensemble_retriever EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.3, 0.7] # 向量检索权重更高 ) # 3. 问题改写提高检索准确率 rewrite_prompt 将以下问题改写为更适合文档检索的形式{question}总结一个完整的 RAG 系统需要组件本文选择原因嵌入模型BGE-small-zh本地运行中文效果好免费向量数据库ChromaDB轻量支持本地持久化大模型DeepSeek性价比最高中文能力强编排框架LangChain生态完善组件丰富整个系统部署在本地服务器即可数据不出门适合企业私有化部署。代码已经过测试可直接运行如有问题欢迎评论区交流 觉得有用点个赞 后续会出 Dify 可视化版本教程。
RAG 实战：用 LangChain + DeepSeek 搭建企业私有知识库问答系统
手把手带你从零搭建一个可用的 RAG 知识库问答系统支持上传 PDF/Word 文档基于文档内容精准回答问题。什么是 RAGRAGRetrieval-Augmented Generation检索增强生成是目前企业落地大模型最主流的方案用户问题 → 向量检索找到相关文档片段→ 大模型基于检索结果生成答案解决的核心问题大模型不了解你的私有数据避免模型幻觉编造不存在的信息答案可溯源有据可查系统架构┌─────────────┐ ┌──────────────┐ ┌─────────────┐ │ 文档上传 │ → │ 文本分割 │ → │ 向量化存储 │ │ PDF/Word/MD │ │ Chunk Split │ │ ChromaDB │ └─────────────┘ └──────────────┘ └─────────────┘ ↓ ┌─────────────┐ ┌──────────────┐ ┌─────────────┐ │ 最终回答 │ ← │ DeepSeek │ ← │ 向量检索 │ │ 含来源引用 │ │ 生成回答 │ │ Top-K 结果 │ └─────────────┘ └──────────────┘ └─────────────┘环境准备pip install langchain langchain-community langchain-openai pip install chromadb pip install pypdf python-docx pip install sentence-transformers一、文档加载与分割from langchain.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter import os def load_documents(file_path: str): 根据文件类型加载文档 ext os.path.splitext(file_path)[1].lower() if ext .pdf: loader PyPDFLoader(file_path) elif ext in [.docx, .doc]: loader Docx2txtLoader(file_path) elif ext .txt or ext .md: loader TextLoader(file_path, encodingutf-8) else: raise ValueError(f不支持的文件类型: {ext}) return loader.load() def split_documents(documents, chunk_size500, chunk_overlap50): 将文档切割成小块 splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, # 每块最大字符数 chunk_overlapchunk_overlap, # 块之间的重叠字符数 separators[\n\n, \n, 。, , , , ], length_functionlen ) return splitter.split_documents(documents) # 使用示例 docs load_documents(company_manual.pdf) chunks split_documents(docs) print(f共切割为 {len(chunks)} 个文本块) # 输出共切割为 156 个文本块二、向量化与存储from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings def create_vector_store(chunks, persist_dir./chroma_db): 创建向量数据库 # 使用本地嵌入模型免费不调用API embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, # 中文效果好的小模型 model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) # 创建并持久化向量库 vector_store Chroma.from_documents( documentschunks, embeddingembeddings, persist_directorypersist_dir ) print(f✅ 向量库创建完成共 {vector_store._collection.count()} 条记录) return vector_store def load_vector_store(persist_dir./chroma_db): 加载已有向量库 embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) return Chroma( persist_directorypersist_dir, embedding_functionembeddings )三、接入 DeepSeek 生成回答from langchain_openai import ChatOpenAI from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate def create_qa_chain(vector_store): 创建问答链 # 使用 DeepSeek兼容 OpenAI API llm ChatOpenAI( modeldeepseek-chat, api_keyYOUR_DEEPSEEK_API_KEY, base_urlhttps://api.deepseek.com, temperature0.1, # 低温度回答更准确 ) # 自定义提示词要求基于文档回答 prompt_template 你是一个专业的知识库助手。请严格基于以下检索到的文档内容回答用户问题。 如果文档中没有相关信息请明确说明根据现有文档无法找到相关信息不要编造答案。 检索到的文档内容 {context} 用户问题{question} 请给出准确、简洁的回答并在回答末尾注明信息来源文档名称和页码 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 创建检索问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervector_store.as_retriever( search_typesimilarity, search_kwargs{k: 4} # 检索最相关的4个文档块 ), chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue # 返回来源文档 ) return qa_chain def ask(qa_chain, question: str): 提问并获取带来源的回答 result qa_chain({query: question}) answer result[result] sources result[source_documents] print(f\n❓ 问题{question}) print(f\n 回答{answer}) print(f\n 参考来源) seen set() for doc in sources: source doc.metadata.get(source, 未知来源) page doc.metadata.get(page, ) key f{source}-{page} if key not in seen: seen.add(key) print(f - {source} (f 第{page1}页 if page ! else )) return answer四、完整系统整合import os class KnowledgeBase: def __init__(self, persist_dir./chroma_db): self.persist_dir persist_dir self.vector_store None self.qa_chain None def build(self, file_paths: list): 从文件列表构建知识库 all_chunks [] for path in file_paths: print(f 加载文档{path}) docs load_documents(path) chunks split_documents(docs) all_chunks.extend(chunks) print(f 切割为 {len(chunks)} 块) print(f\n⏳ 正在向量化 {len(all_chunks)} 个文本块...) self.vector_store create_vector_store(all_chunks, self.persist_dir) self.qa_chain create_qa_chain(self.vector_store) print(✅ 知识库构建完成) def load(self): 加载已有知识库 self.vector_store load_vector_store(self.persist_dir) self.qa_chain create_qa_chain(self.vector_store) print(✅ 知识库加载完成) def chat(self, question: str) - str: 提问 if not self.qa_chain: raise RuntimeError(知识库未初始化请先调用 build() 或 load()) return ask(self.qa_chain, question) def add_document(self, file_path: str): 向已有知识库添加文档 docs load_documents(file_path) chunks split_documents(docs) self.vector_store.add_documents(chunks) print(f✅ 已添加 {len(chunks)} 个文本块) # 使用示例 # 第一次构建知识库 kb KnowledgeBase() kb.build([ company_manual.pdf, # 公司手册 product_docs.docx, # 产品文档 faq.md # 常见问题 ]) # 后续使用直接加载 kb KnowledgeBase() kb.load() # 提问 kb.chat(公司的年假政策是什么) kb.chat(产品支持哪些操作系统) kb.chat(如何申请报销)五、实际效果展示❓ 问题年假是几天 回答根据公司员工手册年假天数与工龄挂钩 - 工龄1-3年5天 - 工龄3-10年10天 - 工龄10年以上15天 年假需提前3个工作日申请经直属领导审批后方可使用。 参考来源 - company_manual.pdf 第12页六、性能优化建议# 1. 批量处理嵌入减少 API 调用 embeddings HuggingFaceEmbeddings( encode_kwargs{ normalize_embeddings: True, batch_size: 64 # 批量处理 } ) # 2. 混合检索相似度 关键词 from langchain.retrievers import BM25Retriever, EnsembleRetriever bm25_retriever BM25Retriever.from_documents(chunks) bm25_retriever.k 4 vector_retriever vector_store.as_retriever(search_kwargs{k: 4}) # 混合检索效果更好 ensemble_retriever EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.3, 0.7] # 向量检索权重更高 ) # 3. 问题改写提高检索准确率 rewrite_prompt 将以下问题改写为更适合文档检索的形式{question}总结一个完整的 RAG 系统需要组件本文选择原因嵌入模型BGE-small-zh本地运行中文效果好免费向量数据库ChromaDB轻量支持本地持久化大模型DeepSeek性价比最高中文能力强编排框架LangChain生态完善组件丰富整个系统部署在本地服务器即可数据不出门适合企业私有化部署。代码已经过测试可直接运行如有问题欢迎评论区交流 觉得有用点个赞 后续会出 Dify 可视化版本教程。
