LangChain大模型应用开发：基于RAG实现文档问答

介绍

大家好，博主又来给大家分享知识了。随着大模型应用的不断发展，很多开发者都在探索如何更好地利用相关工具进行开发。那么这次给大家分享的内容是使用LangChain进行大模型应用开发中的基于RAG实现文档问答的功能。

好了，我们直接进入正题。

RAG

RAG即检索增强生成(Retrieval-Augmented Generation)，是一种结合信息检索和语言生成的技术。它在处理用户输入时，先从外部知识库中检索相关信息，再利用大语言模型将这些信息整合，生成回答。这种方式让模型在生成内容时能借助最新外部知识，提高回答的准确性和时效性，减少 “幻觉” 问题，广泛应用于问答系统、智能客服等场景。

大语言模型可以对广泛的主题进行推理，但它们的知识仅限于训练时截止日期前的公开数据。如果我们想构建能够对私有数据或模型截止日期后引入的数据进行推理的人工智能应用，我们需要用特定信息来增强模型的知识。检索适当信息并将其插入模型提示的过程被称为检索增强生成(RAG)。

LangChain拥有多个不同的组件，这些组件能够为构建问答应用程序提供支持，同时，对于基于检索增强生成(RAG)技术的更广泛领域应用，LangChain的组件也可以起到协助构建的作用。

下图展示的是基于检索增强生成(RAG)技术实现文档问答的流程。

我们看到，上图表达了从知识文本处理到最终输出回答的完整过程：

文本分块：知识文本被分割成一个个小的文本块(chunks)，便于后续处理。
嵌入生成：这些文本块被送入嵌入模型，将文本转化为向量形式，以便在向量空间中进行相似度计算。
向量存储：生成的向量被存储在向量数据库中，为后续的检索做准备。
用户查询：用户输入查询内容，该查询也会经过类似的向量化处理。
检索与回答：系统根据用户查询向量，在向量数据库中检索召回相关的文本块向量，然后将相关信息输入到大语言模型(LLM)中，由大语言模型生成回答输出给用户。此外，图中虚线连接的记忆库，用于存储历史交互信息，辅助模型更好地回答问题。

RAG工作流

一个典型的RAG(检索增强生成)应用主要由两大关键部分构成：

索引构建(Indexing)：此环节旨在从各类数据源，如文档、数据库、网页等，收集数据。接着，对数据进行预处理，像文本分块、去除噪声等，然后借助嵌入模型将其转化为向量形式，构建索引存储在向量数据库中。这部分工作一般在离线状态下完成，能提前为后续的检索做足准备。
检索与生成(Retrieval and generation)：当用户输入查询指令后，系统先将查询内容向量化，然后依据相似度算法，在已建立的索引中快速检索出相关数据。筛选整合这些数据后，传递给大语言模型，模型基于输入信息生成最终答案反馈给用户。

从原始数据到得出答案的最常见完整顺序如下：

索引(Indexing)

加载(Load)：首先我们需要加载数据。这是通过文档加载器Document Loaders完成的。
分割(Split)：文本分割器Text splitters将大型文档(Documents)分成更小的块(chunks)。这对于索引数据和将其传递给模型都很有用，因为大块数据更难搜索，而且不适合模型有限的上下文窗口。
存储(Store)：我们需要一个地方来存储和索引我们的分割(splits)，以便后续可以对其进行搜索。这通常使用向量存储VectorStore和嵌入模型Embeddings model来完成。

检索和生成(Retrieval and generation)

检索 (Retrieve)：给定用户输入，使用检索器 Retriever 从存储中检索相关的文本片段。
生成 (Generate)： ChatModel 使用包含问题和检索到的数据的提示来生成答案。

文档问答

实现流程

基于RAG的文档问答APP实现流程如下：

用户通过RAG客户端上传txt格式文档。上传时可设置进度条显示，提升用户体验。
服务器端接收文件，并进行存储备份，同时记录文件相关信息，如上传时间、用户标识等。
服务器端程序按既定编码格式读取文件内容，若读取失败则反馈错误信息给用户。
采用合适的文本分割算法，将文件内容拆分成合适长度的文本块，确保不超出Embedding模型的token限制，同时尽量保持语义完整。
利用Embedding模型将文本块转化为向量形式，存储到向量数据库中，并基于数据库特性生成高效的检索器。
完成上述步骤后，程序进入可交互状态，提示用户可以开始提问。
用户输入问题，大模型调用检索器在向量数据库中检索相关文本块，依据相关性和语义匹配度筛选、整合内容，组织成答案回复用户。

代码实现

实现文件上传

Streamlit是一个开源的Python库，用于构建数据科学和机器学习Web应用。它可以让我们仅用纯Python代码，就能快速搭建互动性强且用户友好的数据应用。

使用Streamlit，我们能够轻松实现丰富多样的用户交互功能。它提供了一系列直观且易于使用的组件，极大地增强了用户与应用之间的互动性。

首先，我们要安装Streamlit。安装命令：

 pip install streamlit

然后，我们使用Streamlit实现文件上传，我这里为了讲解只实现了txt文件上传，其实这里可以在type参数里面设置多个文件类型，在后面的检索器方法里面针对每个类型进行处理即可。

完整代码

# 导入Streamlit库，别名为st，用于构建交互式Web应用
import streamlit as st# 上传txt文件，允许上传多个文件
uploaded_files = st.sidebar.file_uploader(label="上传txt文件",  # 在侧边栏创建一个文件上传组件，设置标签为“上传txt文件”type=["txt"],  # 只允许上传txt类型文件accept_multiple_files=True,  # 且允许多选
)if not uploaded_files:  # 如果没有上传文件st.info("请先上传按TXT文档。")  # 在页面显示提示信息，告知用户先上传TXT文档st.stop()  # 停止应用后续代码的执行

运行结果

如何运行

这里大家注意一下，上述完整代码的运行并不是传统的在PyCharm中按Ctrl+Shift+F10的运行方式。若要运行代码需要在PyCharm的终端下输入命令：streamlit run 你的Python文件.py。

博主的工程目录为：D:\Software\PythonProjects\LangChain>，博主要执行的Python文件名字为：streamlit_app.py。所以博主运行该Python文件的命令是streamlit run streamlit_app.py。

运行说明

执行完上述命令后，电脑上的默认浏览器会打开并展示运行结果界面。若要关闭此浏览器，先要在上述终端(你执行streamlit命令的终端)按下Ctrl+c，当终端出现Stopping...后，再去关闭浏览器页面。
若先关闭了浏览器页面，此时回到终端(你执行streamlit命令的终端)按下Ctrl+c后没有任何反应。此时大家要再次点击执行完streamlit run出现的链接地址(一个是Local URL，一个是Network URL)，随便点击一个地址。等到浏览器打开页面后，再执行上面的步骤就可以结束终端运行了。

实现检索器

我们实现一个检索器，它用来对上传的文件进行处理，构建一个基于这些文件内容的检索器。

演示代码

# 导入os模块，用于处理文件和目录路径等操作系统相关操作
import os
# 导入tempfile模块，用于创建临时文件和目录
import tempfile
# 从langchain.agents模块导入create_react_agent函数和AgentExecutor类，用于创建和执行代理
from langchain.agents import create_react_agent, AgentExecutor
# 从langchain_community.document_loaders模块导入TextLoader类，用于加载文本文件
from langchain_community.document_loaders import TextLoader
# 从langchain_community.embeddings模块导入OpenAIEmbeddings类，用于生成文本的嵌入向量
from langchain_community.embeddings import OpenAIEmbeddings
# 从langchain_community.vectorstores模块导入FAISS类，用于存储和检索向量数据
from langchain_community.vectorstores import FAISS
# 从langchain_core.prompts模块导入PromptTemplate类，用于创建提示模板
from langchain_core.prompts import PromptTemplate
# 从langchain_openai模块导入ChatOpenAI类，用于调用OpenAI的聊天模型
from langchain_openai import ChatOpenAI
# 从langchain_text_splitters模块导入RecursiveCharacterTextSplitter类，用于分割文本
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 从langchain.tools.retriever模块导入create_retriever_tool函数，用于创建检索工具
from langchain.tools.retriever import create_retriever_toolclass LangChainBasedOnRAG:def __init__(self):# 初始化检索器，初始值为Noneself.retriever = None# 初始化工具列表，初始值为Noneself.tools = None# 初始化代理执行器，初始值为Noneself.agent_executor = Nonedef configure_retriever(self, uploaded_files):# 读取上传的文档，并写入一个临时目录docs = []  # 用于存储加载的文档# 创建一个临时目录，指定目录位置为 D 盘temp_dir = tempfile.TemporaryDirectory(dir=r"D:\\")# 遍历上传的文件列表for file in uploaded_files:# 拼接临时文件的完整路径temp_filepath = os.path.join(temp_dir.name, file.name)# 以二进制写入模式打开临时文件with open(temp_filepath, "wb") as f:# 将文件内容写入临时文件f.write(file.getvalue())# 创建一个TextLoader实例，用于加载临时文件loader = TextLoader(temp_filepath, encoding="utf-8")# 加载文件内容并添加到docs列表中docs.extend(loader.load())# 创建一个RecursiveCharacterTextSplitter实例，设置块大小和重叠部分text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=30)# 对文档进行分割splits = text_splitter.split_documents(docs)# 创建一个OpenAIEmbeddings实例，用于生成文本的嵌入向量embeddings = OpenAIEmbeddings()# 使用FAISS存储文档的嵌入向量vector = FAISS.from_documents(docs, embeddings)# 将向量存储转换为检索器self.retriever = vector.as_retriever()# 调用setup_tools方法设置工具self.setup_tools()# 调用setup_agent方法设置代理self.setup_agent()

创建检索工具

我们利用LangChain中的create_retriever_tool函数，基于已配置好的检索器创建一个名为 “文档检索” 的工具，该工具可根据用户问题从文档中检索信息并用于回复。

演示代码

    def setup_tools(self):# 创建一个检索工具tool = create_retriever_tool(self.retriever,  # 指定检索器"文档检索",  # 工具名称"用于检索用户提出的问题，并基于检索到的文档内容进行回复.",  # 工具描述)# 将工具添加到工具列表中self.tools = [tool]

创建智能体执行器

我们配置和创建一个基于语言模型的代理及其执行器，用于根据用户输入从文档中检索信息并回答问题。

演示代码

    def setup_agent(self):# 定义代理的指令instructions = """您是一个设计用于查询文档来回答问题的代理。您可以使用文档检索工具，并基于检索内容来回答问题您可能不查询文档就知道答案，但是您仍然应该查询文档来获得答案。如果您从文档中找不到任何信息用于回答问题，则只需返回“抱歉，这个问题我还不知道。”作为答案。"""# 定义基础提示模板base_prompt_template = """{instructions}TOOLS:------You have access to the following tools:{tools}To use a tool, please use the following format:•```Thought: Do I need to use a tool? YesAction: the action to take, should be one of [{tool_names}]Action Input: the input to the actionObservation: the result of the action•```When you have a response to say to the Human, or if you do not need to use a tool, you MUST use the format:•```Thought: Do I need to use a tool? NoFinal Answer: [your response here]•```Begin!New input: {input}{agent_scratchpad}"""# 根据基础提示模板创建PromptTemplate实例base_prompt = PromptTemplate.from_template(base_prompt_template)# 对基础提示进行部分填充，填入指令信息prompt = base_prompt.partial(instructions=instructions)# 创建llm，使用gpt-3.5-turbo模型chat_model = ChatOpenAI(model="gpt-3.5-turbo")# 创建react Agentagent = create_react_agent(chat_model, self.tools, prompt)# 创建代理执行器self.agent_executor = AgentExecutor(agent=agent, tools=self.tools, verbose=False)

展示智能体回复

我们还是使用streamlit库将智能体对用户提问的回答显示在前端页面上。

演示代码

# 如果用户有输入查询内容if self.user_query:# 将用户的消息添加到st.session_state的消息列表中st.session_state.messages.append({"role": "user", "content": self.user_query})# 创建用户角色的聊天消息显示区域with st.chat_message("user"):# 在聊天消息区域显示用户输入的内容st.write(self.user_query)# 创建一个StreamlitCallbackHandler实例，用于处理回调st_cb = StreamlitCallbackHandler(st.container())# 配置回调信息，将StreamlitCallbackHandler实例放入列表中config = {"callbacks": [st_cb]}# 调用LangChain实例的agent_executor_invoke方法处理用户查询response = st.session_state.langchain_based_on_rag.agent_executor_invoke(self.user_query, config)# 从响应中提取助手的回复内容assistant_reply = response["output"]# 将助手的回复消息添加到st.session_state的消息列表中st.session_state.messages.append({"role": "assistant", "content": assistant_reply})# 创建助手角色的聊天消息显示区域with st.chat_message("assistant"):# 在聊天消息区域显示助手的回复内容st.write(assistant_reply)

完整代码

streamlit_app.py

# 导入Streamlit库，用于构建交互式 Web 应用
import streamlit as st
# 从langchain的回调模块中导入StreamlitCallbackHandler，用于在Streamlit中处理回调
from langchain.callbacks.streamlit import StreamlitCallbackHandler
# 从自定义模块中导入基于RAG的LangChain类
from langchain_based_on_rag import LangChainBasedOnRAGclass StreamlitApp:def __init__(self):# 初始化用户查询，初始值为Noneself.user_query = None# 初始化上传的文件，初始值为Noneself.uploaded_files = None# 检查st.session_state中是否存在'messages'键，如果不存在则初始化为空列表if 'messages' not in st.session_state:st.session_state.messages = []def start_chat(self):# 再次检查st.session_state中是否存在'messages'键，如果不存在则初始化为空列表if 'messages' not in st.session_state:st.session_state.messages = []# 遍历st.session_state中的消息列表for message in st.session_state.messages:# 创建对应角色(用户或助手)的聊天消息显示区域with st.chat_message(message["role"]):# 在聊天消息区域显示消息内容st.write(message["content"])# 如果用户有输入查询内容if self.user_query:# 将用户的消息添加到st.session_state的消息列表中st.session_state.messages.append({"role": "user", "content": self.user_query})# 创建用户角色的聊天消息显示区域with st.chat_message("user"):# 在聊天消息区域显示用户输入的内容st.write(self.user_query)# 创建一个StreamlitCallbackHandler实例，用于处理回调st_cb = StreamlitCallbackHandler(st.container())# 配置回调信息，将StreamlitCallbackHandler实例放入列表中config = {"callbacks": [st_cb]}# 调用LangChain实例的agent_executor_invoke方法处理用户查询response = st.session_state.langchain_based_on_rag.agent_executor_invoke(self.user_query, config)# 从响应中提取助手的回复内容assistant_reply = response["output"]# 将助手的回复消息添加到st.session_state的消息列表中st.session_state.messages.append({"role": "assistant", "content": assistant_reply})# 创建助手角色的聊天消息显示区域with st.chat_message("assistant"):# 在聊天消息区域显示助手的回复内容st.write(assistant_reply)def run_app(self):# 在侧边栏创建一个文件上传器，允许用户上传多个txt文件self.uploaded_files = st.sidebar.file_uploader(label="上传txt文件",type=["txt"],accept_multiple_files=True,)# 如果用户没有上传文件if not self.uploaded_files:# 如果st.session_state中存在'uploaded_files'键，则删除该键值对if 'uploaded_files' in st.session_state:del st.session_state.uploaded_files# 如果 st.session_state 中存在'langchain_based_on_rag'键，则删除该键值对if 'langchain_based_on_rag' in st.session_state:del st.session_state.langchain_based_on_rag# 在页面上显示提示信息，提醒用户先上传TXT文档st.info("请先上传TXT文档。")# 停止应用后续代码的执行st.stop()else:# 将上传的文件信息保存到st.session_state中st.session_state.uploaded_files = self.uploaded_files# 如果 st.session_state 中不存在'langchain_based_on_rag'键if 'langchain_based_on_rag' not in st.session_state:# 创建一个LangChainBasedOnRAG类的实例并保存到st.session_state中st.session_state.langchain_based_on_rag = LangChainBasedOnRAG()# 调用LangChainBasedOnRAG实例的configure_retriever方法，配置检索器st.session_state.langchain_based_on_rag.configure_retriever(self.uploaded_files)# 向st.session_state的消息列表中添加一条助手的欢迎消息st.session_state.messages.append({"role": "assistant", "content": "您好，我是您的智能助手"})# 遍历st.session_state中的消息列表for message in st.session_state.messages:# 创建对应角色(用户或助手)的聊天消息显示区域with st.chat_message(message["role"]):# 在聊天消息区域显示消息内容st.write(message["content"])# 创建一个聊天输入框，提示用户开始提问self.user_query = st.chat_input(placeholder="请开始提问吧!")else:# 创建一个聊天输入框，提示用户开始提问self.user_query = st.chat_input(placeholder="请开始提问吧!")# 调用 start_chat 方法开始聊天交互self.start_chat()if __name__ == "__main__":# 检查st.session_state中是否存在'streamlit_app'键，如果不存在则创建StreamlitApp类的实例并保存if 'streamlit_app' not in st.session_state:st.session_state.streamlit_app = StreamlitApp()# 此处代码有误，应该是调用run_app方法而不是run_capp方法，下面为修正注释# 调用StreamlitApp实例的run_app方法来运行应用st.session_state.streamlit_app.run_app()

langchain_based_on_rag.py

# 导入os模块，用于处理文件和目录路径等操作系统相关操作
import os
# 导入tempfile模块，用于创建临时文件和目录
import tempfile
# 从langchain.agents模块导入create_react_agent函数和AgentExecutor类，用于创建和执行代理
from langchain.agents import create_react_agent, AgentExecutor
# 从langchain_community.document_loaders模块导入TextLoader类，用于加载文本文件
from langchain_community.document_loaders import TextLoader
# 从langchain_community.embeddings模块导入OpenAIEmbeddings类，用于生成文本的嵌入向量
from langchain_community.embeddings import OpenAIEmbeddings
# 从langchain_community.vectorstores模块导入FAISS类，用于存储和检索向量数据
from langchain_community.vectorstores import FAISS
# 从langchain_core.prompts模块导入PromptTemplate类，用于创建提示模板
from langchain_core.prompts import PromptTemplate
# 从langchain_openai模块导入ChatOpenAI类，用于调用OpenAI的聊天模型
from langchain_openai import ChatOpenAI
# 从langchain_text_splitters模块导入RecursiveCharacterTextSplitter类，用于分割文本
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 从langchain.tools.retriever模块导入create_retriever_tool函数，用于创建检索工具
from langchain.tools.retriever import create_retriever_toolclass LangChainBasedOnRAG:def __init__(self):# 初始化检索器，初始值为Noneself.retriever = None# 初始化工具列表，初始值为Noneself.tools = None# 初始化代理执行器，初始值为Noneself.agent_executor = Nonedef configure_retriever(self, uploaded_files):# 读取上传的文档，并写入一个临时目录docs = []  # 用于存储加载的文档# 创建一个临时目录，指定目录位置为 D 盘temp_dir = tempfile.TemporaryDirectory(dir=r"D:\\")# 遍历上传的文件列表for file in uploaded_files:# 拼接临时文件的完整路径temp_filepath = os.path.join(temp_dir.name, file.name)# 以二进制写入模式打开临时文件with open(temp_filepath, "wb") as f:# 将文件内容写入临时文件f.write(file.getvalue())# 创建一个TextLoader实例，用于加载临时文件loader = TextLoader(temp_filepath, encoding="utf-8")# 加载文件内容并添加到docs列表中docs.extend(loader.load())# 创建一个RecursiveCharacterTextSplitter实例，设置块大小和重叠部分text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=30)# 对文档进行分割splits = text_splitter.split_documents(docs)# 创建一个OpenAIEmbeddings实例，用于生成文本的嵌入向量embeddings = OpenAIEmbeddings()# 使用FAISS存储文档的嵌入向量vector = FAISS.from_documents(docs, embeddings)# 将向量存储转换为检索器self.retriever = vector.as_retriever()# 调用setup_tools方法设置工具self.setup_tools()# 调用setup_agent方法设置代理self.setup_agent()def setup_tools(self):# 创建一个检索工具tool = create_retriever_tool(self.retriever,  # 指定检索器"文档检索",  # 工具名称"用于检索用户提出的问题，并基于检索到的文档内容进行回复.",  # 工具描述)# 将工具添加到工具列表中self.tools = [tool]def setup_agent(self):# 定义代理的指令instructions = """您是一个设计用于查询文档来回答问题的代理。您可以使用文档检索工具，并基于检索内容来回答问题您可能不查询文档就知道答案，但是您仍然应该查询文档来获得答案。如果您从文档中找不到任何信息用于回答问题，则只需返回“抱歉，这个问题我还不知道。”作为答案。"""# 定义基础提示模板base_prompt_template = """{instructions}TOOLS:------You have access to the following tools:{tools}To use a tool, please use the following format:•```Thought: Do I need to use a tool? YesAction: the action to take, should be one of [{tool_names}]Action Input: the input to the actionObservation: the result of the action•```When you have a response to say to the Human, or if you do not need to use a tool, you MUST use the format:•```Thought: Do I need to use a tool? NoFinal Answer: [your response here]•```Begin!New input: {input}{agent_scratchpad}"""# 根据基础提示模板创建PromptTemplate实例base_prompt = PromptTemplate.from_template(base_prompt_template)# 对基础提示进行部分填充，填入指令信息prompt = base_prompt.partial(instructions=instructions)# 创建llm，使用gpt-3.5-turbo模型chat_model = ChatOpenAI(model="gpt-3.5-turbo")# 创建react Agentagent = create_react_agent(chat_model, self.tools, prompt)# 创建代理执行器self.agent_executor = AgentExecutor(agent=agent, tools=self.tools, verbose=False)def agent_executor_invoke(self, user_input, config):# 调用代理执行器处理用户输入response = self.agent_executor.invoke({"input": user_input, "chat_history": []}, config=config)return responseif __name__ == "__main__":# 创建LangChainBasedOnRAG类的实例langchain_based_on_rag = LangChainBasedOnRAG()

公司规章制度.txt

公司上班时间是早上9点到下午6点。

运行结果

运行说明

这是博主的运行命令，请大家参考。如果大家有对该命令有疑问可以先翻到实现文件上传/如何运行和实现文件上传/运行说明查看内容。

(venv) PS D:\Software\PythonProjects\LangChain> streamlit run .\streamlit_app.py

博主在写这篇博文的时候streamlit使用的版本是1.36.0。

(venv) PS D:\Software\PythonProjects\LangChain> pip show streamlit
Name: streamlit
Version: 1.36.0
Summary: A faster way to build and share data apps
Home-page: https://streamlit.io
Author: Snowflake Inc
Author-email: hello@streamlit.io
License: Apache License 2.0
Location: d:\software\pythonprojects\langchain\venv\lib\site-packages
Requires: altair, blinker, cachetools, click, gitpython, numpy, packaging, pandas, pillow, protobuf, pyarrow, pydeck, requests, rich, tenacity, toml, tornado, typing-extensions, watchdog
Required-by:

结束

好了，以上就是本次分享的全部内容。不知道大家是否掌握了基于RAG实现文档问答的功能。

在LangChain中基于检索增强生成(Retrieval Augmented Generation, RAG)实现文档问答功能，其核心思路是将文档数据进行处理和存储，构建一个可检索的数据库，当用户提出问题时，先从数据库中检索相关信息，再结合语言模型生成回答。

传统的语言模型(如GPT等)在回答问题时，主要依赖于预训练时学到的知识，对于一些特定领域或最新的信息可能无法准确回答。而基于RAG的文档问答系统，在用户提问时，会先从已有的文档库中检索相关信息，然后结合这些具体的文档内容来生成回答，使得回答更加准确、具体且可追溯，能够更好地处理特定领域的专业知识和具体事实性问题。

在LangChain中基于RAG实现文档问答功能，能够充分发挥语言模型和文档检索的优势，提高问答系统的性能、可靠性、可解释性和灵活性，同时降低成本，适用于各种不同的应用场景。这也是博主本次要分享的目的。

那么本次分享就到这里了。最后，博主还是那句话：请大家多去大胆的尝试和使用，成功总是在不断的失败中试验出来的，敢于尝试就已经成功了一半。如果大家对博主分享的内容感兴趣或有帮助，请点赞和关注。大家的点赞和关注是博主持续分享的动力🤭，博主也希望让更多的人学习到新的知识。