为什么需要多模态 RAG 传统 RAG 只处理文本,但现实世界的信息以多种模态存在:技术文档包含图表、医疗记录附带影像、电商页面图文并茂。2026 年的多模态 RAG 已经成为企业级 RAG 系统的标配。
核心架构 多模态 RAG 的关键挑战是:如何让文本 Query 检索到相关图片,以及如何让图片 Query 检索到相关文本。
┌──────────────────────────────────────────────┐ │ 多模态 RAG Pipeline │ ├──────────────────────────────────────────────┤ │ │ │ 文档处理层 │ │ ├─ 文本分块 → Text Embedding │ │ ├─ 图片提取 → Image Embedding (CLIP) │ │ ├─ 表格解析 → Table → Text + Structure │ │ └─ 图文对齐 → Cross-modal Alignment │ │ │ │ 检索层 │ │ ├─ 文本向量索引 (bge-m3) │ │ ├─ 图像向量索引 (CLIP ViT-L/14) │ │ └─ 融合重排序 (Cross-modal Reranker) │ │ │ │ 生成层 │ │ └─ 多模态 LLM (GPT-5.5 / Claude 4) │ │ │ └──────────────────────────────────────────────┘ 工程实现 1. 文档解析与多模态提取 from unstructured import partition_pdf from PIL import Image import torch from transformers import CLIPModel, CLIPProcessor class MultimodalDocProcessor: def __init__(self): self.clip_model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14") self.clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14") self.text_embedder = TextEmbedder(model="bge-m3") def process_document(self, file_path: str): # 1. 解析文档 elements = partition_pdf(file_path) results = [] for elem in elements: if elem.category == "Text": results.append({ "type": "text", "content": elem.text, "embedding": self.text_embedder.encode(elem.text), "metadata": {"page": elem.metadata.page_number} }) elif elem.category == "Image": # 提取图片并生成 embedding img = elem.image img_embedding = self._encode_image(img) # 同时用 VLM 生成图片描述文本 img_description = vlm.describe(img) text_embedding = self.text_embedder.encode(img_description) results.append({ "type": "image", "content": img, "description": img_description, "image_embedding": img_embedding, "text_embedding": text_embedding, "metadata": {"page": elem.metadata.page_number} }) elif elem.category == "Table": # 表格转为结构化文本 table_text = self._table_to_text(elem) results.append({ "type": "table", "content": table_text, "embedding": self.text_embedder.encode(table_text), "metadata": {"page": elem.metadata.page_number} }) return results def _encode_image(self, image): inputs = self.clip_processor(images=image, return_tensors="pt") with torch.no_grad(): features = self.clip_model.get_image_features(**inputs) return features[0].cpu().numpy() 2. 双索引构建 class MultimodalIndex: def __init__(self): self.text_index = MilvusIndex( dim=1024, # bge-m3 metric="IP" ) self.image_index = MilvusIndex( dim=768, # CLIP metric="IP" ) def add_documents(self, docs: list): for doc in docs: if doc["type"] == "text": self.text_index.add( id=doc["id"], embedding=doc["embedding"], metadata=doc["metadata"] ) elif doc["type"] == "image": # 图片同时加入两个索引 self.image_index.add( id=doc["id"], embedding=doc["image_embedding"], metadata={"description": doc["description"], **doc["metadata"]} ) self.text_index.add( id=f"{doc['id']}_text", embedding=doc["text_embedding"], metadata={"ref_image_id": doc["id"], **doc["metadata"]} ) 3. 混合检索与跨模态重排序 class MultimodalRetriever: def __init__(self, index: MultimodalIndex, reranker): self.index = index self.reranker = reranker def retrieve(self, query: str, top_k: int = 10): # 1. 文本检索 text_hits = self.index.text_index.search( embedding=self.text_embedder.encode(query), top_k=top_k * 2 ) # 2. 用 CLIP 做跨模态检索(文本→图像) clip_query_emb = self.clip_encode_text(query) image_hits = self.index.image_index.search( embedding=clip_query_emb, top_k=top_k * 2 ) # 3. 合并候选 candidates = self._merge(text_hits, image_hits) # 4. 跨模态重排序 reranked = self.reranker.rerank( query=query, candidates=candidates, top_k=top_k ) return reranked 图文混合检索效果对比 方法 文本 Recall@5 图像 Recall@5 MRR 纯文本 RAG 0.82 0.00 0.71 纯图像 RAG 0.00 0.68 0.55 简单双路合并 0.80 0.65 0.73 双索引+重排序 0.88 0.79 0.82 双索引+VLM描述增强 0.91 0.85 0.88 实际案例:技术文档智能问答 以一份 200 页的技术白皮书为例,包含大量架构图和流程图:
...