2025视觉问答技术全面指南：从基础原理到前沿应用

# 视觉问答模型示例实现
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import ViTModel, BertModel

class VisualQuestionAnsweringModel(nn.Module):
    def __init__(self, image_model_name, text_model_name, num_answers):
        super().__init__()
        # 初始化图像编码器
        self.image_encoder = ViTModel.from_pretrained(image_model_name)
        # 初始化文本编码器
        self.text_encoder = BertModel.from_pretrained(text_model_name)
        
        # 图像特征投影层
        self.image_proj = nn.Linear(self.image_encoder.config.hidden_size, 768)
        # 文本特征投影层
        self.text_proj = nn.Linear(self.text_encoder.config.hidden_size, 768)
        
        # 多模态融合注意力层
        self.multihead_attn = nn.MultiheadAttention(embed_dim=768, num_heads=12)
        
        # 前馈网络用于推理
        self.fc1 = nn.Linear(768 * 2, 1024)
        self.fc2 = nn.Linear(1024, 512)
        self.fc3 = nn.Linear(512, num_answers)  # 答案分类头
        
        # Dropout层用于防止过拟合
        self.dropout = nn.Dropout(0.3)
        
    def forward(self, image_inputs, text_inputs):
        # 提取图像特征
        image_outputs = self.image_encoder(**image_inputs)
        image_features = image_outputs.last_hidden_state  # [batch_size, num_patches, hidden_dim]
        image_features = self.image_proj(image_features)  # 投影到相同维度
        
        # 提取文本特征
        text_outputs = self.text_encoder(**text_inputs)
        text_features = text_outputs.last_hidden_state  # [batch_size, seq_len, hidden_dim]
        text_features = self.text_proj(text_features)  # 投影到相同维度
        
        # 调整注意力机制的输入格式（需要seq_len在前）
        image_features = image_features.permute(1, 0, 2)  # [num_patches, batch_size, hidden_dim]
        text_features = text_features.permute(1, 0, 2)  # [seq_len, batch_size, hidden_dim]
        
        # 应用交叉注意力
        # 以文本特征为查询，图像特征为键和值
        attn_output, attn_weights = self.multihead_attn(
            query=text_features,
            key=image_features,
            value=image_features
        )
        
        # 调整回原来的格式
        attn_output = attn_output.permute(1, 0, 2)  # [batch_size, seq_len, hidden_dim]
        
        # 获取文本特征的池化表示（使用[CLS]标记）
        text_cls = attn_output[:, 0, :]  # [batch_size, hidden_dim]
        
        # 获取图像特征的池化表示（平均池化）
        image_avg = torch.mean(image_features.permute(1, 0, 2), dim=1)  # [batch_size, hidden_dim]
        
        # 融合文本和图像特征
        combined_features = torch.cat([text_cls, image_avg], dim=1)  # [batch_size, 2*hidden_dim]
        
        # 通过前馈网络进行推理
        x = F.relu(self.fc1(combined_features))
        x = self.dropout(x)
        x = F.relu(self.fc2(x))
        x = self.dropout(x)
        logits = self.fc3(x)  # [batch_size, num_answers]
        
        return logits

# 示例使用代码
# 初始化处理器
# from transformers import AutoProcessor, AutoTokenizer
# image_processor = AutoProcessor.from_pretrained("google/vit-base-patch16-224")
# text_tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
# 
# # 准备图像输入
# image = ...  # 加载图像
# image_inputs = image_processor(images=image, return_tensors="pt")
# # 准备文本输入
# question = "What color is the shirt?"
# text_inputs = text_tokenizer(question, return_tensors="pt")
# 
# # 初始化模型
# model = VisualQuestionAnsweringModel(
#     image_model_name="google/vit-base-patch16-224",
#     text_model_name="bert-base-uncased",
#     num_answers=1000  # 假设有1000个可能的答案
# )
# 
# # 模型推理
# with torch.no_grad():
#     logits = model(image_inputs, text_inputs)
#     predicted_answer_id = torch.argmax(logits, dim=1).item()

# 2025年基于大型多模态基础模型的视觉问答系统示例实现
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoProcessor, AutoModelForCausalLM, AutoModelForVisualQuestionAnswering, pipeline
import numpy as np
from PIL import Image
import requests
from tqdm import tqdm
import time
import os
import json

class AdvancedVisualQuestionAnswering:
    def __init__(self, model_name="microsoft/phi-3-vision-128k-instruct", device=None):
        # 设置设备
        self.device = device if device is not None else ("cuda" if torch.cuda.is_available() else "cpu")
        
        # 加载预训练的多模态模型和处理器
        print(f"加载多模态模型: {model_name}")
        
        try:
            # 尝试加载专用的VQA模型
            self.model = AutoModelForVisualQuestionAnswering.from_pretrained(
                model_name,
                device_map="auto" if torch.cuda.is_available() else None,
                trust_remote_code=True
            )
            self.processor = AutoProcessor.from_pretrained(
                model_name,
                trust_remote_code=True
            )
            self.use_generic_model = False
        except Exception as e:
            print(f"无法加载专用的VQA模型，使用通用多模态模型: {e}")
            # 使用通用多模态模型
            try:
                self.model = AutoModelForCausalLM.from_pretrained(
                    model_name,
                    device_map="auto" if torch.cuda.is_available() else None,
                    trust_remote_code=True
                )
                self.processor = AutoProcessor.from_pretrained(
                    model_name,
                    trust_remote_code=True
                )
                self.use_generic_model = True
            except Exception as e2:
                raise RuntimeError(f"无法加载模型: {e2}")
        
        # 设置模型为评估模式
        self.model.eval()
        
        # 获取模型信息
        self.model_name = model_name
        
        # 初始化pipeline用于快速推理（对于支持的模型）
        self.pipeline = None
        if not self.use_generic_model:
            try:
                self.pipeline = pipeline(
                    "visual-question-answering",
                    model=self.model,
                    image_processor=self.processor.image_processor,
                    tokenizer=self.processor.tokenizer,
                    device=self.device
                )
            except Exception:
                pass
    
    def load_image(self, image_path_or_url):
        # 加载图像
        if isinstance(image_path_or_url, str) and image_path_or_url.startswith(('http://', 'https://')):
            # 从URL加载图像
            try:
                response = requests.get(image_path_or_url, stream=True)
                response.raise_for_status()  # 检查请求是否成功
                image = Image.open(response.raw).convert('RGB')
                return image
            except Exception as e:
                raise ValueError(f"无法从URL加载图像: {e}")
        else:
            # 从本地路径加载图像
            try:
                image = Image.open(image_path_or_url).convert('RGB')
                return image
            except Exception as e:
                raise ValueError(f"无法从本地路径加载图像: {e}")
    
    def answer_question(self, image, question, max_length=1024, num_return_sequences=1, temperature=0.7):
        # 回答关于给定图像的问题
        
        # 确保图像是PIL格式
        if not isinstance(image, Image.Image):
            if isinstance(image, str):
                # 尝试从路径或URL加载图像
                image = self.load_image(image)
            else:
                raise TypeError("图像必须是PIL.Image对象或图像路径/URL字符串")
        
        if not self.use_generic_model and self.pipeline is not None:
            # 如果有专用的VQA pipeline，使用它
            try:
                result = self.pipeline(image=image, question=question)
                # 格式化结果
                if isinstance(result, list) and len(result) > 0:
                    return {
                        "answer": result[0]["answer"],
                        "confidence": result[0]["score"],
                        "model": self.model_name,
                        "question": question
                    }
                else:
                    return {
                        "answer": result["answer"],
                        "confidence": result["score"],
                        "model": self.model_name,
                        "question": question
                    }
            except Exception as e:
                print(f"使用pipeline失败，回退到通用方法: {e}")
                # 失败时回退到通用方法
        
        # 构建提示
        if self.use_generic_model:
            # 对于通用多模态模型，需要构建适当的提示
            prompt = f"<|image|>\n根据上面的图像，回答以下问题: {question}\n回答: "
        else:
            # 对于专用VQA模型，直接使用问题
            prompt = question
        
        # 准备输入
        inputs = self.processor(
            images=image if not self.use_generic_model else None,
            text=prompt if self.use_generic_model else question,
            return_tensors="pt",
            max_length=max_length,
            truncation=True
        )
        
        # 移至设备
        if not hasattr(self.model, "hf_device_map") or self.model.hf_device_map is None:
            inputs = {k: v.to(self.device) for k, v in inputs.items()}
        
        # 使用模型生成回答
        with torch.no_grad():
            if not self.use_generic_model:
                # 对于专用VQA模型
                outputs = self.model(**inputs)
                logits = outputs.logits
                
                # 获取答案
                answer_indices = logits.argmax(dim=-1)
                answer = self.processor.tokenizer.decode(answer_indices[0], skip_special_tokens=True)
            else:
                # 对于通用多模态模型
                outputs = self.model.generate(
                    **inputs,
                    max_length=max_length,
                    num_return_sequences=num_return_sequences,
                    temperature=temperature,
                    top_p=0.95,
                    do_sample=True
                )
                
                # 解码生成的回答
                answers = []
                for i in range(num_return_sequences):
                    # 提取生成的回答（去除输入部分）
                    answer = self.processor.tokenizer.decode(outputs[i], skip_special_tokens=True)
                    
                    # 尝试从生成文本中提取回答部分
                    if "回答:" in answer:
                        answer = answer.split("回答:")[-1].strip()
                    elif "答案是" in answer:
                        answer = answer.split("答案是")[-1].strip()
                    elif "根据图像" in answer:
                        answer = answer.split("根据图像")[-1].strip()
                    
                    answers.append(answer)
                
                answer = answers[0] if num_return_sequences == 1 else answers
        
        # 返回结果
        if num_return_sequences == 1:
            return {
                "answer": answer,
                "model": self.model_name,
                "question": question,
                "prompt": prompt if self.use_generic_model else None
            }
        else:
            return {
                "answers": answers,
                "model": self.model_name,
                "question": question,
                "prompt": prompt if self.use_generic_model else None
            }

# 示例使用代码
# # 初始化视觉问答系统
# vqa_system = AdvancedVisualQuestionAnswering(
#     model_name="microsoft/phi-3-vision-128k-instruct"
# )
# print("模型信息:", vqa_system.get_model_info())
# 
# # 示例图像（可以是本地路径或URL）
# image_path = "https://images.unsplash.com/photo-1542291026-7eec264c27ff"
# 
# try:
#     # 加载图像
#     image = vqa_system.load_image(image_path)
#     print(f"图像已加载: {image.size}, {image.mode}")
#     
#     # 示例问题
#     questions = [
#         "图像中有什么动物？",
#         "图像中的动物是什么颜色的？"
#     ]
#     
#     # 回答单个问题
#     question = questions[0]
#     result = vqa_system.answer_question(image, question)
#     print(f"问题: {question}")
#     print(f"回答: {result['answer']}")
# except Exception as e:
#     print(f"运行视觉问答系统失败: {e}")

# 视觉问答模型知识蒸馏优化示例
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoProcessor, AutoModelForCausalLM, AutoModelForVisualQuestionAnswering
import numpy as np
from PIL import Image
import requests
from tqdm import tqdm
import time
import os
import json
from torch.utils.data import DataLoader, Dataset

class VQADistillationStudent(nn.Module):
    def __init__(self, image_encoder, text_encoder, fusion_dim=768, dropout=0.1):
        super().__init__()
        # 学生模型的简化架构
        self.image_encoder = image_encoder
        self.text_encoder = text_encoder
        self.fusion_layer = nn.Linear(fusion_dim * 2, fusion_dim)
        self.dropout = nn.Dropout(dropout)
        self.output_layer = nn.Linear(fusion_dim, 1000)  # 假设有1000个可能的答案类别
        self.layer_norm = nn.LayerNorm(fusion_dim)
    
    def forward(self, pixel_values, input_ids, attention_mask=None):
        # 提取图像特征
        image_features = self.image_encoder(pixel_values).last_hidden_state[:, 0, :]  # [CLS] token
        
        # 提取文本特征
        text_output = self.text_encoder(input_ids, attention_mask=attention_mask)
        text_features = text_output.last_hidden_state[:, 0, :]  # [CLS] token
        
        # 融合特征
        combined = torch.cat([image_features, text_features], dim=-1)
        fused = self.fusion_layer(combined)
        fused = F.gelu(fused)
        fused = self.layer_norm(fused)
        fused = self.dropout(fused)
        
        # 生成输出
        logits = self.output_layer(fused)
        
        return logits

class VQADistillationTrainer:
    def __init__(self, teacher_model_name="microsoft/phi-3-vision-128k-instruct", student_config=None, device=None):
        # 设置设备
        self.device = device if device is not None else ("cuda" if torch.cuda.is_available() else "cpu")
        
        # 加载教师模型
        print(f"加载教师模型: {teacher_model_name}")
        try:
            self.teacher_model = AutoModelForVisualQuestionAnswering.from_pretrained(
                teacher_model_name,
                device_map="auto" if torch.cuda.is_available() else None,
                trust_remote_code=True
            )
        except Exception:
            # 如果无法加载专用VQA模型，使用通用多模态模型
            self.teacher_model = AutoModelForCausalLM.from_pretrained(
                teacher_model_name,
                device_map="auto" if torch.cuda.is_available() else None,
                trust_remote_code=True
            )
        
        self.teacher_processor = AutoProcessor.from_pretrained(teacher_model_name, trust_remote_code=True)
        self.teacher_model.eval()  # 设置为评估模式
        
        # 创建学生模型
        if student_config is None:
            student_config = {
                "fusion_dim": 768,
                "dropout": 0.1
            }
        
        # 为了简化示例，我们使用教师模型的部分组件来构建学生模型
        self.student_model = VQADistillationStudent(
            image_encoder=self.get_simplified_image_encoder(),
            text_encoder=self.get_simplified_text_encoder(),
            fusion_dim=student_config["fusion_dim"],
            dropout=student_config["dropout"]
        )
        
        self.student_model.to(self.device)
        
        # 初始化优化器和损失函数
        self.optimizer = torch.optim.AdamW(self.student_model.parameters(), lr=5e-5)
        self.temperature = 2.0  # 知识蒸馏的温度参数
        self.alpha = 0.7  # 软标签损失的权重
        self.beta = 0.3   # 硬标签损失的权重
        
        # 知识蒸馏特有的损失函数
        self.soft_criterion = nn.KLDivLoss(reduction="batchmean")
        self.hard_criterion = nn.CrossEntropyLoss()
    
    def get_teacher_logits(self, image, question):
        # 获取教师模型的输出logits
        with torch.no_grad():
            inputs = self.teacher_processor(
                images=image,
                text=question,
                return_tensors="pt"
            )
            
            # 移至设备
            if not hasattr(self.teacher_model, "hf_device_map") or self.teacher_model.hf_device_map is None:
                inputs = {k: v.to(self.device) for k, v in inputs.items()}
            
            # 获取教师模型输出
            try:
                # 尝试专用VQA模型的输出格式
                outputs = self.teacher_model(**inputs)
                teacher_logits = outputs.logits
            except Exception:
                # 尝试通用多模态模型的输出格式
                outputs = self.teacher_model.generate(
                    **inputs,
                    max_length=100,
                    output_scores=True,
                    return_dict_in_generate=True
                )
                # 假设最后一个token的分数是答案的分数
                teacher_logits = outputs.scores[-1].mean(dim=0).unsqueeze(0)
        
        return teacher_logits
    
    def train_step(self, image, question, answer_label=None):
        # 单个训练步骤
        # 获取教师模型的软标签
        teacher_logits = self.get_teacher_logits(image, question)
        teacher_probs = F.softmax(teacher_logits / self.temperature, dim=-1)
        
        # 准备学生模型的输入
        student_inputs = self.teacher_processor(
            images=image,
            text=question,
            return_tensors="pt"
        )
        
        # 移至设备
        student_inputs = {k: v.to(self.device) for k, v in student_inputs.items()}
        
        # 学生模型前向传播
        student_logits = self.student_model(
            student_inputs["pixel_values"],
            student_inputs["input_ids"],
            student_inputs["attention_mask"] if "attention_mask" in student_inputs else None
        )
        
        # 计算软标签损失（知识蒸馏损失）
        student_log_probs = F.log_softmax(student_logits / self.temperature, dim=-1)
        soft_loss = self.soft_criterion(student_log_probs, teacher_probs)
        
        # 计算硬标签损失（如果有标签）
        hard_loss = 0
        if answer_label is not None:
            if isinstance(answer_label, str):
                # 简单地将答案字符串转换为哈希值作为类别索引
                answer_index = hash(answer_label) % 1000
                answer_tensor = torch.tensor([answer_index], device=self.device)
            else:
                answer_tensor = answer_label.to(self.device)
            
            hard_loss = self.hard_criterion(student_logits, answer_tensor)
        
        # 总损失
        total_loss = self.alpha * soft_loss + self.beta * hard_loss
        
        # 反向传播和优化
        self.optimizer.zero_grad()
        total_loss.backward()
        self.optimizer.step()
        
        return {
            "total_loss": total_loss.item(),
            "soft_loss": soft_loss.item(),
            "hard_loss": hard_loss.item() if answer_label is not None else 0
        }

# 示例使用代码
# # 初始化知识蒸馏训练器
# distillation_trainer = VQADistillationTrainer(
#     teacher_model_name="microsoft/phi-3-vision-128k-instruct"
# )
# 
# # 示例图像、问题和答案
# image_path = "https://images.unsplash.com/photo-1542291026-7eec264c27ff"
# image = distillation_trainer.load_image(image_path)
# question = "图像中有什么动物？"
# answer = "狗"
# 
# # 单步训练
# result = distillation_trainer.train_step(image, question, answer)
# print(f"训练结果: 总损失={result['total_loss']:.4f}")

# 视觉问答模型知识蒸馏示例
import torch
import torch.nn as nn
import torch.nn.functional as F

class VQADistillationTrainer:
    def __init__(self, teacher_model, student_model, temperature=2.0, alpha=0.7):
        self.teacher_model = teacher_model
        self.student_model = student_model
        self.temperature = temperature  # 蒸馏温度
        self.alpha = alpha  # 软标签和硬标签的权重
        self.criterion = nn.CrossEntropyLoss()
        self.optimizer = torch.optim.Adam(self.student_model.parameters(), lr=1e-4)
        
    def train_step(self, image_inputs, text_inputs, labels):
        # 教师模型不参与梯度更新
        with torch.no_grad():
            teacher_logits = self.teacher_model(image_inputs, text_inputs)
        
        # 学生模型前向传播
        student_logits = self.student_model(image_inputs, text_inputs)
        
        # 计算软标签损失
        soft_teacher_probs = F.softmax(teacher_logits / self.temperature, dim=1)
        soft_student_log_probs = F.log_softmax(student_logits / self.temperature, dim=1)
        soft_loss = F.kl_div(soft_student_log_probs, soft_teacher_probs, reduction='batchmean') * (self.temperature ** 2)
        
        # 计算硬标签损失
        hard_loss = self.criterion(student_logits, labels)
        
        # 总损失
        loss = self.alpha * soft_loss + (1 - self.alpha) * hard_loss
        
        # 反向传播和优化
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()
        
        return loss.item()

# 示例用法
# teacher_model = ...  # 预训练好的大型VQA模型
# student_model = ...  # 小型VQA模型
# 
# trainer = VQADistillationTrainer(teacher_model, student_model)
# 
# # 训练循环
# # for epoch in range(num_epochs):
# #     for batch in dataloader:
# #         image_inputs, text_inputs, labels = batch
# #         loss = trainer.train_step(image_inputs, text_inputs, labels)