基于 LangChainFaiss的本地知识库问答系统实战1、项目概述本文介绍如何使用LangChain框架构建一个基于本地文档的问答系统RAG。相比原生实现LangChain 提供了更简洁的 API 和更强大的组件生态让开发者能够快速搭建生产级的文档问答应用。1.1 什么是 LangChainLangChain 是一个用于开发大语言模型LLM应用的 Python 框架它提供了文档加载器支持 PDF、Word、TXT 等多种格式文本分割器智能分割长文档向量存储集成 FAISS、Qdrant、Chroma 等向量数据库检索链自动完成检索-生成流程模型封装统一接口调用各种 LLM1.2 系统架构2、核心组件详解2.1 文档加载与分割fromlangchain_community.document_loadersimportPyPDFLoader,Docx2txtLoader,TextLoaderfromlangchain.text_splitterimportRecursiveCharacterTextSplitterdefload_and_split_docs(doc_folder):用LangChain Loader加载多格式文档并用分割器处理docs[]# 遍历文件夹加载所有文档forfileinos.listdir(doc_folder):file_pathos.path.join(doc_folder,file)iffile.endswith(.pdf):loaderPyPDFLoader(file_path)docs.extend(loader.load())# 自动按页分割带页码信息eliffile.endswith(.docx):loaderDocx2txtLoader(file_path)docs.extend(loader.load())# 自动按段落分割eliffile.endswith(.txt):loaderTextLoader(file_path,encodingutf-8)docs.extend(loader.load())# 自动按行分割# 分割文档按500字符重叠50字符text_splitterRecursiveCharacterTextSplitter(chunk_size500,chunk_overlap50,length_functionlen)split_docstext_splitter.split_documents(docs)returnsplit_docs关键参数说明参数说明推荐值chunk_size每个文本块的最大长度500-1000chunk_overlap相邻块的重叠字符数50-100length_function计算长度的函数len2.2 自定义 Embedding 类由于直接使用HuggingFaceEmbeddings加载本地 BGE 模型存在兼容性问题我们自定义一个 Embedding 类fromlangchain_core.embeddingsimportEmbeddingsfromtransformersimportAutoModel,AutoTokenizerclassLocalBGEEmbeddings(Embeddings):自定义本地 BGE Embedding 类def__init__(self,model_path):self.modelAutoModel.from_pretrained(model_path).cuda()self.model.eval()self.tokenizerAutoTokenizer.from_pretrained(model_path)defembed_documents(self,texts):将文档列表转换为向量列表withtorch.no_grad():encodedself.tokenizer(texts,paddingTrue,truncationTrue,max_length512,return_tensorspt)encoded{k:v.cuda()fork,vinencoded.items()}outputself.model(**encoded)# Mean Poolingembeddingsself._mean_pooling(output,encoded[attention_mask])# L2 归一化embeddingstorch.nn.functional.normalize(embeddings,p2,dim1)# 返回 Python 列表returnembeddings.cpu().numpy().tolist()defembed_query(self,text):将查询文本转换为向量resultself.embed_documents([text])returnresult[0]def_mean_pooling(self,model_output,attention_mask):Mean Pooling 操作token_embeddingsmodel_output[0]input_mask_expandedattention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()returntorch.sum(token_embeddings*input_mask_expanded,1)/torch.clamp(input_mask_expanded.sum(1),min1e-9)为什么选择 Mean PoolingBGE 模型输出的是每个 token 的向量[batch, seq_len, hidden_dim]需要通过 Pooling 得到句子向量输入: [batch, seq_len, 1024] ↓ Mean Pooling 输出: [batch, 1024]2.3 向量存储FAISSfromlangchain_community.vectorstoresimportFAISSdefinit_vector_store(split_docs):初始化 FAISS 向量存储embeddingsLocalBGEEmbeddings(/path/to/bge-large-zh-v1.5)# 使用 FAISS 存储向量vector_storeFAISS.from_documents(documentssplit_docs,embeddingembeddings)returnvector_storeFAISS vs Qdrant特性FAISSQdrant部署方式本地内存本地/远程服务持久化支持支持元数据过滤有限强大适用场景原型开发生产环境2.4 QA 链构建fromlangchain.chainsimportRetrievalQAfromlangchain_community.llmsimportHuggingFacePipelinefromtransformersimportpipelinedefinit_qa_chain(vector_store):初始化 RetrievalQA 链# 4位量化配置quantization_configBitsAndBytesConfig(load_in_4bitTrue,bnb_4bit_use_double_quantTrue,bnb_4bit_quant_typenf4,bnb_4bit_compute_dtypetorch.bfloat16)# 加载模型modelAutoModelForCausalLM.from_pretrained(/path/to/deepseek-llm-7b-base,quantization_configquantization_config,device_mapauto)tokenizerAutoTokenizer.from_pretrained(/path/to/deepseek-llm-7b-base)# 创建 Pipelinellm_pipelinepipeline(text-generation,modelmodel,tokenizertokenizer,max_new_tokens512,temperature0.7,top_p0.9)llmHuggingFacePipeline(pipelinellm_pipeline)# 创建 QA 链qa_chainRetrievalQA.from_chain_type(llmllm,chain_typestuff,# 简单填充式链retrievervector_store.as_retriever(top_k3),return_source_documentsTrue# 返回源文档用于溯源)returnqa_chainchain_type 说明类型说明适用场景stuff直接填充所有检索到的文档文档片段较短map_reduce分别处理每个文档后汇总文档片段较长refine迭代优化答案需要高质量答案3、完整代码# encodingutf-8importosimporttorchfromlangchain_community.document_loadersimport(PyPDFLoader,Docx2txtLoader,TextLoader)fromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain_core.embeddingsimportEmbeddingsfromlangchain_community.vectorstoresimportFAISSfromlangchain.chainsimportRetrievalQAfromlangchain_community.llmsimportHuggingFacePipelinefromtransformersimport(AutoModel,AutoTokenizer,AutoModelForCausalLM,BitsAndBytesConfig,pipeline)classLocalBGEEmbeddings(Embeddings):自定义本地 BGE Embedding 类def__init__(self,model_path):self.modelAutoModel.from_pretrained(model_path).cuda()self.model.eval()self.tokenizerAutoTokenizer.from_pretrained(model_path)defembed_documents(self,texts):withtorch.no_grad():encodedself.tokenizer(texts,paddingTrue,truncationTrue,max_length512,return_tensorspt)encoded{k:v.cuda()fork,vinencoded.items()}outputself.model(**encoded)# Mean Poolingtoken_embeddingsoutput[0]input_mask_expandedencoded[attention_mask].unsqueeze(-1).expand(token_embeddings.size()).float()embeddingstorch.sum(token_embeddings*input_mask_expanded,1)/torch.clamp(input_mask_expanded.sum(1),min1e-9)# L2 归一化embeddingstorch.nn.functional.normalize(embeddings,p2,dim1)returnembeddings.cpu().numpy().tolist()defembed_query(self,text):resultself.embed_documents([text])returnresult[0]defload_and_split_docs(doc_folder):加载并分割文档docs[]forfileinos.listdir(doc_folder):file_pathos.path.join(doc_folder,file)iffile.endswith(.pdf):loaderPyPDFLoader(file_path)docs.extend(loader.load())eliffile.endswith(.docx):loaderDocx2txtLoader(file_path)docs.extend(loader.load())eliffile.endswith(.txt):loaderTextLoader(file_path,encodingutf-8)docs.extend(loader.load())text_splitterRecursiveCharacterTextSplitter(chunk_size500,chunk_overlap50,length_functionlen)returntext_splitter.split_documents(docs)definit_vector_store(split_docs):初始化向量存储embeddingsLocalBGEEmbeddings(/path/to/bge-large-zh-v1.5)returnFAISS.from_documents(documentssplit_docs,embeddingembeddings)definit_qa_chain(vector_store):初始化 QA 链quantization_configBitsAndBytesConfig(load_in_4bitTrue,bnb_4bit_use_double_quantTrue,bnb_4bit_quant_typenf4,bnb_4bit_compute_dtypetorch.bfloat16)modelAutoModelForCausalLM.from_pretrained(/path/to/deepseek-llm-7b-base,quantization_configquantization_config,device_mapauto)tokenizerAutoTokenizer.from_pretrained(/path/to/deepseek-llm-7b-base)llm_pipelinepipeline(text-generation,modelmodel,tokenizertokenizer,max_new_tokens512,temperature0.7,top_p0.9)llmHuggingFacePipeline(pipelinellm_pipeline)returnRetrievalQA.from_chain_type(llmllm,chain_typestuff,retrievervector_store.as_retriever(top_k3),return_source_documentsTrue)defmain():# 配置路径DOC_FOLDER/path/to/docs# 加载文档print(正在加载文档...)split_docsload_and_split_docs(DOC_FOLDER)# 初始化向量库print(正在生成向量...)vector_storeinit_vector_store(split_docs)# 初始化 QA 链print(正在加载模型...)qa_chaininit_qa_chain(vector_store)# 交互式问答print(\n系统就绪输入问题输入 quit 退出)whileTrue:queryinput(\n问题: )ifquery.lower()quit:breakresultqa_chain({query:query})print(f\n回答:{result[result]})print(\n参考来源:)fori,docinenumerate(result[source_documents],1):print(f{i}.{doc.metadata[source]})if__name____main__:torch.set_num_threads(1)main()4、运行效果5、常见问题与解决方案5.1 sentence-transformers 版本冲突问题ImportError: cannot import name cached_download解决直接使用transformers.AutoModel加载 BGE 模型绕过sentence-transformers。5.2 FAISS 向量维度错误问题ValueError: too many values to unpack解决确保embed_documents返回List[List[float]]embed_query返回List[float]。5.3 CUDA 内存不足问题RuntimeError: CUDA out of memory解决使用 4-bit 量化加载模型减小chunk_size减少同时处理的文本量使用torch.set_num_threads(1)限制线程数6、总结本文介绍了如何使用 LangChain 构建本地知识库问答系统核心要点文档处理使用 LangChain 的 Loader 和 Splitter 简化文档处理流程向量生成自定义 Embedding 类解决本地模型加载问题向量存储FAISS 适合快速原型Qdrant 适合生产环境问答链RetrievalQA 自动完成检索和生成流程相比原生实现LangChain 版本代码更简洁、更易维护且能方便地替换各个组件如换用其他向量库或 LLM。
多模态大模型学习笔记（十九）——基于 LangChain+Faiss的本地知识库问答系统实战
基于 LangChainFaiss的本地知识库问答系统实战1、项目概述本文介绍如何使用LangChain框架构建一个基于本地文档的问答系统RAG。相比原生实现LangChain 提供了更简洁的 API 和更强大的组件生态让开发者能够快速搭建生产级的文档问答应用。1.1 什么是 LangChainLangChain 是一个用于开发大语言模型LLM应用的 Python 框架它提供了文档加载器支持 PDF、Word、TXT 等多种格式文本分割器智能分割长文档向量存储集成 FAISS、Qdrant、Chroma 等向量数据库检索链自动完成检索-生成流程模型封装统一接口调用各种 LLM1.2 系统架构2、核心组件详解2.1 文档加载与分割fromlangchain_community.document_loadersimportPyPDFLoader,Docx2txtLoader,TextLoaderfromlangchain.text_splitterimportRecursiveCharacterTextSplitterdefload_and_split_docs(doc_folder):用LangChain Loader加载多格式文档并用分割器处理docs[]# 遍历文件夹加载所有文档forfileinos.listdir(doc_folder):file_pathos.path.join(doc_folder,file)iffile.endswith(.pdf):loaderPyPDFLoader(file_path)docs.extend(loader.load())# 自动按页分割带页码信息eliffile.endswith(.docx):loaderDocx2txtLoader(file_path)docs.extend(loader.load())# 自动按段落分割eliffile.endswith(.txt):loaderTextLoader(file_path,encodingutf-8)docs.extend(loader.load())# 自动按行分割# 分割文档按500字符重叠50字符text_splitterRecursiveCharacterTextSplitter(chunk_size500,chunk_overlap50,length_functionlen)split_docstext_splitter.split_documents(docs)returnsplit_docs关键参数说明参数说明推荐值chunk_size每个文本块的最大长度500-1000chunk_overlap相邻块的重叠字符数50-100length_function计算长度的函数len2.2 自定义 Embedding 类由于直接使用HuggingFaceEmbeddings加载本地 BGE 模型存在兼容性问题我们自定义一个 Embedding 类fromlangchain_core.embeddingsimportEmbeddingsfromtransformersimportAutoModel,AutoTokenizerclassLocalBGEEmbeddings(Embeddings):自定义本地 BGE Embedding 类def__init__(self,model_path):self.modelAutoModel.from_pretrained(model_path).cuda()self.model.eval()self.tokenizerAutoTokenizer.from_pretrained(model_path)defembed_documents(self,texts):将文档列表转换为向量列表withtorch.no_grad():encodedself.tokenizer(texts,paddingTrue,truncationTrue,max_length512,return_tensorspt)encoded{k:v.cuda()fork,vinencoded.items()}outputself.model(**encoded)# Mean Poolingembeddingsself._mean_pooling(output,encoded[attention_mask])# L2 归一化embeddingstorch.nn.functional.normalize(embeddings,p2,dim1)# 返回 Python 列表returnembeddings.cpu().numpy().tolist()defembed_query(self,text):将查询文本转换为向量resultself.embed_documents([text])returnresult[0]def_mean_pooling(self,model_output,attention_mask):Mean Pooling 操作token_embeddingsmodel_output[0]input_mask_expandedattention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()returntorch.sum(token_embeddings*input_mask_expanded,1)/torch.clamp(input_mask_expanded.sum(1),min1e-9)为什么选择 Mean PoolingBGE 模型输出的是每个 token 的向量[batch, seq_len, hidden_dim]需要通过 Pooling 得到句子向量输入: [batch, seq_len, 1024] ↓ Mean Pooling 输出: [batch, 1024]2.3 向量存储FAISSfromlangchain_community.vectorstoresimportFAISSdefinit_vector_store(split_docs):初始化 FAISS 向量存储embeddingsLocalBGEEmbeddings(/path/to/bge-large-zh-v1.5)# 使用 FAISS 存储向量vector_storeFAISS.from_documents(documentssplit_docs,embeddingembeddings)returnvector_storeFAISS vs Qdrant特性FAISSQdrant部署方式本地内存本地/远程服务持久化支持支持元数据过滤有限强大适用场景原型开发生产环境2.4 QA 链构建fromlangchain.chainsimportRetrievalQAfromlangchain_community.llmsimportHuggingFacePipelinefromtransformersimportpipelinedefinit_qa_chain(vector_store):初始化 RetrievalQA 链# 4位量化配置quantization_configBitsAndBytesConfig(load_in_4bitTrue,bnb_4bit_use_double_quantTrue,bnb_4bit_quant_typenf4,bnb_4bit_compute_dtypetorch.bfloat16)# 加载模型modelAutoModelForCausalLM.from_pretrained(/path/to/deepseek-llm-7b-base,quantization_configquantization_config,device_mapauto)tokenizerAutoTokenizer.from_pretrained(/path/to/deepseek-llm-7b-base)# 创建 Pipelinellm_pipelinepipeline(text-generation,modelmodel,tokenizertokenizer,max_new_tokens512,temperature0.7,top_p0.9)llmHuggingFacePipeline(pipelinellm_pipeline)# 创建 QA 链qa_chainRetrievalQA.from_chain_type(llmllm,chain_typestuff,# 简单填充式链retrievervector_store.as_retriever(top_k3),return_source_documentsTrue# 返回源文档用于溯源)returnqa_chainchain_type 说明类型说明适用场景stuff直接填充所有检索到的文档文档片段较短map_reduce分别处理每个文档后汇总文档片段较长refine迭代优化答案需要高质量答案3、完整代码# encodingutf-8importosimporttorchfromlangchain_community.document_loadersimport(PyPDFLoader,Docx2txtLoader,TextLoader)fromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain_core.embeddingsimportEmbeddingsfromlangchain_community.vectorstoresimportFAISSfromlangchain.chainsimportRetrievalQAfromlangchain_community.llmsimportHuggingFacePipelinefromtransformersimport(AutoModel,AutoTokenizer,AutoModelForCausalLM,BitsAndBytesConfig,pipeline)classLocalBGEEmbeddings(Embeddings):自定义本地 BGE Embedding 类def__init__(self,model_path):self.modelAutoModel.from_pretrained(model_path).cuda()self.model.eval()self.tokenizerAutoTokenizer.from_pretrained(model_path)defembed_documents(self,texts):withtorch.no_grad():encodedself.tokenizer(texts,paddingTrue,truncationTrue,max_length512,return_tensorspt)encoded{k:v.cuda()fork,vinencoded.items()}outputself.model(**encoded)# Mean Poolingtoken_embeddingsoutput[0]input_mask_expandedencoded[attention_mask].unsqueeze(-1).expand(token_embeddings.size()).float()embeddingstorch.sum(token_embeddings*input_mask_expanded,1)/torch.clamp(input_mask_expanded.sum(1),min1e-9)# L2 归一化embeddingstorch.nn.functional.normalize(embeddings,p2,dim1)returnembeddings.cpu().numpy().tolist()defembed_query(self,text):resultself.embed_documents([text])returnresult[0]defload_and_split_docs(doc_folder):加载并分割文档docs[]forfileinos.listdir(doc_folder):file_pathos.path.join(doc_folder,file)iffile.endswith(.pdf):loaderPyPDFLoader(file_path)docs.extend(loader.load())eliffile.endswith(.docx):loaderDocx2txtLoader(file_path)docs.extend(loader.load())eliffile.endswith(.txt):loaderTextLoader(file_path,encodingutf-8)docs.extend(loader.load())text_splitterRecursiveCharacterTextSplitter(chunk_size500,chunk_overlap50,length_functionlen)returntext_splitter.split_documents(docs)definit_vector_store(split_docs):初始化向量存储embeddingsLocalBGEEmbeddings(/path/to/bge-large-zh-v1.5)returnFAISS.from_documents(documentssplit_docs,embeddingembeddings)definit_qa_chain(vector_store):初始化 QA 链quantization_configBitsAndBytesConfig(load_in_4bitTrue,bnb_4bit_use_double_quantTrue,bnb_4bit_quant_typenf4,bnb_4bit_compute_dtypetorch.bfloat16)modelAutoModelForCausalLM.from_pretrained(/path/to/deepseek-llm-7b-base,quantization_configquantization_config,device_mapauto)tokenizerAutoTokenizer.from_pretrained(/path/to/deepseek-llm-7b-base)llm_pipelinepipeline(text-generation,modelmodel,tokenizertokenizer,max_new_tokens512,temperature0.7,top_p0.9)llmHuggingFacePipeline(pipelinellm_pipeline)returnRetrievalQA.from_chain_type(llmllm,chain_typestuff,retrievervector_store.as_retriever(top_k3),return_source_documentsTrue)defmain():# 配置路径DOC_FOLDER/path/to/docs# 加载文档print(正在加载文档...)split_docsload_and_split_docs(DOC_FOLDER)# 初始化向量库print(正在生成向量...)vector_storeinit_vector_store(split_docs)# 初始化 QA 链print(正在加载模型...)qa_chaininit_qa_chain(vector_store)# 交互式问答print(\n系统就绪输入问题输入 quit 退出)whileTrue:queryinput(\n问题: )ifquery.lower()quit:breakresultqa_chain({query:query})print(f\n回答:{result[result]})print(\n参考来源:)fori,docinenumerate(result[source_documents],1):print(f{i}.{doc.metadata[source]})if__name____main__:torch.set_num_threads(1)main()4、运行效果5、常见问题与解决方案5.1 sentence-transformers 版本冲突问题ImportError: cannot import name cached_download解决直接使用transformers.AutoModel加载 BGE 模型绕过sentence-transformers。5.2 FAISS 向量维度错误问题ValueError: too many values to unpack解决确保embed_documents返回List[List[float]]embed_query返回List[float]。5.3 CUDA 内存不足问题RuntimeError: CUDA out of memory解决使用 4-bit 量化加载模型减小chunk_size减少同时处理的文本量使用torch.set_num_threads(1)限制线程数6、总结本文介绍了如何使用 LangChain 构建本地知识库问答系统核心要点文档处理使用 LangChain 的 Loader 和 Splitter 简化文档处理流程向量生成自定义 Embedding 类解决本地模型加载问题向量存储FAISS 适合快速原型Qdrant 适合生产环境问答链RetrievalQA 自动完成检索和生成流程相比原生实现LangChain 版本代码更简洁、更易维护且能方便地替换各个组件如换用其他向量库或 LLM。
