多模态技术范式演进：从视觉对齐到视觉原语的完整路线

L = -Σ log(exp(sim(I_i, T_i) / τ) / Σ_j exp(sim(I_i, T_j) / τ))

import torch
import torch.nn as nn
import torch.nn.functional as F
from torchvision.models import resnet18, ResNet18_Weights

class SimpleCLIP(nn.Module):
    def __init__(self, embed_dim=256):
        super().__init__()
        # 图像编码器：ResNet-18 去除分类头
        resnet = resnet18(weights=ResNet18_Weights.IMAGENET1K_V1)
        self.image_encoder = nn.Sequential(*list(resnet.children())[:-1])
        self.image_proj = nn.Linear(512, embed_dim)
        
        # 文本编码器：简单的词嵌入 + 平均池化
        self.vocab_size = 30000
        self.text_embed = nn.Embedding(self.vocab_size, embed_dim)
        self.text_proj = nn.Linear(embed_dim, embed_dim)
        
        # 可学习的温度参数（参考 CLIP 论文）
        self.logit_scale = nn.Parameter(torch.ones([]) * np.log(1 / 0.07))
    
    def encode_image(self, images):
        # images: (batch, 3, 224, 224)
        features = self.image_encoder(images).squeeze(-1).squeeze(-1)
        return F.normalize(self.image_proj(features), dim=-1)
    
    def encode_text(self, text_ids):
        # text_ids: (batch, seq_len)
        embeddings = self.text_embed(text_ids).mean(dim=1)
        return F.normalize(self.text_proj(embeddings), dim=-1)
    
    def forward(self, images, text_ids):
        image_features = self.encode_image(images)
        text_features = self.encode_text(text_ids)
        
        # 计算图文相似度矩阵
        logit_scale = self.logit_scale.exp()
        logits_per_image = logit_scale * image_features @ text_features.T
        logits_per_text = logits_per_image.T
        
        # InfoNCE 损失（双向）
        labels = torch.arange(len(images), device=images.device)
        loss_img = F.cross_entropy(logits_per_image, labels)
        loss_txt = F.cross_entropy(logits_per_text, labels)
        return (loss_img + loss_txt) / 2

from torch.utils.data import DataLoader
from torchvision.datasets import CocoCaptions
from torchvision.transforms import Compose, Resize, CenterCrop, ToTensor, Normalize

# 数据预处理
transform = Compose([
    Resize(224), CenterCrop(224),
    ToTensor(), Normalize(mean=[0.485, 0.456, 0.406],
                          std=[0.229, 0.224, 0.225])
])

# 简化版数据集（实际应使用完整的 COCO/Flickr30k）
dataset = CocoCaptions(root='coco/val2017/', annFile='coco/annotations/captions_val2017.json')
dataloader = DataLoader(dataset, batch_size=64, shuffle=True)

# 训练
model = SimpleCLIP(embed_dim=256).cuda()
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=0.01)

for epoch in range(10):
    model.train()
    total_loss = 0
    for images, captions in dataloader:
        images = torch.stack([transform(img) for img in images]).cuda()
        # 简化的 tokenization（实际应用应使用 CLIP tokenizer）
        text_ids = torch.randint(0, 30000, (len(captions), 16)).cuda()
        
        loss = model(images, text_ids)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    
    print(f"Epoch {epoch+1}, Loss: {total_loss/len(dataloader):.4f}")

def zero_shot_classify(model, image, class_names):
    """使用训练好的模型进行零样本分类"""
    model.eval()
    
    # 编码图像
    with torch.no_grad():
        image_features = model.encode_image(image.unsqueeze(0))
    
    # 编码类别文本模板
    templates = [f"a photo of a {name}" for name in class_names]
    # 简化的文本编码（实际应用需要完整的 tokenizer）
    text_features = torch.stack([
        model.encode_text(torch.randint(0, 30000, (1, 16)).cuda())
        for _ in class_names
    ]).squeeze(1)
    
    # 计算相似度并取最大值
    similarities = (image_features @ text_features.T).squeeze(0)
    probs = F.softmax(similarities * model.logit_scale.exp(), dim=0)
    
    predicted_idx = probs.argmax().item()
    return class_names[predicted_idx], probs[predicted_idx].item()

# 使用示例
classes = ["cat", "dog", "car", "bird", "tree"]
predicted_class, confidence = zero_shot_classify(model, test_image, classes)
print(f"预测: {predicted_class} (置信度: {confidence:.2%})")

import torch
import torch.nn as nn
from timm import create_model

class VisualPrimitiveEncoder(nn.Module):
    """视觉原语编码器：从图像中提取多尺度原语特征"""
    
    def __init__(self, num_primitives=64, hidden_dim=768):
        super().__init__()
        self.num_primitives = num_primitives
        
        # 使用 ViT 作为基础编码器
        self.backbone = create_model('vit_base_patch16_224', pretrained=True)
        self.backbone.head = nn.Identity()
        
        # 原语解码头：将 backbone 特征分解为多个原语通道
        self.primitive_heads = nn.ModuleDict({
            'geometric': nn.Sequential(
                nn.Linear(hidden_dim, hidden_dim),
                nn.GELU(),
                nn.Linear(hidden_dim, num_primitives // 4)  # 16 几何原语
            ),
            'texture': nn.Sequential(
                nn.Linear(hidden_dim, hidden_dim),
                nn.GELU(),
                nn.Linear(hidden_dim, num_primitives // 4)  # 16 材质原语
            ),
            'spatial': nn.Sequential(
                nn.Linear(hidden_dim, hidden_dim),
                nn.GELU(),
                nn.Linear(hidden_dim, num_primitives // 4)  # 16 空间原语
            ),
            'temporal': nn.Sequential(
                nn.Linear(hidden_dim, hidden_dim),
                nn.GELU(),
                nn.Linear(hidden_dim, num_primitives // 4)  # 16 时序原语
            ),
        })
    
    def forward(self, images):
        # images: (batch, 3, 224, 224)
        # 提取 backbone 特征（patch 级别）
        features = self.backbone.forward_features(images)  # (batch, num_patches, hidden_dim)
        
        # 将特征分解为不同类别的原语
        primitives = {}
        for name, head in self.primitive_heads.items():
            # 对每个 patch 提取原语
            # 全局平均池化得到图级别原语
            global_feat = features.mean(dim=1)  # (batch, hidden_dim)
            primitives[name] = head(global_feat)  # (batch, num_primitives/4)
        
        return features, primitives
    
    def compute_primitive_loss(self, images, masked_images):
        """多尺度自监督预训练损失"""
        # 编码原始图像和遮罩图像
        original_features, original_primitives = self(images)
        masked_features, masked_primitives = self(masked_images)
        
        # 重建损失：遮罩版本应能重建原始原语
        recon_loss = 0
        for name in self.primitive_heads:
            recon_loss += F.mse_loss(
                masked_primitives[name],
                original_primitives[name]
            )
        
        # 对比损失：原始图像的不同增强视图应在原语空间中对齐
        # （这里省略，与 CLIP 的对比损失类似）
        
        return recon_loss

# 训练循环
def train_primitive_encoder(model, dataloader, epochs=100):
    optimizer = torch.optim.AdamW(model.parameters(), lr=3e-4, weight_decay=0.05)
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs)
    
    for epoch in range(epochs):
        model.train()
        total_loss = 0
        
        for batch in dataloader:
            images = batch['image'].cuda()
            # 生成遮罩图像（随机遮罩 40% 的 patch）
            masked_images = apply_random_mask(images, mask_ratio=0.4)
            
            loss = model.compute_primitive_loss(images, masked_images)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        
        scheduler.step()
        avg_loss = total_loss / len(dataloader)
        print(f"Epoch {epoch+1}, Loss: {avg_loss:.4f}, LR: {scheduler.get_last_lr()[0]:.6f}")
    
    return model

def apply_random_mask(images, mask_ratio=0.4):
    **"""随机遮罩**：模拟 MAE 的遮罩策略"""
    batch_size, channels, height, width = images.shape
    patch_size = 16
    num_patches = (height // patch_size) * (width // patch_size)
    
    # 随机选择要遮罩的 patch
    num_mask = int(num_patches * mask_ratio)
    mask = torch.ones(batch_size, num_patches, device=images.device)
    
    for i in range(batch_size):
        indices = torch.randperm(num_patches)[:num_mask]
        mask[i, indices] = 0
    
    # 将遮罩应用到图像（简化版，实际应用应按 patch 遮罩）
    return images * 0 + images  # 简化实现