目录一、构建知识库二、调用回顾 初识Agent 这篇内容Agent是要在LLM的帮助下通过调用工具来做成事情。而且也给了一个简单的公式Agent 大语言模型LLM 工具 知识库。在上一篇 Agent调用工具 中我们已经学习了Agent如何调用工具如何用LCEL编排本篇内容就是学习一下基于 LangChain 的多文件 RAG 应用也就是知识库的部分。RAG检索增强生成我们在前面 如何与大模型交互从提示词工程到RAG 和 RAG的工作流程与目标 这两篇内容中详细的介绍过先检索再回复是为了让大模型在回答我们的问题时“有据可依”减少大模型的幻觉。RAG包含两个步骤1.构建知识库2.调用。一、构建知识库# 步骤 1加载文档并创建索引 def load_documents_and_create_index(file_dir: str ./docs, persist_dir: str ./langchain_storage): 加载文档文件夹中的所有文件并创建向量索引 # 创建embedding模型 embeddings DashScopeEmbeddings( modeltext-embedding-v1, dashscope_api_keyDASHSCOPE_API_KEY, ) # 检查索引是否已存在如果存在加载向量数据库 if os.path.exists(persist_dir): try: # 从存储中加载索引 vector_store FAISS.load_local( persist_dir, embeddings, allow_dangerous_deserializationTrue ) print(从存储加载索引成功) return vector_store except Exception as e: print(f加载索引失败: {e}将重新创建索引) # 如果索引不存在创建新索引 if not os.path.exists(file_dir): print(f文档目录 {file_dir} 不存在) return None # Step 1: 加载目录下的所有 txt 文件 loader DirectoryLoader( file_dir, glob**/*.txt, loader_clsTextLoader, loader_kwargs{encoding: utf-8} ) documents loader.load() print(f加载了 {len(documents)} 个文档) if not documents: print(没有找到任何文档) return None # Step 2:文本分割 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, length_functionlen, ) chunks text_splitter.split_documents(documents) print(f文本被分割成 {len(chunks)} 个块) # Step 3: 创建向量索引 vector_store FAISS.from_documents(chunks, embeddings) # 保存索引 os.makedirs(persist_dir, exist_okTrue) vector_store.save_local(persist_dir) print(f索引已保存到 {persist_dir}) return vector_storeRAG的工作流程与目标 还记得我们前面怎么写构建知识库吗代码中其实也是这个逻辑函数一进来检查向量库索引是否已经建好了如果建好了加载索引后就直接返回vector_store如果没有建好那就走如下的三步Step 1加载目录下的所有 txt 文件》整理文档DocumentsStep 2文本分割》文本切片Chunking每个文本块最大1000个字符数相邻两个块之间的重叠200个字符数防止关键信息被截断在边界。Setp 3创建向量索引并保存》存入向量 数据库 Vector DataBase 二、调用# 创建问答链 qa_chain create_qa_chain(llm) # Step 1: 执行查询 query 介绍下雇主责任险 print(f\n用户查询: {query}\n) # Step 2: 相似度搜索找到相关文档 docs vector_store.similarity_search(query, k5) # 显示召回的文档内容 print( 召回的文档内容 ) if docs: for i, doc in enumerate(docs): print(f\n文档片段 {i1}:) print(f内容: {doc.page_content[:200]}...) print(f来源: {doc.metadata.get(source, 未知)}) else: print(没有召回任何文档内容) print(\n) # 格式化上下文 context \n\n.join(doc.page_content for doc in docs) # step 3: 执行问答链 print( AI 回复 ) response qa_chain.invoke({context: context, question: query}) print(response) print(\n)Step 1执行查询》用户提问QueryStep 2相似度搜索找到相关文档》向量数据库相似度检索RetriveStep 3执行问答链》提示词拼接Augment,交给大模型LLM生成答案Generate--------------------------------------------------------------------------------------------------------------完整代码如下#!/usr/bin/env python # coding: utf-8 基于 LangChain 的多文件 RAG 应用 支持加载 docs 文件夹下的多种格式文件进行问答 import os from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.embeddings import DashScopeEmbeddings from langchain_community.vectorstores import FAISS from langchain_community.chat_models import ChatTongyi from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 获取 API Key DASHSCOPE_API_KEY os.getenv(DASHSCOPE_API_KEY) if not DASHSCOPE_API_KEY: raise ValueError(请设置环境变量 DASHSCOPE_API_KEY) # 步骤 1加载文档并创建索引 def load_documents_and_create_index(file_dir: str ./docs, persist_dir: str ./langchain_storage): 加载文档文件夹中的所有文件并创建向量索引 # 创建embedding模型 embeddings DashScopeEmbeddings( modeltext-embedding-v1, dashscope_api_keyDASHSCOPE_API_KEY, ) # 检查索引是否已存在 if os.path.exists(persist_dir): try: # 从存储中加载索引 vector_store FAISS.load_local( persist_dir, embeddings, allow_dangerous_deserializationTrue ) print(从存储加载索引成功) return vector_store except Exception as e: print(f加载索引失败: {e}将重新创建索引) # 如果索引不存在创建新索引 if not os.path.exists(file_dir): print(f文档目录 {file_dir} 不存在) return None # 加载目录下的所有 txt 文件 loader DirectoryLoader( file_dir, glob**/*.txt, loader_clsTextLoader, loader_kwargs{encoding: utf-8} ) documents loader.load() print(f加载了 {len(documents)} 个文档) if not documents: print(没有找到任何文档) return None # 文本分割 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, length_functionlen, ) chunks text_splitter.split_documents(documents) print(f文本被分割成 {len(chunks)} 个块) # 创建向量索引 vector_store FAISS.from_documents(chunks, embeddings) # 保存索引 os.makedirs(persist_dir, exist_okTrue) vector_store.save_local(persist_dir) print(f索引已保存到 {persist_dir}) return vector_store # 步骤 2创建问答链 def create_qa_chain(llm): 创建 QA 问答链 (LangChain 1.x LCEL 写法) # QA Prompt 模板 qa_prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的AI助手。 根据以下上下文内容回答用户的问题。如果上下文中没有相关信息请如实说明。 你总是用中文回复用户。 上下文内容: {context}), (human, {question}) ]) # 创建问答链 (LCEL 管道语法) qa_chain qa_prompt | llm | StrOutputParser() return qa_chain # 步骤 3主函数 def main(): 主函数 # 配置 LLM llm ChatTongyi( model_namedeepseek-v3, dashscope_api_keyDASHSCOPE_API_KEY ) # 加载文档并创建索引 vector_store load_documents_and_create_index() if vector_store is None: print(无法创建索引程序退出) return # 创建问答链 qa_chain create_qa_chain(llm) # 执行查询 query 介绍下雇主责任险 print(f\n用户查询: {query}\n) # 相似度搜索找到相关文档 docs vector_store.similarity_search(query, k5) # 显示召回的文档内容 print( 召回的文档内容 ) if docs: for i, doc in enumerate(docs): print(f\n文档片段 {i1}:) print(f内容: {doc.page_content[:200]}...) print(f来源: {doc.metadata.get(source, 未知)}) else: print(没有召回任何文档内容) print(\n) # 格式化上下文 context \n\n.join(doc.page_content for doc in docs) # 执行问答链 print( AI 回复 ) response qa_chain.invoke({context: context, question: query}) print(response) print(\n) if __name__ __main__: main()执行结果内容太长图没有截全