强推!计算机博士通俗易懂讲透slowfast行为识别算法,入门到跑通代码,草履虫都能学会!
最近在视频行为识别项目中,发现很多同学对SlowFast算法既感兴趣又觉得难以入手。网上的资料要么过于理论化,要么代码不完整跑不通。本文将从零开始,用最通俗的语言拆解SlowFast的核心思想,并提供完整的代码实现和调参技巧,让即使是刚入门的小白也能轻松上手。
1. 行为识别与SlowFast算法背景
1.1 什么是视频行为识别
视频行为识别是计算机视觉领域的一个重要分支,旨在让计算机能够理解视频中人物的动作和行为。与图像识别不同,视频行为识别需要处理时间维度上的信息,这就涉及到对视频帧序列的分析和理解。
在实际应用中,视频行为识别技术广泛应用于智能监控、人机交互、体育分析、医疗康复等领域。比如在安防监控中,系统可以自动识别出异常行为(如打架、跌倒);在智能家居中,可以通过手势识别控制家电;在体育训练中,可以分析运动员的技术动作。
1.2 SlowFast算法的诞生背景
传统的视频行为识别方法主要面临两个挑战:一是如何有效捕捉时间维度上的运动信息,二是如何在计算效率和识别精度之间取得平衡。SlowFast算法正是针对这些问题提出的创新解决方案。
该算法由Facebook AI Research在2019年提出,其核心思想很直观:人类视觉系统在处理动态信息时,既有对整体场景的慢速感知,也有对快速变化的敏感捕捉。SlowFast网络通过两个并行的通路来模拟这一机制,分别处理不同的时间尺度信息。
2. SlowFast算法核心原理详解
2.1 双通路架构设计
SlowFast网络最核心的创新就是其双通路设计。这两个通路各有特点,相互补充:
慢通路(Slow Pathway):
- 处理低帧率的视频帧序列(通常为原始视频的1/16帧率)
- 通道数较多,提取丰富的空间特征
- 主要负责捕捉静态场景信息和缓慢变化的特征
- 类似于人类视觉中对整体环境的感知
快通路(Fast Pathway):
- 处理高帧率的视频帧序列(通常与原始视频帧率相同)
- 通道数较少,计算量相对较小
- 专门负责捕捉快速运动的细节信息
- 类似于人类视觉中对快速变化的敏感反应
2.2 时间尺度与通道设计
两个通路在时间尺度上存在显著差异。假设原始视频帧率为30fps,Slow通路可能只处理2fps的帧序列,而Fast通路处理30fps的完整帧序列。这种设计使得网络能够同时捕捉长期的时间依赖关系和短期的快速运动。
在通道数设计上,Slow通路的通道数通常是Fast通路的4-8倍。这是因为空间特征相对复杂,需要更多的通道来编码,而时间特征相对简单,可以用较少的通道处理。
2.3 横向连接与信息融合
两个通路之间通过横向连接进行信息交互,主要包括两种类型的连接:
时间维度连接:将Fast通路的高时间分辨率特征下采样到与Slow通路相同的时间尺度,然后进行融合。
特征维度连接:通过1×1卷积调整通道数,确保两个通路的特征图能够正确拼接。
这种融合机制使得网络能够综合利用空间细节和时间动态信息,大大提升了行为识别的准确性。
3. 环境准备与依赖安装
3.1 基础环境要求
在开始代码实现之前,我们需要准备好开发环境。以下是推荐的基础配置:
- 操作系统:Ubuntu 18.04+ 或 Windows 10+(推荐Linux环境)
- Python版本:3.7或3.8(3.6可能存在兼容性问题)
- 深度学习框架:PyTorch 1.7+ 或 TensorFlow 2.4+
- GPU支持:CUDA 10.2+,cuDNN 7.6.5+(可选但强烈推荐)
3.2 依赖包安装
创建并激活Python虚拟环境后,安装必要的依赖包:
# 创建虚拟环境 python -m venv slowfast_env source slowfast_env/bin/activate # Linux/Mac # 或者 slowfast_env\Scripts\activate # Windows # 安装PyTorch(根据CUDA版本选择) pip install torch==1.9.0+cu111 torchvision==0.10.0+cu111 torchaudio==0.9.0 -f https://download.pytorch.org/whl/torch_stable.html # 安装其他依赖 pip install opencv-python pillow matplotlib numpy scipy sklearn pandas tqdm pip install pytorchvideo # Facebook官方视频处理库 pip install fvcore iopath # 深度学习工具库3.3 项目结构准备
建议的项目目录结构如下:
slowfast-project/ ├── configs/ # 配置文件 ├── datasets/ # 数据集目录 ├── models/ # 模型定义 ├── utils/ # 工具函数 ├── train.py # 训练脚本 ├── test.py # 测试脚本 └── requirements.txt # 依赖列表4. SlowFast模型代码实现
4.1 基础模块定义
首先实现SlowFast网络的基础构建模块:
import torch import torch.nn as nn import torch.nn.functional as F from torchvision.models.video import r3d_18 class SlowFastBase(nn.Module): """SlowFast网络基础模块""" def __init__(self, slow_temporal_stride=16, fast_temporal_stride=2, alpha=4, beta=8, fusion_method='concat'): super(SlowFastBase, self).__init__() self.slow_temporal_stride = slow_temporal_stride self.fast_temporal_stride = fast_temporal_stride self.alpha = alpha # 时间尺度比例 self.beta = beta # 通道数比例 self.fusion_method = fusion_method def lateral_connection(self, fast_feat, slow_feat): """横向连接:将Fast通路特征融合到Slow通路""" # 时间维度下采样 batch_size, channels, time, height, width = fast_feat.shape target_time = slow_feat.shape[2] if time != target_time: # 使用3D自适应池化进行时间维度下采样 fast_feat = F.adaptive_avg_pool3d( fast_feat, (target_time, height, width) ) # 通道数调整 if fast_feat.shape[1] != slow_feat.shape[1]: fast_feat = nn.Conv3d( fast_feat.shape[1], slow_feat.shape[1], kernel_size=1, stride=1, padding=0 )(fast_feat) return fast_feat4.2 残差块实现
class Bottleneck3D(nn.Module): """3D残差瓶颈块""" def __init__(self, in_channels, out_channels, stride=1, expansion=4): super(Bottleneck3D, self).__init__() mid_channels = out_channels // expansion self.conv1 = nn.Conv3d(in_channels, mid_channels, kernel_size=1, bias=False) self.bn1 = nn.BatchNorm3d(mid_channels) self.conv2 = nn.Conv3d(mid_channels, mid_channels, kernel_size=3, stride=stride, padding=1, bias=False) self.bn2 = nn.BatchNorm3d(mid_channels) self.conv3 = nn.Conv3d(mid_channels, out_channels, kernel_size=1, bias=False) self.bn3 = nn.BatchNorm3d(out_channels) self.relu = nn.ReLU(inplace=True) # shortcut连接 self.shortcut = nn.Sequential() if stride != 1 or in_channels != out_channels: self.shortcut = nn.Sequential( nn.Conv3d(in_channels, out_channels, kernel_size=1, stride=stride, bias=False), nn.BatchNorm3d(out_channels) ) def forward(self, x): residual = x out = self.conv1(x) out = self.bn1(out) out = self.relu(out) out = self.conv2(out) out = self.bn2(out) out = self.relu(out) out = self.conv3(out) out = self.bn3(out) out += self.shortcut(residual) out = self.relu(out) return out4.3 完整的SlowFast网络实现
class SlowFast(nn.Module): """完整的SlowFast网络实现""" def __init__(self, num_classes=400, slow_temporal_stride=16, fast_temporal_stride=2, alpha=4, beta=8): super(SlowFast, self).__init__() self.slow_temporal_stride = slow_temporal_stride self.fast_temporal_stride = fast_temporal_stride self.alpha = alpha self.beta = beta # Slow通路初始卷积 self.slow_conv1 = nn.Conv3d(3, 64, kernel_size=(1, 7, 7), stride=(1, 2, 2), padding=(0, 3, 3), bias=False) self.slow_bn1 = nn.BatchNorm3d(64) self.slow_relu = nn.ReLU(inplace=True) self.slow_maxpool = nn.MaxPool3d(kernel_size=(1, 3, 3), stride=(1, 2, 2), padding=(0, 1, 1)) # Fast通路初始卷积(通道数较少) fast_in_channels = 3 fast_out_channels = 64 // beta self.fast_conv1 = nn.Conv3d(fast_in_channels, fast_out_channels, kernel_size=(5, 7, 7), stride=(1, 2, 2), padding=(2, 3, 3), bias=False) self.fast_bn1 = nn.BatchNorm3d(fast_out_channels) self.fast_relu = nn.ReLU(inplace=True) self.fast_maxpool = nn.MaxPool3d(kernel_size=(1, 3, 3), stride=(1, 2, 2), padding=(0, 1, 1)) # 残差块配置 self.res_layers = self._make_res_layers() # 分类头 self.avgpool = nn.AdaptiveAvgPool3d((1, 1, 1)) self.dropout = nn.Dropout(0.5) self.fc = nn.Linear(2048 + 256, num_classes) # Slow + Fast特征拼接 def _make_res_layers(self): """创建残差层""" # 简化实现,实际应根据论文配置完整的残差块 layers = nn.ModuleDict() # 这里添加各阶段的残差块... return layers def forward(self, x): # 输入x的形状: (batch, channels, time, height, width) batch_size, _, time, height, width = x.shape # 生成Slow和Fast通路的输入 slow_x = x[:, :, ::self.slow_temporal_stride, :, :] # 时间下采样 fast_x = x[:, :, ::self.fast_temporal_stride, :, :] # 完整时间分辨率 # Slow通路前向传播 slow_out = self.slow_conv1(slow_x) slow_out = self.slow_bn1(slow_out) slow_out = self.slow_relu(slow_out) slow_out = self.slow_maxpool(slow_out) # Fast通路前向传播 fast_out = self.fast_conv1(fast_x) fast_out = self.fast_bn1(fast_out) fast_out = self.fast_relu(fast_out) fast_out = self.fast_maxpool(fast_out) # 残差块处理(简化) # slow_out, fast_out = self.res_layers(slow_out, fast_out) # 特征融合 slow_out = self.avgpool(slow_out) fast_out = self.avgpool(fast_out) # 展平 slow_out = slow_out.view(slow_out.size(0), -1) fast_out = fast_out.view(fast_out.size(0), -1) # 特征拼接 combined = torch.cat([slow_out, fast_out], dim=1) combined = self.dropout(combined) output = self.fc(combined) return output5. 数据预处理与加载
5.1 视频数据预处理
视频行为识别的数据预处理相对复杂,需要处理时间维度:
import torch from torch.utils.data import Dataset, DataLoader import cv2 import os from PIL import Image import numpy as np class VideoDataset(Dataset): """视频数据集类""" def __init__(self, video_paths, labels, clip_length=32, frame_rate=30, crop_size=224, is_training=True): self.video_paths = video_paths self.labels = labels self.clip_length = clip_length self.frame_rate = frame_rate self.crop_size = crop_size self.is_training = is_training # 数据增强变换 if is_training: self.transform = self._get_train_transform() else: self.transform = self._get_val_transform() def _get_train_transform(self): """训练时数据增强""" from torchvision import transforms return transforms.Compose([ transforms.ToPILImage(), transforms.RandomResizedCrop(self.crop_size), transforms.RandomHorizontalFlip(0.5), transforms.ColorJitter(0.2, 0.2, 0.2), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) def _get_val_transform(self): """验证时变换""" from torchvision import transforms return transforms.Compose([ transforms.ToPILImage(), transforms.Resize(256), transforms.CenterCrop(self.crop_size), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) def __len__(self): return len(self.video_paths) def __getitem__(self, idx): video_path = self.video_paths[idx] label = self.labels[idx] # 读取视频帧 frames = self._load_video_frames(video_path) # 时间维度采样 if len(frames) > self.clip_length: # 均匀采样 indices = np.linspace(0, len(frames)-1, self.clip_length, dtype=int) frames = [frames[i] for i in indices] else: # 重复最后一帧补全 while len(frames) < self.clip_length: frames.append(frames[-1]) frames = frames[:self.clip_length] # 应用变换 transformed_frames = [] for frame in frames: transformed_frame = self.transform(frame) transformed_frames.append(transformed_frame) # 转换为tensor: (C, T, H, W) -> (T, C, H, W) video_tensor = torch.stack(transformed_frames).permute(1, 0, 2, 3) return video_tensor, label def _load_video_frames(self, video_path): """从视频文件加载帧""" frames = [] cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise ValueError(f"无法打开视频文件: {video_path}") success, frame = cap.read() while success: # 转换BGR到RGB frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) frames.append(frame_rgb) success, frame = cap.read() cap.release() return frames5.2 数据加载器配置
def create_data_loaders(data_dir, batch_size=8, num_workers=4): """创建训练和验证数据加载器""" # 假设数据目录结构为: # data_dir/ # train/ # class1/ # class2/ # val/ # class1/ # class2/ train_video_paths = [] train_labels = [] val_video_paths = [] val_labels = [] # 遍历目录收集数据(简化实现) # 实际项目中需要根据具体数据集结构实现 train_dataset = VideoDataset(train_video_paths, train_labels, is_training=True) val_dataset = VideoDataset(val_video_paths, val_labels, is_training=False) train_loader = DataLoader( train_dataset, batch_size=batch_size, shuffle=True, num_workers=num_workers, pin_memory=True ) val_loader = DataLoader( val_dataset, batch_size=batch_size, shuffle=False, num_workers=num_workers, pin_memory=True ) return train_loader, val_loader6. 模型训练与调优
6.1 训练脚本实现
import torch import torch.nn as nn import torch.optim as optim from torch.optim.lr_scheduler import CosineAnnealingLR import time import os from tqdm import tqdm class SlowFastTrainer: """SlowFast模型训练器""" def __init__(self, model, train_loader, val_loader, device, config): self.model = model.to(device) self.train_loader = train_loader self.val_loader = val_loader self.device = device self.config = config # 损失函数和优化器 self.criterion = nn.CrossEntropyLoss() self.optimizer = optim.SGD( model.parameters(), lr=config['lr'], momentum=0.9, weight_decay=1e-4 ) # 学习率调度器 self.scheduler = CosineAnnealingLR( self.optimizer, T_max=config['epochs'] ) # 训练记录 self.train_losses = [] self.val_accuracies = [] self.best_accuracy = 0.0 def train_epoch(self, epoch): """训练一个周期""" self.model.train() running_loss = 0.0 correct = 0 total = 0 pbar = tqdm(self.train_loader, desc=f'Epoch {epoch+1} Training') for batch_idx, (inputs, targets) in enumerate(pbar): inputs, targets = inputs.to(self.device), targets.to(self.device) self.optimizer.zero_grad() outputs = self.model(inputs) loss = self.criterion(outputs, targets) loss.backward() self.optimizer.step() running_loss += loss.item() _, predicted = outputs.max(1) total += targets.size(0) correct += predicted.eq(targets).sum().item() # 更新进度条 pbar.set_postfix({ 'Loss': f'{loss.item():.3f}', 'Acc': f'{100.*correct/total:.2f}%' }) epoch_loss = running_loss / len(self.train_loader) epoch_acc = 100. * correct / total return epoch_loss, epoch_acc def validate(self, epoch): """验证模型""" self.model.eval() correct = 0 total = 0 with torch.no_grad(): for inputs, targets in tqdm(self.val_loader, desc=f'Epoch {epoch+1} Validation'): inputs, targets = inputs.to(self.device), targets.to(self.device) outputs = self.model(inputs) _, predicted = outputs.max(1) total += targets.size(0) correct += predicted.eq(targets).sum().item() accuracy = 100. * correct / total return accuracy def train(self): """完整训练流程""" print("开始训练SlowFast模型...") for epoch in range(self.config['epochs']): start_time = time.time() # 训练 train_loss, train_acc = self.train_epoch(epoch) # 验证 val_acc = self.validate(epoch) # 学习率调整 self.scheduler.step() # 记录结果 self.train_losses.append(train_loss) self.val_accuracies.append(val_acc) # 保存最佳模型 if val_acc > self.best_accuracy: self.best_accuracy = val_acc self.save_checkpoint(epoch, True) epoch_time = time.time() - start_time print(f'Epoch {epoch+1}/{self.config["epochs"]}: ' f'Train Loss: {train_loss:.4f}, ' f'Train Acc: {train_acc:.2f}%, ' f'Val Acc: {val_acc:.2f}%, ' f'Time: {epoch_time:.2f}s') def save_checkpoint(self, epoch, is_best=False): """保存模型检查点""" checkpoint = { 'epoch': epoch, 'model_state_dict': self.model.state_dict(), 'optimizer_state_dict': self.optimizer.state_dict(), 'best_accuracy': self.best_accuracy, 'train_losses': self.train_losses, 'val_accuracies': self.val_accuracies } filename = f'checkpoint_epoch_{epoch+1}.pth' if is_best: filename = 'best_model.pth' torch.save(checkpoint, os.path.join(self.config['save_dir'], filename))6.2 训练配置与启动
def main(): """主训练函数""" # 配置参数 config = { 'lr': 0.01, 'epochs': 100, 'batch_size': 8, 'save_dir': './checkpoints', 'clip_length': 32, 'frame_rate': 30 } # 创建目录 os.makedirs(config['save_dir'], exist_ok=True) # 设备设置 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f'使用设备: {device}') # 创建模型 model = SlowFast(num_classes=400) # 假设400个行为类别 # 创建数据加载器 train_loader, val_loader = create_data_loaders('./data', config['batch_size']) # 创建训练器 trainer = SlowFastTrainer(model, train_loader, val_loader, device, config) # 开始训练 trainer.train() if __name__ == '__main__': main()7. 模型推理与部署
7.1 单视频推理实现
class SlowFastInference: """SlowFast模型推理类""" def __init__(self, model_path, device='cuda'): self.device = torch.device(device if torch.cuda.is_available() else 'cpu') self.model = self.load_model(model_path) self.model.eval() # 预处理配置 self.clip_length = 32 self.crop_size = 224 def load_model(self, model_path): """加载训练好的模型""" checkpoint = torch.load(model_path, map_location='cpu') model = SlowFast(num_classes=checkpoint.get('num_classes', 400)) model.load_state_dict(checkpoint['model_state_dict']) return model.to(self.device) def preprocess_video(self, video_path): """预处理视频为模型输入格式""" # 读取视频帧 cap = cv2.VideoCapture(video_path) frames = [] while True: ret, frame = cap.read() if not ret: break frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) frames.append(frame_rgb) cap.release() # 帧采样 if len(frames) > self.clip_length: indices = np.linspace(0, len(frames)-1, self.clip_length, dtype=int) frames = [frames[i] for i in indices] else: while len(frames) < self.clip_length: frames.append(frames[-1]) frames = frames[:self.clip_length] # 应用变换 transform = transforms.Compose([ transforms.ToPILImage(), transforms.Resize(256), transforms.CenterCrop(self.crop_size), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) processed_frames = [] for frame in frames: processed_frame = transform(frame) processed_frames.append(processed_frame) # 转换为tensor: (C, T, H, W) video_tensor = torch.stack(processed_frames).permute(1, 0, 2, 3) return video_tensor.unsqueeze(0) # 添加batch维度 def predict(self, video_path): """对单个视频进行行为识别""" # 预处理 input_tensor = self.preprocess_video(video_path) input_tensor = input_tensor.to(self.device) # 推理 with torch.no_grad(): outputs = self.model(input_tensor) probabilities = F.softmax(outputs, dim=1) predicted_class = outputs.argmax(dim=1).item() confidence = probabilities[0][predicted_class].item() return predicted_class, confidence # 使用示例 if __name__ == '__main__': inference = SlowFastInference('best_model.pth') # 对测试视频进行预测 video_path = 'test_video.mp4' class_id, confidence = inference.predict(video_path) # 假设有类别名称映射 class_names = ['走路', '跑步', '跳跃', '挥手', ...] # 根据实际数据集定义 print(f'预测行为: {class_names[class_id]}, 置信度: {confidence:.3f}')7.2 批量推理与性能优化
def batch_inference(model, video_paths, batch_size=4): """批量视频推理""" results = [] for i in range(0, len(video_paths), batch_size): batch_paths = video_paths[i:i+batch_size] batch_tensors = [] # 预处理批次视频 for path in batch_paths: tensor = inference.preprocess_video(path) batch_tensors.append(tensor) # 堆叠批次 batch_tensor = torch.cat(batch_tensors, dim=0).to(inference.device) # 批量推理 with torch.no_grad(): outputs = model(batch_tensor) probabilities = F.softmax(outputs, dim=1) predicted_classes = outputs.argmax(dim=1).cpu().numpy() confidences = probabilities.max(dim=1)[0].cpu().numpy() # 收集结果 for j, (class_id, conf) in enumerate(zip(predicted_classes, confidences)): results.append({ 'video_path': batch_paths[j], 'predicted_class': class_id, 'confidence': conf }) return results8. 常见问题与解决方案
8.1 内存不足问题
问题现象:训练时出现内存溢出错误,特别是处理长视频时。
解决方案:
- 减小批次大小(batch_size)
- 使用梯度累积技术
- 采用混合精度训练
- 使用数据并行训练
# 梯度累积示例 def train_with_gradient_accumulation(model, dataloader, accumulation_steps=4): optimizer.zero_grad() for i, (inputs, targets) in enumerate(dataloader): outputs = model(inputs) loss = criterion(outputs, targets) loss = loss / accumulation_steps # 归一化损失 loss.backward() if (i + 1) % accumulation_steps == 0: optimizer.step() optimizer.zero_grad()8.2 过拟合问题
问题现象:训练准确率很高,但验证准确率停滞不前。
解决方案:
- 增加数据增强
- 使用更严格的正则化
- 早停策略
- 标签平滑技术
# 标签平滑实现 class LabelSmoothingCrossEntropy(nn.Module): def __init__(self, smoothing=0.1): super(LabelSmoothingCrossEntropy, self).__init__() self.smoothing = smoothing def forward(self, logits, targets): confidence = 1.0 - self.smoothing log_probs = F.log_softmax(logits, dim=-1) nll_loss = -log_probs.gather(dim=-1, index=targets.unsqueeze(1)) nll_loss = nll_loss.squeeze(1) smooth_loss = -log_probs.mean(dim=-1) loss = confidence * nll_loss + self.smoothing * smooth_loss return loss.mean()8.3 训练不收敛问题
问题现象:损失值波动大,长时间不下降。
解决方案:
- 检查学习率设置
- 验证数据预处理是否正确
- 检查模型初始化
- 使用梯度裁剪
# 梯度裁剪示例 torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)9. 性能优化技巧
9.1 模型压缩与加速
知识蒸馏:使用大模型指导小模型训练
class KnowledgeDistillationLoss(nn.Module): def __init__(self, alpha=0.7, temperature=3): super().__init__() self.alpha = alpha self.temperature = temperature self.kl_loss = nn.KLDivLoss(reduction='batchmean') def forward(self, student_logits, teacher_logits, targets): # 软目标损失 soft_loss = self.kl_loss( F.log_softmax(student_logits/self.temperature, dim=1), F.softmax(teacher_logits/self.temperature, dim=1) ) * (self.temperature ** 2) # 硬目标损失 hard_loss = F.cross_entropy(student_logits, targets) return self.alpha * soft_loss + (1 - self.alpha) * hard_loss9.2 多尺度训练策略
class MultiScaleTraining: """多尺度训练增强模型鲁棒性""" def __init__(self, scales=[224, 256, 288]): self.scales = scales def get_random_scale(self): return random.choice(self.scales) def apply_scale(self, frames, target_scale): # 动态调整帧尺寸 resized_frames = [] for frame in frames: h, w = frame.shape[:2] # 保持长宽比调整尺寸 scale_factor = target_scale / min(h, w) new_h, new_w = int(h * scale_factor), int(w * scale_factor) resized_frame = cv2.resize(frame, (new_w, new_h)) resized_frames.append(resized_frame) return resized_frames10. 实际项目应用建议
10.1 工业级部署考虑
模型服务化:使用TorchServe或Triton进行模型部署
# 简单的Flask API服务示例 from flask import Flask, request, jsonify import base64 import cv2 import numpy as np app = Flask(__name__) inference_engine = SlowFastInference('best_model.pth') @app.route('/predict', methods=['POST']) def predict(): try: # 接收base64编码的视频或视频路径 video_data = request.json['video'] if 'base64' in video_data: # 解码base64视频 video_bytes = base64.b64decode(video_data['base64']) # 保存临时文件或直接处理 # ... 处理逻辑 else: video_path = video_data['path'] class_id, confidence = inference_engine.predict(video_path) return jsonify({ 'success': True, 'prediction': class_id, 'confidence': float(confidence), 'class_name': class_names[class_id] }) except Exception as e: return jsonify({'success': False, 'error': str(e)}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)10.2 持续学习与模型更新
增量学习策略:适应新的行为类别
class IncrementalLearning: """增量学习处理新类别""" def __init__(self, base_model, old_classes, new_classes): self.base_model = base_model self.old_classes = old_classes self.new_classes = new_classes self.total_classes = old_classes + new_classes # 扩展分类层 self.model = self.extend_classifier(base_model, self.total_classes) def extend_classifier(self, model, num_classes): # 替换最后的全连接层 in_features = model.fc.in_features model.fc = nn.Linear(in_features, num_classes) return model def fine_tune(self, new_data_loader, epochs=50): # 只训练新添加的权重,冻结其他层 for name, param in self.model.named_parameters(): if 'fc' not in name: # 只训练全连接层 param.requires_grad = False # 微调训练...通过本文的详细讲解和完整代码实现,相信你已经对SlowFast行为识别算法有了深入的理解。从理论基础到代码实践,从模型训练到实际部署,我们覆盖了完整的技术链路。在实际项目中,建议先从小的数据集开始实验,逐步调整参数优化性能。
行为识别技术正在快速发展,SlowFast作为其中的经典算法,为后续的研究奠定了重要基础。掌握这一技术将为你在计算机视觉领域的深入发展提供有力支撑。