检索增强生成（RAG）系统和多模态大型语言模型（LLM）正在迅速发展，其应用场景广泛，从优化搜索体验到生成复杂内容无所不包。这些方法不断得到完善，以拓展人工智能的能力边界。但是，如果你能结合它们的优势，构建一个不仅能处理文本还能无缝处理图像的RAG系统，那会怎样呢？

现在，想象一下，更进一步，创建这样一个系统，而不依赖于LangChain或LlamaIndex等预建框架，又会如何？

无框架的方法让你拥有完全的控制权，可以根据你的确切需求定制应用程序，而不受外部依赖或版本限制的约束。

在这篇文章中，我们将带你深入了解这一过程。从比较文本和图像提取工具，到从零开始构建一个稳健的多模态RAG系统，我们将逐步引导你，使你能够创建自己的定制解决方案。

第一部分：多模态RAG系统的数据摄入

任何RAG系统的基础都是可靠的数据。为了系统能够检索并生成准确的结果，其处理的内容需要是干净、完整且结构良好的。在处理PDF时，这意味着不仅要提取文本，还要提取图像和表格，确保不丢失任何有价值的信息。

为PDF处理选择合适的工具至关重要。市面上有许多库可供选择，决定哪个适合你的需求可能会让人感到困惑。有些库注重速度，有些则专注于准确性，而少数库在这两者之间取得了平衡。为了帮助你做出明智的决定，我们评估了三个流行的库：PyMuPDF、PDFium和PDFPlumber。

为RAG系统中的PDF处理选择合适的工具

PDFPlumber：

pdfplumber能够从PDF中提取相当准确的文本和表格。然而，它有一个显著的缺点：对于复杂的PDF，处理起来可能非常耗时。例如，从100页的PDF中提取内容可能需要长达1分钟10到30秒的时间。这是一个相当长的延迟，尤其是在处理多个PDF时。

代码示例：

!pip install pdfplumber

import pdfplumber
import time
def read_text_pdfplumber(pdf_path):
 """ Extracts text and tables from pdf using pdfplumber """
  text_results=[]
  tables=[]
  with pdfplumber.open(pdf_path) as pdf:
      for page in pdf.pages:
          # Text extraction
          text = page.extract_text()
          text_results.append(text)
          # Table extraction
          table = page.extract_tables()
          tables.append(table)
  return(text_results,tables)

对于一个622页的PDF，该脚本花了152.35秒来提取文本和表格。

PDFium：

pdfium是一个快速的库，只需5到15秒就能将PDF内容转换为文本。由于其将PDF页面转换为图像然后再从中提取文本的方法，文本提取相当准确。这使得它成为快速文本提取的绝佳选择。然而，它有一个限制：它不支持明确地将图像或表格与文本分开。

代码示例：

!pip install pypdfium2

import pypdfium2 as pdfium
def read_text_pdfium(path):
   """ Extracts text from pdf using pdfium """
   
   pdf = pdfium.PdfDocument(path)
   version = pdf.get_version()  # get the PDF standard version
   n_pages = len(pdf)
   pdfium_text=""
   for i in range(len(pdf)):
    page = pdf[i]  # load a page
     width, height = page.get_size()
     textpage = page.get_textpage()
     text_part = textpage.get_text_bounded(left=50, bottom=100, right=width, top=height)
     pdfium_text+=text_part
   
   return pdfium_text

对于一个622页的PDF，该脚本花了5到10秒来提取文本。

PyMuPDF

如果你在寻找从PDF中提取内容的最佳工具之一，那么PyMuPDF难以被超越。它速度极快，处理一个PDF只需4到7秒。它不仅能提取文本，还能处理图像，使其成为一个功能全面的选择。如果你需要在多个PDF中快速获得可靠的结果，那么PyMuPDF绝对值得考虑。

代码示例：

!pip install pymupdf

import fitz  # PyMuPDF library
def read_text_pymupdf(path):
    """Extracts text from a PDF using PyMuPDF."""
    doc = fitz.open(path)
    text_results = []
    for page in doc:
        text = page.get_text()
        text_results.append(text)
    return text_results

在这里，我将使用PyMuPDF从PDF中提取图像。提取的图像将保存在输出目录中。

代码示例：

import fitz  # PyMuPDF
import os
def extract_images_from_pdf(pdf_path, output_folder):
  """  Extract images from pdf using PyMuPDF """
    pdf_document = fitz.open(pdf_path)
    
    os.makedirs(output_folder, exist_ok=True)
    for page_number in range(len(pdf_document)):
        page = pdf_document[page_number]
        images = page.get_images(full=True)
                for image_index, img in enumerate(images):
                    xref = img[0]
                    base_image = pdf_document.extract_image(xref)
                    image_bytes = base_image["image"]
                    image_ext = base_image["ext"]
                    image_filename = f"page_{page_number+1}_image_{image_index+1}.{image_ext}"
                    image_path = os.path.join(output_folder, image_filename)
                    with open(image_path, "wb") as image_file:
                        image_file.write(image_bytes)
                    print(f"Saved: {image_path}")
            pdf_document.close()

对于一个622页的PDF，PyMuPDF仅在5到6秒内就提取了文本，与其他库相比具有显著的性能优势。这种效率使得它成为处理大型或多个PDF的绝佳选择。然而，理想的库取决于你的特定用例和需求。

PDF库处理时间比较

下表比较了使用三个库（PDFium、PyMuPDF和PDFPlumber）处理不同页数PDF所需的时间。

根据性能分析，PyMuPDF在处理页数超过1,000的PDF时，始终优于PDFiumPDFPlumber，速度比PDFium快2.3倍，比PDFPlumber快59倍。

关键要点

• PyMuPDF速度最快，非常适合大规模或时间敏感的任务。
• PDFium在速度和准确性之间取得了平衡，但缺乏图像提取功能。
• PDFPlumber速度较慢，但在从复杂PDF中提取详细表格方面表现出色。

库性能比较

第二部分：嵌入文本和图像

现在我们已经处理了PDF内容，下一步是构建一个多模态RAG系统。这涉及将提取的文本和图像集成到一个检索增强的管道中。

对于内容提取，我将使用PyMuPDF，因为它具有卓越的速度和处理文本和图像的能力。为了构建多模态检索器，我将利用QdrantDB作为向量存储。

处理和嵌入文本和图像

为了优化嵌入性能和检索准确性，文本内容被分割成更小、更易管理的块。在这里，我们使用LangChain的RecursiveCharacterTextSplitter，因为其简单性和可配置性。然而，文本分割可以通过多种方式完成，例如编写自定义分割器或利用NLTK或spaCy等库。

在这个例子中，我们为每个块添加元数据，包括PDF路径和唯一的UUID。元数据在检索过程中有助于保持上下文。

安装依赖项：

#for colab to load openai client
%%capture
!pip install openai==1.55.3 httpx==0.27.2 --force-reinstall --quiet

!pip install transformers torch pillow
!pip install --upgrade nltk
!pip install sentence-transformers
!pip install --upgrade qdrant-client fastembed Pillow
!pip install -U langchain-community

将文本分割成块

我们将为每个块添加相关的元数据，元数据可以是任何内容，可以是有助于检索块的内容。但在这里，我只是为文本添加了PDF路径和一个唯一的UUID作为元数据。

from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1024,
    chunk_overlap=20,
    length_function=len,
    is_separator_regex=False,
)
#here text_results contain all text from pdf
doc_texts = text_splitter.create_documents(text_results)
import uuid
for i in range(len(doc_texts)):
  unique_id = str(uuid.uuid4())
  doc_texts[i].metadata['document_info'] = pdf_path
  doc_texts[i].metadata['uuid'] = unique_id
  print(doc_texts[0].metadata)

嵌入文本块

分割后，使用Nomic文本嵌入模型对文本块进行嵌入。这将每个块转换为适合存储在检索器中的向量表示。

from transformers import AutoTokenizer, AutoModel
# Load the tokenizer and model
text_tokenizer = AutoTokenizer.from_pretrained("nomic-ai/nomic-embed-text-v1.5", trust_remote_code=True)
text_model = AutoModel.from_pretrained("nomic-ai/nomic-embed-text-v1.5", trust_remote_code=True)
def get_text_embeddings(text):
    inputs = text_tokenizer(text, return_tensors="pt", padding=True, truncation=True)
    outputs = text_model(**inputs)
    embeddings = outputs.last_hidden_state.mean(dim=1)
    return embeddings[0].detach().numpy()
# Example usage
text = "This is a test sentence."
embeddings = get_text_embeddings(text)
print(embeddings[:5])
#embedding text
texts_embeded = [get_text_embeddings(document.page_content) for document in doc_texts]
text_embeddings_size=len(texts_embeded[0])
text_embeddings_size
#output: 768

嵌入图像

从指定目录中读取图像，并使用Nomic Vision嵌入模型进行嵌入。使用last_hidden_state输出的平均值来表示每个图像，作为固定长度的向量。

以下是加载模型和测试图像的示例代码：

from transformers import AutoModel, AutoProcessor
from PIL import Image
import torch
model = AutoModel.from_pretrained("nomic-ai/nomic-embed-vision-v1.5", trust_remote_code=True)
processor = AutoProcessor.from_pretrained("nomic-ai/nomic-embed-vision-v1.5")
# Load the image

#testing an image on loaded model 
image = Image.open("/content/drive/MyDrive/images/Screenshot 2024-09-08 025257.png")
inputs = processor(images=image, return_tensors="pt")
model.eval()
#testing model
with torch.no_grad():
    outputs = model(**inputs)
embeddings = outputs.last_hidden_state
print(embeddings)

对于我们的多模态RAG系统，将使用PIL从目录中读取图像，然后使用加载的视觉模型进行嵌入，代码如下。

import os
import numpy as np
image_embeddings = []
image_files = os.listdir(output_directory)
for img in image_files:
    try:
        image = Image.open(os.path.join(output_directory, img))
        inputs = processor(images=image, return_tensors="pt")
        model.eval()
        with torch.no_grad():
            outputs = model(**inputs)
        embeddings = outputs.last_hidden_state
        print(f"Image: {img}, Embedding shape: {embeddings.shape}")
        if embeddings.size(0) > 0:  
            image_embedding = embeddings.mean(dim=1).squeeze().cpu().numpy()
            image_embeddings.append(image_embedding)
        else:
            print(f"Skipping image {img} due to empty embeddings.")
    except Exception as e:
        print(f"Error processing {img}: {e}")
#setting image embedding model length:
image_embeddings_size=len(image_embeddings[0])

关键点：

• 工具灵活性：虽然这里使用了Nomic模型，但可以根据特定需求替换为Sentence Transformers或自定义模型等替代方案。
• 元数据的重要性：添加唯一标识符等相关元数据可以确保可追溯性并提高检索准确性。

准备好文本和图像嵌入后，下一步是将它们集成到一个统一的检索管道中。这将使系统能够跨模态无缝地检索和生成响应。

设置Qdrant以进行多模态检索

准备好文本和图像嵌入后，下一步是初始化Qdrant客户端并配置其以进行多模态检索。Qdrant作为向量数据库，允许在文本和图像嵌入之间进行高效的相似度搜索。

初始化Qdrant客户端

在这里，我们使用内存配置来测试Qdrant。对于生产环境，请将:memory:替换为持久数据库路径，以确保可扩展性。

from qdrant_client import QdrantClient, models
client = QdrantClient(":memory:")

为文本和图像创建单独的集合

我们创建两个单独的集合：一个用于文本，另一个用于图像。每个集合都根据使用的嵌入模型配置向量参数（大小和距离）。

# Check if the collection exists, and create it if not
if not client.collection_exists("images"):  #creating a Collection
    client.create_collection(
        collection_name="images",
        vectors_config=models.VectorParams(
        size=image_embeddings_size,  # Vector size is defined by model being used for embedding
        distance=models.Distance.COSINE,
    ),
   )
    
if not client.collection_exists("text"):
 client.create_collection(
        collection_name ="text",
      vectors_config=models.VectorParams(
        size=text_embeddings_size,  # Vector size is defined by model being used for embedding
        distance=models.Distance.COSINE,
    ),
 )

将嵌入上传到Qdrant

创建集合后，我们将嵌入和元数据填充到其中。元数据确保在检索过程中保持上下文和可追溯性。

上传文本嵌入

我们通过上传文本嵌入及其相关元数据（如内容和唯一标识符）来填充文本集合。

client.upload_points(
    collection_name="text","text",
    points=[
        models.PointStruct(
            id=doc.metadata['uuid'],
            vector=np.array(texts_embeded[idx]),
            payload={ #save meta data and content as payload
                "metadata": doc.metadata,
                "content": doc.page_content
            }
        )
        for idx, doc in enumerate(doc_texts)
    ]
)

上传图像嵌入

对于图像，每个嵌入在有效载荷元数据中与其对应的图像路径相关联。我们存储图像路径，以便在多模态LLM检索到图像时可以重新加载。这种有序的方法允许在集合内高效搜索图像。

#for image embeddings
# Ensure that image_embeddings are not empty
if len(image_embeddings) > 0:
    # Upload points to Qdrant
    client.upload_points(
        collection_name="images",
        points=[
            models.PointStruct(
                id=str(uuid.uuid4()),  # unique id of a point
                vector= np.array(image_embeddings[idx])  ,
                payload={"image_path": output_directory+'/'+str(image_files[idx])}  # Image path as metadata
            )
            for idx in range(len(image_files))  
    )
else:
    print("No valid embeddings found, nothing to upload.")

第三部分：创建检索器

最后，我们通过设置一个多模态检索器，将所有内容整合在一起，使我们能够检索与用户查询匹配的文本块和图像。检索器被配置为基于相似度分数，从每个集合（文本和图像）中返回前3个结果。

为什么使用单独的集合？

文本和图像嵌入在不同的特征空间中运作，它们的相似度分数不能直接比较。通过分离集合并运行独立查询，我们确保文本和图像结果都能根据适当的权重和相关性被检索出来。这种方法避免了由于一种模态（例如文本）的相似度分数较高而主导结果的情况，这可能会掩盖来自另一种模态的相关结果。

使用独立集合确保：

• 文本块根据其与查询的语义相似性被检索。
• 图像根据其视觉相似性被检索，独立于文本分数。

这种分离对于多模态系统至关重要，因为在这些系统中，两种模态对于满足用户意图同等重要。

检索器函数

import matplotlib.pyplot as plt
import matplotlib.image as mpimg
def MultiModalRetriever(user_query):
    """
    Retrieve multimodal results (text and images).
    Parameters:
    - user_query: The user's query string.
  
    """
    query = get_text_embeddings(user_query)
    # Retrieve text hits
    text_hits = client.query_points(
        collection_name="text",
        query=query,
        limit=3, #set your own limit according to the requirements
    ).points
    # Retrieve image hits
    Image_hits = client.query_points(
        collection_name="images",
        query=query,
        limit=3, #set your own limit according to the requirements
    ).points
    return text_hits, Image_hits

现在构建一个查看器来显示检索结果：

import matplotlib.pyplot as plt
import matplotlib.image as mpimg
#text_trunc_length: Maximum number of characters to display for text hits before truncating.
def MultiModalRetrieverDisplay(text_hits,Image_hits, text_trunc_length=150):
   """
    Displays text and image results from a multimodal retriever.  
    Parameters:
    ----------
    text_hits :List of text results with `id`, `payload`, and `score`.
    Image_hits :List of image results with `id`, `payload`, and `score`.
    text_trunc_length : int, Maximum length for displayed text content (default is 150).
    Displays:
    --------
    - Text results in bold with IDs, truncated content, and scores.
    - Image results in a matplotlib plot with scores in titles.
    """
    
    print("\\033[1mText Results:\\033[0m")
    for i, hit in enumerate(text_hits, 1):
        print("NODEID:",hit.id)
        content = hit.payload['content']
        truncated_content = content[:text_trunc_length] + "..." if len(content) > text_trunc_length else content
        bold_truncated_content = f"\\033[1m{truncated_content}"
        print(f"{i}. {bold_truncated_content} | Score: {hit.score}")
   
    print("\\nImage Results:")
    fig, axes = plt.subplots(1, len(Image_hits), figsize=(15, 5))  # Adjust figsize as needed
    for ax, hit in zip(axes, Image_hits):
        image_path = hit.payload['image_path']
        print(f"Displaying image: {image_path} | Score: {hit.score}")
        img = mpimg.imread(image_path)
        ax.imshow(img)
        ax.axis('off')
        ax.set_title(f"Score: {hit.score}", fontsize=10)
    plt.suptitle("Image Results", fontsize=16)
    plt.tight_layout()
    plt.show()

测试多模态检索器和查看器功能

text_hits, Image_hits=MultiModalRetriever("how much of acre burned in 2022?")"how much of acre burned in 2022?")
MultiModalRetrieverDisplay(text_hits,Image_hits, text_trunc_length=150)

多模态LLM集成的RAG管道

要构建一个多模态RAG系统管道，你可以集成任何多模态LLM来处理文本和图像。其工作原理如下：

1. 准备数据：将文本块（上下文列表）和图像分别整理成文件路径列表。
2. 处理图像：将图像文件路径转换为编码的图像数据。然后，这些编码的图像将根据模型的要求作为URL或编码对象传递给模型。
3. 模型输入：将文本块和编码图像组合成单一的输入格式。多模态LLM将使用文本块作为上下文，图像作为视觉数据来生成答案。

这种方法使模型能够无缝处理文本和视觉信息，提供全面且准确的响应。

构建RAG函数：

from openai import ChatCompletion
import openai
import base64
from base64 import b64decode
import os
def MultiModalRAG(
    context: list,
    images: list,
    user_query: str,
    client: client,
    model: str = "yourMLLM"):  # The name of multimodal LLM you are using
    generation_prompt = f"""
    Based on the given context, answer the user query: {user_query},
    context can be tables, texts or Images. Provide the answer from contexts.
    user query: {user_query}
    
    Contexts: {context}\\n
    Output:
    """
    # Helper function to encode an image as a base64 string
    def encode_image(image_path):
        if image_path:
            with open(image_path, "rb") as image_file:
                return base64.b64encode(image_file.read()).decode()
        return None
     
    
    image_paths = images
    messages = [
        {
            "role": "system",
            "content": "You are a helpful assistant."
        },
        {
            "role": "user",
            "content": generation_prompt,
        }
    ]
    
    # Encode images and add them to the messages if present
    for image_path in image_paths:
        img_base64 = encode_image(image_path)
        if img_base64:
            # Adding the image base64 string as part of the text content
            messages.append({
                "role": "user",
                "content": [
                  {
                      "type": "image_url",
                      "image_url": {
                          "url": f_"data:image/jpeg;base64,{img_base64}"  # Send the base64-encoded image
                      },
                  },
              ],
            })
     
    # Create the chat completion
    chat_completion = client.chat.completions.create(
        messages=messages,
        model=model,
        temperature=0.5,
        top_p=0.99,
    )  
    
    return chat_completion.choices[0].message.content

我们的多模态RAG函数接受图像路径作为输入，打开并编码这些图像，然后将它们传递给多模态LLM以回答用户查询。这些图像路径存储在有效载荷中，如图像嵌入集合步骤中所述。当检索器获取图像嵌入时，我们使用它们的有效载荷来获取多模态LLM所需的图像路径。

def RAG(query):
  text_hits, Image_hits=MultiModalRetriever(query)
  retrieved_images=[i.payload['image_path'] for i in Image_hits]
  answer=MultiModalRAG(text_hits,retrieved_images,query,openaiclient)
  return(answer)
RAG("""What percentage of nationwide acreage burned by wildfires in 
2022 was on federal lands, and how does this compare to the 10-year average?""")

第四部分：结论：让多模态数据为你所用

构建一个多模态RAG系统，旨在将复杂、多样的数据转化为真正有用的东西。从使用PyMuPDF等工具提取内容，到使用Nomic Vision和Text等模型嵌入文本和图像，并在Qdrant中进行组织，每一步都在构建一个无缝工作的系统。最后添加多模态LLM，将一切整合在一起，使系统能够提供不仅准确而且上下文丰富、全面的答案。

概述的过程展示了如何：

• 提取多模态数据。
• 将其嵌入并存储在Qdrant中。
• 检索数据以进行基于多模态LLM的处理。

这个过程展示了AI如何弥合不同类型数据（文本和视觉）之间的鸿沟，以解决现实世界的问题。采用正确的方法，RAG系统不仅仅是一个工具；它成为获取有意义见解和做出更明智决策的途径。