实验 3实现 RAG 构建与多个 PDF 聊天应用

使用 Streamlit、Langchain 和 OpenAI

这是文章 “与 PDF 聊天” 的第二部分，发布在“与一切聊天”系列中。这部分将重点介绍如何使用 RAG 构建一个支持多个 PDF 的聊天应用。

对于那些不太了解的人，“与一切聊天”系列专注于为您提供构建 LLM 应用程序的技术知识和技巧。我创建的所有应用程序都使用流行的框架：Streamlit、Langchain 和 OpenAI（LLM 模型）。

您可以在这里找到“与一切聊天”系列：我的 GitHub

难度等级：中级 🎖️🎖️

本文将要涵盖的主要内容：

1. 为什么 RAG 重要？
2. RAG 实际是如何工作的
3. 应用 RAG 构建一个“与多个 PDF 聊天”的应用
4. 演示（对于那些想先看到结果以获得动力的人。）

1. 为什么 RAG 重要？

在使用像 GPT、Claude、BERT 等 LLM 模型时，您可能遇到以下问题：

• 模型虚构信息并生成不真实的答案 (幻觉效应)
• 您希望模型 在您自己的数据源上运行，从文本数据（pdf、doc 等）到各种类型的媒体数据（视频、音频、照片等）
• 有时，您处理非常长的内容，导致 昂贵的处理成本 或您的账户简单地被封锁 1-2 小时（影响您的生产力）

如果您对上述问题感同身受，是时候了解 RAG 了。

什么是 RAG？

RAG (检索增强生成) 是一种通过从外部资源获取事实来提高生成 AI 模型准确性和可靠性的技术。RAG 是一种重要的技术，因为：

• 几乎每个 LLM 都可以使用 RAG 连接几乎任何外部资源。
• RAG 使您的应用程序对我们的用户更可靠（“信任”）。为什么？因为 RAG 为模型提供了可以引用的来源，就像研究论文中的脚注一样，因此用户可以检查任何声明。这建立了信任。
• 减少模型错误猜测的可能性，这种现象有时称为幻觉。
• 最后，RAG 使得这种方法 比使用额外数据集重新训练模型更快且成本更低。并且它允许用户动态更换新的来源。

2. RAG 实际工作原理

从高层次来看，整个 RAG 过程分为两个阶段：

• 阶段 1（预处理）： 将外部源加载到我们的系统中
• 阶段 2（推理）： 在 LLM 的支持下为用户的查询生成答案

让我们深入了解这两个阶段

阶段 1：预处理阶段

1. 从外部源加载数据到文本： 这个过程在 langchain 框架的支持下很容易完成。

• 在上面的例子中，我使用 PyPDFLoader 从 PDF 文件加载数据
• 在实践中，我还使用其他文档加载器从 YouTube、CSV、Confluence 等加载数据
• 你可以访问 langchain 官方网站 获取所有文档加载器的列表

2. 将文本转换为块： 将文本拆分为小块，以优化信息的处理、存储和检索，同时提高 AI 系统的质量和准确性。

3. 将块嵌入向量数据库： 向量数据库被认为是 RAG 实现堆栈中最常见的方式（在本文中，我也选择了一个可以轻松在本地磁盘上运行的向量数据库，即 Chroma）。选择最佳向量数据库的内容在本文中未涉及。如果你想深入了解这个主题，请查看 SuperLinked 的列表。

阶段 2：推理运行时阶段

1. 输入查询（问题）： 输入查询的用户需要得到回答或处理。

2. 嵌入查询： 使用嵌入模型将查询转换为嵌入。

3. 信息检索： 使用查询的向量表示从包含嵌入文本的大型数据库中查找相关文本。

4. 文本生成器： 提取的文本将与原始查询一起用于生成新的答案或文本。生成器（通常是大型语言模型）将结合这些文本段落中的信息生成反馈。

5. 输出（答案）： 创建的文本将作为完整答案返回给用户。

3. 应用 RAG 构建“与多个 PDF 聊天”应用

一旦你了解了 RAG 的工作原理，就该和我一起构建一个“与多个 PDF 聊天”的应用了。并看看我如何将 RAG 应用到实际应用中。

设置

首先，我们需要安装所有包

import streamlit as st
from langchain_community.document_loaders import PyPDFLoader
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain_community.vectorstores import Chroma
import chromadb
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.retrieval import create_retrieval_chain

将 OPENAI_API_KEY 替换为您自己的密钥，在示例中，我使用了 streamlit secret：

# Replace it with your OPENAI API KEY if you use streamlit secrets
OPENAI_API_KEY = st.secrets["OPENAI_API_KEY"]

在 ChromaDB 中创建名为 “chat-with-pdf” 的集合

# Load Vector datasse
native_db = chromadb.PersistentClient("./chroma_db")
db = Chroma(client=native_db, collection_name="chat-with-pdf", embedding_function=OpenAIEmbeddings())

为了与集合一起工作，我们将使用函数 get_collection

@st.cache_resource
def get_collection():
    print("DEBUG: call get_collection()")
    collection = None
    try:
        # Delete all documents
        native_db.delete_collection("chat-with-pdf")
    except:
        pass
    finally:
        collection = native_db.get_or_create_collection("chat-with-pdf", embedding_function=OpenAIEmbeddingFunction(api_key=OPENAI_API_KEY))
    return collection

阶段 1：预处理阶段

当用户上传一些新文件时：

• 加载所有新上传的文件
• 将其内容拆分为块
• 将块嵌入到 ChromaDB 中，并保存到 ChromaDB

# Load, transform and embed new files into Vector Database
def add_files(uploaded_files):
    collection = get_collection()

    # old_filenames: contains a list of names of files being used
    # uploaded_filenames: contains a list of names of uploaded files
    old_filenames = st.session_state.old_filenames
    uploaded_filename = [file.name for file in uploaded_files]
    new_files = [file for file in uploaded_files if file.name not in old_filenames]

    for file in new_files:
        # Step 1: load uploaded file
        temp_file = f"./temp/{file.name}.pdf"
        with open(temp_file, "wb") as f:
            f.write(file.getvalue())
        loader = PyPDFLoader(temp_file)
        pages = loader.load()

        # Step 2: split content in to chunks
        text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
        chunks = text_splitter.split_documents(pages)

        # Step 3: embed chunks into Vector Store
        # collection.add(ids=file.name,documents=chunks)
        for index, chunk in enumerate(chunks):
            collection.upsert(
                ids=[chunk.metadata.get("source") + str(index)], metadatas=chunk.metadata,
                documents=chunk.page_content
            )

否则，当用户从列表中删除文件时，我们需要删除所有相关的块

# Remove all relevant chunks of the removed files
def remove_files(uploaded_files):
    collection = get_collection()

    # old_filenames: contains a list of names of files being used
    # uploaded_filenames: contains a list of names of uploaded files
    old_filenames = st.session_state.old_filenames
    uploaded_filename = [file.name for file in uploaded_files]

    # Step 1: Get the list of file that was removed from upload files
    deleted_filenames = [name for name in old_filenames if name not in uploaded_filename]

    # Step 2: Remove all relevant chunks of the removed files
    if len(deleted_filenames) > 0:
        all_chunks = collection.get()

        ids = all_chunks["ids"]
        metadatas = all_chunks["metadatas"]

        if len(metadatas) > 0:
            deleted_ids = []
            for name in deleted_filenames:
                for index, metadata in enumerate(metadatas):
                    if metadata['source'] == f"./temp/{name}.pdf":
                        deleted_ids.append(ids[index])
            collection.delete(ids=deleted_ids)

将两个方法合并为一个函数

# Return chunks after having any change in the file list
def refresh_chunks(uploaded_files):
    # old_filenames: contains a list of names of files being used
    # uploaded_filenames: contains a list of names of uploaded files
    old_filenames = st.session_state.old_filenames
    uploaded_filename = [file.name for file in uploaded_files]

    if len(old_filenames) < len(uploaded_filename):
        add_files(uploaded_files)
    elif len(old_filenames) > len(uploaded_filename):
        remove_files(uploaded_files)

    # Step 3: Save the state
    st.session_state.old_filenames = uploaded_filename

第2阶段：推理运行时阶段

• 创建一个与ChromaDB协作的检索器。检索器是一个接口，能够根据非结构化查询返回文档。
• 创建一个链条将所有内容连接在一起

# Init langchain
llm = ChatOpenAI(api_key=OPENAI_API_KEY)
prompt = ChatPromptTemplate.from_template("""
Based on the provided context only, find the best answer for my question. Format the answer in markdown format
<context>
{context}
</context>
Question:{input}
""")
document_chain = create_stuff_documents_chain(llm, prompt)
retriever = db.as_retriever()
retriever_chain = create_retrieval_chain(retriever, document_chain)

每当你想使用 retriever_chain 时，可以调用 invoke. 例如：

response = retriever_chain.invoke({"input": "How to create a prompt for a new digital product"})

最终结果：

4. 展示案例

视频演示：

奖励 🎁

您可以轻松地使用 RAG 搜索实用示例。这里有一个有趣的例子，展示如何在文本到 SQL 中使用 RAG：VannaAI

VannaAI 在我撰写本文时拥有超过 9k 的星标和 658 个分支。详细信息可以在 这里 找到 . VannaAI 正在将 RAG 应用于两件事情：

• 使用检索增强来帮助您使用 LLM 生成准确的 SQL 查询，以便查询您的数据库。
• 您可以在自己的数据上训练 RAG “模型”，然后提出问题，这将返回可以设置为自动在您的数据库上运行的 SQL 查询。”

在你离开之前！🤟

如果你觉得这篇文章对你有帮助，并希望表示支持，请为我的文章鼓掌 10 次。 👏 这真的会激励我，并将这篇文章推荐给更多人。