多模态RAG 2026实践：跨越文本、图像与视频的检索

文本→图像: "找一张展示AI架构的图"
图像→文本: "这张图对应的技术文档在哪？"
图像→图像: "找类似的图片"

┌─────────────────────────────────────────────┐
│             多模态查询处理                    │
│  (文本/图像/音频 → 统一表示)                  │
├─────────────────────────────────────────────┤
│             跨模态检索引擎                    │
│  ┌─────────┬─────────┬─────────┐           │
│  │文本检索  │图像检索  │音频检索  │           │
│  └─────────┴─────────┴─────────┘           │
├─────────────────────────────────────────────┤
│             多模态融合层                      │
│  (跨模态排序、上下文组装)                     │
├─────────────────────────────────────────────┤
│             多模态生成                        │
│  (文本+图像+表格 → 统一回答)                  │
└─────────────────────────────────────────────┘

class MultiModalEncoder:
    def __init__(self):
        self.text_encoder = load_model("text-embedding-3-large")
        self.image_encoder = load_model("clip-vit-large")
        self.audio_encoder = load_model("audio-embedding-model")
    
    async def encode(self, content, modality):
        """编码不同模态的内容"""
        if modality == "text":
            return await self.text_encoder.embed(content)
        elif modality == "image":
            return await self.image_encoder.encode(content)
        elif modality == "audio":
            return await self.audio_encoder.encode(content)
        elif modality == "mixed":
            # 混合内容：分别编码后融合
            return await self.encode_mixed(content)
    
    async def encode_mixed(self, content):
        """编码混合模态内容"""
        embeddings = []
        for item in content:
            emb = await self.encode(item["data"], item["modality"])
            embeddings.append({"embedding": emb, "modality": item["modality"]})
        
        # 使用跨模态对齐模型将不同模态的embedding映射到统一空间
        unified = await self.align_embeddings(embeddings)
        return unified

class CrossModalAligner:
    def __init__(self):
        self.clip_model = load_model("clip-large")
    
    async def align(self, text_embedding, image_embedding):
        """将文本和图像embedding对齐到同一空间"""
        # CLIP模型已经将文本和图像映射到同一空间
        # 只需确保使用相同的模型编码
        return text_embedding, image_embedding
    
    async def compute_cross_modal_similarity(self, text_emb, image_emb):
        """计算跨模态相似度"""
        return cosine_similarity(text_emb, image_emb)

class TextToImageRetriever:
    async def search(self, text_query, top_k=10):
        """用文本查询检索图像"""
        # 1. 用CLIP文本编码器编码查询
        query_embedding = await self.clip.encode_text(text_query)
        
        # 2. 在图像向量库中搜索
        results = await self.image_vector_store.search(
            query_embedding, top_k=top_k
        )
        
        # 3. 添加元数据
        for result in results:
            result.modality = "image"
            result.query = text_query
        
        return results

class ImageToTextRetriever:
    async def search(self, image_query, top_k=10):
        """用图像查询检索文本"""
        # 1. 用CLIP图像编码器编码查询
        query_embedding = await self.clip.encode_image(image_query)
        
        # 2. 在文本向量库中搜索
        results = await self.text_vector_store.search(
            query_embedding, top_k=top_k
        )
        
        return results

class MultiModalRetriever:
    async def retrieve(self, query, top_k=10):
        """多模态混合检索"""
        results = []
        
        # 1. 根据查询类型执行不同检索
        if query.has_text and query.has_image:
            # 文本+图像查询：分别检索后融合
            text_results = await self.text_to_all_search(query.text, top_k)
            image_results = await self.image_to_all_search(query.image, top_k)
            results = self.fuse(text_results, image_results)
        elif query.has_text:
            results = await self.text_to_all_search(query.text, top_k)
        elif query.has_image:
            results = await self.image_to_all_search(query.image, top_k)
        
        # 2. 跨模态重排
        reranked = await self.cross_modal_rerank(results, query)
        
        return reranked[:top_k]
    
    async def cross_modal_rerank(self, results, query):
        """跨模态重排"""
        for result in results:
            # 计算与查询的跨模态相关性
            if query.has_text and result.modality == "image":
                score = await self.clip.score(query.text, result.image)
            elif query.has_image and result.modality == "text":
                score = await self.clip.score(result.text, query.image)
            else:
                score = result.similarity
            
            result.final_score = 0.5 * result.similarity + 0.5 * score
        
        results.sort(key=lambda x: -x.final_score)
        return results

class MultiModalDocumentParser:
    async def parse(self, document):
        """解析多模态文档（如PDF、PPT）"""
        chunks = []
        
        # 1. 提取文本
        text_blocks = await self.extract_text(document)
        
        # 2. 提取图像
        images = await self.extract_images(document)
        
        # 3. 提取表格
        tables = await self.extract_tables(document)
        
        # 4. 关联文本和图像
        for i, text_block in enumerate(text_blocks):
            chunk = {
                "id": f"chunk-{i}",
                "text": text_block.text,
                "page": text_block.page,
                "position": text_block.position
            }
            
            # 找到同一页附近的图像
            related_images = [
                img for img in images 
                if img.page == text_block.page 
                and self.is_nearby(img.position, text_block.position)
            ]
            if related_images:
                chunk["images"] = related_images
            
            # 找到相关表格
            related_tables = [
                tbl for tbl in tables
                if tbl.page == text_block.page
            ]
            if related_tables:
                chunk["tables"] = related_tables
            
            chunks.append(chunk)
        
        return chunks

class ImageCaptionGenerator:
    async def generate_caption(self, image):
        """为图像生成描述"""
        # 1. 基础描述
        basic_caption = await self.vlm.generate(
            image=image,
            prompt="描述这张图片的内容"
        )
        
        # 2. 详细描述
        detailed_caption = await self.vlm.generate(
            image=image,
            prompt="详细描述这张图片，包括物体、场景、文字、数据等信息"
        )
        
        # 3. 结构化描述
        structured = await self.vlm.generate(
            image=image,
            prompt="""
            以JSON格式描述这张图片：
            {
                "type": "图表/照片/截图/...",
                "main_subject": "主要内容",
                "details": ["细节1", "细节2", ...],
                "text_in_image": "图片中的文字",
                "data": {如果有数据}
            }
            """
        )
        
        return {
            "basic": basic_caption,
            "detailed": detailed_caption,
            "structured": structured
        }

class TableUnderstanding:
    async def process_table(self, table):
        """处理表格数据"""
        # 1. 转换为结构化格式
        structured = self.to_structured(table)
        
        # 2. 生成表格描述
        description = await self.llm.generate(
            f"描述以下表格的内容和含义:\n{structured}"
        )
        
        # 3. 生成表格embedding
        embedding = await self.text_encoder.embed(description)
        
        return {
            "structured_data": structured,
            "description": description,
            "embedding": embedding,
            "raw": table
        }

class MultiModalContextAssembler:
    async def assemble(self, question, retrieved_items):
        """组装多模态上下文"""
        context_parts = []
        
        for item in retrieved_items:
            if item.modality == "text":
                context_parts.append({
                    "type": "text",
                    "content": item.text
                })
            elif item.modality == "image":
                context_parts.append({
                    "type": "image",
                    "content": item.image,
                    "caption": item.caption  # 图像描述
                })
            elif item.modality == "table":
                context_parts.append({
                    "type": "table",
                    "content": item.structured_data,
                    "description": item.description
                })
        
        # 按相关性排序
        context_parts.sort(key=lambda x: -x.get("relevance", 0))
        
        return context_parts

class MultiModalGenerator:
    async def generate(self, question, context):
        """基于多模态上下文生成回答"""
        # 构建多模态prompt
        prompt_parts = [f"问题: {question}\n\n参考信息:\n"]
        
        for i, item in enumerate(context):
            if item["type"] == "text":
                prompt_parts.append(f"[文档{i+1}]\n{item['content']}\n")
            elif item["type"] == "image":
                prompt_parts.append(f"[图像{i+1}]\n{item['caption']}\n")
                # 某些VLM支持直接传入图像
            elif item["type"] == "table":
                prompt_parts.append(f"[表格{i+1}]\n{item['description']}\n")
        
        prompt_parts.append("\n请基于以上信息回答问题。")
        
        # 使用多模态LLM生成
        if any(item["type"] == "image" for item in context):
            # 使用视觉语言模型
            answer = await self.vlm.generate(
                text="\n".join(prompt_parts),
                images=[item["content"] for item in context if item["type"] == "image"]
            )
        else:
            # 使用纯文本LLM
            answer = await self.llm.generate("\n".join(prompt_parts))
        
        return answer

class VideoRAGProcessor:
    async def process_video(self, video):
        """处理视频为可检索的片段"""
        # 1. 关键帧提取
        keyframes = await self.extract_keyframes(video, interval=2.0)
        
        # 2. 语音转文字
        transcript = await self.asr.transcribe(video.audio)
        
        # 3. 为每个关键帧生成描述
        for frame in keyframes:
            frame.caption = await self.vlm.generate(
                image=frame.image,
                prompt="描述这个视频帧的内容"
            )
        
        # 4. 对齐文本和帧
        aligned_segments = self.align_transcript_frames(transcript, keyframes)
        
        # 5. 构建可检索的片段
        segments = []
        for seg in aligned_segments:
            segments.append({
                "id": seg.id,
                "timestamp": seg.timestamp,
                "text": seg.text,
                "image": seg.frame.image,
                "caption": seg.frame.caption,
                "text_embedding": await self.encode(seg.text),
                "image_embedding": await self.encode(seg.frame.image)
            })
        
        return segments

class VideoRetriever:
    async def search(self, query, top_k=5):
        """视频片段检索"""
        # 文本查询 → 检索视频片段
        query_emb = await self.encode(query)
        
        # 同时在文本embedding和图像embedding上搜索
        text_results = await self.text_index.search(query_emb, top_k=top_k*2)
        image_results = await self.image_index.search(query_emb, top_k=top_k*2)
        
        # 融合
        fused = self.fuse(text_results, image_results)
        
        return fused[:top_k]

class MultiModalRAGEvaluator:
    async def evaluate(self, test_cases):
        metrics = {
            "cross_modal_retrieval_accuracy": [],
            "multimodal_answer_accuracy": [],
            "image_grounding_accuracy": [],
            "hallucination_rate": []
        }
        
        for case in test_cases:
            result = await self.system.query(case.question, case.image_query)
            
            # 跨模态检索准确率
            metrics["cross_modal_retrieval_accuracy"].append(
                self.eval_retrieval(result.retrieved, case.relevant)
            )
            
            # 多模态答案准确率
            metrics["multimodal_answer_accuracy"].append(
                await self.eval_answer(result.answer, case.expected)
            )
            
            # 图像定位准确率（答案是否正确引用了图像信息）
            if case.requires_image:
                metrics["image_grounding_accuracy"].append(
                    self.eval_image_grounding(result.answer, case.image_info)
                )
        
        return {k: np.mean(v) for k, v in metrics.items()}