RAG 多模态检索：文本+图像+表格的统一检索方案引言传统的RAG系统主要处理文本数据，但现实世界的信息以多种形式存在：产品图片、数据表格、扫描文档、技术图表等。多模态RAG（Multimodal RAG）打破了单一文本的限制，让AI能够理解和检索图像、表格、文档等多种类型的信息，真正实现"所见即所得"的智能检索。为什么需要多模态RAG？信息多样性：企业知识库包含PDF、图片、Excel等多种格式语义完整性：图表往往比文字更直观地表达信息检索准确性：某些信息只能通过视觉特征识别用户体验：用户希望用图片搜索图片，用表格搜索数据多模态RAG架构整体架构┌─────────────────────────────────────────────────────────────┐ │ Multimodal RAG System │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 文本处理 │ │ 图像处理 │ │ 表格处理 │ │ │ │ Text Engine │ │ Image Engine │ │ Table Engine │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ Multimodal Embedding Space │ │ │ │ (统一向量空间) │ │ │ └─────────────────────────────────────────────────────┘ │ │ │ │ │ ┌────────────────┼────────────────┐ │ │ ▼ ▼ ▼ │ │ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ │ │ 文本检索 │ │ 图像检索 │ │ 表格检索 │ │ │ └────────────┘ └────────────┘ └────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────┐ │ │ │ 结果融合 │ │ │ │ Fusion │ │ │ └──────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘核心挑战挑战说明解决方案表示差异文本、图像、表格的向量空间不同统一嵌入模型 / 跨模态对齐检索融合如何合并不同模态的检索结果多模态融合策略存储管理多模态数据存储和索引分层存储 + 元数据管理查询理解用户查询可能是文本或图像查询类型识别 + 多模态查询文本检索增强基础文本处理fromlangchain_text_splittersimportRecursiveCharacterTextSplitterfromlangchain_openaiimportOpenAIEmbeddingsfromlangchain_community.vectorstoresimportChromafromlangchain_core.documentsimportDocumentclassTextProcessor:"""文本处理器"""def__init__(self,chunk_size:int=500,chunk_overlap:int=50,embedding_model:str="text-embedding-3-small"):self.splitter=RecursiveCharacterTextSplitter(chunk_size=chunk_size,chunk_overlap=chunk_overlap,separators=["\n\n","\n","。","！","？","；"," ",""])self.embeddings=OpenAIEmbeddings(model=embedding_model)defprocess_text(self,text:str,metadata:dict=None)-list:"""处理文本"""# 分割文本chunks=self.splitter.split_text(text)# 创建文档documents=[Document(page_content=chunk,metadata={**(metadataor{}),"chunk_index":i,"modality":"text"})fori,chunkinenumerate(chunks)]returndocumentsdefprocess_file(self,file_path:str)-list:"""处理文件"""withopen(file_path,'r',encoding='utf-8')asf:text=f.read()returnself.process_text(text,{"source":file_path})# 使用processor=TextProcessor()docs=processor.process_text("这是一段很长的文本..."*100,{"source":"example","author":"test"})print(f"生成{len(docs)}个文本块")语义分块fromtypingimportListimportnumpyasnpclassSemanticTextSplitter:"""语义分块器 - 按语义边界分割"""def__init__(self,embeddings:OpenAIEmbeddings,similarity_threshold:float=0.7,min_chunk_size:int=100):self.embeddings=embeddings self.similarity_threshold=similarity_threshold self.min_chunk_size=min_chunk_sizedefsplit_by_semantics(self,text:str)-List[str]:"""按语义分割文本"""# 1. 按句子分割sentences=self._split_sentences(text)iflen(sentences)2:return[text]# 2. 计算句子嵌入sentence_embeddings=self.embeddings.embed_documents(sentences)# 3. 计算相邻句子相似度similarities=[]foriinrange(len(sentence_embeddings)-1):sim=self._cosine_similarity(sentence_embeddings[i],sentence_embeddings[i+1])similarities.append(sim)# 4. 找到语义边界（相似度低的地方）boundaries=[0]fori,siminenumerate(similarities):ifsimself.similarity_threshold:boundaries.append(i+1)boundaries.append(len(sentences))# 5. 合并句子为块chunks=[]foriinrange(len(boundaries)-1):start=boundaries[i]end=boundaries[i+1]chunk="".join(sentences[start:end])# 确保块大小iflen(chunk)=self.min_chunk_sizeori==len(boundaries)-2:chunks.append(chunk)elifchunks:# 合并到前一个块chunks[-1]+=chunkreturnchunksdef_split_sentences(self,text:str)-List[str]:"""分割句子"""importre# 中英文句子分割sentences=re.split(r'([。！？.!?])',text)# 重新组合标点sentences=[sentences[i]+(sentences[i+1]ifi+1len(sentences)else'')foriinrange(0,len(sentences)-1,2)]return[s.strip()forsinsentencesifs.strip()]def_cosine_similarity(self,vec1:List[float],vec2:List[float])-float:"""计算余弦相似度"""vec1=np.array(vec1)vec2=np.array(vec2)returnnp.dot(vec1,vec2)/(np.linalg.norm(vec1)*np.linalg.norm(vec2))# 使用semantic_splitter=SemanticTextSplitter(OpenAIEmbeddings())chunks=semantic_splitter.split_by_semantics("第一段讲AI技术。第二段讲机器学习。第三段讲深度学习应用。")图像检索实现图像嵌入生成fromtypingimportList,Dictimportbase64fromPILimportImageimportioclassImageProcessor:"""图像处理器"""def__init__(self,model:str="clip-ViT-B-32"):# 使用sentence-transformers的CLIP模型fromsentence_transformersimportSentenceTransformer self.model=SentenceTransformer(model)defencode_image(self,image_path:str)-str:"""图像转base64"""withopen(image_path,"rb")asf:returnbase64.b64encode(f.read()).decode()defget_image_embedding(self,image_path:str)-List[float]:"""获取图像嵌入"""fromPILimportImage img=Image.open(image_path)embedding=self.model.encode(img)returnembedding.tolist()defget_text_embedding(self,text:str)-List[float]:"""获取文本嵌入（用于跨模态检索）"""embedding=self.model.encode(text)returnembedding.tolist()defprocess_image(self,image_path:str,metadata:dict=None)