多模态 LLM：视觉-语言模型

# 多模态数据加载示例：同时处理图像和文本
from torch.utils.data import Dataset
from PIL import Image
import torch

class ImageTextDataset(Dataset):
    """同时加载图像-文本对的多模态数据集"""
    def __init__(self, image_dir, captions_file, transform):
        self.image_dir = image_dir
        self.captions = load_json(captions_file)
        self.transform = transform
        self.samples = list(self.captions.items())

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

    def __getitem__(self, idx):
        img_path, caption = self.samples[idx]
        image = self.transform(Image.open(img_path).convert("RGB"))
        return {"image": image, "text": caption}

# 模态融合策略对比
import torch.nn as nn

class EarlyFusion(nn.Module):
    """早期融合：在输入层拼接模态"""
    def __init__(self, vision_dim=768, text_dim=768, hidden=1024):
        super().__init__()
        self.fusion = nn.Linear(vision_dim + text_dim, hidden)

    def forward(self, img_feat, text_feat):
        combined = torch.cat([img_feat, text_feat], dim=-1)
        return self.fusion(combined)

class LateFusion(nn.Module):
    """晚期融合：各模态独立处理后再融合"""
    def __init__(self, vision_dim=768, text_dim=768):
        super().__init__()
        self.vision_proj = nn.Linear(vision_dim, 512)
        self.text_proj = nn.Linear(text_dim, 512)

    def forward(self, img_feat, text_feat):
        v = self.vision_proj(img_feat)
        t = self.text_proj(text_feat)
        return v + t  # 在共享空间相加

# CLIP 对比损失计算
import torch
import torch.nn as nn
import torch.nn.functional as F

def clip_loss(image_features, text_features, logit_scale):
    """CLIP 的 InfoNCE 对比损失"""
    # 归一化特征向量
    image_features = F.normalize(image_features, dim=-1)
    text_features = F.normalize(text_features, dim=-1)

    # 计算相似度矩阵 (batch_size x batch_size)
    logits = logit_scale * image_features @ text_features.t()

    # 对角线标签：每个图像对应的正确文本
    labels = torch.arange(len(logits), device=logits.device)

    # 双向损失：图文 + 文图
    loss_i = F.cross_entropy(logits, labels)
    loss_t = F.cross_entropy(logits.t(), labels)
    return (loss_i + loss_t) / 2

# 使用示例
logit_scale = nn.Parameter(torch.ones([]) * np.log(1 / 0.07))
loss = clip_loss(img_feat, txt_feat, logit_scale.exp())

# CLIP 零样本图像分类
import torch
import clip
from PIL import Image

device = "cuda" if torch.cuda.is_available() else "cpu"
model, preprocess = clip.load("ViT-B/32", device=device)

# 准备候选类别
classes = ["a photo of a dog", "a photo of a cat",
           "a photo of a bird", "a photo of a car"]
text_tokens = clip.tokenize(classes).to(device)

# 编码图像和文本
image = preprocess(Image.open("test.jpg")).unsqueeze(0).to(device)
with torch.no_grad():
    image_features = model.encode_image(image)
    text_features = model.encode_text(text_tokens)

# 计算相似度并选最高分
image_features /= image_features.norm(dim=-1, keepdim=True)
text_features /= text_features.norm(dim=-1, keepdim=True)
similarity = (100.0 * image_features @ text_features.T).softmax(dim=-1)
print(f"Predicted: {classes[similarity.argmax().item()]}")

# LLaVA 架构核心：视觉特征投影到 LLM 词嵌入空间
import torch.nn as nn

class LlavaProjector(nn.Module):
    """两层 MLP 投影器：视觉特征 → LLM 词嵌入"""
    def __init__(self, vision_dim=1024, llm_dim=4096):
        super().__init__()
        self.linear_1 = nn.Linear(vision_dim, llm_dim)
        self.act = nn.GELU()
        self.linear_2 = nn.Linear(llm_dim, llm_dim)

    def forward(self, vision_features):
        # vision_features: [batch, num_patches, vision_dim]
        hidden = self.linear_1(vision_features)
        hidden = self.act(hidden)
        return self.linear_2(hidden)

class SimpleLLaVA(nn.Module):
    """简化的 LLaVA 模型"""
    def __init__(self, clip_model, llm_model):
        super().__init__()
        self.vision_encoder = clip_model.vision_model
        self.projector = LlavaProjector(vision_dim=1024, llm_dim=4096)
        self.llm = llm_model

    def forward(self, image, input_ids, attention_mask):
        # 1. 视觉编码
        vision_out = self.vision_encoder(image)
        # 2. 投影到 LLM 空间
        image_tokens = self.projector(vision_out.last_hidden_state)
        # 3. 拼接文本嵌入
        text_embeds = self.llm.get_input_embeddings()(input_ids)
        # 4. 联合输入 LLM
        combined = torch.cat([image_tokens, text_embeds], dim=1)
        return self.llm(inputs_embeds=combined, attention_mask=attention_mask)

# LLaVA 训练脚本（两阶段训练）
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

# ===== 阶段 1：冻结 LLM，只训练投影层 =====
def train_projector_stage(model, dataloader, epochs=1):
    """阶段 1：预训练投影层，冻结视觉编码器和 LLM"""
    model.vision_encoder.requires_grad_(False)
    model.llm.requires_grad_(False)
    optimizer = torch.optim.AdamW(model.projector.parameters(), lr=1e-3)

    for epoch in range(epochs):
        for batch in dataloader:
            loss = model(batch["image"], batch["input_ids"],
                         batch["attention_mask"])
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()

# ===== 阶段 2：全模型微调 =====
def train_full_model(model, dataloader, epochs=3):
    """阶段 2：解冻 LLM 后 5 层，全模型微调"""
    for param in model.llm.parameters():
        param.requires_grad = False
    # 只解冻最后 5 层
    for layer in model.llm.model.layers[-5:]:
        for param in layer.parameters():
            param.requires_grad = True
    # ... 训练逻辑同上

# 多模态指令数据格式
def build_multimodal_instruction(image_path, question, answer):
    """构建单条多模态指令样本"""
    return {
        "image": image_path,
        "conversations": [
            {
                "from": "human",
                "value": "<image>\n" + question
            },
            {
                "from": "gpt",
                "value": answer
            }
        ]
    }

# 示例：视觉问答指令
sample = build_multimodal_instruction(
    image_path="coco/train2017/000000123.jpg",
    question="图中有几个人在跑步？",
    answer="图中有 3 个人在公园的小路上跑步，其中两人并肩跑在前面，一人在后面追赶。"
)

# 多模态指令微调（使用 transformers Trainer）
from transformers import TrainingArguments, Trainer

# 自定义数据整理器
class LLaVADataCollator:
    def __init__(self, tokenizer, image_processor):
        self.tokenizer = tokenizer
        self.image_processor = image_processor

    def __call__(self, features):
        # 处理图像
        images = [f["image"] for f in features]
        pixel_values = self.image_processor(images, return_tensors="pt")

        # 处理文本对话
        conversations = [f["conversations"] for f in features]
        texts = [self._format_conversation(conv) for conv in conversations]
        batch_encoding = self.tokenizer(texts, padding=True,
                                        return_tensors="pt")
        return {batch_encoding, pixel_values}

training_args = TrainingArguments(
    output_dir="./llava-sft",
    per_device_train_batch_size=2,
    gradient_accumulation_steps=16,
    learning_rate=2e-5,
    num_train_epochs=3,
    fp16=True,
    logging_steps=50,
    save_strategy="epoch",
)

# 视觉问答推理
from transformers import AutoProcessor, LlavaForConditionalGeneration
from PIL import Image
import torch

model_id = "llava-hf/llava-1.5-7b-hf"
processor = AutoProcessor.from_pretrained(model_id)
model = LlavaForConditionalGeneration.from_pretrained(
    model_id, torch_dtype=torch.float16, device_map="auto"
)

def ask_about_image(image_path, question):
    """视觉问答：根据图片回答问题"""
    image = Image.open(image_path)
    prompt = f"USER: <image>\n{question}\nASSISTANT:"
    inputs = processor(text=prompt, images=image, return_tensors="pt")
    inputs = {k: v.to("cuda") for k, v in inputs.items()}

    with torch.no_grad():
        output = model.generate(**inputs, max_new_tokens=100)
    answer = processor.decode(output[0], skip_special_tokens=True)
    return answer.split("ASSISTANT:")[-1].strip()

result = ask_about_image("photo.jpg", "图中有几个人？他们穿什么颜色的衣服？")
print(f"回答: {result}")

# 图像描述生成
def generate_image_caption(image_path, style="detailed"):
    """生成不同风格的图像描述"""
    image = Image.open(image_path)

    prompts = {
        "brief": "USER: <image>\n用一句话描述这张图片\nASSISTANT:",
        "detailed": "USER: <image>\n请详细描述这张图片中的所有细节，包括物体、颜色、位置和动作\nASSISTANT:",
        "creative": "USER: <image>\n以富有创意和诗意的方式描述这张图片\nASSISTANT:",
        "technical": "USER: <image>\n以技术分析的角度描述这张图片的构图、光影和色彩\nASSISTANT:",
    }

    prompt = prompts.get(style, prompts["detailed"])
    inputs = processor(text=prompt, images=image, return_tensors="pt")

    output = model.generate(**inputs, max_new_tokens=200,
                            do_sample=True, temperature=0.7)
    return processor.decode(output[0], skip_special_tokens=True)

# 视频帧采样策略
import cv2
import numpy as np

def uniform_sampling(video_path, num_frames=8):
    """均匀采样：等间隔抽取帧"""
    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    indices = np.linspace(0, total_frames - 1, num_frames, dtype=int)
    frames = []
    for idx in indices:
        cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
        _, frame = cap.read()
        frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
    cap.release()
    return frames

def keyframe_sampling(video_path, max_frames=8):
    """关键帧采样：基于帧间差异选择代表性帧"""
    cap = cv2.VideoCapture(video_path)
    frames, prev_gray = [], None
    scores = []
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        if prev_gray is not None:
            diff = cv2.absdiff(gray, prev_gray).mean()
            scores.append((len(frames), diff))
        frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
        prev_gray = gray
    cap.release()
    # 选择差异最大的帧作为关键帧
    scores.sort(key=lambda x: x[1], reverse=True)
    indices = sorted([x[0] for x in scores[:max_frames]])
    return [frames[i] for i in indices]

# 视频理解管线
from transformers import AutoProcessor, AutoModelForVision2Seq
from PIL import Image

class VideoUnderstandingPipeline:
    """视频理解管线：采样 → 编码 → 生成"""
    def __init__(self, model_id="llava-hf/llava-1.5-7b-hf"):
        self.processor = AutoProcessor.from_pretrained(model_id)
        self.model = AutoModelForVision2Seq.from_pretrained(
            model_id, torch_dtype=torch.float16, device_map="auto"
        )

    def analyze_video(self, video_path, question, num_frames=8):
        """分析视频并回答问题"""
        # 1. 采样关键帧
        frames = uniform_sampling(video_path, num_frames)
        pil_frames = [Image.fromarray(f) for f in frames]

        # 2. 构建 prompt
        prompt = f"USER: <image>\n这段视频展示了什么？{question}\nASSISTANT:"

        # 3. 处理并推理
        inputs = self.processor(text=prompt, images=pil_frames,
                                return_tensors="pt")
        inputs = {k: v.to("cuda") for k, v in inputs.items()}

        output = self.model.generate(**inputs, max_new_tokens=200)
        return self.processor.decode(output[0], skip_special_tokens=True)

# 使用 HuggingFace 进行多模态推理
from transformers import pipeline
from PIL import Image

# 方式一：使用高层 Pipeline API（最简单）
vqa_pipe = pipeline("visual-question-answering",
                     model="llava-hf/llava-1.5-7b-hf",
                     device=0)

image = Image.open("scene.jpg")
result = vqa_pipe(image, "图中有什么动物？",
                   max_new_tokens=50)
print(f"回答: {result[0]['answer']} (置信度: {result[0]['score']:.2f})")

# 方式二：使用低层 API（更灵活）
from transformers import AutoProcessor, LlavaForConditionalGeneration

processor = AutoProcessor.from_pretrained("llava-hf/llava-1.5-7b-hf")
model = LlavaForConditionalGeneration.from_pretrained(
    "llava-hf/llava-1.5-7b-hf",
    torch_dtype=torch.float16,
    device_map="auto",
)

prompt = "USER: <image>\n描述这张图片\nASSISTANT:"
inputs = processor(text=prompt, images=image, return_tensors="pt")
output = model.generate(**inputs, max_new_tokens=200, do_sample=True)
print(processor.decode(output[0], skip_special_tokens=True))

# 多模型对比评测框架
from transformers import AutoProcessor, AutoModelForVision2Seq
import torch
import time

class MultimodalBenchmark:
    """多模态模型对比评测"""
    def __init__(self):
        self.models = {
            "llava-7b": "llava-hf/llava-1.5-7b-hf",
            "qwen2-vl": "Qwen/Qwen2-VL-7B-Instruct",
            "phi3-vision": "microsoft/Phi-3-vision-128k-instruct",
        }
        self.loaded = {}

    def load_model(self, name):
        if name not in self.loaded:
            model_id = self.models[name]
            proc = AutoProcessor.from_pretrained(model_id,
                                                  trust_remote_code=True)
            model = AutoModelForVision2Seq.from_pretrained(
                model_id, torch_dtype=torch.float16,
                device_map="auto", trust_remote_code=True
            )
            self.loaded[name] = (proc, model)
        return self.loaded[name]

    def evaluate(self, name, image, question):
        proc, model = self.load_model(name)
        prompt = f"USER: <image>\n{question}\nASSISTANT:"
        inputs = proc(text=prompt, images=image, return_tensors="pt")
        start = time.time()
        output = model.generate(**inputs, max_new_tokens=100)
        elapsed = time.time() - start
        return proc.decode(output[0], skip_special_tokens=True), elapsed