Python_LangChain_RAG知识库

Python LangChain 搭建企业级 RAG 知识库完整代码CSDN 专栏Python 实战与解决问题作者[小汤包快跑]发布时间2026年标签Python, LangChain, RAG, 知识库, 向量数据库, 大模型一、前言你是否遇到过这样的场景公司有大量内部文档员工查找信息效率低下客服需要快速回答产品问题但文档分散各处想做一个智能问答系统但不知道如何下手RAGRetrieval-Augmented Generation检索增强生成就是解决这类问题的最佳方案。今天我将带你用Python LangChain搭建一个企业级 RAG 知识库让 AI 能够基于你的私有文档回答问题。本文包含完整代码可直接运行二、什么是 RAG2.1 RAG 核心原理用户提问 → 检索相关文档 → 将文档问题一起送给大模型 → 生成答案为什么需要 RAG✅ 大模型有知识截止日期无法获取最新信息✅ 企业私有数据不能上传到公网大模型✅ 减少大模型幻觉提高回答准确性2.2 RAG 架构图┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 文档加载 │────▶│ 文本分割 │────▶│ 向量化存储 │ │ (PDF/Word) │ │ (Chunking) │ │ (Vector DB) │ └─────────────┘ └─────────────┘ └──────┬──────┘ │ ┌─────────────┐ ┌─────────────┐ ┌──────▼──────┐ │ 生成答案 │◀────│ 构建Prompt │◀────│ 相似度检索 │ │ (LLM生成) │ │ (ContextQA) │ │ (Top-K) │ └─────────────┘ └─────────────┘ └─────────────┘三、环境准备3.1 安装依赖pipinstalllangchain langchain-community langchain-openai pipinstallchromadb sentence-transformers pipinstallpypdf unstructured3.2 目录结构rag_project/ ├── documents/ # 存放知识库文档 │ ├── 产品手册.pdf │ └── 常见问题.docx ├── vector_store/ # 向量数据库存储 ├── rag_system.py # 主程序 └── config.py # 配置文件四、完整代码实现4.1 配置文件 (config.py)# config.pyimportos# API 配置DEEPSEEK_API_KEYyour-deepseek-api-keyDEEPSEEK_BASE_URLhttps://api.deepseek.com/v1# 模型配置EMBEDDING_MODELBAAI/bge-large-zh-v1.5# 中文嵌入模型LLM_MODELdeepseek-chat# 向量数据库配置VECTOR_DB_PATH./vector_store# RAG 参数CHUNK_SIZE500# 文本块大小CHUNK_OVERLAP50# 文本块重叠大小TOP_K5# 检索文档数量4.2 主程序 (rag_system.py) 企业级 RAG 知识库系统 基于 LangChain ChromaDB DeepSeek importosfromtypingimportList,Optionalfromlangchain_community.document_loadersimport(PyPDFLoader,UnstructuredWordDocumentLoader,TextLoader,DirectoryLoader)fromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain_community.embeddingsimportHuggingFaceEmbeddingsfromlangchain_community.vectorstoresimportChromafromlangchain_openaiimportChatOpenAIfromlangchain.chainsimportRetrievalQAfromlangchain.promptsimportPromptTemplatefromlangchain.schemaimportDocument# 配置 DEEPSEEK_API_KEYyour-deepseek-api-keyDEEPSEEK_BASE_URLhttps://api.deepseek.com/v1classRAGKnowledgeBase:RAG 知识库类def__init__(self,persist_directory:str./vector_store,embedding_model:strBAAI/bge-large-zh-v1.5): 初始化 RAG 知识库 Args: persist_directory: 向量数据库持久化目录 embedding_model: 嵌入模型名称 self.persist_directorypersist_directory self.embedding_model_nameembedding_model# 初始化嵌入模型print(正在加载嵌入模型...)self.embeddingsHuggingFaceEmbeddings(model_nameembedding_model,model_kwargs{device:cpu},encode_kwargs{normalize_embeddings:True})# 初始化向量数据库ifos.path.exists(persist_directory):print(正在加载已有向量数据库...)self.vector_storeChroma(persist_directorypersist_directory,embedding_functionself.embeddings)else:print(创建新的向量数据库...)self.vector_storeNone# 初始化大模型print(正在连接大模型...)self.llmChatOpenAI(modeldeepseek-chat,base_urlDEEPSEEK_BASE_URL,api_keyDEEPSEEK_API_KEY,temperature0.7,max_tokens2000)print(✅ 知识库初始化完成)defload_documents(self,directory:str)-List[Document]: 加载目录中的所有文档 Args: directory: 文档目录路径 Returns: 文档列表 documents[]# 支持的文件类型loaders{.pdf:PyPDFLoader,.docx:UnstructuredWordDocumentLoader,.txt:TextLoader,.md:TextLoader}print(f正在扫描目录:{directory})forroot,dirs,filesinos.walk(directory):forfileinfiles:file_pathos.path.join(root,file)file_extos.path.splitext(file)[1].lower()iffile_extinloaders:try:print(f 正在加载:{file})loaderloaders[file_ext](file_path)docsloader.load()# 添加元数据fordocindocs:doc.metadata[source]filedoc.metadata[file_path]file_path documents.extend(docs)print(f ✅ 成功加载{len(docs)}页/段)exceptExceptionase:print(f ❌ 加载失败:{file}, 错误:{e})print(f\n共加载{len(documents)}个文档片段)returndocumentsdefsplit_documents(self,documents:List[Document],chunk_size:int500,chunk_overlap:int50)-List[Document]: 将文档分割成小块 Args: documents: 原始文档列表 chunk_size: 每块大小字符数 chunk_overlap: 块之间重叠大小 Returns: 分割后的文档列表 print(f\n正在分割文档 (块大小:{chunk_size}, 重叠:{chunk_overlap})...)text_splitterRecursiveCharacterTextSplitter(chunk_sizechunk_size,chunk_overlapchunk_overlap,length_functionlen,separators[\n\n,\n,。,, ,])chunkstext_splitter.split_documents(documents)print(f✅ 分割完成共{len(chunks)}个文本块)returnchunksdefcreate_vector_store(self,documents:List[Document]): 创建向量数据库 Args: documents: 文档列表 print(\n正在创建向量数据库...)self.vector_storeChroma.from_documents(documentsdocuments,embeddingself.embeddings,persist_directoryself.persist_directory)# 持久化保存self.vector_store.persist()print(f✅ 向量数据库创建完成已保存到:{self.persist_directory})defadd_documents(self,directory:str): 添加新文档到知识库 Args: directory: 文档目录路径 # 加载文档documentsself.load_documents(directory)ifnotdocuments:print(⚠️ 未找到可加载的文档)return# 分割文档chunksself.split_documents(documents)# 创建或更新向量数据库ifself.vector_storeisNone:self.create_vector_store(chunks)else:print(\n正在添加新文档到向量数据库...)self.vector_store.add_documents(chunks)self.vector_store.persist()print(✅ 新文档添加完成)defquery(self,question:str,top_k:int5,return_source:boolTrue)-dict: 查询知识库 Args: question: 用户问题 top_k: 检索文档数量 return_source: 是否返回来源文档 Returns: 包含答案和来源的字典 ifself.vector_storeisNone:return{answer:知识库为空请先添加文档,sources:[]}# 自定义 Prompt 模板prompt_template你是一个专业的知识库问答助手。请基于以下参考信息回答问题。 参考信息 {context} 用户问题{question} 回答要求 1. 基于参考信息给出准确、专业的回答 2. 如果参考信息不足以回答问题请明确说明 3. 回答要简洁明了突出重点 4. 可以适当使用列表、表格等格式增强可读性 请回答PROMPTPromptTemplate(templateprompt_template,input_variables[context,question])# 创建检索链qa_chainRetrievalQA.from_chain_type(llmself.llm,chain_typestuff,retrieverself.vector_store.as_retriever(search_kwargs{k:top_k}),return_source_documentsreturn_source,chain_type_kwargs{prompt:PROMPT})# 执行查询print(f\n 正在查询:{question})resultqa_chain.invoke({query:question})response{answer:result[result],sources:[]}ifreturn_sourceandsource_documentsinresult:fordocinresult[source_documents]:response[sources].append({content:doc.page_content[:200]...,source:doc.metadata.get(source,未知)})returnresponsedefsimilarity_search(self,query:str,k:int5)-List[Document]: 相似度搜索仅检索不生成答案 Args: query: 查询文本 k: 返回结果数量 Returns: 相似文档列表 ifself.vector_storeisNone:return[]returnself.vector_store.similarity_search(query,kk)# 使用示例 defmain():主函数# 初始化知识库kbRAGKnowledgeBase(persist_directory./vector_store,embedding_modelBAAI/bge-large-zh-v1.5)# 添加文档首次运行# kb.add_documents(./documents)# 交互式查询print(\n*50)print( 企业知识库问答系统)print(*50)print(输入 quit 退出\n)whileTrue:questioninput(\n请输入问题: ).strip()ifquestion.lower()quit:print(再见)breakifnotquestion:continue# 查询resultkb.query(question,top_k5)# 显示答案print(\n-*50)print( 回答:)print(-*50)print(result[answer])# 显示来源ifresult[sources]:print(\n 参考来源:)fori,sourceinenumerate(result[sources],1):print(f [{i}]{source[source]})if__name____main__:main()五、快速开始5.1 准备文档将你的文档PDF、Word、TXT放入documents目录。5.2 首次运行# 修改 main 函数取消注释添加文档的代码defmain():kbRAGKnowledgeBase()# 首次运行添加文档kb.add_documents(./documents)# 取消注释# ... 后续代码5.3 运行程序python rag_system.py六、进阶功能6.1 支持更多文档类型fromlangchain_community.document_loadersimport(CSVLoader,JSONLoader,UnstructuredHTMLLoader)# CSV 文件loaderCSVLoader(file_pathdata.csv)# JSON 文件loaderJSONLoader(file_pathdata.json,jq_schema.[])# HTML 文件loaderUnstructuredHTMLLoader(file_pathpage.html)6.2 使用其他向量数据库# FAISS本地、轻量fromlangchain_community.vectorstoresimportFAISS vector_storeFAISS.from_documents(chunks,embeddings)# Pinecone云端fromlangchain_pineconeimportPineconeVectorStore vector_storePineconeVectorStore.from_documents(chunks,embeddings,index_namemy-index)# Milvus企业级fromlangchain_milvusimportMilvus vector_storeMilvus.from_documents(chunks,embeddings)6.3 多轮对话支持fromlangchain.memoryimportConversationBufferMemoryfromlangchain.chainsimportConversationalRetrievalChain# 添加记忆功能memoryConversationBufferMemory(memory_keychat_history,return_messagesTrue)# 使用 ConversationalRetrievalChainqa_chainConversationalRetrievalChain.from_llm(llmllm,retrievervector_store.as_retriever(),memorymemory)七、性能优化建议7.1 文本分割策略场景chunk_sizechunk_overlap说明问答系统300-50050较小的块提高精度摘要生成1000-2000100较大的块保留上下文代码文档500-800100中等大小平衡精度和上下文7.2 检索优化# 使用 MMR最大边际相关性检索retrievervector_store.as_retriever(search_typemmr,search_kwargs{k:5,fetch_k:20,lambda_mult:0.5})# 使用相似度阈值过滤retrievervector_store.as_retriever(search_typesimilarity_score_threshold,search_kwargs{score_threshold:0.7})7.3 嵌入模型选择模型语言特点适用场景BAAI/bge-large-zh-v1.5中文中文效果优秀中文知识库text-embedding-3-small多语言OpenAI 官方多语言混合m3e-base中文轻量快速资源受限环境八、常见问题Q1: 加载模型时内存不足# 使用更小的模型embedding_modelBAAI/bge-small-zh-v1.5# 或限制并发os.environ[TOKENIZERS_PARALLELISM]falseQ2: 回答质量不高优化文本分割调整 chunk_size 和 chunk_overlap增加 top_k检索更多相关文档优化 Prompt给模型更明确的指令清洗数据去除文档中的噪声页眉页脚等Q3: 如何更新知识库# 添加新文档kb.add_documents(./new_documents)# 删除所有数据重新创建importshutil shutil.rmtree(./vector_store)kbRAGKnowledgeBase()# 重新初始化kb.add_documents(./documents)九、总结本文介绍了如何用LangChain Python搭建企业级 RAG 知识库。核心流程文档加载支持 PDF、Word、TXT 等多种格式文本分割将长文档切分成适合检索的小块向量化使用嵌入模型将文本转为向量存储检索使用向量数据库高效存储和检索问答生成结合检索结果和大模型生成答案完整代码已上传 GitHub[你的仓库链接]十、下篇预告下一篇文章我将介绍如何用Python AI 自动化生成周报/日报让 AI 帮你自动整理工作内容、生成报告。关注专栏不错过精彩内容推荐阅读LangChain RAG 官方教程ChromaDB 文档BGE 嵌入模型有问题欢迎在评论区留言我会逐一回复版权声明本文为博主原创文章转载请注明出处。