多模态学习（二）：视觉问答与图文生成

import json
from PIL import Image

class VQADataset:
    """VQA v2 数据集加载器"""
    def __init__(self, questions_path: str, annotations_path: str):
        with open(questions_path) as f:
            self.questions = json.load(f)["questions"]
        with open(annotations_path) as f:
            self.annotations = json.load(f)["annotations"]

        # 构建 answer-to-label 映射
        self.answer2label = {
            a["answer"]: a["answer_id"]
            for a in json.load(open("answer2label.json"))["answers"]
        }

    def __getitem__(self, idx: int):
        q = self.questions[idx]
        a = self.annotations[idx]
        # 统计众数答案作为标签
        answer_counts = {}
        for annot in a["multiple_choice_answer"]:
            answer_counts[annot] = answer_counts.get(annot, 0) + 1
        label = max(answer_counts, key=answer_counts.get)
        return q["question"], a["image_id"], label

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

from torchvision.datasets import VQAv2
from torch.utils.data import DataLoader

# 使用 torchvision 内置 VQA v2 数据集
dataset = VQAv2(
    root="./data",
    annFile="./data/annotations/v2_mscoco_train2014_annotations.json",
    quesFile="./data/annotations/v2_OpenEnded_mscoco_train2014_questions.json",
)

dataloader = DataLoader(
    dataset,
    batch_size=32,
    shuffle=True,
    collate_fn=lambda batch: {
        "images": [b[0] for b in batch],
        "questions": [b[1] for b in batch],
        "targets": [b[2] for b in batch],
    },
)

print(f"VQA v2 训练集样本数: {len(dataset)}")

import torch
import torch.nn as nn
import torchvision.models as models

class EarlyVQA(nn.Module):
    """早期 CNN+RNN VQA 模型"""
    def __init__(self, vocab_size: int, embed_dim: int,
                 hidden_dim: int, num_answers: int):
        super().__init__()
        # CNN: 预训练 ResNet 提取视觉特征
        resnet = models.resnet101(pretrained=True)
        self.visual_encoder = nn.Sequential(
            *list(resnet.children())[:-2],  # 去掉 FC 和 GAP
            nn.AdaptiveAvgPool2d(1),
            nn.Flatten()
        )
        # 冻结 CNN 前几层
        for name, param in self.visual_encoder.named_parameters():
            if "layer1" in name or "conv1" in name:
                param.requires_grad = False

        # RNN: LSTM 编码问题
        self.word_embed = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(embed_dim, hidden_dim,
                            batch_first=True)

        # 融合与分类
        self.fusion = nn.Linear(2048 + hidden_dim, 1024)
        self.classifier = nn.Sequential(
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(1024, num_answers)
        )

    def forward(self, images, questions, lengths):
        # 视觉特征 [B, 2048]
        v_feat = self.visual_encoder(images)
        # 文本特征 [B, hidden_dim]
        q_embed = self.word_embed(questions)
        _, (h_n, _) = self.lstm(q_embed)
        q_feat = h_n[-1]
        # 拼接融合
        fused = torch.cat([v_feat, q_feat], dim=1)
        return self.classifier(self.fusion(fused))

import torch
import torch.nn as nn

class MCBPool(nn.Module):
    """Multimodal Compact Bilinear Pooling (简化版)"""
    def __init__(self, input_dim: int, output_dim: int,
                 sum_pool: bool = True):
        super().__init__()
        self.output_dim = output_dim
        self.sum_pool = sum_pool
        # Count Sketch 投影矩阵
        self.hash_v = nn.Parameter(
            torch.randint(0, output_dim, (input_dim,)),
            requires_grad=False
        )
        self.sign_v = nn.Parameter(
            torch.randint(0, 2, (input_dim,)) * 2 - 1,
            requires_grad=False
        )
        self.hash_q = nn.Parameter(
            torch.randint(0, output_dim, (input_dim,)),
            requires_grad=False
        )
        self.sign_q = nn.Parameter(
            torch.randint(0, 2, (input_dim,)) * 2 - 1,
            requires_grad=False
        )

    def count_sketch(self, x, hash_fn, sign_fn):
        """将输入投影到 sketch 空间"""
        B, D = x.shape
        output = torch.zeros(B, self.output_dim, device=x.device)
        idx = hash_fn.expand(B, -1)
        signs = sign_fn.expand(B, -1)
        output.scatter_add_(1, idx, x * signs)
        return output

    def forward(self, v, q):
        # 对视觉和文本特征分别做 Count Sketch
        fft_v = torch.fft.rfft(self.count_sketch(v, self.hash_v, self.sign_v))
        fft_q = torch.fft.rfft(self.count_sketch(q, self.hash_q, self.sign_q))
        # 频域卷积 = 频域逐元素相乘
        fused = torch.fft.irfft(fft_v * fft_q)
        # 开方 + 归一化 (sqrt + l2 norm)
        fused = torch.sign(fused) * torch.sqrt(torch.abs(fused) + 1e-8)
        if self.sum_pool:
            return fused.sum(dim=1)
        return fused

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

class StackedAttention(nn.Module):
    """堆叠注意力网络 (SAN)"""
    def __init__(self, hidden_dim: int, num_stacks: int = 2):
        super().__init__()
        self.num_stacks = num_stacks
        # 注意力打分网络
        self.w_q = nn.Linear(hidden_dim, hidden_dim)
        self.w_v = nn.Linear(hidden_dim, hidden_dim)
        self.w_a = nn.Linear(hidden_dim, 1)
        self.tanh = nn.Tanh()
        self.fc = nn.Linear(hidden_dim, hidden_dim)
        self.dropout = nn.Dropout(0.5)

    def forward(self, v: torch.Tensor, q: torch.Tensor):
        """v: [B, N, D] 图像区域特征, q: [B, D] 问题特征"""
        u = q.clone()  # 初始化为问题特征
        for _ in self.num_stacks:
            # 计算注意力分数
            h_v = self.w_v(v)  # [B, N, D]
            h_q = self.w_q(u).unsqueeze(1)  # [B, 1, D]
            scores = self.w_a(self.tanh(h_v + h_q)).squeeze(-1)  # [B, N]
            attn = F.softmax(scores, dim=-1)  # [B, N]
            # 加权求和
            v_att = torch.bmm(attn.unsqueeze(1), v).squeeze(1)  # [B, D]
            # 残差更新
            u = u + self.dropout(self.fc(v_att))
        return u  # 返回注意力增强后的问题表示

import torch
import torch.nn as nn

class BottomUpTopDown(nn.Module):
    """自顶向下注意力 + 目标检测区域特征"""
    def __init__(self, visual_dim: int, text_dim: int,
                 num_heads: int = 16):
        super().__init__()
        self.num_heads = num_heads
        head_dim = visual_dim // num_heads
        # 问题到注意力的投影
        self.w_q = nn.Linear(text_dim, visual_dim)
        # key/value 投影
        self.w_k = nn.Linear(visual_dim, visual_dim)
        self.w_v = nn.Linear(visual_dim, visual_dim)
        self.scale = head_dim ** -0.5

        # 输出投影
        self.out_proj = nn.Linear(visual_dim, visual_dim)
        self.layer_norm = nn.LayerNorm(visual_dim)

    def forward(self, regions: torch.Tensor, question: torch.Tensor):
        """regions: [B, K, D] K 个检测区域, question: [B, D]"""
        B, K, D = regions.shape
        # 多头注意力
        q = self.w_q(question).view(B, 1, self.num_heads, -1).transpose(1, 2)
        k = self.w_k(regions).view(B, K, self.num_heads, -1).transpose(1, 2)
        v = self.w_v(regions).view(B, K, self.num_heads, -1).transpose(1, 2)

        attn = (q @ k.transpose(-2, -1)) * self.scale
        attn = F.softmax(attn, dim=-1)
        context = (attn @ v).transpose(1, 2).reshape(B, D)

        return self.layer_norm(question + self.out_proj(context))

import torch
import torch.nn as nn
from transformers import ViTModel, BertModel

class ViLTStyleVQA(nn.Module):
    """简化版 ViLT 风格的 VQA 模型"""
    def __init__(self, num_answers: int, hidden_dim: int = 768):
        super().__init__()
        # 图像 patch 线性投影（替代 CNN）
        self.patch_embed = nn.Conv2d(
            in_channels=3, out_channels=hidden_dim,
            kernel_size=16, stride=16  # 16x16 patch
        )
        self.type_embed = nn.Embedding(2, hidden_dim)  # 0=image, 1=text
        self.pos_embed = nn.Parameter(torch.randn(1, 256, hidden_dim))

        # 共享 Transformer 编码器
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=hidden_dim, nhead=12, dim_feedforward=3072,
            dropout=0.1, batch_first=True
        )
        self.transformer = nn.TransformerEncoder(
            encoder_layer, num_layers=12
        )

        # CLS token
        self.cls_token = nn.Parameter(torch.randn(1, 1, hidden_dim))

        # 答案分类头
        self.classifier = nn.Linear(hidden_dim, num_answers)

    def forward(self, images: torch.Tensor, input_ids: torch.Tensor,
                text_embeds: torch.Tensor, attention_mask: torch.Tensor):
        B = images.shape[0]
        # 图像 patch 嵌入 [B, 3, 224, 224] -> [B, 196, 768]
        img_tokens = self.patch_embed(images).flatten(2).transpose(1, 2)
        img_type = self.type_embed(torch.zeros(B, img_tokens.size(1),
                                               dtype=torch.long, device=images.device))

        # 文本嵌入
        txt_type = self.type_embed(torch.ones(B, text_embeds.size(1),
                                              dtype=torch.long, device=images.device))

        # 拼接 + 位置编码
        cls = self.cls_token.expand(B, -1, -1)
        tokens = torch.cat([cls, img_tokens + img_type,
                            text_embeds + txt_type], dim=1)
        tokens = tokens + self.pos_embed[:, :tokens.size(1), :]

        # 扩展 attention mask
        full_mask = torch.cat([
            torch.ones(B, 1 + img_tokens.size(1), device=images.device),
            attention_mask
        ], dim=1)

        output = self.transformer(tokens, src_key_padding_mask=~full_mask.bool())
        return self.classifier(output[:, 0])  # CLS token 输出

import torch
import torch.nn as nn

class CrossModalAttention(nn.Module):
    """跨模态自注意力模块 - 视觉和文本 token 统一处理"""
    def __init__(self, d_model: int, nhead: int = 8, dropout: float = 0.1):
        super().__init__()
        self.self_attn = nn.MultiheadAttention(
            d_model, nhead, dropout=dropout, batch_first=True
        )
        self.norm1 = nn.LayerNorm(d_model)
        self.ffn = nn.Sequential(
            nn.Linear(d_model, d_model * 4),
            nn.GELU(),
            nn.Dropout(dropout),
            nn.Linear(d_model * 4, d_model),
            nn.Dropout(dropout),
        )
        self.norm2 = nn.LayerNorm(d_model)

    def forward(self, x: torch.Tensor, mask: torch.Tensor = None):
        """x: [B, N, D] 包含所有模态的 token"""
        # 自注意力（所有 token 互相可见）
        attn_out, _ = self.self_attn(x, x, x, key_padding_mask=mask)
        x = self.norm1(x + attn_out)
        # FFN
        x = self.norm2(x + self.ffn(x))
        return x

# 使用示例
cross_attn = CrossModalAttention(d_model=768, nhead=12)
# 拼接视觉和文本 token
multimodal_tokens = torch.cat([visual_tokens, text_tokens], dim=1)
# [B, N_img + N_txt, 768]
output = cross_attn(multimodal_tokens)
print(f"跨模态输出形状: {output.shape}")

from transformers import LxmertTokenizer, LxmertModel
import torch

# 加载预训练的 LXMERT 模型
tokenizer = LxmertTokenizer.from_pretrained("unc-nlp/lxmert-base-uncased")
model = LxmertModel.from_pretrained("unc-nlp/lxmert-base-uncased")

# 准备输入
image_features = torch.randn(1, 36, 2048)  # Faster R-CNN 提取
image_features_bbox = torch.randn(1, 36, 4)  # 边界框坐标
question = "What color is the car?"

inputs = tokenizer(question, return_tensors="pt")

# 前向传播
output = model(
    input_ids=inputs["input_ids"],
    visual_feats=image_features,
    visual_pos=image_features_bbox,
)

# 获取跨模态表示
cross_encoded = output.language_output  # [1, seq_len, 768]
pooled_output = output.pooled_output    # [1, 768]
print(f"语言输出: {cross_encoded.shape}")
print(f"池化输出: {pooled_output.shape}")

import torch
import torch.nn as nn
from transformers import PreTrainedModel

class VILBERTForVQA(PreTrainedModel):
    """基于 ViLBERT 的 VQA 微调模型"""
    def __init__(self, config, num_answers: int):
        super().__init__(config)
        from transformers import VilBertModel
        self.vilbert = VilBertModel(config)

        # VQA 分类头
        self.classifier = nn.Sequential(
            nn.Linear(config.hidden_size, config.hidden_size * 2),
            nn.GELU(),
            nn.LayerNorm(config.hidden_size * 2),
            nn.Dropout(0.1),
            nn.Linear(config.hidden_size * 2, num_answers)
        )

    def forward(self, input_ids, visual_feats, visual_pos,
                attention_mask=None, token_type_ids=None,
                labels=None):
        outputs = self.vilbert(
            input_ids=input_ids,
            visual_feats=visual_feats,
            visual_pos=visual_pos,
            attention_mask=attention_mask,
        )
        # 取 pooled output 用于分类
        pooled = outputs[1]  # [B, hidden_size]
        logits = self.classifier(pooled)

        if labels is not None:
            loss = nn.CrossEntropyLoss()(logits, labels)
            return {"loss": loss, "logits": logits}
        return {"logits": logits}

def vqa_score(pred_answer: str, gt_answers: list) -> float:
    """VQA 官方评估分数计算
    每个问题有 10 个标注答案，计算预测答案的准确度
    """
    from collections import Counter
    answer_counts = Counter(gt_answers)
    # VQA 规则: min(count/3, 1.0)
    pred_count = answer_counts.get(pred_answer, 0)
    return min(pred_count / 3.0, 1.0)

# 示例
gt = ["yes", "yes", "yes", "yes", "yes",
      "yes", "yes", "no", "no", "maybe"]
print(f"预测 yes 得分: {vqa_score('yes', gt):.2f}")    # 1.00
print(f"预测 no 得分: {vqa_score('no', gt):.2f}")      # 0.67
print(f"预测 maybe 得分: {vqa_score('maybe', gt):.2f}")  # 0.33
print(f"预测 no idea 得分: {vqa_score('no idea', gt):.2f}")  # 0.00

import torch
import numpy as np
from collections import defaultdict

class VQAEvaluator:
    """完整的 VQA 评估工具"""
    def __init__(self):
        self.results = []
        self.question_type_scores = defaultdict(list)

    def add_result(self, question_id: int, question_type: str,
                   pred: str, gt_answers: list):
        score = self._compute_score(pred, gt_answers)
        self.results.append({
            "question_id": question_id,
            "question_type": question_type,
            "score": score
        })
        self.question_type_scores[question_type].append(score)

    def _compute_score(self, pred: str, gt: list) -> float:
        from collections import Counter
        counts = Counter(gt)
        return min(counts.get(pred, 0) / 3.0, 1.0)

    def overall_accuracy(self) -> float:
        """整体准确率"""
        return np.mean([r["score"] for r in self.results])

    def per_type_accuracy(self) -> dict:
        """每种问题类型的准确率"""
        return {
            qtype: np.mean(scores)
            for qtype, scores in self.question_type_scores.items()
        }

    def summary(self) -> str:
        """生成评估报告"""
        lines = [f"Overall Accuracy: {self.overall_accuracy():.2%}"]
        for qtype, acc in sorted(self.per_type_accuracy().items()):
            lines.append(f"  {qtype}: {acc:.2%}")
        return "\n".join(lines)

from transformers import OFAForConditionalGeneration, OFATokenizer
from PIL import Image
import torch

# 加载 OFA 模型（统一多模态模型）
model_name = "OFA-Sys/ofa-base"
model = OFAForConditionalGeneration.from_pretrained(model_name)
tokenizer = OFATokenizer.from_pretrained(model_name)
model.eval()

def vqa_inference(image_path: str, question: str,
                  num_beams: int = 4) -> str:
    """使用 OFA 进行 VQA 推理"""
    image = Image.open(image_path).convert("RGB")
    # OFA 使用特定的 prompt 格式
    prompt = f" what is the answer to the question '{question}'?"
    inputs = tokenizer(
        [prompt],
        images=image,
        return_tensors="pt"
    )
    # 生成答案
    with torch.no_grad():
        outputs = model.generate(
            input_ids=inputs["input_ids"],
            pixel_values=inputs["pixel_values"],
            max_new_tokens=10,
            num_beams=num_beams,
        )
    answer = tokenizer.decode(outputs[0], skip_special_tokens=True)
    return answer.strip()

# 使用示例
result = vqa_inference("photo.jpg", "How many people are in the image?")
print(f"Answer: {result}")

from transformers import Blip2ForConditionalGeneration, Blip2Processor
from PIL import Image
import torch

# 加载 BLIP-2 模型
processor = Blip2Processor.from_pretrained(
    "Salesforce/blip2-opt-2.7b"
)
model = Blip2ForConditionalGeneration.from_pretrained(
    "Salesforce/blip2-opt-2.7b",
    torch_dtype=torch.float16,  # 半精度推理
)
model.eval().cuda()

def blip2_vqa(image_path: str, question: str) -> str:
    """使用 BLIP-2 进行 VQA 推理"""
    image = Image.open(image_path).convert("RGB")
    # VQA prompt
    prompt = f"Question: {question} Answer:"
    inputs = processor(image, prompt, return_tensors="pt").to(
        "cuda", torch.float16
    )
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=20,
            num_beams=5,
            do_sample=False,
        )
    answer = processor.decode(outputs[0], skip_special_tokens=True)
    return answer.strip()

# 批量推理
questions = ["What is this?", "What color?", "How many?"]
answers = [blip2_vqa("photo.jpg", q) for q in questions]
for q, a in zip(questions, answers):
    print(f"Q: {q} | A: {a}")