跨模态嵌入：连接人工智能模态

在共享的嵌入空间中统一文本、图像和音频

目录

跨模态嵌入 代表了人工智能领域的一项突破，它使不同数据类型能够在统一的表示空间中实现理解和推理。

这项技术为现代多模态应用（从图像搜索到内容生成）提供了动力。

跨模态嵌入 此图片来自文章： CrossCLR: 用于多模态视频表示的跨模态对比学习，作者 Mohammadreza Zolfaghari 等人

理解跨模态嵌入

跨模态嵌入是向量表示，它们将来自不同模态（如文本、图像、音频和视频）的信息编码到共享的嵌入空间中。与传统的单模态嵌入不同，跨模态方法学习一个统一的表示，其中语义相似的概念无论其原始格式如何都会聚在一起。

什么是跨模态嵌入？

从根本上说，跨模态嵌入解决了一个AI中的关键挑战：如何在不同类型的数据之间进行比较和关联。传统的图像分类器只能处理图像，而文本模型只处理文本。跨模态嵌入通过将不同模态投影到一个共同的向量空间中来弥合这一差距，其中：

  • 猫的图像和单词“猫”具有相似的嵌入向量
  • 语义关系在模态之间得以保留
  • 距离度量（余弦相似度、欧几里得距离）测量跨模态相似度

这种统一的表示使强大的功能成为可能，例如使用文本查询搜索图像、从图像生成标题，甚至在没有任务特定训练的情况下进行零样本分类。

跨模态学习的架构

现代跨模态系统通常采用双编码器架构并使用对比学习目标：

双编码器：分别对每个模态进行编码的神经网络。例如，CLIP 使用：

  • 用于图像的视觉变换器（ViT）或 ResNet
  • 基于变换器的语言编码器

对比学习：模型通过最大化匹配对（如图像及其标题）之间的相似性，同时最小化不匹配对之间的相似性来学习。对比损失函数可以表示为：

$$ \mathcal{L} = -\log \frac{\exp(\text{sim}(v_i, t_i) / \tau)}{\sum_{j=1}^{N} \exp(\text{sim}(v_i, t_j) / \tau)} $$

其中 $v_i$ 是图像嵌入，$t_i$ 是文本嵌入，$\text{sim}$ 是相似度（通常为余弦），$\tau$ 是温度参数。

关键技术和模型

CLIP：开创性的视觉-语言理解

OpenAI 的 CLIP（对比语言-图像预训练）通过在互联网上训练 4 亿个图像-文本对，彻底改变了该领域。模型架构包括：

import torch
from transformers import CLIPProcessor, CLIPModel
from PIL import Image

# 加载预训练的 CLIP 模型
model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

# 准备输入
image = Image.open("example.jpg")
texts = ["a photo of a cat", "a photo of a dog", "a photo of a bird"]

# 处理并获取嵌入
inputs = processor(text=texts, images=image, return_tensors="pt", padding=True)

# 获取跨模态相似度分数
outputs = model(**inputs)
logits_per_image = outputs.logits_per_image
probs = logits_per_image.softmax(dim=1)

print(f"概率: {probs}")

CLIP 的关键创新在于规模和简洁性。通过在没有人工标注的大规模网络数据上训练，它实现了显著的零样本迁移能力。CLIP 与传统视觉模型有何不同？与在固定标签集上训练的监督分类器不同，CLIP 从自然语言监督中学习，使其能够适应任何可以用文本描述的视觉概念。

ImageBind：扩展到六种模态

Meta 的 ImageBind 将跨模态嵌入扩展到视觉和语言以外，包括：

  • 音频（环境声音、语音）
  • 深度信息
  • 热成像
  • IMU（运动传感器）数据

这创建了一个真正多模态的嵌入空间，其中所有六种模态都对齐。关键见解是图像作为“绑定”模态——通过将图像与其他模态配对并利用现有的视觉-语言对齐，ImageBind 在训练时无需所有可能的模态对即可创建统一的空间。

开源替代方案

生态系统已扩展为多个开源实现：

OpenCLIP：一个社区实现，提供更大的模型和多样的训练配方：

import open_clip

model, _, preprocess = open_clip.create_model_and_transforms(
    'ViT-L-14', 
    pretrained='laion2b_s32b_b82k'
)
tokenizer = open_clip.get_tokenizer('ViT-L-14')

LAION-5B 模型：在庞大的开源 LAION-5B 数据集上训练，提供与专有模型相当或更好的性能的替代方案。

对于有兴趣使用最先进的开源文本嵌入解决方案的开发者，Qwen3 嵌入与重排序模型在 Ollama 上 提供了出色的多语言性能和易于本地部署。

实施策略

构建跨模态搜索系统

跨模态嵌入用于语义搜索的实用实现涉及多个组件。跨模态嵌入的主要应用是什么？它们支持从电子商务产品搜索到内容审核和创意工具的用例。

import numpy as np
from typing import List, Tuple
import faiss
from transformers import CLIPModel, CLIPProcessor

class CrossModalSearchEngine:
    def __init__(self, model_name: str = "openai/clip-vit-base-patch32"):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.image_index = None
        self.image_metadata = []
        
    def encode_images(self, images: List) -> np.ndarray:
        """对图像进行编码以生成嵌入"""
        inputs = self.processor(images=images, return_tensors="pt", padding=True)
        with torch.no_grad():
            embeddings = self.model.get_image_features(**inputs)
        return embeddings.cpu().numpy()
    
    def encode_text(self, texts: List[str]) -> np.ndarray:
        """对文本查询进行编码以生成嵌入"""
        inputs = self.processor(text=texts, return_tensors="pt", padding=True)
        with torch.no_grad():
            embeddings = self.model.get_text_features(**inputs)
        return embeddings.cpu().numpy()
    
    def build_index(self, image_embeddings: np.ndarray):
        """构建FAISS索引以实现高效的相似性搜索"""
        dimension = image_embeddings.shape[1]
        
        # 对嵌入进行归一化以进行余弦相似性
        faiss.normalize_L2(image_embeddings)
        
        # 创建索引（使用HNSW进行大规模部署）
        self.image_index = faiss.IndexHNSWFlat(dimension, 32)
        self.image_index.hnsw.efConstruction = 40
        self.image_index.add(image_embeddings)
    
    def search(self, query: str, k: int = 10) -> List[Tuple[int, float]]:
        """使用文本查询搜索图像"""
        query_embedding = self.encode_text([query])
        faiss.normalize_L2(query_embedding)
        
        distances, indices = self.image_index.search(query_embedding, k)
        return list(zip(indices[0], distances[0]))

# 使用示例
engine = CrossModalSearchEngine()

# 从图像集合中构建索引
image_embeddings = engine.encode_images(image_collection)
engine.build_index(image_embeddings)

# 使用文本进行搜索
results = engine.search("sunset over mountains", k=5)

针对特定任务进行微调

尽管预训练模型在一般用途上表现良好，但特定领域的应用会从微调中受益：

from transformers import CLIPModel, CLIPProcessor, AdamW
import torch.nn as nn

class FineTuneCLIP:
    def __init__(self, model_name: str, num_epochs: int = 10):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.num_epochs = num_epochs
        
    def contrastive_loss(self, image_embeddings, text_embeddings, temperature=0.07):
        """计算InfoNCE对比损失"""
        # 归一化嵌入
        image_embeddings = nn.functional.normalize(image_embeddings, dim=1)
        text_embeddings = nn.functional.normalize(text_embeddings, dim=1)
        
        # 计算相似度矩阵
        logits = torch.matmul(image_embeddings, text_embeddings.T) / temperature
        
        # 标签是沿对角线（匹配对）
        labels = torch.arange(len(logits), device=logits.device)
        
        # 对称损失
        loss_i = nn.functional.cross_entropy(logits, labels)
        loss_t = nn.functional.cross_entropy(logits.T, labels)
        
        return (loss_i + loss_t) / 2
    
    def train(self, dataloader, learning_rate=5e-6):
        """在特定领域的数据上进行微调"""
        optimizer = AdamW(self.model.parameters(), lr=learning_rate)
        
        self.model.train()
        for epoch in range(self.num_epochs):
            total_loss = 0
            for batch in dataloader:
                images, texts = batch['images'], batch['texts']
                
                # 处理输入
                inputs = self.processor(
                    text=texts, 
                    images=images, 
                    return_tensors="pt", 
                    padding=True
                )
                
                # 前向传播
                outputs = self.model(**inputs, return_loss=True)
                loss = outputs.loss
                
                # 反向传播
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()
                
                total_loss += loss.item()
            
            avg_loss = total_loss / len(dataloader)
            print(f"Epoch {epoch+1}/{self.num_epochs}, Loss: {avg_loss:.4f}")

生产部署考虑因素

优化推理性能

如何优化跨模态嵌入以进行生产？性能优化对于实际部署至关重要：

模型量化：减少模型大小并提高推理速度：

import torch
from torch.quantization import quantize_dynamic

# 对CPU推理进行动态量化
quantized_model = quantize_dynamic(
    model, 
    {torch.nn.Linear}, 
    dtype=torch.qint8
)

ONNX 转换：导出为ONNX以优化推理：

import torch.onnx

dummy_input = processor(text=["sample"], return_tensors="pt")
torch.onnx.export(
    model,
    tuple(dummy_input.values()),
    "clip_model.onnx",
    input_names=['input_ids', 'attention_mask'],
    output_names=['output'],
    dynamic_axes={
        'input_ids': {0: 'batch_size'},
        'attention_mask': {0: 'batch_size'}
    }
)

批量处理：通过批量处理最大化GPU利用率：

def batch_encode(items: List, batch_size: int = 32):
    """为了效率，批量处理项目"""
    embeddings = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i+batch_size]
        batch_embeddings = encode_batch(batch)
        embeddings.append(batch_embeddings)
    return np.concatenate(embeddings, axis=0)

可扩展的向量存储

对于大规模应用，向量数据库是必不可少的。除了模型本身，基础设施也很重要：

FAISS（Facebook AI Similarity Search）：高效的相似性搜索库

  • 支持数十亿个向量
  • 多种索引类型（flat、IVF、HNSW）
  • 可用GPU加速

Milvus：开源向量数据库

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 定义模式
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=512),
    FieldSchema(name="metadata", dtype=DataType.JSON)
]
schema = CollectionSchema(fields, description="图像嵌入")

# 创建集合
collection = Collection("images", schema)

# 创建索引
index_params = {
    "metric_type": "IP",  # 内积（余弦相似度）
    "index_type": "IVF_FLAT",
    "params": {"nlist": 1024}
}
collection.create_index("embedding", index_params)

Pinecone/Weaviate：提供易于扩展和维护的托管向量数据库服务。

高级用例和应用

零样本分类

跨模态嵌入使在没有任务特定训练的情况下进行分类成为可能。这种能力超越了传统的计算机视觉方法——例如，如果您对更专业的使用TensorFlow进行目标检测 感兴趣，这代表了一种用于特定检测任务的补充监督学习方法。

def zero_shot_classify(image, candidate_labels: List[str], model, processor):
    """将图像分类到任意类别"""
    # 创建文本提示
    text_inputs = [f"a photo of a {label}" for label in candidate_labels]
    
    # 获取嵌入
    inputs = processor(
        text=text_inputs, 
        images=image, 
        return_tensors="pt", 
        padding=True
    )
    outputs = model(**inputs)
    
    # 计算概率
    logits = outputs.logits_per_image
    probs = logits.softmax(dim=1)
    
    # 返回排序后的预测
    sorted_indices = probs.argsort(descending=True)[0]
    return [(candidate_labels[idx], probs[0][idx].item()) for idx in sorted_indices]

# 示例用法
labels = ["cat", "dog", "bird", "fish", "horse"]
predictions = zero_shot_classify(image, labels, model, processor)
print(f"最佳预测: {predictions[0]}")

多模态RAG（检索增强生成）

将跨模态嵌入与语言模型结合，创建强大的多模态RAG系统。一旦检索到相关文档，使用嵌入模型重新排序 可以通过根据相关性重新排序检索到的候选结果，显著提高结果质量：

class MultimodalRAG:
    def __init__(self, clip_model, llm_model):
        self.clip = clip_model
        self.llm = llm_model
        self.knowledge_base = []
    
    def add_documents(self, images, texts, metadata):
        """将多模态文档添加到知识库"""
        image_embeds = self.clip.encode_images(images)
        text_embeds = self.clip.encode_text(texts)
        
        # 存储组合信息
        for i, (img_emb, txt_emb, meta) in enumerate(
            zip(image_embeds, text_embeds, metadata)
        ):
            self.knowledge_base.append({
                'image_embedding': img_emb,
                'text_embedding': txt_emb,
                'metadata': meta,
                'index': i
            })
    
    def retrieve(self, query: str, k: int = 5):
        """检索相关多模态内容"""
        query_embedding = self.clip.encode_text([query])[0]
        
        # 计算相似度
        similarities = []
        for doc in self.knowledge_base:
            # 平均相似度跨模态
            img_sim = np.dot(query_embedding, doc['image_embedding'])
            txt_sim = np.dot(query_embedding, doc['text_embedding'])
            combined_sim = (img_sim + txt_sim) / 2
            similarities.append((combined_sim, doc))
        
        # 返回前k个
        similarities.sort(reverse=True)
        return [doc for _, doc in similarities[:k]]
    
    def answer_query(self, query: str):
        """使用检索到的多模态上下文回答查询"""
        retrieved_docs = self.retrieve(query)
        
        # 从检索到的文档中构造上下文
        context = "\n".join([doc['metadata']['text'] for doc in retrieved_docs])
        
        # 使用LLM生成答案
        prompt = f"Context:\n{context}\n\nQuestion: {query}\n\nAnswer:"
        answer = self.llm.generate(prompt)
        
        return answer, retrieved_docs

如果您正在用Go实现生产级RAG系统，您可能会发现这篇关于使用Ollama和Qwen3嵌入模型在Go中重新排序文本文档 的指南特别有用，以优化检索质量。

内容审核与安全

跨模态嵌入在检测跨模态的不当内容方面表现出色：

class ContentModerator:
    def __init__(self, model, processor):
        self.model = model
        self.processor = processor
        
        # 定义安全类别
        self.unsafe_categories = [
            "暴力内容",
            "成人内容",
            "仇恨图像",
            "血腥暴力",
            "显性材料"
        ]
        
        self.safe_categories = [
            "适合工作",
            "家庭友好",
            "教育内容"
        ]
    
    def moderate_image(self, image, threshold: float = 0.3):
        """检查图像是否包含不当内容"""
        # 组合所有类别
        all_categories = self.unsafe_categories + self.safe_categories
        text_inputs = [f"包含 {cat} 的图像" for cat in all_categories]
        
        # 获取预测
        inputs = self.processor(
            text=text_inputs, 
            images=image, 
            return_tensors="pt"
        )
        outputs = self.model(**inputs)
        probs = outputs.logits_per_image.softmax(dim=1)[0]
        
        # 检查不安全类别
        unsafe_scores = probs[:len(self.unsafe_categories)]
        max_unsafe_score = unsafe_scores.max().item()
        
        return {
            'is_safe': max_unsafe_score < threshold,
            'confidence': 1 - max_unsafe_score,
            'flagged_categories': [
                self.unsafe_categories[i] 
                for i, score in enumerate(unsafe_scores) 
                if score > threshold
            ]
        }

最佳实践和常见陷阱

数据预处理

适当的预处理对于最佳性能至关重要：

from torchvision import transforms

# 标准CLIP预处理
clip_transform = transforms.Compose([
    transforms.Resize(224, interpolation=transforms.InterpolationMode.BICUBIC),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(
        mean=[0.48145466, 0.4578275, 0.40821073],
        std=[0.26862954, 0.26130258, 0.27577711]
    )
])

处理偏见和公平性

跨模态模型可能会从训练数据中继承偏见：

缓解策略：

  • 在多样化的群体上进行评估
  • 在微调期间使用去偏技术
  • 实施公平感知的检索
  • 在生产中进行定期审计和监控

嵌入质量评估

在生产中监控嵌入质量：

def assess_embedding_quality(embeddings: np.ndarray):
    """计算嵌入质量的指标"""
    # 计算平均成对距离
    distances = np.linalg.norm(
        embeddings[:, None] - embeddings[None, :], 
        axis=2
    )
    avg_distance = distances.mean()
    
    # 检查聚类（低内部聚类距离）
    from sklearn.cluster import KMeans
    kmeans = KMeans(n_clusters=10)
    labels = kmeans.fit_predict(embeddings)
    
    # 计算轮廓系数
    from sklearn.metrics import silhouette_score
    score = silhouette_score(embeddings, labels)
    
    return {
        'avg_pairwise_distance': avg_distance,
        'silhouette_score': score
    }

Docker 部署示例

打包您的跨模态应用程序以实现轻松部署：

FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu22.04

# 安装Python和依赖项
RUN apt-get update && apt-get install -y python3-pip

WORKDIR /app

# 安装依赖项
COPY requirements.txt .
RUN pip3 install --no-cache-dir -r requirements.txt

# 复制应用程序代码
COPY . .

# 暴露API端口
EXPOSE 8000

# 运行API服务器
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

# api.py - 用于跨模态嵌入的FastAPI服务器
from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel
import torch
from PIL import Image
import io

app = FastAPI()

# 启动时加载模型
@app.on_event("startup")
async def load_model():
    global model, processor
    model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
    processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
    if torch.cuda.is_available():
        model = model.cuda()

class TextQuery(BaseModel):
    text: str

@app.post("/embed/text")
async def embed_text(query: TextQuery):
    inputs = processor(text=[query.text], return_tensors="pt", padding=True)
    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}
    
    with torch.no_grad():
        embeddings = model.get_text_features(**inputs)
    
    return {"embedding": embeddings.cpu().numpy().tolist()}

@app.post("/embed/image")
async def embed_image(file: UploadFile = File(...)):
    image = Image.open(io.BytesIO(await file.read()))
    inputs = processor(images=image, return_tensors="pt")
    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}
    
    with torch.no_grad():
        embeddings = model.get_image_features(**inputs)
    
    return {"embedding": embeddings.cpu().numpy().tolist()}

@app.post("/similarity")
async def compute_similarity(
    text: TextQuery, 
    file: UploadFile = File(...)
):
    # 获取两个嵌入
    image = Image.open(io.BytesIO(await file.read()))
    inputs = processor(text=[text.text], images=image, return_tensors="pt", padding=True)
    
    if torch.cuda.is_available():
        inputs = {k: v.cuda() for k, v in inputs.items()}
    
    outputs = model(**inputs)
    similarity = outputs.logits_per_image[0][0].item()
    
    return {"similarity": similarity}

未来方向

跨模态嵌入领域正在迅速发展：

更大的模态覆盖：未来的模型可能会纳入额外的模态，如触觉（触觉反馈）、气味和味觉数据，以实现真正全面的多模态理解。

改进的效率：对蒸馏和高效架构的研究将使强大的跨模态模型在边缘设备上可用。

更好的对齐：更精确地对齐模态的高级技术，包括循环一致性损失和对抗训练。

组合理解：从简单的物体识别向理解跨模态的复杂关系和组合发展。

时间建模：通过嵌入空间中的显式时间推理，更好地处理视频和时间序列数据。

有用的链接

跨模态嵌入代表了人工智能系统处理和理解信息方式的范式转变。通过打破不同类型数据之间的壁垒，这些技术使更自然和强大的AI应用成为可能。无论您是构建搜索系统、内容审核工具还是创意应用，掌握跨模态嵌入都为多模态AI的创新开辟了广阔的可能性。