#认证作者激励计划#

传统RAG系统在处理纯文本应用场景中已展现出显著效果，然而现实世界的信息载体往往呈现多模态特征。文档中普遍包含图像、表格、图表等承载关键信息的视觉元素，这些多模态内容的有效处理正是多模态RAG系统的核心价值所在。

多模态RAG最优方案选择

经过系统性研究和实验验证，我们将介绍一个在RAG系统中处理多模态内容的最佳实现方案。该方案在性能表现、准确性指标和实现复杂度之间实现了优化平衡。

图1：多模态RAG系统整体架构图，展示从文档处理到向量化存储的完整工作流程

架构优势分析

架构采用模态特定处理与后期融合相结合的技术路线。相比其他技术方案，该架构具有以下显著优势：

首先，在模态信息保留方面，该方法避免了统一嵌入方法可能导致的模态特有信息丢失问题，通过针对各模态优化的专用工具实现精确的内容类型处理。其次，系统具备良好的灵活性和模块化特征，支持单独组件的升级优化（例如更换更高性能的图像理解模型），而无需重构整个系统架构。

在检索精度方面，研究数据表明，该方法在处理复杂多模态查询时的性能相比统一方法提升23%。同时，该架构基于广泛可用的开源工具和模型构建，确保了大多数组织的技术可达性和实施可行性。

多模态文档处理工作流程

以下详细阐述推荐工作流程的各个环节，说明各组件如何协同工作以构建统一的系统架构：

图2：多模态RAG方法的连接工作流程图

1、结构保留的文档分割

该模块的核心功能是将文档分解为可管理的片段，同时保持其逻辑结构和不同内容类型之间的关联关系。

结构感知分割对于系统性能至关重要，它确保相关内容（如图像及其标题）在分割过程中保持关联，这对准确理解和检索具有决定性作用。

import fitz # PyMuPDF 
def split_pdf_by_structure(pdf_path): 
"""根据PDF文档的逻辑结构进行拆分。""" 
doc = fitz.open(pdf_path) 
sections = [] 

# 提取文档结构（简化示例）
toc = doc.get_toc() 
if toc: 
# 使用目录进行结构化拆分
for i, (level, title, page) in enumerate(toc): 
next_page = toc[i+1][2] if i < len(toc)-1 else len(doc) 
section = { 
"title": title, 
"start_page": page-1, # 0 索引
"end_page": next_page-1, 
"level": level 
} 
sections.append(section) 
else: 
# 回退到页面级拆分
for i in range(len(doc)): 
sections.append({ 
"title": f"Page {i+1}", 
"start_page": i, 
"end_page": i, 
"level": 1 
}) 

return sections, doc

研究结果表明，在分割过程中保持文档结构能够显著提升多模态内容的检索质量指标。

2、模态特定内容提取

该模块采用针对特定模态优化的专用工具处理各类内容（文本、图像、表格）。

不同内容类型需要采用相应的处理技术才能有效提取其信息内容，通用方法往往产生次优结果。

def extract_multimodal_content(sections, doc): 
"""使用专用工具从每种模态中提取内容。""" 
extracted_content = [] 

for section in sections: 
section_content = { 
"title": section["title"], 
"level": section["level"], 
"text_elements": [], 
"images": [], 
"tables": [] 
} 

# 处理节中的每个页面
for page_num in range(section["start_page"], section["end_page"] + 1): 
page = doc[page_num] 

# 使用 PyMuPDF 的文本提取功能提取文本
text_blocks = page.get_text("blocks") 
for block in text_blocks: 
if block[6] == 0: # 文本块
section_content["text_elements"].append({ 
"text": block[4], 
"bbox": block[:4], 
"page": page_num 
}) 

# 使用 PyMuPDF 的图像提取功能提取图像
image_list = page.get_images(full=True) 
for img_index, img_info in enumerate(image_list): 
xref = img_info[0] 
base_image = doc.extract_image(xref) 
image_data = { 
"image_data": base_image["image"], 
"extension": base_image["ext"], 
"bbox": page.get_image_bbox(img_info), 
"page": page_num 
} 
section_content["images"].append(image_data) 

# 使用专门的表格提取工具提取表格
# 在此示例中，我们将使用简化方法
tables = extract_tables_from_page(page) 
for table in tables: 
section_content["tables"].append({ 
"data": table, 
"page": page_num 
}) 

extracted_content.append(section_content) 

return extracted_content 

def extract_tables_from_page(page): 
""" 
使用专门的表格检测从页面中提取表格。
在生产系统中，您将使用专用的表格提取
库，如 Camelot、Tabula 或深度学习模型。
""" 
# 为说明目的简化实现
tables = [] 
# 使用启发式或机器学习来识别表格区域
# 然后从这些区域提取结构化数据
return tables

3、关系保留的HTML转换

该模块将提取的多模态内容转换为结构化HTML格式，同时保留内容元素间的关联关系。

HTML作为标准化格式能够有效表示混合模态内容并保持结构完整性，为后续处理提供理想的数据基础。

from bs4 import BeautifulSoup 
import os 
import base64 

def convert_to_structured_html(extracted_content, output_dir): 
"""将提取的多模态内容转换为保留关系的结构化 HTML。""" 
os.makedirs(output_dir, exist_ok=True) 
html_files = [] 

for section in extracted_content: 
# 为此部分创建一个新的 HTML 文档
soup = BeautifulSoup("<article></article>", "html.parser") 
article = soup.find("article") 

# 添加节标题
header = soup.new_tag(f"h{section['level']}") 
header.string = section["title"] 
article.append(header) 

# 按页面和位置对所有元素进行排序
all_elements = [] 

# 添加文本元素
for text_elem in section["text_elements"]: 
all_elements.append({ 
"type": "text", 
"data": text_elem, 
"page": text_elem["page"], 
"y_pos": text_elem["bbox"][1] # 用于排序的 y 坐标
}) 

# 添加图像
for i, img_data_item in enumerate(section["images"]): 
# 将图像保存到文件
img_filename = f"{section['title'].replace(' ', '_')}_img_{i}.{img_data_item['extension']}" 
img_path = os.path.join(output_dir, img_filename) 
with open(img_path, "wb") as f: 
f.write(img_data_item["image_data"]) 

all_elements.append({ 
"type": "image", 
"data": { 
"path": img_path, 
"bbox": img_data_item["bbox"] 
}, 
"page": img_data_item["page"], 
"y_pos": img_data_item["bbox"][1] # 用于排序的 y 坐标
}) 

# 添加表格
for i, table in enumerate(section["tables"]): 
all_elements.append({ 
"type": "table", 
"data": table["data"], 
"page": table["page"], 
"y_pos": 0 # 在生产环境中会使用实际位置
}) 

# 按页面然后按 y 位置对元素进行排序
all_elements.sort(key=lambda x: (x["page"], x["y_pos"])) 

# 按正确顺序将元素添加到 HTML
for elem in all_elements: 
if elem["type"] == "text": 
p = soup.new_tag("p") 
p.string = elem["data"]["text"] 
article.append(p) 

elif elem["type"] == "image": 
figure = soup.new_tag("figure") 
img_tag = soup.new_tag("img", src=elem["data"]["path"]) 
figure.append(img_tag) 

# 查找潜在的标题（图像正下方的文本元素）
idx = all_elements.index(elem) 
if idx + 1 < len(all_elements) and all_elements[idx + 1]["type"] == "text": 
next_elem = all_elements[idx + 1] 
if next_elem["page"] == elem["page"] and next_elem["y_pos"] - elem["y_pos"] < 50: 
# 这段文字很可能是一个标题
figcaption = soup.new_tag("figcaption") 
figcaption.string = next_elem["data"]["text"] 
figure.append(figcaption) 

article.append(figure) 

elif elem["type"] == "table": 
# 将表格数据转换为 HTML 表格
table_tag = soup.new_tag("table") 
for row_data in elem["data"]: 
tr = soup.new_tag("tr") 
for cell in row_data: 
td = soup.new_tag("td") 
td.string = str(cell) 
tr.append(td) 
table_tag.append(tr) 

article.append(table_tag) 

# 保存 HTML 文件
html_filename = f"{section['title'].replace(' ', '_')}.html" 
html_path = os.path.join(output_dir, html_filename) 
with open(html_path, "w", encoding="utf-8") as f: 
f.write(str(soup)) 

html_files.append(html_path) 

return html_files

在实施过程中，建议使用语义HTML5标签（如<figure>、<figcaption>、<table>、<section>）来保留不同内容元素的语义含义，而非仅关注其视觉呈现效果。

4、关系保留的语义分块

HTML转换为多模态内容的标准化表示提供了统一的处理基础，同时保持了结构完整性。

该模块将HTML内容划分为语义完整的片段，同时维护不同元素间的关联关系。

有效的分块策略对检索质量具有决定性影响。过大的块会降低检索精度，而过小的块则会丢失重要的上下文信息。

from bs4 import BeautifulSoup 
import networkx as nx 

def create_semantic_chunks_with_relationships(html_files, max_chunk_size=1000): 
"""创建语义块，同时保留元素之间的关系。""" 
chunks = [] 
relationship_graph = nx.DiGraph() 

for html_file in html_files: 
with open(html_file, "r", encoding="utf-8") as f: 
html_content = f.read() 

soup = BeautifulSoup(html_content, "html.parser") 

# 提取节标题
section_title = soup.find(["h1", "h2", "h3", "h4", "h5", "h6"]).get_text() 
section_id = f"section_{len(chunks)}" 

# 将节节点添加到关系图
relationship_graph.add_node(section_id, type="section", title=section_title) 

# 查找用于分块的语义边界
boundaries = soup.find_all(["h1", "h2", "h3", "h4", "h5", "h6", "section"]) 

if len(boundaries) <= 1: 
# 没有内部分界线，处理整个部分
current_chunk = { 
"id": f"chunk_{len(chunks)}", 
"html": str(soup), 
"text": soup.get_text(separator=" ", strip=True), 
"parent": section_id 
} 
chunks.append(current_chunk) 
relationship_graph.add_node(current_chunk["id"], type="chunk") 
relationship_graph.add_edge(section_id, current_chunk["id"], relation="contains") 
else: 
# 处理每个子部分
for i in range(len(boundaries) - 1): 
start = boundaries[i] 
end = boundaries[i + 1] 

# 收集开始和结束之间的所有元素
elements = [] 
current = start.next_sibling 
while current and current != end: 
if current.name: # 跳过 NavigableString
elements.append(current) 
current = current.next_sibling 

# 从这些元素创建块
if elements: 
chunk_soup = BeautifulSoup("<div></div>", "html.parser") 
chunk_div = chunk_soup.find("div") 

# 添加标题
chunk_div.append(start.copy()) 

# 添加所有元素
for element in elements: 
chunk_div.append(element.copy()) 

# 检查块是否太大
chunk_text = chunk_div.get_text(separator=" ", strip=True) 
if len(chunk_text) > max_chunk_size: 
# 进一步拆分此块
sub_chunks = split_large_chunk(chunk_div, max_chunk_size) 
for sub_chunk in sub_chunks: 
sub_id = f"chunk_{len(chunks)}" 
sub_chunk_obj = { 
"id": sub_id, 
"html": str(sub_chunk), 
"text": sub_chunk.get_text(separator=" ", strip=True), 
"parent": section_id 
} 
chunks.append(sub_chunk_obj) 
relationship_graph.add_node(sub_id, type="chunk") 
relationship_graph.add_edge(section_id, sub_id, relation="contains") 
else: 
# 按原样添加块
chunk_id = f"chunk_{len(chunks)}" 
chunk_obj = { 
"id": chunk_id, 
"html": str(chunk_div), 
"text": chunk_text, 
"parent": section_id 
} 
chunks.append(chunk_obj) 
relationship_graph.add_node(chunk_id, type="chunk") 
relationship_graph.add_edge(section_id, chunk_id, relation="contains") 

# 为图像和表格添加特殊处理，以确保它们正确连接
process_special_elements(soup, chunks, relationship_graph) 

return chunks, relationship_graph 

def split_large_chunk(chunk_div, max_chunk_size): 
"""根据段落将大块拆分为较小的块。""" 
# 为简洁起见，省略了实现细节
return [chunk_div] # 占位符

def process_special_elements(soup, chunks, graph): 
"""处理图像和表格以确保正确的关系。""" 
# 为简洁起见，省略了实现细节
pass

在实施中，建议使用图数据结构显式表示块间关系。这种方法支持更复杂的检索策略，能够沿着关系链路查找相关内容。

5、多模态向量化与存储

该模块将语义块转换为向量表示，并将其存储在向量数据库中以实现高效检索。

不同模态需要采用相应的向量化方法才能有效捕获其语义内容特征。

图3：推荐方法采用模态特定处理和后期融合的技术架构

from sentence_transformers import SentenceTransformer 
from PIL import Image 
import torch 
import chromadb 
import json 

def vectorize_and_store_multimodal_chunks(chunks, relationship_graph, output_dir): 
"""使用特定模态模型对块进行矢量化，并与关系一起存储。""" 
# 初始化嵌入模型
text_embedder = SentenceTransformer("all-MiniLM-L6-v2") 
image_embedder = SentenceTransformer("clip-ViT-B-32") 

# 初始化向量数据库
client = chromadb.Client() 
collection = client.create_collection(name="multimodal_docs") 

# 处理每个块
for chunk in chunks: 
# 解析 HTML
soup = BeautifulSoup(chunk["html"], "html.parser") 

# 提取用于嵌入的文本
text_content = soup.get_text(separator=" ", strip=True) 

# 提取用于多模态嵌入的图像
images = soup.find_all("img") 
image_embeddings = [] 

for img_tag in images: 
try: 
# 加载图像并生成嵌入
img_path = img_tag["src"] 
img_embedding = image_embedder.encode(Image.open(img_path)) 
image_embeddings.append(img_embedding) 
except Exception as e: 
print(f"Error processing image {img_tag.get('src', 'unknown')}: {e}") 

# 生成文本嵌入
text_embedding = text_embedder.encode(text_content) 

# 合并嵌入（简化方法）
# 在生产环境中，您将使用更复杂的融合技术
final_embedding = text_embedding 
if image_embeddings: 
# 平均图像嵌入
avg_img_embedding = sum(image_embeddings) / len(image_embeddings) 
# 与文本嵌入连接并规范化
final_embedding = torch.cat([ 
torch.tensor(text_embedding), 
torch.tensor(avg_img_embedding) 
]).mean(dim=0).numpy() 

# 获取关系元数据
relationships = [] 
for edge in relationship_graph.edges(chunk["id"]): 
source, target = edge 
relationships.append({ 
"source": source, 
"target": target, 
"relation": relationship_graph.edges[edge].get("relation", "related") 
}) 

# 存储在向量数据库中
collection.add( 
ids=[chunk["id"]], 
embeddings=[final_embedding.tolist()], 
metadatas=[{ 
"html_content": chunk["html"], 
"parent": chunk.get("parent", ""), 
"relationships": json.dumps(relationships) 
}], 
documents=[text_content] 
) 

# 保存关系图以供检索
nx.write_gpickle(relationship_graph, f"{output_dir}/relationships.gpickle") 

return collection

对于生产系统，建议考虑使用更复杂的融合方法（如交叉注意力机制或门控融合），以替代简单的串联或平均方法来组合不同模态的嵌入向量。

检索流程：系统集成实现

在完成多模态RAG系统构建后，以下展示其查询处理机制：

def retrieve_multimodal_content(query, collection, relationship_graph, k=5): 
"""根据查询检索相关的多模态内容。""" 
# 分析查询以确定相关模态
query_modalities = analyze_query_modalities(query) 

# 生成查询嵌入
if "image" in query_modalities: 
# 对于有关图像的查询，请使用图像感知嵌入器
query_embedding = image_text_embedder.encode(query) # 假设 image_text_embedder 已定义
else: 
# 对于纯文本查询，请使用文本嵌入器
query_embedding = text_embedder.encode(query) # 假设 text_embedder 已定义

# 执行初始检索
results = collection.query( 
query_embeddings=[query_embedding.tolist()], 
n_results=k 
) 

# 利用关系感知增强结果
enhanced_results = enhance_with_relationships( 
results, relationship_graph, query, collection 
) 

return enhanced_results 

def analyze_query_modalities(query): 
"""分析查询以确定其针对的模态。""" 
# 基于关键字的简单方法
image_keywords = ["image", "picture", "photo", "figure", "diagram", "chart"] 
table_keywords = ["table", "data", "row", "column", "cell"] 

modalities = ["text"] 

if any(keyword in query.lower() for keyword in image_keywords): 
modalities.append("image") 

if any(keyword in query.lower() for keyword in table_keywords): 
modalities.append("table") 

return modalities 

def enhance_with_relationships(results, graph, query, collection): 
"""使用关系信息增强检索结果。""" 
enhanced_results = [] 
retrieved_ids = set() 

for i, result_id in enumerate(results["ids"][0]): 
retrieved_ids.add(result_id) 
enhanced_results.append({ 
"id": result_id, 
"text": results["documents"][0][i], 
"metadata": results["metadatas"][0][i], 
"score": results["distances"][0][i] if "distances" in results else 1.0 - i/len(results["ids"][0]) 
}) 

# 查找可能相关的相关块
for result in enhanced_results[:]: # 复制以避免在迭代期间修改
# 从元数据中获取关系
relationships = json.loads(result["metadata"].get("relationships", "[]")) 

for rel in relationships: 
related_id = rel["target"] 
if related_id not in retrieved_ids: 
# 检查此相关块是否与查询相关
related_metadata = collection.get(ids=[related_id]) 
if related_metadata and related_metadata["ids"]: 
related_text = related_metadata["documents"][0] 

# 简单相关性检查（在生产环境中会更复杂）
if any(term in related_text.lower() for term in query.lower().split()): 
retrieved_ids.add(related_id) 
enhanced_results.append({ 
"id": related_id, 
"text": related_text, 
"metadata": related_metadata["metadatas"][0], 
"score": result["score"] * 0.9, # 相关内容的得分略低
"relation": "related to " + result["id"] 
}) 

# 按分数排序
enhanced_results.sort(key=lambda x: x["score"], reverse=True) 

return enhanced_results

方法优势对比分析

推荐方案相比其他技术路线在以下关键维度具有显著优势：

在混合模态处理能力方面，通过使用专用工具处理各模态后进行结果整合，能够捕获每种内容类型的独特特征。在关系保留机制上，通过显式建模和保留内容元素间的关系，维护了准确理解和检索所需的上下文信息。

在自适应检索能力方面，检索过程能够根据查询的模态需求进行适应性调整，确保无论内容格式如何都能检索到最相关的信息。在实际可行性层面，该方法基于广泛可用的工具和模型实现，为大多数组织提供了良好的技术可达性。

总结

本文提出的多模态RAG方法采用模态特定处理、后期融合和关系保留的技术架构，在性能表现、准确性指标和实现复杂度之间实现了最佳平衡。通过遵循该技术路线，能够构建一个有效处理复杂文档中全部信息的RAG系统。

在后续研究中，我们将重点探讨多模态RAG系统从实验阶段向生产就绪阶段的迁移方法，着重关注系统可扩展性、监控机制和持续优化策略等关键技术问题。

参考文献

Lang Mei, Siyu Mo, Zhihan Yang, Chong Chen. “A Survey of Multimodal Retrieval-Augmented Generation.” arXiv:2504.08748, March 2025.

“Multimodal Queries Require Multimodal RAG: Researchers from KAIST and DeepAuto.ai Propose UniversalRAG.” MarkTechPost, May 2025.

Shah, Suruchi and Dharmapuram, Suraj. “Bridging Modalities: Multimodal RAG for Advanced Information Retrieval.” InfoQ, April 2025.

“How Multimodal RAG unlocks human-like reasoning in real-time.” LinkedIn Pulse, March 2025.

“Trends in Active Retrieval Augmented Generation: 2025 and Beyond.” Signity Solutions, February 2025.

作者：Ashwindevelops