112_跨模态微调：文本与图像联合优化

多模态学习发展时间线：
2014-2018: 早期阶段 - 特征拼接与浅层融合
2019-2020: 发展阶段 - Vision-Language Transformer出现
2021-2023: 突破阶段 - CLIP等对比学习模型兴起
2024-2025: 成熟阶段 - 跨模态微调技术体系化

# CLIP模型架构简化示意
class CLIP(nn.Module):
    def __init__(self, text_encoder, image_encoder, projection_dim=512):
        super().__init__()
        self.text_encoder = text_encoder  # 文本Transformer
        self.image_encoder = image_encoder  # ResNet或ViT
        
        # 投影层，将不同模态映射到共享空间
        self.text_projection = nn.Linear(text_encoder.output_dim, projection_dim)
        self.image_projection = nn.Linear(image_encoder.output_dim, projection_dim)
        
        # 温度参数
        self.logit_scale = nn.Parameter(torch.ones([]) * np.log(1 / 0.07))
    
    def forward(self, text_input, image_input):
        # 编码文本和图像
        text_features = self.text_encoder(text_input)
        image_features = self.image_encoder(image_input)
        
        # 投影到共享空间
        text_embeddings = self.text_projection(text_features)
        image_embeddings = self.image_projection(image_features)
        
        # 归一化特征
        text_embeddings = text_embeddings / text_embeddings.norm(dim=-1, keepdim=True)
        image_embeddings = image_embeddings / image_embeddings.norm(dim=-1, keepdim=True)
        
        # 计算温度缩放的相似度矩阵
        logit_scale = self.logit_scale.exp()
        logits_per_text = logit_scale * text_embeddings @ image_embeddings.t()
        logits_per_image = logits_per_text.t()
        
        return logits_per_text, logits_per_image

# CLIP损失函数计算
import torch.nn.functional as F

def clip_loss(logits_per_text, logits_per_image):
    # 生成真实标签（对角线）
    batch_size = logits_per_text.shape[0]
    labels = torch.arange(batch_size, device=logits_per_text.device)
    
    # 计算文本侧和图像侧的交叉熵损失
    text_loss = F.cross_entropy(logits_per_text, labels)
    image_loss = F.cross_entropy(logits_per_image, labels)
    
    # 返回平均损失
    return (text_loss + image_loss) / 2

def supervised_contrastive_loss(features, labels, temperature=0.1):
    # 归一化特征
    features = F.normalize(features, dim=1)
    
    # 计算相似度矩阵
    sim_matrix = torch.matmul(features, features.t()) / temperature
    
    # 创建掩码：相同标签为正样本
    mask = torch.eq(labels.unsqueeze(0), labels.unsqueeze(1)).float()
    mask = mask.fill_diagonal_(0)  # 排除自身
    
    # 计算正样本对数概率
    exp_sim = torch.exp(sim_matrix)
    sum_exp = torch.sum(exp_sim, dim=1, keepdim=True) - exp_sim.diag().unsqueeze(1)
    log_prob = sim_matrix - torch.log(sum_exp)
    
    # 计算损失
    mean_log_prob_pos = (mask * log_prob).sum(1) / mask.sum(1)
    loss = -mean_log_prob_pos.mean()
    
    return loss

# 模态间翻译约束示例
class CrossModalTranslation(nn.Module):
    def __init__(self, text_dim, image_dim):
        super().__init__()
        # 文本到图像的翻译网络
        self.text_to_image = nn.Sequential(
            nn.Linear(text_dim, (text_dim + image_dim) // 2),
            nn.ReLU(),
            nn.Linear((text_dim + image_dim) // 2, image_dim)
        )
        
        # 图像到文本的翻译网络
        self.image_to_text = nn.Sequential(
            nn.Linear(image_dim, (text_dim + image_dim) // 2),
            nn.ReLU(),
            nn.Linear((text_dim + image_dim) // 2, text_dim)
        )
    
    def forward(self, text_features, image_features):
        # 双向翻译
        translated_text = self.image_to_text(image_features)
        translated_image = self.text_to_image(text_features)
        
        # 计算重构损失
        text_recon_loss = F.mse_loss(translated_text, text_features)
        image_recon_loss = F.mse_loss(translated_image, image_features)
        
        return (text_recon_loss + image_recon_loss) / 2

# 动态温度参数实现
class DynamicTemperature(nn.Module):
    def __init__(self, initial_temperature=0.07, min_temp=0.01, max_temp=0.1):
        super().__init__()
        self.base_temp = nn.Parameter(torch.tensor(initial_temperature))
        self.min_temp = min_temp
        self.max_temp = max_temp
        self.global_step = 0
    
    def forward(self, logits_per_text, logits_per_image, epoch=None):
        # 根据训练进度或样本难度动态调整温度
        if epoch is not None:
            # 基于训练轮数的动态调整
            progress = min(1.0, epoch / 100.0)
            temperature = self.min_temp + (self.max_temp - self.min_temp) * (1 - progress)
        else:
            # 基于当前batch难度的动态调整
            text_mean_pos = logits_per_text.diag().mean()
            image_mean_pos = logits_per_image.diag().mean()
            mean_pos = (text_mean_pos + image_mean_pos) / 2
            
            # 难度自适应调整
            if mean_pos > 20:
                temperature = min(self.max_temp, self.base_temp * 1.1)
            elif mean_pos < 10:
                temperature = max(self.min_temp, self.base_temp * 0.9)
            else:
                temperature = self.base_temp
        
        self.global_step += 1
        return temperature

# 多模态模型中的LoRA实现示例
class LoRALayer(nn.Module):
    def __init__(self, in_dim, out_dim, rank=8, alpha=16):
        super().__init__()
        self.W = nn.Linear(in_dim, out_dim, bias=False)
        self.lora_A = nn.Linear(in_dim, rank, bias=False)
        self.lora_B = nn.Linear(rank, out_dim, bias=False)
        self.alpha = alpha / rank  # 缩放因子
        
        # 初始化
        nn.init.normal_(self.lora_A.weight, std=1e-4)
        nn.init.zeros_(self.lora_B.weight)
    
    def forward(self, x):
        # 原始权重 + LoRA权重
        return self.W(x) + self.alpha * self.lora_B(self.lora_A(x))

# 为CLIP模型的文本和图像编码器添加LoRA
class CLIPWithLoRA(CLIP):
    def __init__(self, text_encoder, image_encoder, projection_dim=512, lora_rank=8):
        super().__init__(text_encoder, image_encoder, projection_dim)
        
        # 为文本编码器添加LoRA
        for layer in self.text_encoder.transformer.layers:
            layer.self_attn.q_proj = LoRALayer(layer.self_attn.q_proj.in_features,
                                              layer.self_attn.q_proj.out_features,
                                              rank=lora_rank)
            layer.self_attn.k_proj = LoRALayer(layer.self_attn.k_proj.in_features,
                                              layer.self_attn.k_proj.out_features,
                                              rank=lora_rank)
            layer.self_attn.v_proj = LoRALayer(layer.self_attn.v_proj.in_features,
                                              layer.self_attn.v_proj.out_features,
                                              rank=lora_rank)
        
        # 为图像编码器添加LoRA
        # 类似地为ViT或ResNet的关键层添加LoRA...

# 图文检索任务的微调示例
class RetrievalFinetuner:
    def __init__(self, clip_model, learning_rate=5e-5):
        self.model = clip_model
        self.optimizer = torch.optim.AdamW(
            self.model.parameters(),
            lr=learning_rate,
            weight_decay=0.01
        )
        self.scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
            self.optimizer,
            T_max=100
        )
    
    def train_step(self, images, texts, hard_negatives=None):
        self.model.train()
        self.optimizer.zero_grad()
        
        # 计算基本的CLIP损失
        logits_per_text, logits_per_image = self.model(texts, images)
        loss = clip_loss(logits_per_text, logits_per_image)
        
        # 如果有难负样本，计算额外的损失
        if hard_negatives is not None:
            hard_logits = self.model.compute_similarity(
                self.model.encode_text(texts),
                self.model.encode_image(hard_negatives)
            )
            # 确保正样本得分高于难负样本
            margin = 0.5
            hard_loss = F.relu(margin + hard_logits.diag() - logits_per_text.diag()).mean()
            loss += 0.5 * hard_loss
        
        loss.backward()
        self.optimizer.step()
        self.scheduler.step()
        
        return loss.item()

# 域对抗训练在多模态微调中的应用
class DomainAdversarialTrainer:
    def __init__(self, clip_model, domain_discriminator):
        self.model = clip_model
        self.domain_discriminator = domain_discriminator
        self.clip_optimizer = torch.optim.AdamW(self.model.parameters())
        self.disc_optimizer = torch.optim.AdamW(self.domain_discriminator.parameters())
    
    def train_step(self, source_images, source_texts, target_images, target_texts):
        # 1. 训练领域判别器
        self.disc_optimizer.zero_grad()
        
        # 提取源域和目标域的特征
        source_features = self.model.encode_text(source_texts)
        target_features = self.model.encode_text(target_texts)
        
        # 判别器预测
        source_preds = self.domain_discriminator(source_features.detach())
        target_preds = self.domain_discriminator(target_features.detach())
        
        # 计算判别器损失
        disc_source_loss = F.binary_cross_entropy(source_preds, torch.ones_like(source_preds))
        disc_target_loss = F.binary_cross_entropy(target_preds, torch.zeros_like(target_preds))
        disc_loss = disc_source_loss + disc_target_loss
        
        disc_loss.backward()
        self.disc_optimizer.step()
        
        # 2. 训练CLIP模型（对抗训练）
        self.clip_optimizer.zero_grad()
        
        # 计算CLIP损失
        logits_per_text, logits_per_image = self.model(
            torch.cat([source_texts, target_texts]),
            torch.cat([source_images, target_images])
        )
        clip_loss_val = clip_loss(logits_per_text, logits_per_image)
        
        # 计算对抗损失（目标是混淆判别器）
        all_features = self.model.encode_text(torch.cat([source_texts, target_texts]))
        adv_preds = self.domain_discriminator(all_features)
        
        # 源域应该被预测为0，目标域应该被预测为1（对抗）
        source_adv_labels = torch.zeros(source_preds.shape[0], device=source_preds.device)
        target_adv_labels = torch.ones(target_preds.shape[0], device=target_preds.device)
        adv_labels = torch.cat([source_adv_labels, target_adv_labels])
        
        adv_loss = F.binary_cross_entropy(adv_preds, adv_labels)
        
        # 总损失
        total_loss = clip_loss_val - 0.1 * adv_loss  # 权重系数控制对抗强度
        
        total_loss.backward()
        self.clip_optimizer.step()
        
        return clip_loss_val.item(), disc_loss.item(), adv_loss.item()

# 多模态数据增强示例
class MultimodalDataAugmentation:
    def __init__(self):
        # 图像增强变换
        self.image_transforms = transforms.Compose([
            transforms.RandomResizedCrop(224, scale=(0.8, 1.0)),
            transforms.RandomHorizontalFlip(p=0.5),
            transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2),
            transforms.RandomRotation(10),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
        ])
        
        # 同义词词典（简化示例）
        self.synonyms = {
            '狗': ['犬', '小狗', '宠物狗'],
            '猫': ['猫咪', '小猫', '宠物猫']
        }
    
    def augment_image(self, image):
        return self.image_transforms(image)
    
    def augment_text(self, text):
        # 同义词替换增强
        words = text.split()
        augmented_words = []
        
        for word in words:
            if word in self.synonyms and random.random() < 0.3:  # 30%概率替换
                augmented_words.append(random.choice(self.synonyms[word]))
            else:
                augmented_words.append(word)
        
        return ' '.join(augmented_words)
    
    def augment_pair(self, image, text, num_augmentations=3):
        # 为一对样本生成多个增强版本
        augmented_pairs = []
        
        for _ in range(num_augmentations):
            aug_image = self.augment_image(image)
            aug_text = self.augment_text(text)
            augmented_pairs.append((aug_image, aug_text))
        
        return augmented_pairs

# 跨模态微调配置示例
config = {
    # 模型配置
    'model_name': 'openai/clip-vit-large-patch14',  # 基础模型
    'finetune_method': 'lora',  # 微调方法：'full', 'lora', 'adapter', 'prompt'
    'lora_rank': 8,  # LoRA的秩参数
    'freeze_encoder': True,  # 是否冻结编码器
    
    # 训练配置
    'batch_size': 64,  # 批量大小
    'gradient_accumulation_steps': 4,  # 梯度累积步数
    'max_epochs': 50,  # 最大训练轮数
    'early_stopping_patience': 10,  # 早停耐心值
    
    # 优化器配置
    'optimizer': 'adamw',  # 优化器
    'learning_rate': 5e-5,  # 初始学习率
    'weight_decay': 0.01,  # 权重衰减
    'lr_scheduler': 'cosine',  # 学习率调度器
    'warmup_ratio': 0.1,  # 预热比例
    'gradient_clip_val': 1.0,  # 梯度裁剪值
    
    # 损失函数配置
    'loss_type': 'contrastive',  # 损失函数类型
    'temperature': 0.07,  # 温度参数
    'use_dynamic_temperature': True,  # 是否使用动态温度
    
    # 数据配置
    'image_size': 224,  # 图像大小
    'max_length': 77,  # 文本最大长度
    'use_augmentation': True,  # 是否使用数据增强
    
    # 硬件配置
    'precision': '16-mixed',  # 精度：'32-true', '16-mixed', 'bf16-mixed'
    'num_workers': 4,  # 数据加载工作线程数
    'pin_memory': True,  # 是否固定内存
    
    # 评估配置
    'evaluation_strategy': 'epoch',  # 评估策略
    'save_strategy': 'epoch',  # 保存策略
    'logging_strategy': 'steps',  # 日志策略
    'logging_steps': 100,  # 日志记录步数
}

# 训练监控与可视化示例
import wandb
from pytorch_lightning import Trainer
from pytorch_lightning.loggers import WandbLogger
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint

# 初始化Wandb记录器
wandb_logger = WandbLogger(
    project="cross-modal-finetuning",
    name="clip-lora-retrieval",
    config=config
)

# 早停回调
early_stopping = EarlyStopping(
    monitor="val_recall_at_1",  # 监控的指标
    patience=config['early_stopping_patience'],
    mode="max"  # 最大化评估指标
)

# 模型检查点回调
checkpoint_callback = ModelCheckpoint(
    monitor="val_recall_at_1",
    dirpath="./checkpoints/",
    filename="clip-best-{epoch:02d}-{val_recall_at_1:.4f}",
    save_top_k=3,  # 保存前3个最佳模型
    mode="max"
)

# 初始化训练器
trainer = Trainer(
    max_epochs=config['max_epochs'],
    accelerator="gpu",
    devices=1,
    precision=config['precision'],
    logger=wandb_logger,
    callbacks=[early_stopping, checkpoint_callback],
    gradient_clip_val=config['gradient_clip_val'],
    accumulate_grad_batches=config['gradient_accumulation_steps'],
    log_every_n_steps=config['logging_steps']
)

# 训练模型
trainer.fit(model, train_dataloader, val_dataloader)

# 评估模型
trainer.test(model, test_dataloader)

# 视觉提示学习示例
class VisualPromptLearner(nn.Module):
    def __init__(self, num_tokens=10, hidden_size=768):
        super().__init__()
        # 可学习的视觉提示
        self.prompt_tokens = nn.Parameter(
            torch.randn(1, num_tokens, hidden_size)
        )
        # 用于初始化提示的线性层
        self.init_proj = nn.Linear(hidden_size, hidden_size)
        # 视觉提示的位置嵌入
        self.prompt_pos_embedding = nn.Parameter(
            torch.randn(1, num_tokens, hidden_size)
        )
    
    def forward(self, image_features, prompt_init=None):
        # 如果提供了初始化特征，则使用它来初始化提示
        if prompt_init is not None:
            with torch.no_grad():
                # 使用初始特征来引导提示初始化
                self.prompt_tokens.data = self.init_proj(prompt_init).mean(dim=0, keepdim=True)
        
        # 将提示与图像特征拼接
        batch_size = image_features.shape[0]
        prompts = self.prompt_tokens.expand(batch_size, -1, -1)
        prompt_pos = self.prompt_pos_embedding.expand(batch_size, -1, -1)
        
        # 应用位置编码到提示
        prompts_with_pos = prompts + prompt_pos
        
        # 将提示添加到图像特征的开头
        combined_features = torch.cat([prompts_with_pos, image_features], dim=1)
        
        return combined_features