RAG 多模态检索方案：图片 + 文本的联合 Embedding 工程实践

RAG 多模态检索方案图片 文本的联合 Embedding 工程实践一、用户上传了截图问这个报错怎么解决系统检索不到用户上传了一张报错截图并问这个怎么解决。RAG 系统试图对截图中的文字做 OCR 提取文本但 OCR 准确率只有 80%把NullPointerException识别成了NullPointerExceplion向量检索自然找不到匹配的文档。传统 RAG 只能处理文本。多模态 RAG 让系统能直接理解图片内容而不依赖不完美的 OCR 中间层。核心思路用多模态 Embedding 模型如 CLIP将图片和文本映射到同一个向量空间实现图文混合检索。二、多模态 RAG 的联合检索架构flowchart TD Q[用户查询: 文字图片] -- SPLIT{内容类型} SPLIT --|文字| TE[文本 Embedding 模型] SPLIT --|图片| IE[图像 Embedding 模型: CLIP] TE -- VEC_T[文本向量] IE -- VEC_I[图像向量] VEC_T -- FUSE[向量融合: 加权平均或拼接] VEC_I -- FUSE FUSE -- SEARCH[向量库检索] subgraph Index[多模态知识库索引] T_DOC[文本文档: 文本向量] I_DOC[截图/图片: 图像向量] T_DOC -- IDX[统一向量索引] I_DOC -- IDX end SEARCH -- IDX IDX -- RESULTS[Top-K 多模态结果] RESULTS -- CONTEXT[拼接: 文本图片描述] CONTEXT -- LLM[多模态 LLM 推理] LLM -- ANSWER[回答]关键技术点CLIP 模型确保图文向量在同一个语义空间、向量融合策略平均/加权/拼接影响检索精度、知识库需要同时为图片和文本建立向量索引。三、Python 实现多模态检索import io import base64 import numpy as np from dataclasses import dataclass from typing import Optional, Union from PIL import Image import openai # 多模态文档 dataclass class MultimodalDocument: 多模态文档文本 可选图片 doc_id: str text_content: str image_base64: Optional[str] None # 截图 text_embedding: Optional[np.ndarray] None image_embedding: Optional[np.ndarray] None modality: str text # text, image, mixed # 多模态 Embedding 编码器 class MultimodalEncoder: 多模态编码器文本 图片统一向量化 def __init__(self, clip_model: str ViT-B/32): self.text_dim 1536 # text-embedding-ada-002 self.image_dim 512 # CLIP ViT-B/32 self.fused_dim self.text_dim self.image_dim def encode_text(self, text: str) - np.ndarray: 文本向量化 response openai.Embedding.create( modeltext-embedding-ada-002, inputtext, ) vec np.array(response[data][0][embedding]) return vec / np.linalg.norm(vec) # L2 归一化 def encode_image(self, image_base64: str) - np.ndarray: 图像向量化使用 CLIP # 实际项目中使用 open_clip 或 transformers # import clip # model, preprocess clip.load(ViT-B/32) # image preprocess(Image.open(io.BytesIO(base64.b64decode(image_base64)))) # with torch.no_grad(): # vec model.encode_image(image.unsqueeze(0)) # return vec.numpy().flatten() # 简化模拟 vec np.random.randn(self.image_dim) return vec / np.linalg.norm(vec) def fuse_embeddings( self, text_vec: np.ndarray, image_vec: Optional[np.ndarray] None, strategy: str concat, text_weight: float 0.6, image_weight: float 0.4, ) - np.ndarray: 融合文本和图像向量 strategy: concat: 拼接 [text_vec, image_vec]保留全部信息 avg: 加权平均维度对齐 if image_vec is None: return text_vec if strategy concat: # 拼接后归一化 fused np.concatenate([text_vec * text_weight, image_vec * image_weight]) return fused / np.linalg.norm(fused) elif strategy avg: # 加权平均 fused text_vec * text_weight image_vec * image_weight return fused / np.linalg.norm(fused) return text_vec # 多模态索引 class MultimodalIndex: 多模态向量索引 def __init__(self): self.vectors: list[np.ndarray] [] self.documents: dict[int, MultimodalDocument] {} def add_document(self, doc: MultimodalDocument, fused_vec: np.ndarray): 添加文档到索引 idx len(self.vectors) self.vectors.append(fused_vec) self.documents[idx] doc return idx def search( self, query_vec: np.ndarray, top_k: int 5, ) - list[tuple[MultimodalDocument, float]]: 检索最相似的文档 if not self.vectors: return [] # 批量计算余弦相似度 vectors_matrix np.stack(self.vectors) similarities np.dot(vectors_matrix, query_vec) # Top-K top_indices np.argsort(similarities)[::-1][:top_k] results [] for idx in top_indices: doc self.documents[int(idx)] score float(similarities[int(idx)]) results.append((doc, score)) return results # 多模态 RAG 引擎 class MultimodalRAGEngine: 多模态 RAG 引擎 def __init__(self): self.encoder MultimodalEncoder() self.index MultimodalIndex() def index_document(self, doc: MultimodalDocument): 索引文档 # 计算文本向量 text_vec None if doc.text_content: text_vec self.encoder.encode_text(doc.text_content) doc.text_embedding text_vec # 计算图像向量 image_vec None if doc.image_base64: image_vec self.encoder.encode_image(doc.image_base64) doc.image_embedding image_vec # 融合向量 if image_vec is not None: fused self.encoder.fuse_embeddings(text_vec, image_vec, concat) else: fused text_vec if fused is not None: self.index.add_document(doc, fused) def query( self, text_query: str, image_query: Optional[str] None, top_k: int 5, ) - list[dict]: 多模态查询 # 编码查询 text_vec self.encoder.encode_text(text_query) image_vec None if image_query: image_vec self.encoder.encode_image(image_query) # 融合查询向量 query_vec self.encoder.fuse_embeddings(text_vec, image_vec, concat) # 检索 results self.index.search(query_vec, top_k) return [ { doc_id: doc.doc_id, text: doc.text_content[:200], has_image: doc.image_base64 is not None, score: score, modality: doc.modality, } for doc, score in results ] # 图片预处理 class ImagePreprocessor: 图片预处理工具 staticmethod def preprocess_for_embedding(image_path: str) - str: 预处理图片返回 base64 img Image.open(image_path) # 1. 调整尺寸CLIP 要求 224x224 img img.resize((224, 224), Image.Resampling.LANCZOS) # 2. 转 RGB处理 PNG 透明度等 if img.mode ! RGB: img img.convert(RGB) # 3. 转 base64 buffer io.BytesIO() img.save(buffer, formatJPEG, quality85) return base64.b64encode(buffer.getvalue()).decode() staticmethod def preprocess_for_display(image_path: str, max_size: int 1024) - str: 预处理用于模型输入的图片大图压缩 img Image.open(image_path) # 保持宽高比缩放到 max_size w, h img.size if max(w, h) max_size: ratio max_size / max(w, h) img img.resize((int(w*ratio), int(h*ratio)), Image.Resampling.LANCZOS) buffer io.BytesIO() img.save(buffer, formatJPEG, quality80) return base64.b64encode(buffer.getvalue()).decode()四、多模态 RAG 的边界与挑战CLIP 的 Vector Space 不完美。CLIP 的图文对齐是基于互联网图文对训练的对于专业领域图片如架构图、流程图、代码截图理解能力有限。微调 CLIP 或使用领域专用的多模态模型是必要的。向量融合的策略影响大。Concat 保留全部信息但维度翻倍检索速度变慢Average 维度不变但信息有损。建议从 Average 开始如果检索精度不够再换 Concat。图片存储成本。知识库中存储原始图片如截图会显著增加存储开销。在送入 Embedding 模型之前压缩图片224x224原始大图只在需要展示时才读取。多模态 LLM 的推理成本。图文混合输入比纯文本消耗更多 Token。在检索阶段用轻量级的 CLIP Embedding只在最后推理阶段才把图片送给多模态 LLM避免重复的图片编码开销。五、总结多模态 RAG 的核心是把图片和文本映射到同一个向量空间。关键步骤用 CLIP 做图像 Embedding 与文本向量做融合 在统一索引中检索。实现路径先验证 CLIP 在你的领域图片上的检索精度确认可用后再建立完整的图文索引。图片预处理是性价比最高的优化——合适的尺寸和压缩比可以降低 50% 的存储和传输开销。