文字、图片、表格一锅端：RAG 多模态检索融合的工程落地

文字、图片、表格一锅端：RAG 多模态检索融合的工程落地

一、纯文本检索的天花板：企业知识库不止有文档

传统 RAG 系统的假设是"知识以文本形式存在"。但企业真实知识库中，大量关键信息以图片、表格、PDF 截图的形式存在。财务报表的核心数据在 Excel 截图里，产品规格书的关键参数在架构图中，培训材料的知识点在 PPT 截图里。

纯文本 RAG 对这些内容完全失明。用户问"Q3 毛利率是多少"，系统在文本语料里找不到答案，因为毛利率数据只存在于一张财务报表截图中。这不是检索精度的问题，而是检索覆盖面的结构性缺失。

多模态 RAG 的目标是将文本、图片、表格统一纳入检索范围，让用户无论通过文字还是图片提问，都能跨模态找到答案。这不是把多个单模态检索拼在一起，而是需要在 Embedding 层面实现跨模态语义对齐。

二、跨模态语义对齐：从异构数据到统一向量空间

多模态 RAG 的核心技术挑战是：文本、图片、表格三种模态的数据，如何映射到同一个向量空间中，使得语义相近的内容（无论模态）在向量空间中距离接近。

graph LR subgraph 数据摄入层 T[文本文档] --> TS[文本分块<br/>Chunk + Metadata] I[图片/PDF截图] --> IS[图片描述生成<br/>VLM Caption] TB[表格/Excel] --> TBS[表格序列化<br/>Markdown/JSON] end subgraph Embedding 层 TS --> TE[文本 Embedding<br/>text-embedding-3] IS --> IE[图片 Embedding<br/>CLIP / 多模态模型] TBS --> TBE[表格 Embedding<br/>文本化后用文本模型] end subgraph 统一向量空间 TE --> VS[向量数据库<br/>Milvus / Qdrant] IE --> VS TBE --> VS end subgraph 检索与融合层 Q[用户查询] --> QE[查询 Embedding] QE --> VS VS --> RR[多路召回 + RRF 融合] RR --> LLM[大模型生成] end style VS fill:#e8f5e9 style RR fill:#fff3e0

文本处理：标准 RAG 流程，分块（Chunking）+ 文本 Embedding。关键点是分块粒度——太粗会引入噪声，太细会丢失上下文。建议按语义段落分块，每块 300-500 Token，保留 50 Token 重叠。

图片处理：这是多模态 RAG 的核心差异点。有两种方案：方案一，用视觉语言模型（VLM）为图片生成文本描述（Caption），然后对描述做文本 Embedding；方案二，用 CLIP 等多模态模型直接生成图片 Embedding。方案一的优势是描述可读、可调试，方案二的优势是保留视觉细节。生产环境建议两者结合——主索引用 Caption 文本 Embedding，辅助索引用图片原生 Embedding。

表格处理：表格是半结构化数据，不能简单按行切分。推荐方案是将表格序列化为 Markdown 或 JSON 格式，保留行列结构，然后作为文本做 Embedding。对于大型表格，按行分组序列化，每组附带表头信息。

三、多模态 RAG 系统实现

3.1 统一文档模型

from dataclasses import dataclass, field from typing import Optional from enum import Enum import hashlib import time class ModalityType(Enum): TEXT = "text" IMAGE = "image" TABLE = "table" @dataclass class DocumentChunk: """统一的文档分块模型，支持多模态""" chunk_id: str modality: ModalityType # 模态类型 content: str # 文本内容 / 图片 Caption / 表格序列化 raw_content: Optional[str] # 原始内容（图片 URL / 原始表格） source_doc_id: str # 来源文档 ID page_number: int = 0 # 页码（PDF 场景） metadata: dict = field(default_factory=dict) embedding: list[float] = field(default_factory=list) # 多模态专用字段 image_caption: str = "" # VLM 生成的图片描述 table_headers: list[str] = field(default_factory=list) # 表格列头 def compute_chunk_id(self) -> str: """基于内容生成唯一 ID，避免重复入库""" raw = f"{self.modality.value}:{self.source_doc_id}:{self.content[:200]}" return hashlib.md5(raw.encode()).hexdigest()[:16]

3.2 多模态文档摄入管线

import base64 from pathlib import Path class MultiModalIngestionPipeline: """多模态文档摄入管线""" def __init__(self, vlm_client, text_embedder, image_embedder, vector_store): self.vlm_client = vlm_client # 视觉语言模型客户端 self.text_embedder = text_embedder # 文本 Embedding 模型 self.image_embedder = image_embedder # 图片 Embedding 模型 self.vector_store = vector_store # 向量数据库 def ingest_document(self, doc_path: str) -> list[DocumentChunk]: """摄入单个文档，自动识别模态并处理""" path = Path(doc_path) chunks = [] if path.suffix == ".pdf": chunks = self._process_pdf(doc_path) elif path.suffix in (".png", ".jpg", ".jpeg"): chunks = self._process_image(doc_path) elif path.suffix in (".xlsx", ".csv"): chunks = self._process_table(doc_path) elif path.suffix in (".md", ".txt"): chunks = self._process_text(doc_path) else: raise ValueError(f"不支持的文件格式: {path.suffix}") # 批量生成 Embedding 并入库 self._embed_and_store(chunks) return chunks def _process_image(self, image_path: str) -> list[DocumentChunk]: """处理图片：生成 Caption + 双路 Embedding""" # 1. 调用 VLM 生成图片描述 with open(image_path, "rb") as f: image_b64 = base64.b64encode(f.read()).decode() caption = self.vlm_client.describe_image( image_b64=image_b64, prompt="请详细描述这张图片中的所有文字、数据和关键信息，" "包括表格内容、数值、标签等。" ) # 2. 构建文档分块 chunk = DocumentChunk( chunk_id="", modality=ModalityType.IMAGE, content=caption, # 用 Caption 作为检索内容 raw_content=image_path, # 保留原始图片路径 source_doc_id=Path(image_path).stem, image_caption=caption, ) chunk.chunk_id = chunk.compute_chunk_id() return [chunk] def _process_table(self, table_path: str) -> list[DocumentChunk]: """处理表格：序列化为 Markdown 格式""" import pandas as pd df = pd.read_excel(table_path) if table_path.endswith(".xlsx") \ else pd.read_csv(table_path) headers = df.columns.tolist() chunks = [] # 大表格按行分组，每组附带表头 group_size = 10 # 每 10 行一组 for start in range(0, len(df), group_size): end = min(start + group_size, len(df)) subset = df.iloc[start:end] # 序列化为 Markdown 表格，保留结构 md_table = subset.to_markdown(index=False) # 在表格前添加表头上下文 full_content = f"表格列: {', '.join(headers)}\n行 {start+1}-{end}:\n{md_table}" chunk = DocumentChunk( chunk_id="", modality=ModalityType.TABLE, content=full_content, raw_content=md_table, source_doc_id=Path(table_path).stem, table_headers=headers, metadata={"row_start": start, "row_end": end}, ) chunk.chunk_id = chunk.compute_chunk_id() chunks.append(chunk) return chunks def _process_text(self, text_path: str) -> list[DocumentChunk]: """处理纯文本文档：按语义段落分块""" with open(text_path, "r", encoding="utf-8") as f: text = f.read() chunks = [] # 简单按段落分块，生产环境建议用递归字符分割器 paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()] for i, para in enumerate(paragraphs): chunk = DocumentChunk( chunk_id="", modality=ModalityType.TEXT, content=para, raw_content=para, source_doc_id=Path(text_path).stem, metadata={"paragraph_index": i}, ) chunk.chunk_id = chunk.compute_chunk_id() chunks.append(chunk) return chunks def _embed_and_store(self, chunks: list[DocumentChunk]) -> None: """批量生成 Embedding 并写入向量数据库""" for chunk in chunks: if chunk.modality == ModalityType.IMAGE and chunk.raw_content: # 图片：双路 Embedding（Caption 文本 + 图片原生） text_emb = self.text_embedder.embed(chunk.content) # image_emb = self.image_embedder.embed(chunk.raw_content) # 生产环境可将两个向量分别存入不同集合 chunk.embedding = text_emb else: # 文本/表格：统一用文本 Embedding chunk.embedding = self.text_embedder.embed(chunk.content) self.vector_store.upsert(chunk)

3.3 多路召回与 RRF 融合

class MultiModalRetriever: """多模态检索器：多路召回 + Reciprocal Rank Fusion""" def __init__(self, vector_store, text_embedder, image_embedder, top_k: int = 5): self.vector_store = vector_store self.text_embedder = text_embedder self.image_embedder = image_embedder self.top_k = top_k def retrieve(self, query: str, top_k: int = None) -> list[DocumentChunk]: """多路召回 + RRF 融合排序""" k = top_k or self.top_k # 路径1：文本查询 -> 文本/表格索引 text_query_emb = self.text_embedder.embed(query) text_results = self.vector_store.search( embedding=text_query_emb, top_k=k * 2, # 多召回一些，融合后截断 filter={"modality": ["text", "table"]}, ) # 路径2：文本查询 -> 图片 Caption 索引 image_results = self.vector_store.search( embedding=text_query_emb, top_k=k, filter={"modality": "image"}, ) # RRF 融合：按排名倒数加权，避免分数尺度不一致 rrf_scores = {} rrf_k = 60 # RRF 平滑参数，值越大排名差异越平滑 for rank, chunk in enumerate(text_results): rrf_scores[chunk.chunk_id] = rrf_scores.get(chunk.chunk_id, 0) + \ 1.0 / (rrf_k + rank + 1) for rank, chunk in enumerate(image_results): rrf_scores[chunk.chunk_id] = rrf_scores.get(chunk.chunk_id, 0) + \ 1.0 / (rrf_k + rank + 1) # 按融合分数排序 all_chunks = {c.chunk_id: c for c in text_results + image_results} sorted_ids = sorted(rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True) return [all_chunks[cid] for cid in sorted_ids[:k]]

四、多模态融合的代价：管线复杂度与对齐精度

多模态 RAG 不是免费的午餐，它在扩展检索覆盖面的同时，引入了显著的工程复杂度。

VLM Caption 的信息损失。视觉语言模型生成的描述不可能 100% 还原图片中的所有细节。一张包含 50 个数据点的折线图，VLM 可能只描述了趋势，丢失了具体数值。解决方案是：对关键图片做 OCR 补充，将 OCR 文本与 Caption 合并后做 Embedding。但这又增加了 OCR 的准确率依赖。

表格序列化的语义断裂。将表格转为 Markdown 后，行列关系被扁平化为文本，模型可能无法正确理解"第 3 行第 2 列"的语义。对于需要精确单元格查询的场景（如"Q3 毛利率"），建议同时维护结构化查询路径（SQL / Pandas），让 LLM 根据查询意图选择检索还是结构化查询。

多路召回的延迟叠加。文本检索和图片检索串行执行时，总延迟等于两者之和。并行执行可以降低延迟，但需要更多的计算资源。生产环境中，建议文本检索走主路径（低延迟），图片检索走异步补充路径（可容忍 200-500ms 延迟）。

向量空间对齐的精度。CLIP 模型在图文对齐上表现良好，但对专业领域（医疗影像、工程图纸）的对齐精度不足。领域场景下需要用领域数据微调多模态 Embedding 模型，这又引入了标注成本和训练开销。

适用边界：如果知识库 90% 以上是纯文本，不需要多模态 RAG，传统文本 RAG 足够；如果知识库包含大量图片和表格（如产品手册、财务报告），多模态 RAG 是必要的工程投入。

五、总结

多模态 RAG 的核心价值在于打破"纯文本检索"的覆盖面限制，将图片、表格纳入统一检索空间。关键技术路径是：图片走 VLM Caption + 文本 Embedding，表格走序列化 + 文本 Embedding，检索走多路召回 + RRF 融合。

落地路线建议：第一步，在现有文本 RAG 基础上新增图片摄入管线，用 VLM 生成 Caption 并入库；第二步，新增表格摄入管线，将 Excel/CSV 序列化为 Markdown 格式；第三步，实现多路召回 + RRF 融合排序，替代单路检索；第四步，对关键图片补充 OCR 文本，提升数值类查询的召回精度；第五步，根据查询意图路由——精确数据查询走结构化查询路径，语义查询走向量检索路径。

多模态 RAG 的投入产出比取决于知识库中非文本内容的占比。在决策前，先统计知识库的模态分布，再决定是否值得引入多模态管线。