构建一个多模态RAG系统用于视觉问答

使用LangChain和GPT-4o构建具有视觉问答能力的多模态RAG聊天机器人。

概述

在本文中，我将指导您构建一个使用 OpenAI 的 GPT-4o 模型的多模态 RAG 聊天应用程序。您将学习以下内容：

• 多模态 RAG 聊天应用程序：创建一个应用程序，通过从 PDF 文档中检索信息来实现视觉问答。
• 无缝解析：使用 Unstructured 库无缝解析文本、表格和图像。
• 性能评估：使用 DeepEval 库提供的各种指标评估聊天机器人的性能。
• Streamlit UI：通过 Streamlit 应用程序演示该应用程序。

为什么要阅读这个？

你是否有兴趣利用像 GPT-4o 这样的先进基础模型的多模态能力来构建自己的 AI 应用程序？那么你来对地方了！

无论你是寻求市场研究报告见解的市场营销专业人士，分析多模态医疗文件的医疗从业者，还是处理复杂法律文件的法律专业人士，这篇文章都为你提供了宝贵的见解。

我将详细解释每个概念，并提供所有代码的详细说明。话虽如此，让我们开始吧！ 🎬

多模态 RAG 的崛起

从基于文本的 RAG 模型过渡到多模态 RAG 系统标志着 AI 能力的重大飞跃。以下是快速概述：

• 起源：RAG 这个术语是在 2021 年 4 月提出的，通过基于文本的知识增强语言输出。
• 进展：随着 2024 年 5 月发布的 GPT-4o 等模型，我们现在可以整合视觉信息，允许同时处理图像、表格和文本。
• 新可能性：这一演变使得更全面和上下文丰富的 AI 应用成为可能。

在本文中，我将展示一个案例研究，使用多模态 RAG 框架对我在 Neurocomputing 上发表的研究文章进行问答。该文章包含文本、表格和图形，我们将探索 GPT-4o 的视觉能力如何回答复杂问题。

让我们开始编码吧！🎬

目录

1. 设置虚拟环境和安装Python库
2. 预处理非结构化数据
3. 文本、表格和图像摘要
4. 多模态检索器
5. 多模态RAG链
6. LLM评估
7. 使用Streamlit的用户界面

设置虚拟环境并安装 Python 库

首先，让我们使用以下命令设置虚拟环境：

python3.10 -m venv venv

现在，让我们安装必要的软件包。您可以在 GitHub 仓库的主目录中的“requirements.txt”中找到它们。

pip install -r requirements.txt

现在，打开一个 Jupyter Notebook，比如“your-project.ipynb”，开始编写您的代码。就这样！我们现在准备进入主要细节。

这是多模态 RAG 项目的工作 GitHub 仓库：

预处理非结构化数据

要构建一个 RAG 应用程序，我们的第一步是将上下文加载到数据库中，这里使用的是 PDF 文档。由于大型语言模型（LLM）的上下文窗口限制，我们无法将整个文档直接存储并传递到提示中。这样做很可能会导致错误，因为它超过了最大标记数。

为了解决这个问题，我们将首先从文档中提取不同的元素——图像、文本和表格。为此任务，我们将使用 Unstructured 库。

安装

首先让我们安装这个包（如果尚未通过 pip 安装的话）

#%brew install tesseract poppler
%pip install -q "unstructured[all-docs]"

请注意，我们还需要系统中的“tesseract”和“poppler”库，以便 unstructured 库能够处理文本提取以及从图像中提取文本。您可以使用 homebrew 安装这两个包（请参见注释行）。

分区和块划分

from unstructured.partition.pdf import partition_pdf

elements = partition_pdf(
    filename="TAGIV.pdf", # mandatory
    strategy="hi_res",                                     # mandatory to use ``hi_res`` strategy
    extract_images_in_pdf=True,                            # mandatory to set as ``True``
    extract_image_block_types=["Image", "Table"],          # optional
    extract_image_block_to_payload=False,                  # optional
    extract_image_block_output_dir="saved_images",  # optional - only works when ``extract_image_block_to_payload=False``
    )

我们将使用 partition_pdf 模块对文档进行分区，并从我们的文件‘TAGIV.pdf’中提取不同的元素。我们将设置 hi_res 策略以提取高质量的图像和表格。可选参数 extract_image_block_types 和 extract_image_block_output_dir 指定仅提取图像和表格，并将其保存到名为 “saved_images” 的目录中。

我们将使用 chunk_by_title 方法对元素进行块划分，该方法用于根据“标题或标题”将提取的元素划分为块。这适用于通常由不同部分和子部分组成的研究文章，例如引言、方法、结果等。

from unstructured.chunking.title import chunk_by_title # might be better for an article 
from typing import Any

chunks = chunk_by_title(elements)

# different category in the document
category_counts = {}

for element in chunks:
    category = str(type(element))
    if category in category_counts:
        category_counts[category] += 1
    else:
        category_counts[category] = 1
    
# Unique_categories will have unique elements
unique_categories = set(category_counts.keys())
category_counts   

{"<class 'unstructured.documents.elements.CompositeElement'>": 200,
 "<class 'unstructured.documents.elements.Table'>": 3,
 "<class 'unstructured.documents.elements.TableChunk'>": 2}

块划分显示有三个独特的类别：

• CompositeElements
• Table
• TableChunk

‘CompositeElements’ 是不同文本的集合，可能是段落、部分、页脚、公式等。还有三个 ‘Table’ 结构，以及两个 ‘TableChunk’，通常表示表格的一部分或片段。因此，可能一个表格跨页分割，只有一部分被划分。

我在文档中确实有四个表格，但只有三个被完全解析。🤔

过滤

接下来，我们将简化文档元素，以便分别处理文本和表格数据以进行进一步处理。为此，我们将定义一个 Pydantic 模型，以标准化文档元素，并根据其类型将其分类为“text”或“table”。

from pydantic import BaseModel

class Element(BaseModel):
    type: str
    text: Any


# 按类型分类
categorized_elements = []
for element in chunks:
    if "unstructured.documents.elements.CompositeElement" in str(type(element)):
        categorized_elements.append(Element(type="text", text=str(element)))
    elif "unstructured.documents.elements.Table" in str(type(element)):
        categorized_elements.append(Element(type="table", text=str(element)))

# 文本
text_elements = [e for e in categorized_elements if e.type == "text"]

# 表格
table_elements = [e for e in categorized_elements if e.type == "table"]

我们将遍历文档元素的块，识别每个元素的类型，并将其附加到分类列表中。最后，我们将此列表过滤为文本和表格元素的单独列表。至此，预处理步骤已完成。

文本、表格和图像摘要

为了为后面使用多向量检索器做准备，我们需要为文本、表格和图像元素创建摘要。这些摘要将存储在向量存储中，以便在我们将输入查询传递到提示中时实现语义搜索。

文本和表格摘要

让我们开始文本和表格摘要。首先，我们将设置一个提示模板，指示AI充当专家研究助理，负责总结表格和文本。接下来，我们将创建一个链，处理每个文本和表格元素，通过这个提示和GPT-4o模型，生成简洁的摘要。

为了提高效率，我们将同时批量处理五个文本或表格元素，使用max_concurrency参数。

%pip install -q langchain langchain-chroma unstructured[all-docs] pydantic lxml langchainhub langchain-openai

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

## 检索器

# 提示
prompt_text = """You are an expert Research Assistant tasked with summarizing tables and texts from research articles. \ 
Give a concise summary of the text. text chunk: {element} """

prompt = ChatPromptTemplate.from_template(prompt_text)

# 摘要链
model = ChatOpenAI(temperature=0, model="gpt-4o")
summarize_chain = {"element": lambda x: x} | prompt | model | StrOutputParser()

# 应用于文本
texts = [i.text for i in text_elements]
text_summaries = summarize_chain.batch(texts, {"max_concurrency": 5})

# 应用于表格
tables = [i.text for i in table_elements]
table_summaries = summarize_chain.batch(tables, {"max_concurrency": 5})

图像摘要

接下来，我们将设置一些函数来帮助我们总结图像。我们将定义三个关键函数：encode_image、image_summarize 和 generate_img_summaries。

1. encode_image：此函数以二进制读取模式（‘rb’）打开图像文件，并返回其 base64 编码的字符串表示。
2. image_summarize：此函数使用一个包含提示的 HumanMessage 对象，指示模型如何总结图像。它还包括 base64 编码的图像数据，格式为数据 URL，以便直接在内容中嵌入图像。
3. generate_img_summaries：此函数处理给定目录中的所有 JPG 图像，为每个图像生成摘要，并返回 base64 编码的图像。

这些函数将使我们能够高效地总结和处理图像，将其无缝集成到我们的多模态 RAG 应用中。

以下是完整代码：

## getting image summaries
import base64
import os

from langchain_core.messages import HumanMessage


def encode_image(image_path):
    """Getting the base64 string"""
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode("utf-8")

def image_summarize(img_base64, prompt):
    """Make image summary"""
    chat = ChatOpenAI(model="gpt-4o", max_tokens=1024)

    msg = chat.invoke(
        [
            HumanMessage(
                content=[
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:image/jpg;base64,{img_base64}"},
                    },
                ]
            )
        ]
    )
    return msg.content


def generate_img_summaries(path):
    """
    Generate summaries and base64 encoded strings for images
    path: Path to list of .jpg files extracted by Unstructured
    """

    # Store base64 encoded images
    img_base64_list = []

    # Store image summaries
    image_summaries = []

    # Prompt
    prompt = """You are an assistant tasked with summarizing images for retrieval. \
    These summaries will be embedded and used to retrieve the raw image. \
    Give a concise summary of the image that is well optimized for retrieval."""

    # Apply to images
    for img_file in sorted(os.listdir(path)):
        if img_file.endswith(".jpg"):
            img_path = os.path.join(path, img_file)
            base64_image = encode_image(img_path)
            img_base64_list.append(base64_image)
            image_summaries.append(image_summarize(base64_image, prompt))

    return img_base64_list, image_summaries


fpath = "saved_images"

# Image summaries
img_base64_list, image_summaries = generate_img_summaries(fpath)

多模态检索器

在我们的摘要准备好后，我们可以创建我们的多模态检索器。

多向量检索器

我们将设置一个多向量检索器，它以vectorstore、docstore、id_key和search_kwargs作为输入。这种方法允许我们单独索引内容摘要，同时存储原始内容，从而促进高效检索。请注意，这只是执行多模态RAG的一种方式；另一种方法可能涉及使用多模态嵌入来嵌入文本和图像，使用CLIP，然后将原始图像和文本块传递给多模态LLM。我可能会在未来的博客文章中探讨这个主题。🙂

我们的检索器利用Chroma vectorstore存储内容摘要的嵌入，使用InMemoryStore存储完整内容。这种设置使得通过摘要进行语义搜索，同时在需要时检索相应的完整内容。每个文档都使用UUID分配一个唯一标识符，这是检索器所需的。

为了简化将摘要添加到vectorstore和将原始内容添加到docstore的过程，我们将添加一个名为add_documents的辅助函数。该函数确保仅添加可用摘要。

import uuid

from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings

def create_multi_vector_retriever(
    vectorstore, text_summaries, texts, table_summaries, tables, image_summaries, images):
    """
    Create retriever that indexes summaries, but returns raw images, table, or texts
    """

    # Initialize the storage layer
    store = InMemoryStore()
    id_key = "doc_id"

    # Create the multi-vector retriever
    retriever = MultiVectorRetriever(
        vectorstore=vectorstore,
        docstore=store,
        id_key=id_key,
        search_kwargs={"k": 2}  # Limit to top 2 results
    )

    # Helper function to add documents to the vectorstore and docstore
    def add_documents(retriever, doc_summaries, doc_contents):
        doc_ids = [str(uuid.uuid4()) for _ in doc_contents]
        summary_docs = [
            Document(page_content=s, metadata={id_key: doc_ids[i]})
            for i, s in enumerate(doc_summaries)
        ]
        retriever.vectorstore.add_documents(summary_docs)
        retriever.docstore.mset(list(zip(doc_ids, doc_contents)))

    # Add texts, tables, and images
    # Check that text_summaries is not empty before adding
    if text_summaries:
        add_documents(retriever, text_summaries, texts)
    # # Check that table_summaries is not empty before adding
    if table_summaries:
        add_documents(retriever, table_summaries, tables)
    # Check that image_summaries is not empty before adding
    if image_summaries:
        add_documents(retriever, image_summaries, images)

    return retriever

创建检索器

现在，让我们使用 OpenAI 嵌入模型分配一个 Chroma 向量存储并创建我们的检索器。

# The vectorstore to use to index the summaries
vectorstore = Chroma(
    collection_name="mm_tagiv_paper", embedding_function=OpenAIEmbeddings()
)

# Create retriever
retriever_multi_vector_img = create_multi_vector_retriever(
    vectorstore,
    text_summaries,
    texts,
    table_summaries,
    tables,
    image_summaries,
    img_base64_list,
)

测试

让我们用这个查询来测试我们的检索器，看看哪些文档被检索到了。

retriever_multi_vector_img.invoke("How is the performance of TAGI-V for the Boston dataset compared to the other methods?")

['TAGI-V are averaged over 3 random seeds. The test log-likelihood values show that TAGI-V performs better than all other methods in 4 out of the 5 datasets. The TAGI-V method is also competitive for RMSE values where it provides the best results in 2 out of the 5 datasets, i.e., Elevators and KeggD, while it is second best for KeggU and Pol. Both PCA+ VI and NL outperform the others in two datasets.']

响应显示它能够从文档中找到具体信息。完美！🚀

多模态 RAG 链

现在我们有了检索器，我们将创建我们的多模态链。我们需要几个辅助函数来管理 base64 编码的图像和文本数据。

辅助函数

• plt_image_base64(img_base64) : 使用 HTML 显示 base64 编码的图像。

def plt_img_base64(img_base64):
    image_html = f'<img src="data:image/jpg;base64,{img_base64}" />'
    display(HTML(image_html))

• looks_like_base64(sb): 检查一个字符串是否看起来是 base64 编码的。

def looks_like_base64(sb):
    return re.match("^[A-Za-z0-9+/]+[=]{0,2}$", sb) is not None

• is_image_data(b64data): 通过检查其头部验证 base64 数据是否代表图像。

def is_image_data(b64data):
    image_signatures = {
        b"\xff\xd8\xff": "jpg",
        b"\x89\x50\x4e\x47\x0d\x0a\x1a\x0a": "png",
        b"\x47\x49\x46\x38": "gif",
        b"\x52\x49\x46\x46": "webp",
    }
    try:
        header = base64.b64decode(b64data)[:8]
        for sig, format in image_signatures.items():
            if header.startswith(sig):
                return True
        return False
    except Exception:
        return False

• resize_base64_image(base64_string, size=(128, 128)): 将 base64 编码的图像调整为指定的尺寸。

def resize_base64_image(base64_string, size=(128, 128)):
    img_data = base64.b64decode(base64_string)
    img = Image.open(io.BytesIO(img_data))
    resized_img = img.resize(size, Image.LANCZOS)
    buffered = io.BytesIO()
    resized_img.save(buffered, format=img.format)
    return base64.b64encode(buffered.getvalue()).decode("utf-8")

• split_image_text_types(docs): 将文档列表分割为 base64 编码的图像和文本。

def split_image_text_types(docs):
    b64_images = []
    texts = []
    for doc in docs:
        if isinstance(doc, Document):
            doc = doc.page_content
        if looks_like_base64(doc) and is_image_data(doc):
            doc = resize_base64_image(doc, size=(1300, 600))
            b64_images.append(doc)
        else:
            texts.append(doc)
    return {"images": b64_images, "texts": texts}

提示函数

img_prompt_func(data_dict) 函数格式化输入数据以供 AI 模型使用。它将文本和图像数据组合成一个单一的提示，其中包括“用户问题”和“聊天记录”。

def img_prompt_func(data_dict):
    formatted_texts = "\n".join(data_dict["context"]["texts"])
    messages = []

    if data_dict["context"]["images"]:
        for image in data_dict["context"]["images"]:
            image_message = {
                "type": "image_url",
                "image_url": {"url": f"data:image/jpg;base64,{image}"},
            }
            messages.append(image_message)
            
    chat_history = data_dict.get("chat_history", [])
    formatted_chat_history = "\n".join([f"{m.type}: {m.content}" for m in chat_history])

    text_message = {
        "type": "text",
        "text": (
            "You are a Research Assistant tasked with answering questions on research articles.\n"
            "You will be given a mixed of text, tables, and image(s) usually of tables, charts or graphs.\n"
            "Use this information to provide accurate information related to the user question. \n"
            f"User-provided question: {data_dict['question']}\n\n"
            "Text and / or tables:\n"
            f"{formatted_texts}"
            "Chat History:\n"
            f"{formatted_chat_history}\n\n"
        ),
    }
    messages.append(text_message)
    return [HumanMessage(content=messages)]

多模态 RAG 链

最后，multi_modal_rag_chain(retriever, memory=None) 函数用于设置我们的 RAG 链。以下是该链的工作原理：

• 它以 RunnableParallel 组件开始，该组件并行检索相关文档，并使用 split_image_text_types 函数将其分为文本和图像。同时，它将用户的问题原封不动地传递，并从内存中检索对话历史。这种并行处理确保所有必要的上下文信息迅速有效地收集。
• 此步骤的输出由 img_prompt_func 格式化为结构化提示，将用户查询、检索到的上下文和聊天历史整合为适合 AI 模型的连贯格式。
• 这个结构化提示随后传递给 GPT-4o 模型，生成基于提供信息的响应。
• 最后，StrOutputParser 确保模型的输出格式化为字符串，准备进一步使用。

这一设计使系统能够灵活处理需要理解和整合文本与视觉数据的复杂查询，同时保持正在进行的对话的上下文。

def multi_modal_rag_chain(retriever, memory=None):
    if memory is None:
        memory = ConversationBufferMemory(return_messages=True, memory_key="chat_history")
    
    model = ChatOpenAI(temperature=0, model="gpt-4o", max_tokens=1024)

    chain = (
        RunnableParallel(
            {
            "context": retriever | RunnableLambda(split_image_text_types),
            "question": RunnablePassthrough(),
            "chat_history": lambda x: memory.load_memory_variables({})["chat_history"]
        })
        | RunnableLambda(img_prompt_func)
        | model
        | StrOutputParser()
    )

    def run_chain(query):
        result = chain.invoke(query)
        memory.save_context({"input": query}, {"output": result})
        return result

    return run_chain


# 创建 RAG 链
chain_mm_rag = multi_modal_rag_chain(retriever=retriever_multi_vector_img)

测试时间！

让我们提出第一个问题，看看我们链条的响应。

# First Question
query = "How is the performance of TAGI-V for the Boston dataset compared to the other methods?"
print(chain_mm_rag(query))

要确定TAGI-V在波士顿数据集上的表现与其他方法相比如何，我们需要查看该数据集提供的具体指标。文本提到，TAGI-V在5个数据集中的4个数据集的测试对数似然值上表现优于所有其他方法，并且在RMSE值上具有竞争力，在5个数据集中的2个数据集中提供了最佳结果。

然而，文本并没有明确指出TAGI-V在波士顿数据集上的表现。要提供准确的答案，我们需要TAGI-V和其他方法在波士顿数据集上的具体测试对数似然值和RMSE值。

根据提供的信息：
- TAGI-V在测试对数似然值上通常表现强劲。
- TAGI-V在RMSE值上具有竞争力，在某些数据集中表现最佳，而在其他数据集中排名第二。

如果波士顿数据集是TAGI-V不是最佳的那些数据集之一，它可能会被PCA+VI或NL超越，这两者在各自的两个数据集中被提及为最佳表现者。

在没有波士顿数据集具体值的情况下，我们可以推断TAGI-V可能具有竞争力，但在这个特定数据集中可能不是最佳表现者。要进行明确的比较，需要所有方法在波士顿数据集上的确切测试对数似然值和RMSE值。

响应是正确的。我可以验证，因为我是文章的作者。😀

让我们尝试第二个问题。

# Second Question
query = "What is the performance of the same method for the Concrete dataset compared to the other methods?"
print(chain_mm_rag(query))

要评估同一方法在混凝土数据集上的表现与其他方法相比如何，我们可以参考提供的图表和表格。以下是详细分析：

### 性能指标：
1. **RMSE（均方根误差）**：
   - 混凝土数据集的RMSE值在图表中显示。比较的方法包括PCA+ESS、PCA+VI、SWAG、TAGI-V、TAGI-V2L、TAGI、PBP、MC-dropout、PBP-MV、VMG、Ensemble、DVI和NN。
   - 从图中可以看出，TAGI-V及其变体（TAGI-V2L、TAGI）在RMSE值上与其他方法具有竞争力。文本中没有明确提供确切的RMSE值，但视觉表现表明TAGI-V表现良好。

2. **训练时间**：
   - 混凝土数据集的训练时间在时间（秒）与RMSE的图表中显示。
   - TAGI-V及其变体（TAGI-V2L、TAGI）相比于PCA+ESS、PCA+VI、PBP-MV和VMG等方法显示出更快的训练时间。TAGI-V显著更快，大约比PCA+ESS和PCA+VI快100倍，比PBP快约10倍，比Ensemble快约3倍。

### 比较分析：
- **TAGI-V**：
  - **RMSE**：TAGI-V显示出具有竞争力的RMSE值，表明良好的预测性能。
  - **训练时间**：TAGI-V的训练时间显著快于大多数其他方法。

- **其他方法**：
  - **PCA+ESS和PCA+VI**：这些方法的RMSE值较高，训练时间较长，相比于TAGI-V。
  - **SWAG、PBP、MC-dropout、PBP-MV、VMG、Ensemble、DVI、NN**：这些方法的RMSE值和训练时间也高于TAGI-V。

### 总结：
TAGI-V在混凝土数据集上表现优越，无论是在RMSE还是训练时间方面。它在RMSE值上具有竞争力，并且在训练时间上显著更快，相比于其他方法。这使得TAGI-V成为混凝土数据集的高效且有效的方法。

所以，它记得我们在询问TAGI-V，表明应用程序现在是对话式的。😎

让我们检查检索到的文档，看看一个问题是否会返回一个base64编码的图像。

# Check retrieval
query = "How is the performance of He compared to modified He for the various datasets such as Boston, Concrete etc.?"
docs = retriever_multi_vector_img.invoke(query, limit=6)

确实有一个检索到的文档是图像文件。🙌🏼

LLM评估

为了评估我们的模型，我们将使用一个名为DeepEval的开源LLM评估框架。该框架提供了多个指标来测试检索到的文档和根据输入查询给出的最终响应。在本次实验中，我们将重点关注以下指标：

• 忠实度指标：衡量模型输出与提供的上下文的对齐程度。
• 上下文相关性指标：评估检索到的上下文与给定查询的相关性。
• 答案相关性指标：评估模型的响应与输入查询的相关性。
• 幻觉指标：检测模型输出是否包含在给定上下文中不存在的信息。

DeepEval还提供了更多指标，鼓励您在库文档中进行探索。

我们将定义一个名为LLM_Metric的类，该类包含每个指标的函数。每个指标的输出不仅是一个分数，还提供了该分数的原因，从而提供对模型性能的更深入的洞察。

class LLM_Metric:
    def __init__(self, query, retrieval_context, actual_output):
        self.query = query
        self.retrieval_context = retrieval_context
        self.actual_output = actual_output

    # Faithfulness
    def get_faithfulness_metric(self):
        metric = FaithfulnessMetric(
            threshold=0.7,
            model="gpt-4o",
            include_reason=True
        )
        test_case = LLMTestCase(
            input=self.query,
            actual_output=self.actual_output,
            retrieval_context=self.retrieval_context
        )

        metric.measure(test_case)
        return metric.score, metric.reason

    # Contextual Relevancy
    def get_contextual_relevancy_metric(self):
        metric = ContextualRelevancyMetric(
            threshold=0.7,
            model="gpt-4o",
            include_reason=True
        )
        test_case = LLMTestCase(
            input=self.query,
            actual_output=self.actual_output,
            retrieval_context=self.retrieval_context
        )
        
        metric.measure(test_case)
        return metric.score, metric.reason
    
    # Answer Relevancy
    def get_answer_relevancy_metric(self):
        metric = AnswerRelevancyMetric(
        threshold=0.7,
        model="gpt-4o",
        include_reason=True
        )
        test_case = LLMTestCase(
            input=self.query,
            actual_output=self.actual_output
        )
        metric.measure(test_case)
        return metric.score, metric.reason
    
    # Hallucination
    def get_hallucination_metric(self):
        metric = HallucinationMetric(threshold=0.5)
        test_case = LLMTestCase(
        input=self.query,
        actual_output=self.actual_output,
        context=self.retrieval_context  
        )
        metric.measure(test_case)
        return metric.score, metric.reason

使用 Streamlit 的用户界面

首先，我们将以模块化的方式构建代码。我们将所有函数放在一个名为 utils 的文件夹中。目录结构如下所示：

advanced-RAG-app/
│
├── utils/
│ ├── __init__.py
│ ├── image_processing.py
│ ├── rag_chain.py
│ ├── rag_evaluation.py
│ └── retriever.py
│
└── main.py
└── requirements.txt

这些函数在之前的帖子中已经解释过。这里我们只是调整了一下结构，以便能够从主应用文件中调用所有函数。

话虽如此！让我们填充我们的 main.py 文件。

import streamlit as st
from unstructured.partition.pdf import partition_pdf
from unstructured.chunking.title import chunk_by_title
from typing import Any
from pydantic import BaseModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from utils.image_processing import generate_img_summaries
from utils.retriever import create_multi_vector_retriever
from utils.rag_chain import multi_modal_rag_chain, plt_img_base64
from utils.rag_evaluation import LLM_Metric
from io import BytesIO
import base64
from PIL import Image
import io

# 初始化会话状态
if 'processed' not in st.session_state:
    st.session_state.processed = False
if 'retriever' not in st.session_state:
    st.session_state.retriever = None
if 'chain' not in st.session_state:
    st.session_state.chain = None

# Streamlit 应用设置
st.set_page_config(page_title='多模态 RAG 应用', page_icon='random', layout='wide', initial_sidebar_state='auto')

def process_document(uploaded_file):
    # 处理 PDF
    with st.spinner('正在处理 PDF...'):
        st.sidebar.info('正在从 PDF 中提取元素...')
        pdf_bytes = uploaded_file.read()
        elements = partition_pdf(
            file=BytesIO(pdf_bytes),
            strategy="hi_res",
            extract_images_in_pdf=True,
            extract_image_block_types=["Image", "Table"],
            extract_image_block_to_payload=False,
            extract_image_block_output_dir="docs/saved_images",
        )
        st.sidebar.success('PDF 元素提取成功！')

    # 按标题创建块
    with st.spinner('正在分块内容...'):
        st.sidebar.info('正在按标题创建块...')
        chunks = chunk_by_title(elements)
        st.sidebar.success('分块完成！')

    # 分类元素
    class Element(BaseModel):
        type: str
        text: Any

    categorized_elements = []
    for element in chunks:
        if "unstructured.documents.elements.CompositeElement" in str(type(element)):
            categorized_elements.append(Element(type="text", text=str(element)))
        elif "unstructured.documents.elements.Table" in str(type(element)):
            categorized_elements.append(Element(type="table", text=str(element)))

    text_elements = [e for e in categorized_elements if e.type == "text"]
    table_elements = [e for e in categorized_elements if e.type == "table"]

    # 提示
    prompt_text = """您是一位专家研究助理，负责总结研究文章中的表格和文本。 \
    请给出文本的简明总结。文本块：{element} """

    prompt = ChatPromptTemplate.from_template(prompt_text)

    # 总结链
    model = ChatOpenAI(temperature=0, model="gpt-4o", max_tokens=1024)
    summarize_chain = {"element": lambda x: x} | prompt | model | StrOutputParser()

    texts = [i.text for i in text_elements]
    text_summaries = summarize_chain.batch(texts, {"max_concurrency": 5})

    tables = [i.text for i in table_elements]
    table_summaries = summarize_chain.batch(tables, {"max_concurrency": 5})

    # 图像总结
    fpath = "docs/saved_images"
    img_base64_list, image_summaries = generate_img_summaries(fpath)

    # 向量存储
    vectorstore = Chroma(
        collection_name="mm_tagiv_paper", embedding_function=OpenAIEmbeddings()
    )

    # 创建检索器
    st.session_state.retriever = create_multi_vector_retriever(
        vectorstore,
        text_summaries,
        texts,
        table_summaries,
        tables,
        image_summaries,
        img_base64_list,
    )

    # 创建 RAG 链
    st.session_state.chain = multi_modal_rag_chain(retriever=st.session_state.retriever)
    st.session_state.processed = True

with st.sidebar:
    # 文件上传
    st.subheader('添加您的 PDF')
    uploaded_file = st.file_uploader("上传 PDF 文件", type=["pdf"])
    if st.button('提交'):
        if uploaded_file is not None:
            process_document(uploaded_file)
            st.success('文档处理成功！')
        else:
            st.error('请先上传 PDF 文件。')

# 查询响应和评估的主页面
st.subheader("RAG 助手")
query = st.text_input("输入您的查询：")

if query and st.session_state.processed:
    # 执行
    retrieval_context = st.session_state.retriever.invoke(query, limit=1)
    actual_output = st.session_state.chain(query)

    # 评估
    llm_metric = LLM_Metric(query, retrieval_context, actual_output)
    faith_score, faith_reason = llm_metric.get_faithfulness_metric()
    relevancy_score, relevancy_reason = llm_metric.get_contextual_relevancy_metric()
    answer_relevancy_score, answer_relevancy_reason = llm_metric.get_answer_relevancy_metric()
    hallucination_score, hallucination_reason = llm_metric.get_hallucination_metric()

    # 显示结果
    st.subheader("查询响应")
    st.write(actual_output)

    st.subheader("评估指标")
    st.write(f"可信度评分：{faith_score}，原因：{faith_reason}")
    st.write(f"上下文相关性评分：{relevancy_score}，原因：{relevancy_reason}")
    st.write(f"答案相关性评分：{answer_relevancy_score}，原因：{answer_relevancy_reason}")
    st.write(f"幻觉评分：{hallucination_score}，原因：{hallucination_reason}")

    
elif query and not st.session_state.processed:
    st.warning("请先上传并处理文档。")

不要感到不知所措！我会详细指导您完成每一个步骤。

首先，应用程序初始化会话状态，以跟踪文档是否已被处理，并存储检索器和 RAG 链对象。

接下来，我们定义 process_document 函数，该函数处理上传的 PDF 文件的核心处理。这包括：

• PDF 提取
• 分块
• 分类
• 总结
• 图像总结
• 向量存储初始化
• 检索器创建
• RAG 链创建

侧边栏允许用户上传 PDF 文件，触发文档处理功能。一旦文档成功处理，主页面允许用户输入查询并查看响应及评估指标。

最后，使用以下命令运行应用：

streamlit run main.py

就这样！

最后的想法

我们已经完成了项目，探讨了如何为PDF文档创建一个多模态的RAG应用程序，以实现视觉问答。以下是我们所涵盖的内容：

• 利用Unstructured库将文档拆分为多个部分。
• 采用多向量检索器将文本、表格和图像摘要存储在向量存储中，同时将原始内容保留在文档存储中。
• 利用DeepEval库评估我们的LLM响应。
• 使用Streamlit构建了一个简单的用户界面。

我很想听听你对这篇文章的看法，请在评论区留言。我希望这能成为你愉快的周末项目。下次见！

À Bientôt 🙂

嗨，我是Bhargob！👋

我是一名机器学习研究员，对构建GenAI应用程序充满热情。

非常感谢所有关注我工作的朋友们。我希望我的文章能激励你在人工智能学习之旅中前行。你可以在LinkedIn和GitHub上与我联系——我始终欢迎与有趣的人和激动人心的生成AI项目合作。如果你需要AI解决方案来推动你的业务，请在LinkedIn上与我联系，我们可以聊聊！

期待与你的联系！

资源

如果您对生成性人工智能的概念不太熟悉，我强烈建议您关注我的LangChain框架初学者系列。从基本概念到高级部署策略，这个系列涵盖了您开始构建自己的AI应用所需的一切。

以下是我们所涵盖内容的快速回顾：

1️⃣ 使用LangChain构建聊天机器人：Python的温和介绍 2️⃣ 创建检索链：使用LangChain的RAG聊天机器人 3️⃣ 使用LangChain创建RAG代理 4️⃣ 使用LangGraph设计RAG代理工作流 5️⃣ LangGraph、FastAPI和Streamlit/Gradio：AI开发的完美组合 6️⃣ 从本地到云：使用Docker和AWS EC2部署LLM应用 7️⃣ 通过Git推送进行云部署：您所需的仅是GitHub Actions

我还发布了一些关于高级AI主题的文章，包括创建一个多代理框架和两个端到端项目：

8️⃣ 构建酒店推荐系统：使用CrewAI、Ollama和Gradio的多代理框架 9️⃣ 使用CrewAI代理创建端到端数据科学项目

我希望这些资源能帮助您在AI的旅程中。祝学习愉快！