当前位置：首页 > news >正文

【RAG 项目实战 07】替换 ConversationalRetrievalChain（单轮问答）

news 2025/6/30 4:31:05

【RAG 项目实战 07】替换 ConversationalRetrievalChain（单轮问答）

NLP Github 项目：

NLP 项目实践：fasterai/nlp-project-practice

介绍：该仓库围绕着 NLP 任务模型的设计、训练、优化、部署和应用，分享大模型算法工程师的日常工作和实战经验
AI 藏经阁：https://gitee.com/fasterai/ai-e-book

介绍：该仓库主要分享了数百本 AI 领域电子书
AI 算法面经：fasterai/nlp-interview-handbook#面经

介绍：该仓库一网打尽互联网大厂NLP算法面经，算法求职必备神器
NLP 剑指Offer：https://gitee.com/fasterai/nlp-interview-handbook

介绍：该仓库汇总了 NLP 算法工程师高频面题

一、RAG 整体流程

检索式问答的系统流程图：

二、RAG 核心模块

2.1 环境配置

# @Author：青松  
# 公众号：FasterAI  
# Python, version 3.10.14  
# Pytorch, version 2.3.0  
# Chainlit, version 1.1.301

2.2 分割文本块并对每一个块建索引

# 配置文件分割器，每个块1000个token，重复100个
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)# 将文件分割成文本块
texts = text_splitter.split_text(text)# 为每个文本块添加元数据
metadatas = [{"source": f"{i}-pl"} for i in range(len(texts))]# 使用异步方式创建 Chroma 向量数据库
vectorstore = await cl.make_async(Chroma.from_texts)(texts, embeddings_model, metadatas=metadatas
)# 将 Chroma 向量数据库转化为检索器
retriever = vectorstore.as_retriever()

2.3 构建 RAG 链

# RAG_Prompt：根据参考内容回答用户问题
rag_template = "你是一个专门处理问答任务的智能助理。请使用给定的参考内容来回答用户的问题，如果你不知道答案，就说你不知道，不要试图编造答案。" \"\n\n用户问题: {question} \n参考内容: {context} \n答案:"rag_prompt = PromptTemplate.from_template(rag_template)# RAG链：根据问题和参考内容生成答案
rag_chain = ({"context": retriever | format_docs, "question": RunnablePassthrough()}| rag_prompt| llm| StrOutputParser()
)

2.4 检索生成

rag_chain = cl.user_session.get("rag_chain")# 使用RAG链处理用户问题
response = rag_chain.invoke(message.content,config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
)# 大模型的回答
await cl.Message(content=response).send()

三、RAG 效果展示

启动程序：

chainlit run rag_app.py -w

系统截图：

在这里插入图片描述

问题一：对第一个问题大模型可以根据文档内容生成答案。
问题二：对第一个问题大模型未能结合聊天历史进行RAG。

四、RAG 完整代码

# @Author：青松
# 公众号：FasterAI
# Python, version 3.10.14
# Pytorch, version 2.3.0
# Chainlit, version 1.1.301import chainlit as cl
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceBgeEmbeddings
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableConfig, RunnablePassthroughimport llm_util
from common import Constants# 获取大模型实例
llm = llm_util.get_llm(Constants.MODEL_NAME['QianFan'])# 获取文本嵌入模型
model_name = "BAAI/bge-small-zh"
# 对模型生成的嵌入进行归一化处理，将它们缩放到具有单位范数（长度为1）的尺度
encode_kwargs = {"normalize_embeddings": True}
embeddings_model = HuggingFaceBgeEmbeddings(model_name=model_name, encode_kwargs=encode_kwargs
)@cl.on_chat_start
async def on_chat_start():""" 监听会话开始事件 """await send_welcome_msg()files = None# 等待用户上传文件while files is None:files = await cl.AskFileMessage(content="Please upload a text file to begin!",accept=["text/plain"],max_size_mb=20,timeout=180,).send()file = files[0]# 发送处理文件的消息msg = cl.Message(content=f"Processing `{file.name}`...", disable_feedback=True)await msg.send()with open(file.path, "r", encoding="utf-8") as f:text = f.read()# 配置文件分割器，每个块1000个token，重复100个text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)# 将文件分割成文本块texts = text_splitter.split_text(text)# 为每个文本块添加元数据metadatas = [{"source": f"{i}-pl"} for i in range(len(texts))]# 使用异步方式创建 Chroma 向量数据库vectorstore = await cl.make_async(Chroma.from_texts)(texts, embeddings_model, metadatas=metadatas)# 将 Chroma 向量数据库转化为检索器retriever = vectorstore.as_retriever()cl.user_session.set("retriever", retriever)# RAG_Prompt：根据参考内容回答用户问题rag_template = "你是一个专门处理问答任务的智能助理。请使用给定的参考内容来回答用户的问题，如果你不知道答案，就说你不知道，不要试图编造答案。" \"\n\n用户问题: {question} \n参考内容: {context} \n答案:"rag_prompt = PromptTemplate.from_template(rag_template)# RAG链：根据问题和参考内容生成答案rag_chain = ({"context": retriever | format_docs, "question": RunnablePassthrough()}| rag_prompt| llm| StrOutputParser())cl.user_session.set("rag_chain", rag_chain)# 通知用户文件已处理完成，更新当前窗口的内容msg.content = f"Processing `{file.name}` done. You can now ask questions!"await msg.update()@cl.on_message
async def on_message(message: cl.Message):""" 监听用户消息事件 """rag_chain = cl.user_session.get("rag_chain")# 使用RAG链处理用户问题response = rag_chain.invoke(message.content,config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),)# 大模型的回答await cl.Message(content=response).send()async def send_welcome_msg():image = cl.Image(url="https://qingsong-1257401904.cos.ap-nanjing.myqcloud.com/wecaht.png")# 发送一个图片await cl.Message(content="**青松** 邀你关注 **FasterAI**， 让每个人的 AI 学习之路走的更容易些！立刻扫码开启 AI 学习、面试快车道 **(^_^)** ",elements=[image],).send()def format_docs(docs):""" 拼接检索到的文本块 """return "\n\n".join(doc.page_content for doc in docs)