【精选优质专栏推荐】

• 《AI 技术前沿》 —— 紧跟 AI 最新趋势与应用
• 《网络安全新手快速入门(附漏洞挖掘案例)》 —— 零基础安全入门必看
• 《BurpSuite 入门教程(附实战图文)》 —— 渗透测试必备工具详解
• 《网安渗透工具使用教程(全)》 —— 一站式工具手册
• 《CTF 新手入门实战教程》 —— 从题目讲解到实战技巧
• 《前后端项目开发(新手必知必会)》 —— 实战驱动快速上手

每个专栏均配有案例与图文讲解，循序渐进，适合新手与进阶学习者，欢迎订阅。

前言

大型语言模型（LLM）发展迅速，已经深度融入我们的日常生活，以至于在各种应用场景中，许多人已对其产生依赖。当人们发现像 ChatGPT 这样的文本生成工具极为实用时，几乎难以避免依赖它们。然而，模型的输出有时可能存在不准确之处，这就需要借助输出增强技术，例如检索增强生成（RAG，Retrieval-Augmented Generation）。

RAG 是一种通过结合外部知识的实时检索来增强 LLM 输出的框架。而多模态 RAG 系统则更进一步，它能够跨多种数据格式（如文本、音频和图像）进行信息检索与处理。

本文将演示如何利用文本、音频和图像数据构建一个多模态 RAG 系统。

实现多模态 RAG 系统

多模态 RAG 系统涉及利用多种类型的数据集，通过访问知识库来实现更高质量的输出。实现方法有多种，但关键在于构建一个在生产环境中稳定运行的系统，而非追求花哨的特性。

在本教程中，我们将通过构建包含图像和音频数据的知识库来增强 RAG 系统。完整代码库可在以下 GitHub 仓库中获取。

其工作流程可概括如下图所示：

目前图像较小，不便于阅读，可点击放大或先保存后放大查看。该工作流程可概括为七个步骤：

1. 提取图像
2. 对图像进行嵌入
3. 存储图像嵌入
4. 处理音频
5. 存储音频嵌入
6. 检索数据
7. 生成并输出响应

由于该流程对计算资源要求较高，我们将使用可访问 GPU 的 Google Colab。具体而言，本教程将使用 A100 GPU，因为对 RAM 的需求较大。

首先，安装本教程所需的所有库：

pip install pdf2image Pillow chromadb torch torchvision torchaudio transformers librosa ipython open-clip-torch qwen_vl_utils

你也可以访问 PyTorch 官方网站，查看适合你系统和环境的版本。

此外，有时从 PDF 中提取图像可能无法正常工作。如果遇到此类问题，应安装以下工具：

apt-get update
apt-get install -y poppler-utils

在环境和工具准备就绪后，我们将导入所有必要的库：

import os
from pdf2image import convert_from_path
from PIL import Image
import chromadb
from chromadb.utils.embedding_functions import OpenCLIPEmbeddingFunction  
import torch
from transformers import CLIPProcessor, CLIPModel, WhisperProcessor, WhisperForConditionalGeneration, Qwen2VLForConditionalGeneration, Qwen2VLProcessor  
import librosa  
from sentence_transformers import SentenceTransformer  
from qwen_vl_utils import process_vision_info
from IPython.display import display, Image as IPImage

在本教程中，我们将使用之前准备好的 PDF 图像数据 和 音频文件（.mp3）。PDF 文件为联合利华的《短食谱》，音频文件为 YouTube 上的 Gordon Ramsay 烹饪音频。

将所有文件放入 dataset 文件夹后，即可开始操作。

首先，我们处理 PDF 文件中的图像数据。为此，需要将 PDF 的每一页提取为图像，示例代码如下：

output_dir = "dataset"
image_output_dir = "extracted_images"

def convert_pdfs_to_images(folder, image_output_dir):
    if not os.path.exists(image_output_dir):
        os.makedirs(image_output_dir)

    pdf_files = [f for f in os.listdir(folder) if f.endswith('.pdf')]
    all_images = {}

    for doc_id, pdf_file in enumerate(pdf_files):
        pdf_path = os.path.join(folder, pdf_file)
        images = convert_from_path(pdf_path, dpi=100)

        image_paths = []
        for i, image in enumerate(images):
            image_path = os.path.join(image_output_dir, f"{doc_id}_page_{i}.png")
            image.save(image_path, "PNG")
            image_paths.append(image_path)

        all_images[doc_id] = image_paths  
    return all_images

all_images = convert_pdfs_to_images(output_dir, image_output_dir)

当 PDF 中的所有图像提取完成后，我们将使用 CLIP 模型生成图像嵌入。CLIP 是 OpenAI 开发的多模态模型，能够理解图像与文本之间的关系。

在本教程的流程中，我们使用 CLIP 生成图像嵌入，并将其存储在 ChromaDB 向量数据库中，以便能够根据文本查询检索相关图像。

生成图像嵌入的示例代码如下：

device = "cuda" if torch.cuda.is_available() else "cpu"
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32").to(device)
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

def embed_images(image_paths):
    embeddings = []
    for path in image_paths:
        image = Image.open(path)
        inputs = processor(images=image, return_tensors="pt", padding=True).to(device)
        with torch.no_grad():
            image_embedding = model.get_image_features(**inputs).cpu().numpy()
        embeddings.append(image_embedding)
    return embeddings

image_embeddings = {}
for doc_id, paths in all_images.items():
    image_embeddings[doc_id] = embed_images(paths)

接下来，我们将处理音频数据，使用 Whisper 模型生成文本转录。Whisper 是 OpenAI 开发的模型，基于 Transformer 架构，可将音频输入转换为文本。

在本教程流程中，Whisper 不用于生成嵌入，而仅负责音频转录。我们会先将音频分块进行转录，然后使用 Sentence Transformers 对转录得到的文本块生成嵌入。

音频转录处理的示例代码如下：

whisper_processor = WhisperProcessor.from_pretrained("openai/whisper-small")
whisper_model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-small").to(device)

# 定义每块文本长度（秒）
def transcribe_audio(audio_path, chunk_length=30):
    audio, sr = librosa.load(audio_path, sr=16000)  
    chunk_size = chunk_length * sr  
    chunks = [audio[i:i + chunk_size] for i in range(0, len(audio), chunk_size)]  
    transcription_chunks = []
    for chunk in chunks:
        inputs = whisper_processor(chunk, sampling_rate=sr, return_tensors="pt").to(device)  
        inputs["attention_mask"] = torch.ones_like(inputs.input_features)
        with torch.no_grad():
            predicted_ids = whisper_model.generate(**inputs, max_length=448)  
        chunk_transcription = whisper_processor.batch_decode(predicted_ids, skip_special_tokens=True)[0]
        transcription_chunks.append(chunk_transcription)

    full_transcription = " ".join(transcription_chunks)
    return full_transcription, transcription_chunks

audio_files = [f for f in os.listdir(output_dir) if f.endswith('.mp3')]
audio_transcriptions = {}
for audio_id, audio_file in enumerate(audio_files):
    audio_path = os.path.join(output_dir, audio_file)
    full_transcription, transcription_chunks = transcribe_audio(audio_path)
    audio_transcriptions[audio_id] = {
        "full_transcription": full_transcription,
        "chunks": transcription_chunks
    }

在所有嵌入生成完成后，我们将把它们存储到 ChromaDB 向量数据库中。由于图像嵌入和音频转录文本嵌入的特性不同，我们会对它们分别处理，并为两类数据分别初始化对应的嵌入函数。

client = chromadb.PersistentClient(path="chroma_db")
embedding_function = OpenCLIPEmbeddingFunction()
text_embedding_model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')

# 删除已有的集合（如有必要）
try:
    client.delete_collection(name="image_collection")
    client.delete_collection(name="audio_collection")
    print("已删除现有集合。")
except Exception as e:
    print(f"集合不存在或无法删除: {e}")

image_collection = client.create_collection(name="image_collection", embedding_function=embedding_function)
audio_collection = client.create_collection(name="audio_collection")

for doc_id, embeddings in image_embeddings.items():
    for i, embedding in enumerate(embeddings):
        image_collection.add(
            ids=[f"image_{doc_id}_{i}"],
            embeddings=[embedding.flatten().tolist()],
            metadatas=[{"doc_id": str(doc_id), "image_path": all_images[doc_id][i]}]
        )

for audio_id, transcription_data in audio_transcriptions.items():
    transcription_chunks = transcription_data["chunks"]
    for chunk_id, chunk in enumerate(transcription_chunks):
        chunk_embedding = text_embedding_model.encode(chunk)
        audio_collection.add(
            ids=[f"audio_{audio_id}_chunk_{chunk_id}"],
            embeddings=[chunk_embedding.tolist()],
            metadatas=[{
                "audio_id": str(audio_id),
                "audio_path": audio_files[audio_id],
                "chunk_id": str(chunk_id)
            }],
            documents=[chunk]
        )

唯一剩下的工作是 在 ChromaDB 向量数据库中设置检索系统。

例如，我们可以使用文本查询，从图像和音频数据中各自检索出前两个最相关的结果。

def retrieve_data(query, top_k=2):

    query_embedding_image = embedding_function([query])[0]  # 图像集合使用 OpenCLIP
    query_embedding_audio = text_embedding_model.encode(query)  # 音频集合使用 SentenceTransformer

    image_results = image_collection.query(
        query_embeddings=[query_embedding_image],
        n_results=top_k
    )

    audio_results = audio_collection.query(
        query_embeddings=[query_embedding_audio.tolist()],
        n_results=top_k
    )

    retrieved_images = [metadata["image_path"] for metadata in image_results["metadatas"][0] if "image_path" in metadata]
    retrieved_chunks = audio_results["documents"][0] if "documents" in audio_results else []

    return retrieved_images, retrieved_chunks

query = "What are the healthiest ingredients to use in the recipe you have?"
retrieved_images, retrieved_chunks = retrieve_data(query)
print("检索到的图像:", retrieved_images)
print("检索到的音频片段:", retrieved_chunks)

输出如下：

检索到的图像: ['extracted_images/0_page_3.png', 'extracted_images/0_page_12.png']

检索到的音频片段: [" Lemon. Zest the lemon. Over. Smells incredible. ...", " and one third of your protein, chicken. ..."]

对于图像检索，系统会返回存储在向量数据库中的图像路径元数据；而对于音频检索，则会返回与文本查询最相关的转录片段。

接下来，我们使用 Qwen-VL 模型设置生成模型。该模型是一种多模态大型语言模型（LLM），能够处理文本和图像输入，并根据多模态数据生成文本响应。

vl_model = Qwen2VLForConditionalGeneration.from_pretrained(
    "Qwen/Qwen2-VL-7B-Instruct",
    torch_dtype=torch.bfloat16,
).cuda().eval()

min_pixels = 256 * 256
max_pixels = 1024 * 1024
vl_model_processor = Qwen2VLProcessor.from_pretrained(
    "Qwen/Qwen2-VL-7B-Instruct",
    min_pixels=min_pixels,
    max_pixels=max_pixels
)

随后，我们将模型配置为接收输入数据，对其进行处理，并生成相应的文本输出：

chat_template = [
    {
        "role": "user",
        "content": [
            {"type": "image", "image": retrieved_images[0]},  # 第一张检索到的图像
            {"type": "image", "image": retrieved_images[1]},  # 第二张检索到的图像
            {"type": "text", "text": query},  # 用户查询
            {"type": "text", "text": "Audio Context: " + " ".join(retrieved_chunks)}  # 包含音频数据
        ],
    }
]

text = vl_model_processor.apply_chat_template(
    chat_template, tokenize=False, add_generation_prompt=True
)

image_inputs, _ = process_vision_info(chat_template)
inputs = vl_model_processor(
    text=[text],
    images=image_inputs,
    padding=True,
    return_tensors="pt",
).to("cuda")

generated_ids = vl_model.generate(**inputs, max_new_tokens=100)
generated_ids_trimmed = [
    out_ids[len(in_ids):] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
]
output_text = vl_model_processor.batch_decode(
    generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
)

print(output_text[0])

输出示例：

在该食谱中最健康的食材是：

1. **柠檬** - 提供清新的柑橘风味，同时富含维生素 C。
2. **帕尔马干酪** - 富含钙和蛋白质。
3. **鸡肉** - 优质瘦蛋白，含必需氨基酸。
4. **橄榄油** - 健康脂肪，富含单不饱和脂肪酸。
5. **果皮** - 增添风味。

如你所见，系统的输出综合考虑了图像和音频数据。

至此，我们已完成多模态 RAG 系统的构建。后续可以根据实际需求，对文件和代码进行相应调整。

总结

检索增强生成（RAG）是一种通过利用外部知识来增强大型语言模型（LLM）输出的框架。在多模态 RAG 系统中，我们不仅使用文本数据，还结合了图像和音频等多种数据类型。

在本文中，我们实现了一个基于 文本、音频和图像 的多模态 RAG 系统，所使用的核心技术包括：

• CLIP：
  用于生成图像嵌入

• Whisper：
  用于音频转录

• SentenceTransformer：
  用于文本嵌入

• ChromaDB：
  用于向量存储

• Qwen-VL：
  用于多模态文本生成