2025年高级视频理解技术：从分类到深度内容分析

# 2025年高级视频分类器实现
import torch
import torch.nn as nn
from transformers import VideoMAEImageProcessor, VideoMAEModel
from transformers import AutoModelForVideoClassification
import numpy as np

class AdvancedVideoClassifier:
    def __init__(self, model_name="microsoft/video-mae-base-finetuned-kinetics400"):
        # 加载预训练的视频分类模型
        self.processor = VideoMAEImageProcessor.from_pretrained(model_name)
        self.model = AutoModelForVideoClassification.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            use_safetensors=True
        )
        
        # 移至GPU（如果可用）
        if torch.cuda.is_available():
            self.model.to("cuda")
        
        # 设置评估模式
        self.model.eval()
        
        # 获取类别标签
        self.id2label = self.model.config.id2label
        self.label2id = self.model.config.label2id
        
        # 设置默认参数
        self.default_params = {
            "num_frames": 16,
            "frame_rate": 8,
            "clip_duration": 2.0  # 秒
        }
    
    def preprocess_video(self, video_frames, **kwargs):
        # 预处理视频帧
        params = {**self.default_params, **kwargs}
        
        # 确保输入是正确的格式
        if isinstance(video_frames, np.ndarray):
            # 如果是numpy数组，转换为PIL图像列表
            video_frames = [self._numpy_to_pil(frame) for frame in video_frames]
        
        # 采样指定数量的帧
        if len(video_frames) > params["num_frames"]:
            # 均匀采样
            indices = np.linspace(0, len(video_frames) - 1, params["num_frames"]).astype(int)
            video_frames = [video_frames[i] for i in indices]
        
        # 使用processor预处理
        inputs = self.processor(
            videos=video_frames,
            return_tensors="pt"
        )
        
        # 移至GPU（如果可用）
        if torch.cuda.is_available():
            inputs = {k: v.to("cuda") for k, v in inputs.items()}
        
        return inputs
    
    def _numpy_to_pil(self, frame):
        # 将numpy数组转换为PIL图像
        from PIL import Image
        if frame.dtype != np.uint8:
            frame = (frame * 255).astype(np.uint8)
        return Image.fromarray(frame)
    
    def classify_video(self, video_frames, top_k=5, **kwargs):
        # 分类视频
        inputs = self.preprocess_video(video_frames, **kwargs)
        
        # 模型推理
        with torch.no_grad():
            outputs = self.model(**inputs)
        
        # 计算概率
        logits = outputs.logits
        probabilities = torch.nn.functional.softmax(logits, dim=-1)
        
        # 获取top-k预测结果
        top_probabilities, top_indices = torch.topk(probabilities, top_k)
        
        # 转换为人类可读的标签
        results = []
        for i in range(top_k):
            label_id = top_indices[0][i].item()
            label = self.id2label[label_id]
            prob = top_probabilities[0][i].item()
            results.append({
                "label": label,
                "probability": prob,
                "label_id": label_id
            })
        
        return results
    
    def extract_features(self, video_frames, **kwargs):
        # 提取视频特征
        inputs = self.preprocess_video(video_frames, **kwargs)
        
        # 访问底层模型以获取特征
        with torch.no_grad():
            # 获取最后一层隐藏状态
            outputs = self.model.base_model(**inputs)
            features = outputs.last_hidden_state.mean(dim=1)  # 平均池化
        
        return features

# 视频动作识别实现
import torch
from transformers import AutoProcessor, AutoModelForVideoClassification
import numpy as np

class AdvancedActionRecognizer:
    def __init__(self, model_name="microsoft/x3d-m"):
        # 加载预训练的动作识别模型
        self.processor = AutoProcessor.from_pretrained(model_name)
        self.model = AutoModelForVideoClassification.from_pretrained(
            model_name,
            torch_dtype=torch.float16,
            use_safetensors=True
        )
        
        # 移至GPU（如果可用）
        if torch.cuda.is_available():
            self.model.to("cuda")
        
        # 设置评估模式
        self.model.eval()
        
        # 获取类别标签
        self.id2label = self.model.config.id2label
        self.label2id = self.model.config.label2id
        
        # 设置默认参数
        self.default_params = {
            "num_frames": 16,
            "frame_rate": 8,
            "clip_duration": 2.0  # 秒
        }
    
    def preprocess_video(self, video_frames, **kwargs):
        # 预处理视频帧
        params = {**self.default_params, **kwargs}
        
        # 确保输入是正确的格式
        if isinstance(video_frames, np.ndarray):
            # 如果是numpy数组，转换为PIL图像列表
            video_frames = [self._numpy_to_pil(frame) for frame in video_frames]
        
        # 采样指定数量的帧
        if len(video_frames) > params["num_frames"]:
            # 均匀采样
            indices = np.linspace(0, len(video_frames) - 1, params["num_frames"]).astype(int)
            video_frames = [video_frames[i] for i in indices]
        
        # 使用processor预处理
        inputs = self.processor(
            videos=video_frames,
            return_tensors="pt"
        )
        
        # 移至GPU（如果可用）
        if torch.cuda.is_available():
            inputs = {k: v.to("cuda") for k, v in inputs.items()}
        
        return inputs
    
    def _numpy_to_pil(self, frame):
        # 将numpy数组转换为PIL图像
        from PIL import Image
        if frame.dtype != np.uint8:
            frame = (frame * 255).astype(np.uint8)
        return Image.fromarray(frame)
    
    def recognize_action(self, video_frames, top_k=5, **kwargs):
        # 识别视频中的动作
        inputs = self.preprocess_video(video_frames, **kwargs)
        
        # 模型推理
        with torch.no_grad():
            outputs = self.model(**inputs)
        
        # 计算概率
        logits = outputs.logits
        probabilities = torch.nn.functional.softmax(logits, dim=-1)
        
        # 获取top-k预测结果
        top_probabilities, top_indices = torch.topk(probabilities, top_k)
        
        # 转换为人类可读的标签
        results = []
        for i in range(top_k):
            label_id = top_indices[0][i].item()
            label = self.id2label[label_id]
            prob = top_probabilities[0][i].item()
            results.append({
                "action": label,
                "probability": prob,
                "label_id": label_id
            })
        
        return results

# 视频分类实现示例
import torch
import numpy as np
from transformers import VideoMAEImageProcessor, AutoModelForVideoClassification

# 加载模型和处理器
model_name = "microsoft/video-mae-base-finetuned-kinetics400"
processor = VideoMAEImageProcessor.from_pretrained(model_name)
model = AutoModelForVideoClassification.from_pretrained(
    model_name,
    torch_dtype=torch.float16,
    use_safetensors=True
)

# 移至GPU（如果可用）
if torch.cuda.is_available():
    model.to("cuda")
    model.eval()

# 假设我们有一个视频帧列表 video_frames（这里仅作示例）
# video_frames = [...]  # 实际应用中从视频文件中提取

# 预处理视频帧
inputs = processor(
    videos=video_frames,
    return_tensors="pt"
)

# 移至GPU（如果可用）
if torch.cuda.is_available():
    inputs = {k: v.to("cuda") for k, v in inputs.items()}

# 模型推理
with torch.no_grad():
    outputs = model(**inputs)

# 计算概率和获取预测结果
logits = outputs.logits
probabilities = torch.nn.functional.softmax(logits, dim=-1)

# 获取top-5预测结果
top_probs, top_indices = torch.topk(probabilities, 5)

# 获取类别标签
id2label = model.config.id2label

# 输出预测结果
print("视频分类结果：")
for i in range(5):
    label_id = top_indices[0][i].item()
    label = id2label[label_id]
    prob = top_probs[0][i].item()
    print(f"{i+1}. {label}: {prob:.4f}")

# 视频理解模型知识蒸馏示例
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoModelForVideoClassification

class VideoKnowledgeDistillation:
    def __init__(self, teacher_model_name, student_model_name):
        # 加载教师模型（大型高精度模型）
        self.teacher_model = AutoModelForVideoClassification.from_pretrained(
            teacher_model_name,
            torch_dtype=torch.float16,
            use_safetensors=True
        )
        for param in self.teacher_model.parameters():
            param.requires_grad = False  # 冻结教师模型参数
        
        # 加载学生模型（小型高效模型）
        self.student_model = AutoModelForVideoClassification.from_pretrained(
            student_model_name,
            torch_dtype=torch.float16,
            use_safetensors=True
        )
        
        # 定义优化器
        self.optimizer = torch.optim.AdamW(
            self.student_model.parameters(),
            lr=1e-5,
            weight_decay=1e-4
        )
        
        # 定义温度参数
        self.temperature = 2.0
        
        # 定义损失权重
        self.ce_loss_weight = 0.5  # 交叉熵损失权重
        self.kd_loss_weight = 0.5  # 知识蒸馏损失权重
        
        # 移至GPU（如果可用）
        if torch.cuda.is_available():
            self.teacher_model = self.teacher_model.to("cuda")
            self.student_model = self.student_model.to("cuda")
    
    def compute_kd_loss(self, student_logits, teacher_logits):
        # 计算知识蒸馏损失（软标签损失）
        student_soft = F.log_softmax(student_logits / self.temperature, dim=-1)
        teacher_soft = F.softmax(teacher_logits / self.temperature, dim=-1)
        
        # 计算KL散度损失
        kd_loss = F.kl_div(student_soft, teacher_soft, reduction="batchmean") * (self.temperature ** 2)
        
        return kd_loss
    
    def train_step(self, inputs, labels):
        # 教师模型推理（不计算梯度）
        with torch.no_grad():
            teacher_outputs = self.teacher_model(**inputs)
            teacher_logits = teacher_outputs.logits
        
        # 学生模型推理
        student_outputs = self.student_model(**inputs)
        student_logits = student_outputs.logits
        
        # 计算交叉熵损失（硬标签）
        ce_loss = F.cross_entropy(student_logits, labels)
        
        # 计算知识蒸馏损失（软标签）
        kd_loss = self.compute_kd_loss(student_logits, teacher_logits)
        
        # 计算总损失
        total_loss = (
            self.ce_loss_weight * ce_loss +
            self.kd_loss_weight * kd_loss
        )
        
        # 反向传播和参数更新
        self.optimizer.zero_grad()
        total_loss.backward()
        self.optimizer.step()
        
        # 计算准确率
        _, predicted = torch.max(student_logits, 1)
        accuracy = (predicted == labels).float().mean()
        
        return {
            'total_loss': total_loss.item(),
            'ce_loss': ce_loss.item(),
            'kd_loss': kd_loss.item(),
            'accuracy': accuracy.item()
        }