RAG 多模态检索入门：图片、表格和结构化数据如何统一进检索管道-拓冰建站

RAG 多模态检索入门：图片、表格和结构化数据如何统一进检索管道

一、深度引言与场景痛点

传统的 RAG 系统有一个致命的盲区：它只认识文字。当你问它"这张架构图里的数据流是怎样的"，它一脸茫然。当你问"财报表格中 Q3 的净利润增长率"，它只能给你一段从文档里搜到的、可能与表格内容风马牛不相及的文字描述。

现实世界的信息是多种形态的——PDF 里有图表、PPT 里有架构图、网页里有表格和代码块、数据库里有结构化记录。如果你只做纯文本检索，意味着你的 RAG 系统戴着"文字眼罩"在处理多模态信息。它会把表格当作无意义的字符串拼接，把图片的 alt 文本当作图片的全部内容，把数值关系的信息丢失殆尽。

多模态 RAG 的核心挑战不是"能不能处理"，而是"如何把异质的模态统一到同一个检索管道中"。我们需要一个范式——无论是图片、表格还是纯文本，最终都能被表示成可比较的向量，参与语义检索。

二、底层机制与原理深度剖析

多模态 RAG 的检索管道包含四个关键步骤：

内容解析与分离：首先要把混合文档中的不同模态内容拆开。对于 PDF，用布局分析模型识别表格区域、图片区域和文本段落。Unstructured、LlamaParse 等工具可以自动完成这步。

模态适配编码：纯文本用文本 Embedding 模型编码；图片用 CLIP 等视觉模型编码成图像向量；表格需要特殊处理——把表格的语义信息提取成文本描述，再用文本 Embedding 编码。

多向量融合检索：多种模态的向量存在于不同的语义空间中，直接做相似度比较毫无意义。核心策略是"数据桥接"——通过文本描述作为中间桥梁，将视觉、结构化信息都转换到文本语义空间中。

上下文拼接：检索到不同模态的内容后，需要按合理的方式组织上下文。比如图片配说明文字，表格转 Markdown，结构化数据格式化成键值对。

flowchart TB subgraph "文档解析层" PDF["原始 PDF/PPT 文档"] --> LP["LlamaParse\n布局分析"] LP --> IMG["图片区域"] LP --> TBL["表格区域"] LP --> TXT["文本段落"] end subgraph "多模态编码层" IMG -->|"CLIP/ViT"| IV["图像向量"] TXT -->|"Text Embedding"| TV["文本向量"] TBL -->|"表格→文本描述"| TD["表格文本"] TD -->|"Text Embedding"| TBV["表格向量"] end subgraph "统一检索层" IV --> MD["多向量索引<br/>Milvus/Qdrant"] TV --> MD TBV --> MD Q["用户查询"] -->|"Text Embedding"| QV["查询向量"] QV --> MD MD -->|"多路召回"| RR["检索结果"] end subgraph "上下文组织层" RR --> ORG["模态感知拼接器"] ORG -->|"图片+描述"| CTX["组织后的上下文"] ORG -->|"表格转Markdown"| CTX ORG -->|"纯文本"| CTX CTX --> LLM["多模态 LLM\n生成回答"] end style PDF fill:#4A90D9,color:#fff style MD fill:#E8A838,color:#fff style LLM fill:#D9534F,color:#fff

这种架构的关键洞察是：用文本描述作为不同模态之间的"通用货币"。图片不需要直接和文本向量做相似度计算，而是让图片被一个描述性的文本片段所代表。

三、生产级代码实现

下面的代码展示了如何构建一个多模态文档处理管道，包括图片描述生成、表格提取和向量化入库。

import asyncio import base64 import hashlib import logging from dataclasses import dataclass, field from enum import Enum from pathlib import Path from llama_index.core import Document, VectorStoreIndex, Settings from llama_index.core.schema import ImageDocument, TextNode from llama_index.multi_modal_llms.openai import OpenAIMultiModal from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.llms.openai import OpenAI from llama_parse import LlamaParse from PIL import Image import io logger = logging.getLogger(__name__) class ContentType(Enum): TEXT = "text" IMAGE = "image" TABLE = "table" @dataclass class MultiModalChunk: """多模态内容块——统一的数据表示。""" content_type: ContentType raw_content: str embedding_text: str # 用于向量化的文本描述 metadata: dict = field(default_factory=dict) image_bytes: bytes | None = None class MultiModalDocumentProcessor: """多模态文档处理器：解析、编码、向量化入库一体化。""" def __init__( self, llm=None, embed_model=None, multimodal_llm=None, ): self.llm = llm or OpenAI(model="gpt-4o-mini") self.embed_model = embed_model or OpenAIEmbedding() self.multimodal_llm = multimodal_llm or OpenAIMultiModal(model="gpt-4o") async def _describe_image(self, image_bytes: bytes, context: str = "") -> str: """用多模态 LLM 为图片生成文本描述，作为桥接向量。""" try: image_b64 = base64.b64encode(image_bytes).decode("utf-8") prompt = ( "请详细描述这张图片的内容。如果是架构图，描述组件和数据流向；" "如果是图表，描述数据趋势和关键数值；如果是普通图片，描述视觉内容。" ) if context: prompt = f"上下文：{context}\n{prompt}" response = await self.multimodal_llm.acomplete( prompt=prompt, image_documents=[ImageDocument(image=image_b64)], ) description = str(response).strip() logger.info("图片描述生成成功，长度=%d", len(description)) return description except Exception as e: logger.error("图片描述生成失败: %s", e) return f"[图片描述生成失败: {type(e).__name__}] 上下文: {context}" @staticmethod def _describe_table(table_text: str) -> str: """将表格结构化数据转为可检索的自然语言描述。""" rows = [r.strip() for r in table_text.strip().split("\n") if r.strip()] if len(rows) < 2: return f"表格内容: {table_text[:200]}" header = rows[0] key_values = [] for row in rows[1:]: parts = [p.strip() for p in row.split("|")] if len(parts) >= 2: key_values.append(f"{parts[0]}: {', '.join(parts[1:])}") return f"表格标题行: {header}。数据: {'; '.join(key_values[:20])}" async def process_document(self, file_path: str) -> list[MultiModalChunk]: """处理混合文档，输出多模态内容块列表。""" chunks: list[MultiModalChunk] = [] file_path_obj = Path(file_path) if not file_path_obj.exists(): raise FileNotFoundError(f"文件不存在: {file_path}") try: # 使用 LlamaParse 解析文档结构 parser = LlamaParse(result_type="markdown") documents = await parser.aload_data(file_path) for doc in documents: # 处理文本部分 if doc.text and doc.text.strip(): chunks.append(MultiModalChunk( content_type=ContentType.TEXT, raw_content=doc.text, embedding_text=doc.text[:2000], metadata={"source": str(file_path_obj.name)}, )) # 处理图片（需要额外提取） if hasattr(doc, 'image_nodes') and doc.image_nodes: for img_node in doc.image_nodes: if img_node.image: description = await self._describe_image( img_node.image, context=doc.text[:500], ) chunks.append(MultiModalChunk( content_type=ContentType.IMAGE, raw_content=description, embedding_text=description, metadata={ "source": str(file_path_obj.name), "content_type": "image", }, image_bytes=img_node.image, )) except FileNotFoundError: raise except Exception as e: logger.error("文档解析失败 [%s]: %s", file_path, e) raise RuntimeError(f"文档解析异常: {file_path}") from e logger.info( "文档 [%s] 解析完成: 文本=%d, 图片=%d, 总计=%d", file_path_obj.name, sum(1 for c in chunks if c.content_type == ContentType.TEXT), sum(1 for c in chunks if c.content_type == ContentType.IMAGE), len(chunks), ) return chunks async def index_chunks(self, chunks: list[MultiModalChunk]) -> VectorStoreIndex: """将多模态内容块向量化并构建检索索引。""" if not chunks: raise ValueError("内容块列表为空，无法构建索引") documents = [] for chunk in chunks: metadata = { **chunk.metadata, "content_type": chunk.content_type.value, } doc = Document( text=chunk.embedding_text, metadata=metadata, ) documents.append(doc) try: index = VectorStoreIndex.from_documents( documents, embed_model=self.embed_model, ) logger.info("多模态索引构建成功，文档数=%d", len(documents)) return index except Exception as e: logger.error("索引构建失败: %s", e) raise async def demo(): processor = MultiModalDocumentProcessor() # 处理混合文档 try: chunks = await processor.process_document("/path/to/report.pdf") for chunk in chunks: print(f"[{chunk.content_type.value}] {chunk.embedding_text[:100]}...") # 构建索引 index = await processor.index_chunks(chunks) # 检索测试 retriever = index.as_retriever(similarity_top_k=5) nodes = await retriever.aretrieve("架构图中的数据流向是怎样的？") for node in nodes: print(f"检索命中 [{node.metadata.get('content_type')}]: {node.text[:80]}...") except FileNotFoundError as e: logger.error("文件未找到: %s", e) except Exception as e: logger.exception("多模态检索流程异常") if __name__ == "__main__": asyncio.run(demo())