How to Build a Multi-Modal RAG Pipeline for Images and Documents in Python

如何使用Python构建多模态RAG管道处理图像和文档 - MachineLearningMastery.com

如何使用Python构建多模态RAG管道处理图像和文档

By

Vinod Chugani

on

2026年10月21日

in

人工智能

0

分享

发布

在本文中，你将学习如何使用Python构建一个多模态RAG管道，该管道能够从复杂文档中提取并推理文本和图像内容。

我们将涵盖的主题包括：

• 如何使用unstructured库将PDF解析为独立的文本和图像元素。
• 如何使用GPT-4o为提取的图像生成可搜索的文本摘要，并将所有内容存储在多向量检索系统中。
• 如何使用包含文本片段和原始图像的多模态提示查询管道，通过GPT-4o获得基于证据的最终答案。

标准的检索增强生成（RAG）在处理纯文本时效果很好。但如果输入的是包含嵌入图表、扫描图示或数据表格的PDF，情况会迅速变得复杂。基于文本的分块会破坏表格格式并跳过图像，导致你的管道只能基于文档的不完整视图回答问题。

多模态RAG管道通过将图像视为与文本同等重要的可检索对象来解决这个问题。在本教程中，我们将使用LangChain、unstructured库和OpenAI API在Python中构建一个管道。我们将解析复杂PDF文件中的文本和图像元素，为每个图像生成可搜索的摘要，将所有内容存储在多向量检索系统中，并使用包含文本和视觉证据的提示查询该系统，最终通过GPT-4o获得答案。

如KDnuggets的《多模态AI指南》所述，现代AI系统设计为能够原生跨内容类型进行推理。本教程将这一原则应用于实际问题：混合文本、表格和嵌入图像的文档。虽然MLM的多模态RAG实现采用适合音频和图像密集型任务的本地PyTorch方法，但本指南使用基于API的架构，可在标准硬件上运行，并且非常适合商业文档使用场景。

先决条件和设置

你需要一个OpenAI API密钥并安装以下软件包。unstructured需要poppler-utils系统依赖项来从PDF中提取图像。

pip install langchain langchain-openai unstructured[pdf] chromadb pillow apt-get install -y poppler-utils

1

2

pip

install

langchain

-

openai

unstructured

[

pdf

]

chromadb

pillow

apt

get

y

poppler

utils

在运行任何代码之前，将API密钥设置为环境变量：

import os os.environ["OPENAI_API_KEY"] = "your-api-key-here"

import

os

.

environ

"OPENAI_API_KEY"

=

"your-api-key-here"

该管道包含四个阶段：解析、摘要、存储和检索。每个阶段都对应下面的代码块之一。

第1步：将文档解析为文本和图像元素

unstructured库在此处承担主要工作。其partition_pdf()函数读取文档并按类型对每个元素进行分类。它将段落、表格和嵌入图像分开处理，而不是将整个文件视为字符流。

from unstructured.partition.pdf import partition_pdf raw_elements = partition_pdf( filename="report.pdf", extract_image_block_types=["Image"], extract_image_block_output_dir="./extracted_images", infer_table_structure=True, chunking_strategy="by_title", max_characters=4000, )

3

4

5

6

7

8

9

10

from

partition

partition_pdf

raw_elements

(

filename

"report.pdf"

,

extract_image_block_types

"Image"

extract_image_block_output_dir

"./extracted_images"

infer_table_structure

True

chunking_strategy

"by_title"

max_characters

4000

)

extract_image_block_output_dir 告诉 unstructured 将提取的图像文件保存在哪里，而 extract_image_block_types 控制哪些元素类型会被提取。接下来，将返回的元素分别归类到文本和图像列表中：

text_elements = [] image_paths = [] for el in raw_elements: if "CompositeElement" in str(type(el)): text_elements.append(str(el)) elif "Image" in str(type(el)): image_paths.append(el.metadata.image_path) print(f"Extracted {len(text_elements)} text chunks, {len(image_paths)} images")

text_elements

image_paths

for

el

:

if

"CompositeElement"

str

type

append

elif

metadata

image_path

print

f

"Extracted {len(text_elements)} text chunks, {len(image_paths)} images"

该循环将复合文本块与图像元素分离，保存图像文件路径以供下一步使用。

第 2 步：为每张图像生成可搜索的摘要

基于文本的向量数据库无法索引原始图像，因此每张图像都需要一个能够通过自然语言查询检索的文本表示。这一步会将每张提取的图像传递给 GPT-4o，并使用专门设计的提示生成密集且可搜索的描述。

import base64 from openai import OpenAI client = OpenAI()

base64

client

准备好客户端后，定义一个函数对每张图像进行 base64 编码，并将其与以检索为导向的提示一起发送给 GPT-4o：

def summarize_image(image_path): with open(image_path, "rb") as f: image_data = base64.b64encode(f.read()).decode("utf-8") response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": [ {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_data}"}}, {"type": "text", "text": "Describe this image in detail for use in a search system."} ]}] ) return response.choices[0].message.content

11

12

13

def

summarize_image

with

open

"rb"

as

image_data

b64encode

read

decode

"utf-8"

response

chat

completions

create

model

"gpt-4o"

messages

{

"role"

"user"

"content"

"type"

"image_url"

"url"

"data:image/png;base64,{image_data}"

}

"text"

"Describe this image in detail for use in a search system."

return

choices

message

content

要对所有提取的图像运行该函数：

image_summaries = [summarize_image(p) for p in image_paths]

image_summaries

p

每个摘要将成为其源图像的可搜索代理。原始图像文件单独保存，当查询匹配其摘要时通过引用检索。

关于成本的说明：在索引时，每个提取的图像都会触发一次 GPT-4o API 调用。如果你处理的是包含大量嵌入式标志或图标的大型文档，建议在传递给 summarize_image() 之前过滤掉低于最小文件大小的图像。

第 3 步：构建多向量存储库

这一步的架构与标准 RAG 不同。MultiVectorRetriever 维护两个存储库：一个用于相似性搜索的向量存储库，保存嵌入的文本块和图像摘要；另一个是文档存储库，保存通过唯一 ID 与这些嵌入相关联的原始源内容。

import uuid from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from langchain.storage import InMemoryStore from langchain.retrievers.multi_vector import MultiVectorRetriever from langchain_core.documents import Document embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma(embedding_function=embeddings) docstore = InMemoryStore() retriever = MultiVectorRetriever(vectorstore=vectorstore, docstore=docstore) text_ids = [str(uuid.uuid4()) for _ in text_elements] image_ids = [str(uuid.uuid4()) for _ in image_summaries]

uuid

langchain_openai

OpenAIEmbeddings

langchain_chroma

Chroma

storage

InMemoryStore

retrievers

multi_vector

MultiVectorRetriever

langchain_core

documents

Document

embeddings

"text-embedding-3-small"

vectorstore

embedding_function

docstore

retriever

text_ids

uuid4

_

image_ids

这会初始化两个存储并为每个元素生成唯一ID。向量存储用于保存可搜索的摘要；文档存储用于保存原始内容。通过ID关联后，向两个存储填充数据：

retriever.vectorstore.add_documents( [Document(page_content=t, metadata={"doc_id": i}) for t, i in zip(text_elements + image_summaries, text_ids + image_ids)] ) raw_docs = ( [Document(page_content=t, metadata={"type": "text"}) for t in text_elements] + [Document(page_content=p, metadata={"type": "image"}) for p in image_paths] ) retriever.docstore.mset(list(zip(text_ids + image_ids, raw_docs)))

add_documents

page_content

t

"doc_id"

i

zip

+

raw_docs

mset

list

将原始内容封装在带有显式类型字段的Document对象中，能使第4步的检索逻辑保持清晰。每个ID都会将向量存储中的嵌入摘要与文档存储中的原始Document对应关联。

第4步：使用多模态检索查询流水线

在查询时，检索器会对问题进行嵌入并向文本和图像摘要中查找最接近的匹配。如果匹配指向图像路径，系统会加载原始文件并与检索到的文本一起传递给GPT-4o。GPT-4o随后会基于两种内容类型生成答案。

def query_pipeline(question): docs = retriever.invoke(question) text_content, image_content = [], [] for doc in docs: if doc.metadata.get("type") == "image": with open(doc.page_content, "rb") as f: img_b64 = base64.b64encode(f.read()).decode("utf-8") image_content.append({"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_b64}"}}) else: text_content.append({"type": "text", "text": doc.page_content})

query_pipeline

question

docs

invoke

text_content

image_content

doc

==

img_b64

"data:image/png;base64,{img_b64}"

else

元数据检查会将每个检索到的Document清晰地路由：图像路径会被打开并转换为base64编码，而文本片段则保持原样。该函数随后会将两者组合成一个单一的多模态GPT-4o调用：

response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": text_content + image_content + [{"type": "text", "text": question}]}] ) return response.choices[0].message.content

使用任何关于文档的问题调用该函数：

answer = query_pipeline("What does the revenue chart show for Q3?") print(answer)

answer

"What does the revenue chart show for Q3?"

GPT-4o会在返回最终答案前，对所有检索到的证据（包括文本和图像）进行推理。

结论

该流程处理标准 RAG 无法应对的场景：关键信息存在于图表、示意图或嵌入式图形中的文档。四个阶段依次为：使用 unstructured 解析、使用 GPT-4o 总结图像、使用多向量检索器存储、使用多模态提示查询。这种模式可扩展至研究论文、财务报告、产品手册以及任何混合文本与视觉元素的文档类型。

在此基础上，可通过添加重排序机制提升检索精度、将 Chroma 向量存储持久化到磁盘以支持生产环境使用，或探索 ColPali 等视觉语言模型（该模型通过将文档页面作为图像进行嵌入，可跳过解析步骤）。

更多相关内容

• 使用 Transformers.js 的多模态浏览器 AI……

• 5 个构建优化 RAG 系统的 Python 库

• 如何使用最少资源构建稳健的 RAG 系统

• 构建 RAG 系统的高级技巧

• 让我们构建一个 RAG 驱动的研究论文助手

• 如何（以及为何）从零开始用 Python 构建 AI 代理

/.entry /think