【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.301

import 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, RunnablePassthrough

import 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)

【动手学 RAG】系列文章：

• 【RAG 项目实战 01】在 LangChain 中集成 Chainlit
• 【RAG 项目实战 02】Chainlit 持久化对话历史
• 【RAG 项目实战 03】优雅的管理环境变量
• 【RAG 项目实战 04】添加多轮对话能力
• 【RAG 项目实战 05】重构：封装代码
• 【RAG 项目实战 06】使用 LangChain 结合 Chainlit 实现文档问答
• 【RAG 项目实战 07】替换 ConversationalRetrievalChain（单轮问答）
• 【RAG 项目实战 08】为 RAG 添加历史对话能力
• More...

【动手部署大模型】系列文章：

• 【模型部署】vLLM 部署 Qwen2-VL 踩坑记 01 - 环境安装
• 【模型部署】vLLM 部署 Qwen2-VL 踩坑记 02 - 推理加速
• 【模型部署】vLLM 部署 Qwen2-VL 踩坑记 03 - 多图支持和输入格式问题
• More...

本文由mdnice多平台发布