第13章 多模态学习¶

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np

class VisionEncoder(nn.Module):  # 继承nn.Module定义网络层
    """简化的Vision Transformer编码器"""
    def __init__(self, image_size=224, patch_size=32, dim=768, depth=12, heads=12):
        super().__init__()  # super()调用父类方法
        num_patches = (image_size // patch_size) ** 2
        self.patch_embed = nn.Conv2d(3, dim, patch_size, patch_size)
        self.cls_token = nn.Parameter(torch.randn(1, 1, dim))
        self.pos_embed = nn.Parameter(torch.randn(1, num_patches + 1, dim))

        encoder_layer = nn.TransformerEncoderLayer(
            d_model=dim, nhead=heads, dim_feedforward=dim * 4,
            batch_first=True, norm_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=depth)
        self.ln = nn.LayerNorm(dim)

    def forward(self, x):
        # x: [B, 3, H, W]
        x = self.patch_embed(x)          # [B, dim, H/P, W/P]
        x = x.flatten(2).transpose(1, 2) # [B, num_patches, dim]
        cls = self.cls_token.expand(x.size(0), -1, -1)
        x = torch.cat([cls, x], dim=1)   # [B, num_patches+1, dim]  # torch.cat沿已有维度拼接张量
        x = x + self.pos_embed
        x = self.transformer(x)
        x = self.ln(x[:, 0])             # 取CLS token
        return x

class TextEncoder(nn.Module):
    """简化的文本编码器"""
    def __init__(self, vocab_size=49408, dim=512, depth=12, heads=8, max_len=77):
        super().__init__()
        self.token_embed = nn.Embedding(vocab_size, dim)
        self.pos_embed = nn.Parameter(torch.randn(1, max_len, dim))

        encoder_layer = nn.TransformerEncoderLayer(
            d_model=dim, nhead=heads, dim_feedforward=dim * 4,
            batch_first=True, norm_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=depth)
        self.ln = nn.LayerNorm(dim)

    def forward(self, tokens):
        x = self.token_embed(tokens) + self.pos_embed[:, :tokens.size(1)]
        x = self.transformer(x)
        # 取EOS token位置（最后一个非padding token）
        x = self.ln(x[torch.arange(x.size(0)), tokens.argmax(dim=-1)])
        return x

class CLIP(nn.Module):
    """完整CLIP模型"""
    def __init__(self, embed_dim=512, vision_dim=768, text_dim=512):
        super().__init__()
        self.visual = VisionEncoder(dim=vision_dim)
        self.text = TextEncoder(dim=text_dim)

        # 投影到共享嵌入空间
        self.image_projection = nn.Linear(vision_dim, embed_dim, bias=False)
        self.text_projection = nn.Linear(text_dim, embed_dim, bias=False)

        # 可学习温度参数（log scale）
        self.logit_scale = nn.Parameter(torch.ones([]) * np.log(1 / 0.07))

    def encode_image(self, image):
        """编码图像为归一化特征"""
        feat = self.visual(image)
        proj = self.image_projection(feat)
        return F.normalize(proj, dim=-1)  # L2归一化

    def encode_text(self, text):
        """编码文本为归一化特征"""
        feat = self.text(text)
        proj = self.text_projection(feat)
        return F.normalize(proj, dim=-1)

    def forward(self, image, text):
        image_features = self.encode_image(image)
        text_features = self.encode_text(text)

        # 缩放余弦相似度
        logit_scale = self.logit_scale.exp()
        logits_per_image = logit_scale * image_features @ text_features.t()
        logits_per_text = logits_per_image.t()

        return logits_per_image, logits_per_text

def clip_loss(logits_per_image, logits_per_text):
    """对称InfoNCE损失"""
    batch_size = logits_per_image.size(0)
    labels = torch.arange(batch_size, device=logits_per_image.device)
    loss_i2t = F.cross_entropy(logits_per_image, labels)  # F.cross_entropy PyTorch函数式交叉熵损失
    loss_t2i = F.cross_entropy(logits_per_text, labels)
    return (loss_i2t + loss_t2i) / 2

# ========== 训练示例 ==========
def train_clip():
    model = CLIP()
    optimizer = torch.optim.AdamW(model.parameters(), lr=5e-4, weight_decay=0.2)

    # 模拟数据
    batch_size = 32
    images = torch.randn(batch_size, 3, 224, 224)
    texts = torch.randint(0, 49408, (batch_size, 77))

    # 前向 + 反向
    logits_per_image, logits_per_text = model(images, texts)
    loss = clip_loss(logits_per_image, logits_per_text)
    loss.backward()  # 反向传播计算梯度
    optimizer.step()  # 更新参数

    print(f"Loss: {loss.item():.4f}")  # 将单元素张量转为Python数值
    print(f"Logit scale: {model.logit_scale.exp().item():.2f}")

# train_clip()

import clip
from PIL import Image

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

# 准备输入
image = preprocess(Image.open("cat.jpg")).unsqueeze(0).to(device)  # unsqueeze增加一个维度  # 移至GPU/CPU
text_templates = [
    "a photo of a cat",
    "a photo of a dog",
    "a photo of a car"
]
text = clip.tokenize(text_templates).to(device)

# 零样本推理
with torch.no_grad():  # 禁用梯度计算，节省内存
    image_features = model.encode_image(image)
    text_features = model.encode_text(text)

    # 归一化
    image_features = F.normalize(image_features, dim=-1)
    text_features = F.normalize(text_features, dim=-1)

    # 计算相似度
    similarity = (100.0 * image_features @ text_features.T).softmax(dim=-1)

for i, cls in enumerate(["cat", "dog", "car"]):  # enumerate同时获取索引和元素
    print(f"{cls}: {similarity[0][i]:.4f}")
# 输出示例: cat: 0.9832  dog: 0.0151  car: 0.0017

def siglip_loss(image_features, text_features, temperature, bias):
    """SigLIP损失函数"""
    # [B, D] @ [D, B] -> [B, B]
    logits = image_features @ text_features.T / temperature + bias

    # 标签矩阵：对角线为1（正样本），其余为-1（负样本）
    B = logits.size(0)
    labels = 2 * torch.eye(B, device=logits.device) - 1  # +1 或 -1

    # Sigmoid二分类损失
    loss = -F.logsigmoid(labels * logits).mean()
    return loss

# 使用Chinese-CLIP进行中文零样本分类
from cn_clip.clip import load_from_name
import cn_clip.clip as clip

device = "cuda" if torch.cuda.is_available() else "cpu"
model, preprocess = load_from_name("ViT-B-16", device=device)

image = preprocess(Image.open("test.jpg")).unsqueeze(0).to(device)
text = clip.tokenize(["一只猫", "一只狗", "一辆车"]).to(device)

with torch.no_grad():
    image_features = model.encode_image(image)
    text_features = model.encode_text(text)

    image_features = F.normalize(image_features, dim=-1)
    text_features = F.normalize(text_features, dim=-1)

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

for label, score in zip(["猫", "狗", "车"], similarity[0]):  # zip按位置配对
    print(f"{label}: {score.item():.4f}")

class QFormer(nn.Module):
    """简化的Q-Former实现"""
    def __init__(self, num_queries=32, dim=768, depth=6, heads=12, visual_dim=1408):
        super().__init__()
        # 可学习查询token
        self.queries = nn.Parameter(torch.randn(1, num_queries, dim))

        # Cross-Attention层（Query与视觉特征交互）
        self.cross_attn_layers = nn.ModuleList([
            nn.MultiheadAttention(dim, heads, batch_first=True)
            for _ in range(depth)
        ])

        # Self-Attention层
        self.self_attn_layers = nn.ModuleList([
            nn.TransformerEncoderLayer(
                d_model=dim, nhead=heads, dim_feedforward=dim * 4,
                batch_first=True, norm_first=True
            )
            for _ in range(depth)
        ])

        # 视觉特征投影（如果维度不匹配）
        self.visual_proj = nn.Linear(visual_dim, dim) if visual_dim != dim else nn.Identity()

        # LayerNorm
        self.cross_attn_norms = nn.ModuleList([nn.LayerNorm(dim) for _ in range(depth)])

    def forward(self, visual_features):
        """
        Args:
            visual_features: [B, num_patches, visual_dim] 来自冻结ViT的特征
        Returns:
            query_output: [B, num_queries, dim] 压缩后的视觉表示
        """
        B = visual_features.size(0)
        visual_features = self.visual_proj(visual_features)

        # 扩展queries到batch维度
        queries = self.queries.expand(B, -1, -1)

        for i in range(len(self.self_attn_layers)):
            # Self-Attention
            queries = self.self_attn_layers[i](queries)

            # Cross-Attention: queries作为Q, 视觉特征作为K,V
            residual = queries
            queries_norm = self.cross_attn_norms[i](queries)
            cross_out, _ = self.cross_attn_layers[i](
                query=queries_norm,
                key=visual_features,
                value=visual_features
            )
            queries = residual + cross_out

        return queries  # [B, 32, dim]

class BLIP2(nn.Module):
    """BLIP-2模型简化版"""
    def __init__(self, visual_dim=1408, llm_dim=2560, num_queries=32, qformer_dim=768):
        super().__init__()
        self.qformer = QFormer(
            num_queries=num_queries,
            dim=qformer_dim,
            visual_dim=visual_dim
        )
        # 投影到LLM输入空间
        self.projection = nn.Linear(qformer_dim, llm_dim)

    def forward(self, visual_features):
        """
        Args:
            visual_features: [B, 257, 1408] 来自冻结ViT-G的特征
        Returns:
            llm_input: [B, 32, 2560] 用于送入冻结LLM
        """
        query_output = self.qformer(visual_features)  # [B, 32, 768]
        llm_input = self.projection(query_output)       # [B, 32, 2560]
        return llm_input

# 使用示例
visual_feats = torch.randn(2, 257, 1408)  # 模拟ViT-G输出
blip2 = BLIP2()
llm_input = blip2(visual_feats)
print(f"LLM输入形状: {llm_input.shape}")  # [2, 32, 2560]

from transformers import Blip2Processor, Blip2ForConditionalGeneration
from PIL import Image

# 加载模型（约8GB显存）
processor = Blip2Processor.from_pretrained("Salesforce/blip2-opt-2.7b")
model = Blip2ForConditionalGeneration.from_pretrained(
    "Salesforce/blip2-opt-2.7b",
    torch_dtype=torch.float16,
    device_map="auto"
)

# === VQA任务 ===
image = Image.open("street_scene.jpg")
question = "How many people are in this image?"
inputs = processor(images=image, text=question, return_tensors="pt").to("cuda", torch.float16)
generated_ids = model.generate(**inputs, max_new_tokens=50)
answer = processor.batch_decode(generated_ids, skip_special_tokens=True)[0]
print(f"Q: {question}")
print(f"A: {answer}")

# === 图像描述 ===
inputs = processor(images=image, return_tensors="pt").to("cuda", torch.float16)
generated_ids = model.generate(**inputs, max_new_tokens=100)
caption = processor.batch_decode(generated_ids, skip_special_tokens=True)[0]
print(f"Caption: {caption}")

# === 指令式对话（InstructBLIP） ===
from transformers import InstructBlipProcessor, InstructBlipForConditionalGeneration

instruct_processor = InstructBlipProcessor.from_pretrained("Salesforce/instructblip-vicuna-7b")
instruct_model = InstructBlipForConditionalGeneration.from_pretrained(
    "Salesforce/instructblip-vicuna-7b",
    torch_dtype=torch.float16,
    device_map="auto"
)

prompt = "Describe this image in detail, including colors, objects, and spatial relationships."
inputs = instruct_processor(images=image, text=prompt, return_tensors="pt").to("cuda", torch.float16)
outputs = instruct_model.generate(**inputs, max_new_tokens=200)
print(instruct_processor.decode(outputs[0], skip_special_tokens=True))

# 使用InternVL2进行多模态对话
from transformers import AutoModel, AutoTokenizer
from PIL import Image

model_name = "OpenGVLab/InternVL2-8B"
model = AutoModel.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    trust_remote_code=True
)
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)

# 单图对话
image = Image.open("diagram.png").convert("RGB")
pixel_values = model.transform(image).unsqueeze(0).to(model.device, torch.bfloat16)

question = "请详细描述这张图片中的内容。"
response = model.chat(tokenizer, pixel_values, question)
print(response)

# 多图对比
images = [Image.open(f"img_{i}.jpg").convert("RGB") for i in range(2)]
pixel_values = torch.stack([  # torch.stack沿新维度拼接张量
    model.transform(img) for img in images
]).to(model.device, torch.bfloat16)

question = "请比较这两张图片的异同。"
response = model.chat(tokenizer, pixel_values, question)
print(response)

from transformers import Qwen2VLForConditionalGeneration, AutoProcessor

# 加载Qwen2-VL
model = Qwen2VLForConditionalGeneration.from_pretrained(
    "Qwen/Qwen2-VL-7B-Instruct",
    torch_dtype=torch.bfloat16,
    device_map="auto"
)
processor = AutoProcessor.from_pretrained("Qwen/Qwen2-VL-7B-Instruct")

# 多模态对话
messages = [
    {
        "role": "user",
        "content": [
            {"type": "image", "image": "test.jpg"},
            {"type": "text", "text": "这张图片里有什么？请用中文详细描述。"}
        ]
    }
]

text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = processor(text=[text], images=[Image.open("test.jpg")], return_tensors="pt").to("cuda")
output_ids = model.generate(**inputs, max_new_tokens=256)
output = processor.batch_decode(output_ids, skip_special_tokens=True)[0]
print(output)

import torch
import torch.nn.functional as F
from PIL import Image
from transformers import CLIPProcessor, CLIPModel
import os

class ImageTextRetrieval:
    """基于CLIP的图文检索系统"""
    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.model.eval()  # eval()评估模式
        self.image_database = []  # (path, embedding)

    def build_index(self, image_dir):
        """构建图像索引"""
        image_paths = [
            os.path.join(image_dir, f)
            for f in os.listdir(image_dir)
            if f.endswith(('.jpg', '.png', '.jpeg'))
        ]

        for path in image_paths:
            image = Image.open(path).convert("RGB")
            inputs = self.processor(images=image, return_tensors="pt")

            with torch.no_grad():
                embedding = self.model.get_image_features(**inputs)
                embedding = F.normalize(embedding, dim=-1)

            self.image_database.append((path, embedding))

        print(f"索引完成: {len(self.image_database)} 张图片")

    def text_to_image(self, query, top_k=5):
        """文本→图像检索"""
        inputs = self.processor(text=query, return_tensors="pt")

        with torch.no_grad():
            text_embedding = self.model.get_text_features(**inputs)
            text_embedding = F.normalize(text_embedding, dim=-1)

        similarities = []
        for path, img_emb in self.image_database:
            sim = (text_embedding @ img_emb.T).item()
            similarities.append((path, sim))

        similarities.sort(key=lambda x: x[1], reverse=True)  # lambda匿名函数
        return similarities[:top_k]

    def image_to_image(self, query_image_path, top_k=5):
        """图像→图像检索（以图搜图）"""
        image = Image.open(query_image_path).convert("RGB")
        inputs = self.processor(images=image, return_tensors="pt")

        with torch.no_grad():
            query_embedding = self.model.get_image_features(**inputs)
            query_embedding = F.normalize(query_embedding, dim=-1)

        similarities = []
        for path, img_emb in self.image_database:
            sim = (query_embedding @ img_emb.T).item()
            similarities.append((path, sim))

        similarities.sort(key=lambda x: x[1], reverse=True)
        return similarities[:top_k]

# 使用示例
retrieval = ImageTextRetrieval()
retrieval.build_index("./image_gallery/")
results = retrieval.text_to_image("a cat sitting on a sofa")
for path, score in results:
    print(f"{path}: {score:.4f}")

"""
多模态RAG系统：结合图像理解和文本检索
"""
import chromadb
from PIL import Image
from transformers import CLIPProcessor, CLIPModel
import torch
import torch.nn.functional as F

class MultimodalRAG:
    """多模态RAG: 支持图像和文本的混合检索"""
    def __init__(self, clip_model="openai/clip-vit-base-patch32"):
        # 视觉-文本编码器
        self.clip = CLIPModel.from_pretrained(clip_model)
        self.processor = CLIPProcessor.from_pretrained(clip_model)
        self.clip.eval()

        # 向量数据库
        self.client = chromadb.Client()
        self.collection = self.client.create_collection(
            name="multimodal_docs",
            metadata={"hnsw:space": "cosine"}
        )
        self.doc_id = 0

    def add_image(self, image_path, description=""):
        """索引图像"""
        image = Image.open(image_path).convert("RGB")
        inputs = self.processor(images=image, return_tensors="pt")

        with torch.no_grad():
            embedding = self.clip.get_image_features(**inputs)
            embedding = F.normalize(embedding, dim=-1)

        self.collection.add(
            embeddings=[embedding[0].numpy().tolist()],
            documents=[description or f"Image: {image_path}"],
            metadatas=[{"type": "image", "path": image_path}],
            ids=[f"doc_{self.doc_id}"]
        )
        self.doc_id += 1

    def add_text(self, text, metadata=None):
        """索引文本"""
        inputs = self.processor(text=text, return_tensors="pt", padding=True, truncation=True)

        with torch.no_grad():
            embedding = self.clip.get_text_features(**inputs)
            embedding = F.normalize(embedding, dim=-1)

        self.collection.add(
            embeddings=[embedding[0].numpy().tolist()],
            documents=[text],
            metadatas=[{"type": "text", **(metadata or {})}],
            ids=[f"doc_{self.doc_id}"]
        )
        self.doc_id += 1

    def query(self, text_query, top_k=5):
        """文本查询，检索相关图像和文本"""
        inputs = self.processor(text=text_query, return_tensors="pt")

        with torch.no_grad():
            query_embedding = self.clip.get_text_features(**inputs)
            query_embedding = F.normalize(query_embedding, dim=-1)

        results = self.collection.query(
            query_embeddings=[query_embedding[0].numpy().tolist()],
            n_results=top_k
        )
        return results

# 使用示例
rag = MultimodalRAG()

# 索引混合内容
rag.add_image("chart.png", "2024年Q3收入增长趋势图")
rag.add_image("architecture.png", "微服务系统架构图")
rag.add_text("2024年第三季度收入同比增长23%，主要得益于AI产品线。")
rag.add_text("系统采用微服务架构，共包含12个核心服务。")

# 查询
results = rag.query("公司收入增长情况")
for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
    print(f"[{meta['type']}] {doc}")

from transformers import pipeline
from PIL import Image

class VQASystem:
    """基于VLM的视觉问答系统"""
    def __init__(self, model_name="Salesforce/blip2-opt-2.7b"):
        self.pipe = pipeline(
            "visual-question-answering",
            model=model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )

    def answer(self, image_path, question):
        """回答关于图像的问题"""
        image = Image.open(image_path).convert("RGB")
        result = self.pipe(image, question)
        return result[0]["answer"]

    def batch_answer(self, image_path, questions):
        """批量回答多个问题"""
        image = Image.open(image_path).convert("RGB")
        answers = {}
        for q in questions:
            result = self.pipe(image, q)
            answers[q] = result[0]["answer"]
        return answers

# 使用示例
vqa = VQASystem()
questions = [
    "What color is the car?",
    "How many people are in the image?",
    "What is the weather like?"
]
results = vqa.batch_answer("street.jpg", questions)
for q, a in results.items():
    print(f"Q: {q}\nA: {a}\n")