4.1 多模态搜索 — 跨模态智能检索技术

本节导读：掌握 Milvus 多模态搜索核心技术，实现文本、图像等跨模态智能检索，为 AI 应用提供强大的数据支撑。

学习目标

• 掌握多模态搜索的基本原理
• 实现文本和图像的嵌入生成
• 学会跨模态搜索和混合检索策略
• 理解语义搜索和重排序技术

核心概念

1. 多模态搜索架构

多模态搜索是指在不同类型的数据（文本、图像、音频等）之间进行跨模态检索的技术：

数据模态分类

• 文本模态：标题、描述、评论等文本内容
• 图像模态：产品图片、用户头像、场景图像
• 音频模态：语音、音乐、环境声音
• 视频模态：视频帧、关键帧、字幕信息

搜索策略

• 单模态搜索：在同一类型数据中检索
• 跨模态搜索：在不同类型数据间检索
• 多模态融合：结合多种模态进行综合搜索

2. 语义搜索原理

语义搜索基于语义相似性而非关键词匹配：

向量化模型

• BERT 系列：文本语义向量化
• CLIP：跨模态对齐
• Sentence-BERT：句子级语义

相似度计算

• 余弦相似度：角度相似性度量
• 欧氏距离：空间距离度量
• 内积：向量关联度计算

分步实战

步骤 1：文本模态搜索

import torch
from transformers import AutoTokenizer, AutoModel
from pymilvus import Collection, connections
import numpy as np
from typing import List, Dict, Any

class TextEmbeddingGenerator:
    """文本嵌入生成器"""
    
    def __init__(self, model_name="sentence-transformers/all-MiniLM-L6-v2"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        
    def generate_embedding(self, text: str) -> np.ndarray:
        """生成文本嵌入"""
        inputs = self.tokenizer(text, return_tensors="pt", padding=True, truncation=True)
        
        with torch.no_grad():
            outputs = self.model(**inputs)
        
        # 使用池化策略获取句子嵌入
        embeddings = self.mean_pooling(outputs.last_hidden_state, inputs['attention_mask'])
        
        return embeddings.numpy()[0]
    
    def mean_pooling(self, token_embeddings, attention_mask):
        """均值池化"""
        input_mask_expanded = attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()
        return torch.sum(token_embeddings * input_mask_expanded, 1) / torch.clamp(input_mask_expanded.sum(1), min=1e-9)

class MultiModalTextSearch:
    """多模态文本搜索系统"""
    
    def __init__(self, milvus_host="localhost", milvus_port="19530"):
        self.milvus_host = milvus_host
        self.milvus_port = milvus_port
        self.embedding_generator = TextEmbeddingGenerator()
        
        # 连接 Milvus
        connections.connect("default", host=milvus_host, port=milvus_port)
        
        # 创建集合
        self.create_text_collection()
    
    def create_text_collection(self):
        """创建文本搜索集合"""
        from pymilvus import CollectionSchema, FieldSchema, DataType
        
        # 定义字段
        fields = [
            FieldSchema("id", DataType.INT64, is_primary=True),
            FieldSchema("content", DataType.VARCHAR, max_length=65535),
            FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=384),
            FieldSchema("metadata", DataType.JSON),
            FieldSchema("timestamp", DataType.INT64)
        ]
        
        schema = CollectionSchema(fields, "文本搜索集合")
        self.text_collection = Collection("text_search", schema)
        
        # 创建索引
        self.create_indexes()
    
    def create_indexes(self):
        """创建搜索索引"""
        # 文本嵌入索引
        index_params = {
            "index_type": "HNSW",
            "params": {"M": 32, "ef": 256},
            "metric_type": "IP"
        }
        
        self.text_collection.create_index("embedding", index_params)
    
    def add_text_document(self, content: str, metadata: Dict[str, Any] = None):
        """添加文本文档"""
        # 生成嵌入
        embedding = self.embedding_generator.generate_embedding(content)
        
        # 插入数据
        data = [
            np.random.randint(0, 1000000),  # id
            content,
            embedding,
            metadata or {},
            int(time.time())
        ]
        
        self.text_collection.insert([data])
        print(f"已添加文本文档: {content[:50]}...")
    
    def search_text(self, query: str, limit: int = 10) -> List[Dict]:
        """文本搜索"""
        # 生成查询嵌入
        query_embedding = self.embedding_generator.generate_embedding(query)
        
        # 执行搜索
        results = self.text_collection.search(
            data=[query_embedding],
            anns_field="embedding",
            param={"metric_type": "IP", "ef": 50},
            limit=limit,
            output_fields=["content", "metadata", "timestamp"]
        )
        
        # 处理结果
        search_results = []
        for hit in results[0]:
            search_results.append({
                "id": hit.id,
                "content": hit.entity.get("content"),
                "score": hit.score,
                "metadata": hit.entity.get("metadata"),
                "timestamp": hit.entity.get("timestamp")
            })
        
        return search_results
​

步骤 2：图像模态搜索

import torch
import torchvision.transforms as transforms
from PIL import Image
import requests
from io import BytesIO
import numpy as np
from typing import List, Dict, Any

class ImageEmbeddingGenerator:
    """图像嵌入生成器"""
    
    def __init__(self, model_name="openai/clip-vit-base-patch32"):
        self.model_name = model_name
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        
        # 加载模型和预处理
        self.model = torch.hub.load('pytorch/vision', 'resnet18', pretrained=True)
        self.model.eval()
        
        # 图像预处理
        self.preprocess = transforms.Compose([
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
        ])
    
    def generate_embedding(self, image_path: str or Image.Image) -> np.ndarray:
        """生成图像嵌入"""
        if isinstance(image_path, str):
            if image_path.startswith("http"):
                # 从URL下载图像
                response = requests.get(image_path)
                image = Image.open(BytesIO(response.content))
            else:
                # 从文件加载图像
                image = Image.open(image_path)
        else:
            image = image_path
        
        # 预处理
        image_tensor = self.preprocess(image)
        image_tensor = image_tensor.unsqueeze(0).to(self.device)
        
        # 生成嵌入
        with torch.no_grad():
            features = self.model(image_tensor)
        
        return features.cpu().numpy()[0]

class MultiModalImageSearch:
    """多模态图像搜索系统"""
    
    def __init__(self, milvus_host="localhost", milvus_port="19530"):
        self.milvus_host = milvus_host
        self.milvus_port = milvus_port
        self.embedding_generator = ImageEmbeddingGenerator()
        
        # 连接 Milvus
        connections.connect("default", host=milvus_host, port=milvus_port)
        
        # 创建集合
        self.create_image_collection()
    
    def create_image_collection(self):
        """创建图像搜索集合"""
        from pymilvus import CollectionSchema, FieldSchema, DataType
        
        fields = [
            FieldSchema("id", DataType.INT64, is_primary=True),
            FieldSchema("image_url", DataType.VARCHAR, max_length=1024),
            FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=512),
            FieldSchema("metadata", DataType.JSON),
            FieldSchema("timestamp", DataType.INT64)
        ]
        
        schema = CollectionSchema(fields, "图像搜索集合")
        self.image_collection = Collection("image_search", schema)
        
        # 创建索引
        index_params = {
            "index_type": "HNSW",
            "params": {"M": 32, "ef": 256},
            "metric_type": "IP"
        }
        
        self.image_collection.create_index("embedding", index_params)
    
    def add_image_document(self, image_url: str, metadata: Dict[str, Any] = None):
        """添加图像文档"""
        try:
            # 生成嵌入
            embedding = self.embedding_generator.generate_embedding(image_url)
            
            # 插入数据
            data = [
                np.random.randint(0, 1000000),  # id
                image_url,
                embedding,
                metadata or {},
                int(time.time())
            ]
            
            self.image_collection.insert([data])
            print(f"已添加图像文档: {image_url}")
            
        except Exception as e:
            print(f"添加图像文档失败: {e}")
    
    def search_image(self, image_url: str, limit: int = 10) -> List[Dict]:
        """图像搜索"""
        try:
            # 生成查询嵌入
            query_embedding = self.embedding_generator.generate_embedding(image_url)
            
            # 执行搜索
            results = self.image_collection.search(
                data=[query_embedding],
                anns_field="embedding",
                param={"metric_type": "IP", "ef": 50},
                limit=limit,
                output_fields=["image_url", "metadata", "timestamp"]
            )
            
            # 处理结果
            search_results = []
            for hit in results[0]:
                search_results.append({
                    "id": hit.id,
                    "image_url": hit.entity.get("image_url"),
                    "score": hit.score,
                    "metadata": hit.entity.get("metadata"),
                    "timestamp": hit.entity.get("timestamp")
                })
            
            return search_results
            
        except Exception as e:
            print(f"图像搜索失败: {e}")
            return []
​

步骤 3：混合搜索系统

class SemanticSearchEngine:
    """语义搜索引擎"""
    
    def __init__(self, milvus_host="localhost", milvus_port="19530"):
        self.milvus_host = milvus_host
        self.milvus_port = milvus_port
        
        # 初始化各个搜索组件
        self.text_search = MultiModalTextSearch(milvus_host, milvus_port)
        self.image_search = MultiModalImageSearch(milvus_host, milvus_port)
        
        # 权重配置
        self.weights = {
            "text": 0.7,
            "image": 0.3
        }
    
    def hybrid_search(self, query: str, top_k: int = 10) -> Dict[str, List[Dict]]:
        """混合搜索"""
        results = {
            "text_results": [],
            "image_results": [],
            "combined_results": []
        }
        
        # 文本搜索
        text_results = self.text_search.search_text(query, limit=top_k)
        results["text_results"] = text_results
        
        # 图像搜索
        image_results = self.image_search.search_image("", limit=top_k//2)
        results["image_results"] = image_results
        
        # 混合排序
        combined_results = self.rank_results(
            text_results, 
            image_results, 
            self.weights
        )
        results["combined_results"] = combined_results
        
        return results
    
    def rank_results(self, text_results: List[Dict], image_results: List[Dict], weights: Dict) -> List[Dict]:
        """混合排序"""
        combined = []
        
        # 处理文本结果
        for result in text_results:
            combined.append({
                "id": f"text_{result['id']}",
                "content": result['content'],
                "score": result['score'] * weights['text'],
                "type": "text",
                "metadata": result['metadata']
            })
        
        # 处理图像结果
        for result in image_results:
            combined.append({
                "id": f"image_{result['id']}",
                "image_url": result['image_url'],
                "score": result['score'] * weights['image'],
                "type": "image",
                "metadata": result['metadata']
            })
        
        # 按分数排序
        combined.sort(key=lambda x: x['score'], reverse=True)
        
        return combined[:10]  # 返回前10个结果
​

常见问题 FAQ

Q1：如何处理大规模数据的实时搜索？

A：

• 增量更新：使用流处理系统实时更新数据
• 分片策略：按时间或类别进行数据分片
• 缓存机制：使用 Redis 缓存热门查询结果
• 负载均衡：部署多个 Milvus 实例实现负载均衡

Q2：如何优化多模态搜索的性能？

A：

• 预计算嵌入：对现有数据预计算嵌入向量
• 批量处理：使用批量搜索减少查询次数
• 索引优化：选择合适的索引类型和参数
• 并行计算：多线程并行处理不同模态

Q3：如何处理跨模态搜索的语义对齐问题？

A：

• 使用 CLIP 模型：CLIP 能够很好地对齐不同模态
• 训练对齐模型：针对特定领域训练跨模态对齐模型
• 相似性度量：使用余弦相似度等合适的相似性度量
• 重排序机制：对初步搜索结果进行语义重排序

最佳实践与避坑

1. 多模态搜索最佳实践

• 统一嵌入空间：确保不同模态在同一个嵌入空间
• 合理的权重分配：根据业务需求调整各模态权重
• 实时性能监控：建立完善的性能监控体系
• 渐进式部署：先实现单一模态，再逐步扩展到多模态

2. 实时流处理避坑指南

• 消息积压处理：合理设置消息处理能力
• 异常处理机制：完善的错误处理和重试机制
• 数据一致性：保证数据的一致性和完整性
• 资源管理：合理配置 CPU、内存等资源

本节小结

通过本节的学习，你已经掌握了 Milvus 多模态搜索的核心技术，包括文本嵌入生成、图像嵌入生成、跨模态搜索和混合检索策略。这些技术将帮助你构建强大的多模态 AI 应用数据引擎。

记住，在实际应用中需要根据具体的业务场景选择合适的技术方案，并不断优化性能和用户体验。建立完善的监控和运维体系，确保系统的稳定性和可靠性。

延伸阅读

• Milvus 多模态搜索官方文档
• 深度学习嵌入模型指南
• 实时流处理系统设计
• 分布式系统架构实践

关键词：Milvus, 多模态搜索, 跨模态检索, 语义搜索, 嵌入模型, 混合检索
难度：高级
预计阅读：60 分钟