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):ifsim<self.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+1<len(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)
与辐射(近场、远场)详解)
