多模态学习（一）：CLIP 视觉-语言预训练

import torch
import torch.nn.functional as F

def compute_modality_gap(image_emb: torch.Tensor, text_emb: torch.Tensor) -> float:
    """计算视觉与语言嵌入空间的模态鸿沟
    通过两个模态嵌入的平均向量之间的欧氏距离来量化
    """
    image_mean = image_emb.mean(dim=0)  # [D]
    text_mean = text_emb.mean(dim=0)    # [D]
    gap = torch.norm(image_mean - text_mean, p=2).item()
    return gap

# 加载 CLIP 模型
import clip
model, preprocess = clip.load("ViT-B/32", device="cpu")
print(f"Image encoder: {model.visual}")
print(f"Text encoder: {model.transformer}")

# 多模态融合的不同范式对比
multimodal_approaches = {
    "Early Fusion": "在输入层拼接多模态特征，如将图像 patch 与 text token 拼接后输入 Transformer",
    "Late Fusion": "各模态独立编码后在决策层融合，如 CLIP 的双塔架构",
    "Intermediate Fusion": "在中间层进行跨模态注意力交互，如 Flamingo、BLIP",
    "Shared Encoder": "共享编码器处理多模态输入，如 ImageBind"
}

for name, desc in multimodal_approaches.items():
    print(f"{name}: {desc}")

# CLIP 选择 Late Fusion 的核心原因：
# 1. 训练效率高，两个编码器可以独立扩展
# 2. 推理灵活，可以分别缓存图像/文本特征
# 3. 零样本能力强，文本端保持开放词汇

import torch
import torch.nn as nn

class CLIPEncoder(nn.Module):
    """CLIP 双塔架构的简化实现"""

    def __init__(self, embed_dim=512):
        super().__init__()
        # 图像塔: ResNet-50 骨干 + 投影头
        self.image_encoder = ResNetEncoder(out_channels=2048)
        self.image_projection = nn.Linear(2048, embed_dim, bias=False)

        # 文本塔: Transformer 编码器 + 投影头
        self.text_encoder = TextTransformer(
            vocab_size=49408, context_length=76,
            embed_dim=512, heads=8, layers=12
        )
        self.text_projection = nn.Linear(512, embed_dim, bias=False)

    def forward(self, images, texts):
        image_features = self.encode_image(images)
        text_features = self.encode_text(texts)
        return image_features, text_features

    def encode_image(self, x):
        x = self.image_encoder(x)           # [B, 2048]
        x = self.image_projection(x)        # [B, embed_dim]
        return F.normalize(x, dim=-1)       # 单位向量

    def encode_text(self, x):
        x = self.text_encoder(x)            # [B, 512]
        x = self.text_projection(x)         # [B, embed_dim]
        return F.normalize(x, dim=-1)       # 单位向量

# CLIP 图像编码器的两种选择对比
import torch.nn as nn

# 方案一: ResNet 编码器
def build_resnet_encoder(arch="resnet50"):
    """ResNet 编码器: 修改全局平均池化后的输出"""
    import torchvision.models as models
    model = getattr(models, arch)(weights=None)
    # 移除原始分类头
    model.fc = nn.Identity()
    # 特征维度: ResNet-50 -> 2048, ResNet-101 -> 2048
    return model

# 方案二: Vision Transformer 编码器
def build_vit_encoder(patch_size=32, embed_dim=768):
    """ViT 编码器: 将图像分割为 patch 序列
    patch_size=32: 每张图片分割为 (224/32)^2 = 49 个 patch
    patch_size=16: 每张图片分割为 (224/16)^2 = 196 个 patch
    """
    return VisionTransformer(
        image_size=224, patch_size=patch_size,
        embed_dim=embed_dim, depth=12, num_heads=12
    )

# 训练规模与性能关系:
# ViT-B/32 (86M参数)  -> 63.2% zero-shot
# ViT-B/16 (151M参数) -> 67.1% zero-shot
# ViT-L/14 (307M参数) -> 75.5% zero-shot
# ViT-H/14 (632M参数) -> 78.0% zero-shot

import torch
import torch.nn as nn
import torch.nn.functional as F

class InfoNCELoss(nn.Module):
    """CLIP 的对比损失实现"""

    def __init__(self, temperature=0.07):
        super().__init__()
        self.logit_scale = nn.Parameter(torch.ones([]) * torch.log(torch.tensor(1.0 / temperature)))

    def forward(self, image_features, text_features):
        """计算对称的 InfoNCE 损失
        image_features: [batch_size, embed_dim] - L2 归一化
        text_features:  [batch_size, embed_dim] - L2 归一化
        """
        # 计算相似度矩阵
        logit_scale = self.logit_scale.exp()
        logits_per_image = logit_scale * image_features @ text_features.t()  # [N, N]
        logits_per_text = logits_per_image.t()                               # [N, N]

        # 标签: 对角线为正样本
        batch_size = image_features.size(0)
        labels = torch.arange(batch_size, device=image_features.device)

        # 双向交叉熵损失
        loss_i = F.cross_entropy(logits_per_image, labels)
        loss_t = F.cross_entropy(logits_per_text, labels)
        loss = (loss_i + loss_t) / 2

        return loss

    def get_temperature(self):
        return self.logit_scale.exp().item()

# 深入理解温度参数的作用
import torch
import matplotlib.pyplot as plt
import numpy as np

def show_temperature_effect():
    """可视化温度参数对相似度分布的影响"""
    sim = torch.linspace(-1, 1, 100)
    temperatures = [0.01, 0.07, 0.5, 1.0]

    for tau in temperatures:
        scaled = sim / tau
        prob = F.softmax(scaled, dim=0)
        plt.plot(sim.numpy(), prob.numpy(), label=f"tau={tau}")

    plt.xlabel("余弦相似度")
    plt.ylabel("softmax 概率")
    plt.title("温度参数对分布的影响")
    plt.legend()
    plt.show()

# 关键洞察:
# tau = 0.01: 分布极尖锐，只有最高相似度获得显著概率
# tau = 0.07: CLIP 默认值，平衡了区分度和梯度流
# tau = 0.5:  分布较平缓，难样本也能获得一定梯度
# tau = 1.0:  近似均匀分布，几乎无法学习

# Batch size 的影响:
# 更大的 batch 提供更多的负样本（N^2 - N 对），
# 但也会增加内存需求。CLIP 训练使用 batch_size=32768。

import torch
import clip

def zero_shot_classification(model, preprocess, image, class_names):
    """使用 CLIP 进行零样本图像分类
    image: PIL Image 或 tensor
    class_names: 类别名称列表
    """
    device = next(model.parameters()).device

    # 1. 构建提示模板
    prompts = [f"a photo of a {c}" for c in class_names]

    # 2. 文本编码
    text_tokens = clip.tokenize(prompts).to(device)
    with torch.no_grad():
        text_features = model.encode_text(text_tokens)
        text_features = text_features / text_features.norm(dim=-1, keepdim=True)

    # 3. 图像编码
    image_input = preprocess(image).unsqueeze(0).to(device)
    with torch.no_grad():
        image_features = model.encode_image(image_input)
        image_features = image_features / image_features.norm(dim=-1, keepdim=True)

    # 4. 计算相似度并预测
    similarity = (100.0 * image_features @ text_features.t()).softmax(dim=-1)
    top_k = 5
    probs, indices = similarity[0].topk(top_k)

    for prob, idx in zip(probs, indices):
        print(f"  {class_names[idx]:>16s}: {prob:.2f}%")

    return class_names[indices[0]]

# 集成提示工程（Ensemble Prompting）提升零样本性能
def ensemble_prompt_classification(model, preprocess, image, class_names):
    """使用多个提示模板集成，显著提升零样本准确率"""

    prompt_templates = [
        "a bad photo of a {}.",
        "a photo of many {}.",
        "a sculpture of a {}.",
        "a photo of the hard to see {}.",
        "a low resolution photo of the {}.",
        "a rendering of a {}.",
        "a cropped photo of the {}.",
        "a close-up photo of a {}.",
        "a black and white photo of a {}.",
        "a clean photo of a {}.",
        "a good photo of a {}.",
    ]

    device = next(model.parameters()).device

    # 构建所有提示
    prompts = [t.format(c) for c in class_names for t in prompt_templates]
    text_tokens = clip.tokenize(prompts).to(device)

    with torch.no_grad():
        text_features = model.encode_text(text_tokens)
        text_features = text_features / text_features.norm(dim=-1, keepdim=True)

    # reshape 并平均
    text_features = text_features.view(len(class_names), len(prompt_templates), -1)
    text_features = text_features.mean(dim=0)  # [n_classes, embed_dim]
    text_features = text_features / text_features.norm(dim=-1, keepdim=True)

    # 图像编码
    image_input = preprocess(image).unsqueeze(0).to(device)
    with torch.no_grad():
        image_features = model.encode_image(image_input)
        image_features /= image_features.norm(dim=-1, keepdim=True)

    similarity = 100.0 * image_features @ text_features.t()
    return similarity.softmax(dim=-1)

# CLIP 训练数据流水线
import torch
from torch.utils.data import Dataset, DataLoader
from torchvision import transforms

class CLIPDataset(Dataset):
    """CLIP 训练数据集: (image, text) 对"""

    def __init__(self, data_path, tokenizer, transform):
        self.samples = load_web_dataset(data_path)  # [(img_path, text), ...]
        self.tokenizer = tokenizer
        self.transform = transform

    def __len__(self):
        return len(self.samples)

    def __getitem__(self, idx):
        img_path, text = self.samples[idx]

        # 图像预处理: 仅随机裁剪和翻转
        # 注意: 不使用颜色抖动等破坏性增强
        image = load_image(img_path)
        image = self.transform(image)

        # 文本编码
        tokens = self.tokenizer(text, context_length=76)

        return image, tokens

# CLIP 的数据增强策略 (极简)
clip_train_transform = transforms.Compose([
    transforms.RandomResizedCrop(224, scale=(0.9, 1.0)),
    transforms.RandomHorizontalFlip(),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.48145466, 0.4578275, 0.40821073],
                         std=[0.26862954, 0.26130258, 0.27577711]),
])

# 训练配置与优化器设置
import torch
import torch.nn as nn

def build_clip_optimizer(model, lr=5e-4, weight_decay=0.2, warmup_steps=2000):
    """CLIP 训练的优化器配置
    - 使用 AdamW，而非原始 Adam
    - 较大的 weight_decay 防止过拟合
    - Warmup 阶段稳定训练初期
    """
    optimizer = torch.optim.AdamW(
        model.parameters(),
        lr=lr,
        weight_decay=weight_decay,
        betas=(0.9, 0.98),
        eps=1e-6
    )

    # Cosine 衰减学习率调度
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
        optimizer, T_max=total_steps, eta_min=0
    )

    # Warmup 学习率
    warmup_scheduler = torch.optim.lr_scheduler.LambdaLR(
        optimizer, lr_lambda=lambda step: min(step / warmup_steps, 1.0)
    )

    return optimizer, scheduler, warmup_scheduler

# 原始 CLIP 训练超参数:
# epochs: 32
# batch_size: 32768 (256 GPU * 128 per GPU)
# lr: 1e-3 (with linear scaling)
# weight_decay: 0.2
# warmup: 2000 steps
# 总训练步数: ~380K
# 硬件: 256x V100 GPU

# 分析 CLIP 的社会偏差
import torch
import clip
import matplotlib.pyplot as plt

def analyze_clip_bias(model, preprocess, test_images, professions, demographics):
    """检测 CLIP 在职业-人口统计关联中的偏差"""
    device = next(model.parameters()).device

    results = {}
    for prof in professions:
        for demo in demographics:
            prompt = clip.tokenize(f"a photo of a {demo} {prof}").to(device)
            with torch.no_grad():
                text_feat = model.encode_text(prompt)
                text_feat = text_feat / text_feat.norm(dim=-1, keepdim=True)

            img_feats = []
            for img in test_images:
                img_input = preprocess(img).unsqueeze(0).to(device)
                with torch.no_grad():
                    feat = model.encode_image(img_input)
                    feat = feat / feat.norm(dim=-1, keepdim=True)
                img_feats.append(feat)

            img_feats = torch.cat(img_feats, dim=0)
            sim = (img_feats @ text_feat.t()).mean().item()
            results[f"{demo}-{prof}"] = sim

    # 可视化偏差
    for prof in professions:
        values = [results[f"{d}-{prof}"] for d in demographics]
        plt.bar(demographics, values)
        plt.title(f"CLIP 对 '{prof}' 的偏差分析")
        plt.ylabel("平均相似度")
        plt.show()

# 已知偏差示例:
# CLIP 倾向于将 "doctor" 与男性关联
# 倾向于将 "nurse" 与女性关联
# 在某些种族分类上表现不均

# 测试 CLIP 的反事实推理能力
import torch
import clip

def test_counterfactual_reasoning(model, preprocess):
    """测试 CLIP 对空间关系和否定的理解"""
    device = next(model.parameters()).device

    # 测试 1: 空间关系
    # 生成/加载包含红球和蓝球的图像
    spatial_tests = [
        ("red ball on left of blue ball", 0.0),  # 期望高相似度
        ("blue ball on left of red ball", 0.0),  # 期望低相似度
    ]

    # 测试 2: 否定理解
    negation_tests = [
        ("a photo without any dog", 0.0),
        ("a photo with a dog", 0.0),
    ]

    # 测试 3: 数量理解
    counting_tests = [
        ("a photo of one cat", 0.0),
        ("a photo of three cats", 0.0),
    ]

    # CLIP 在这些任务上的表现:
    # 空间关系: 准确率接近随机 (约 50%)
    # 否定理解: 经常失败，否定词影响微弱
    # 数量理解: 对 1 vs 2-3 有一定区分力，但对更大数字无效
    print("CLIP 反事实推理能力评估:")
    print("  空间关系: ~50% (随机水平)")
    print("  否定理解: ~55% (轻微优于随机)")
    print("  数量理解: 1 vs 多: ~65%, 2 vs 3: ~55%")
    print("  颜色+形状组合: ~75% (有一定能力)")

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

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

# 2. 零样本图像分类
def classify_image(model, processor, image, labels):
    """使用 CLIP 进行零样本分类"""
    # 构造提示
    texts = [f"a photo of a {label}" for label in labels]

    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)

    for label, prob in zip(labels, probs[0]):
        print(f"  {label}: {prob.item():.2%}")

# 3. 使用
image = Image.open("test.jpg")
classify_image(model, processor, image, ["dog", "cat", "bird", "car", "tree"])

from transformers import CLIPProcessor, CLIPModel
import torch
from PIL import Image
import faiss
import numpy as np

class CLIPIImageSearch:
    """基于 CLIP + FAISS 的图像检索系统"""

    def __init__(self, model_name="openai/clip-vit-base-patch32"):
        self.model = CLIPModel.from_pretrained(model_name)
        self.processor = CLIPProcessor.from_pretrained(model_name)
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model.to(self.device)
        self.index = None
        self.image_ids = []

    def build_index(self, images: list, ids: list):
        """构建图像索引
        images: PIL Image 列表
        ids: 对应的图像标识符
        """
        all_features = []
        for img in images:
            inputs = self.processor(images=img, return_tensors="pt")
            with torch.no_grad():
                feat = self.model.get_image_features(**inputs)
                feat = feat / feat.norm(dim=-1, keepdim=True)
            all_features.append(feat.cpu().numpy())

        features = np.vstack(all_features).astype("float32")

        # 构建 FAISS 索引 (余弦相似度 = 内积 on unit vectors)
        dim = features.shape[1]
        self.index = faiss.IndexFlatIP(dim)
        self.index.add(features)
        self.image_ids = ids
        print(f"索引构建完成: {len(ids)} 张图像, 维度 {dim}")

    def search(self, query_text: str, top_k: int = 5):
        """文本搜索图像"""
        inputs = self.processor(text=query_text, return_tensors="pt", padding=True)
        with torch.no_grad():
            text_feat = self.model.get_text_features(**inputs)
            text_feat = text_feat / text_feat.norm(dim=-1, keepdim=True)

        query_vec = text_feat.cpu().numpy().astype("float32")
        scores, indices = self.index.search(query_vec, top_k)

        results = []
        for score, idx in zip(scores[0], indices[0]):
            results.append({"id": self.image_ids[idx], "score": float(score)})

        return results

# 使用示例:
# searcher = CLIPIImageSearch()
# searcher.build_index(images, image_ids)
# results = searcher.search("a cat sitting on a sofa", top_k=10)