多模态RAG工程实践:图片、表格、文档混合检索的完整方案
你的业务文档里有大量图片、表格、图表,但你的RAG系统只能检索纯文本——这个问题有多严重,取决于你的文档里有多少关键信息藏在图片和表格里。
2026年,多模态RAG已经从研究课题变成了可以落地的工程方案。本文聚焦实践,告诉你怎么处理图片、表格和复杂文档。## 问题的全貌传统RAG的处理流程把文档当纯文本,遇到图片和表格就要么跳过,要么只提取alt text。这意味着:- PDF里的技术图表被完全忽略- Excel表格中的数据关系无法被检索- PPT里的流程图对RAG来说是透明的- 带图片的产品手册,图片中的规格参数全部丢失这些信息往往是最关键的。一个处理制造业文档的RAG系统,如果连图纸都读不懂,说实话价值很有限。## 多模态文档解析### PDF的深度解析用传统的PDF文本提取器(pypdf、pdfminer)只能得到文字,对于富文本PDF远远不够。推荐方案:用视觉语言模型(VLM)对PDF每页截图分析:pythonimport fitz # PyMuPDFimport base64from openai import AsyncOpenAIfrom pathlib import Pathclient = AsyncOpenAI()async def extract_page_content(pdf_path: str, page_num: int) -> dict: """提取PDF单页的完整内容(文字+图表+表格)""" doc = fitz.open(pdf_path) page = doc[page_num] # 高分辨率截图 zoom = 2.0 mat = fitz.Matrix(zoom, zoom) pix = page.get_pixmap(matrix=mat) img_bytes = pix.tobytes("png") img_b64 = base64.b64encode(img_bytes).decode() # 同时提取纯文本(用于快速索引) raw_text = page.get_text() # 用VLM理解整页内容 response = await client.chat.completions.create( model="gpt-4o", messages=[ { "role": "user", "content": [ { "type": "image_url", "image_url": { "url": f"data:image/png;base64,{img_b64}", "detail": "high" } }, { "type": "text", "text": """请完整提取这一页的所有内容,包括:1. 正文文字(保持原有格式和层次)2. 表格(转换为Markdown格式)3. 图表描述(详细说明图表的类型、坐标轴、数据趋势、关键数据点)4. 流程图/架构图(用文字描述各节点和连接关系)5. 公式(用LaTeX格式)输出格式:json{ “text_content”: “正文内容”, “tables”: [{“caption”: “表格标题”, “markdown”: “Markdown表格内容”}], “figures”: [{“type”: “图类型”, “description”: “详细描述”, “key_data”: “关键数据”}], “page_summary”: “本页核心内容一句话总结”}""" } ] } ], max_tokens=4000 ) content = json.loads(response.choices[0].message.content) content["raw_text"] = raw_text content["page_num"] = page_num content["pdf_path"] = pdf_path return contentasync def process_pdf(pdf_path: str) -> list[dict]: """处理整个PDF文档""" doc = fitz.open(pdf_path) tasks = [ extract_page_content(pdf_path, i) for i in range(len(doc)) ] # 并发处理,但控制并发数避免API限流 semaphore = asyncio.Semaphore(3) async def bounded_extract(page_num): async with semaphore: return await extract_page_content(pdf_path, page_num) pages = await asyncio.gather(*[bounded_extract(i) for i in range(len(doc))]) return list(pages)### 图片的向量化存储对于独立的图片,需要同时存储图片向量和文字描述:pythonfrom openai import AsyncOpenAIimport numpy as npclass MultimodalStore: def __init__(self, vector_db): self.db = vector_db self.client = AsyncOpenAI() async def add_image(self, image_path: str, metadata: dict = None) -> str: """添加图片到多模态存储""" with open(image_path, "rb") as f: img_b64 = base64.b64encode(f.read()).decode() # 1. 生成图片描述 description = await self._describe_image(img_b64) # 2. 生成图片的文本向量(基于描述) text_vector = await self._embed_text(description) # 3. 生成图片的视觉向量(如果用clip模型) # visual_vector = clip_model.encode_image(image_path) # 4. 存储 doc_id = self.db.add({ "type": "image", "image_path": image_path, "image_b64": img_b64, "description": description, "vector": text_vector, "metadata": metadata or {} }) return doc_id async def _describe_image(self, img_b64: str) -> str: """生成详细的图片描述,用于文本检索""" response = await self.client.chat.completions.create( model="gpt-4o", messages=[{ "role": "user", "content": [ { "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"} }, { "type": "text", "text": "请详细描述这张图片的内容,包括所有可见的文字、数据、图表元素。描述要详细且精确,用于后续的语义搜索。" } ] }], max_tokens=1000 ) return response.choices[0].message.content### 表格的特殊处理表格数据需要特殊处理——直接向量化Markdown表格效果很差,因为语义主要在行列关系里。pythonclass TableProcessor: async def process_table(self, table_markdown: str, caption: str = "") -> list[dict]: """把表格处理成多个可检索的单元""" chunks = [] # 1. 整表描述(用于概括性查询) table_summary = await self._summarize_table(table_markdown, caption) chunks.append({ "type": "table_summary", "content": table_summary, "table_markdown": table_markdown, "caption": caption }) # 2. 按行/列切分(用于具体数据查询) rows = self._parse_markdown_table(table_markdown) headers = rows[0] if rows else [] for row_idx, row in enumerate(rows[1:], 1): # 每行生成一个自然语言描述 row_text = self._row_to_text(headers, row, caption) chunks.append({ "type": "table_row", "content": row_text, "row_index": row_idx, "data": dict(zip(headers, row)) }) return chunks async def _summarize_table(self, table_md: str, caption: str) -> str: """用LLM生成表格的自然语言摘要""" response = await self.client.chat.completions.create( model="gpt-4o-mini", messages=[{ "role": "user", "content": f"""请用自然语言描述以下表格的内容和关键信息,包括:- 表格主题- 数据范围- 关键的数字/趋势- 重要的行或列表格标题:{caption}表格内容:{table_md}""" }] ) return response.choices[0].message.content def _row_to_text(self, headers: list, row: list, caption: str) -> str: """把表格行转成自然语言描述""" pairs = [f"{h}={v}" for h, v in zip(headers, row) if v.strip()] return f"[{caption}] " + ",".join(pairs)## 混合检索策略处理好文档之后,查询时需要根据问题类型选择合适的检索策略:pythonclass MultimodalRetriever: def __init__(self, text_store, image_store, table_store): self.text = text_store self.images = image_store self.tables = table_store self.router = QueryRouter() async def retrieve(self, query: str, top_k: int = 5) -> list[RetrievedItem]: # 1. 判断查询类型 query_type = await self.router.classify(query) if query_type == "text_only": return await self.text.search(query, top_k=top_k) elif query_type == "table_query": # 表格查询:先找相关表格,再精确检索行 table_results = await self.tables.search(query, top_k=3) return self._expand_table_results(table_results, query) elif query_type == "visual_query": # 视觉查询:同时检索图片和文本 image_results = await self.images.search(query, top_k=3) text_results = await self.text.search(query, top_k=3) return self._merge_results(image_results, text_results) else: # mixed # 全面检索 all_results = await asyncio.gather( self.text.search(query, top_k=3), self.images.search(query, top_k=2), self.tables.search(query, top_k=2) ) return self._rank_and_merge(*all_results, top_k=top_k)class QueryRouter: async def classify(self, query: str) -> str: """判断查询需要哪类信息""" visual_keywords = ["图片", "图表", "图像", "截图", "看起来", "样子", "外观"] table_keywords = ["数据", "统计", "多少", "比较", "排名", "百分比", "数字"] query_lower = query.lower() if any(kw in query_lower for kw in visual_keywords): return "visual_query" elif any(kw in query_lower for kw in table_keywords): return "table_query" # 不确定时用LLM判断 return await self._llm_classify(query)## 回答时的图文混排检索到包含图片的内容后,回答时要同时引用图片:pythonasync def generate_answer_with_images( query: str, retrieved_items: list[RetrievedItem]) -> AnswerWithMedia: # 构建包含图片的上下文 messages = [ {"role": "system", "content": "你是一个知识库助手,请基于提供的材料(包括文字和图片)回答问题。"} ] user_content = [{"type": "text", "text": f"问题:{query}\n\n参考材料:\n"}] for i, item in enumerate(retrieved_items, 1): if item.type == "text": user_content.append({ "type": "text", "text": f"[文字材料{i}] {item.content}" }) elif item.type == "image": user_content.extend([ {"type": "text", "text": f"[图片材料{i}]"}, { "type": "image_url", "image_url": { "url": f"data:image/png;base64,{item.image_b64}", "detail": "high" } } ]) elif item.type == "table": user_content.append({ "type": "text", "text": f"[表格材料{i}]\n{item.table_markdown}" }) user_content.append({ "type": "text", "text": "\n请基于以上材料回答问题,如果引用了图片中的信息,请说明是来自哪张图片。" }) messages.append({"role": "user", "content": user_content}) response = await client.chat.completions.create( model="gpt-4o", messages=messages, max_tokens=2000 ) return AnswerWithMedia( text=response.choices[0].message.content, referenced_images=[item for item in retrieved_items if item.type == "image"] )## 成本控制多模态处理的成本显著高于纯文本RAG,几个控制策略:分级处理:不是所有文档都需要VLM全量解析。先用规则判断文档类型,只对"有大量图表"的文档做视觉解析。缓存描述:图片描述一旦生成,永久缓存,不要每次查询都重新生成。按需加载图片:检索时只返回图片的描述文字和路径,只有在生成最终回答时才加载图片base64,减少不必要的数据传输。小模型初筛:用便宜的模型做第一轮相关性过滤,只把最相关的结果送给贵的模型做最终处理。—本文关键词:多模态RAG、视觉语言模型、PDF解析、表格处理、图文检索
