如果你正在处理监控视频，想自动识别出“打架”“摔倒”“闯入”这类异常行为，但发现传统方法要么误报太多，要么对复杂场景束手无策，那么你遇到的正是视频异常检测领域的核心挑战。

最近，一个名为VQ-VAD的新方法在学术界引起了不小的关注。它没有选择在“异常”这个模糊且难以定义的概念上死磕，而是巧妙地转向了“正常”行为的建模。其核心思路是：先学会如何精准地描述和理解视频中“人”的正常运动模式，任何偏离这个“正常模式库”的行为，都可能是异常。

听起来很直观，但实现起来却需要解决一个关键问题：如何高效、鲁棒地表示“运动”？VQ-VAD 给出的答案是向量量化（Vector-quantized）。这不仅仅是又一个花哨的模型，它代表了一种解决思路的转变——从“大海捞针”式的异常检测，转向“建立正常行为的字典”来进行比对。对于从事安防、智慧城市、工业质检或相关AI算法开发的工程师和研究者来说，理解这种思路及其实现细节，可能比单纯追求更高的准确率数字更有价值。

本文将深入拆解 VQ-VAD 的核心思想、技术实现，并提供一个基于 PyTorch 的简化实践指南。你将了解到：

1. 为什么“以人为中心”和“运动表示”是当前视频异常检测的关键。
2. 向量量化（VQ）如何将连续、高维的运动特征压缩成一个离散的“行为词汇表”。
3. 如何从零开始搭建一个简化版的 VQ-VAD 训练和推理流程。
4. 在实际部署中可能遇到的坑以及对应的优化思路。

1. VQ-VAD 要解决的根本问题是什么？

在深入代码之前，我们必须先厘清视频异常检测（Video Anomaly Detection, VAD）的困境。传统方法，无论是基于手工特征（如光流、轨迹）还是早期深度学习模型，通常面临两大难题：

• “异常”的定义模糊且开放：异常行为是无穷无尽的（打架、偷窃、摔倒、逆行……），你无法收集所有异常样本进行训练。因此，主流方法都采用“半监督”或“无监督”学习，即只使用正常行为的视频进行训练。模型的目标是学习“正常”的模式，任何不符合该模式的事件即被判定为异常。
• 背景复杂性与语义鸿沟：监控场景中，光照变化、摄像机抖动、无关物体（如飘动的旗帜）都会产生剧烈的像素变化，但这些并非语义上的“异常”。模型很容易被这些“视觉异常”干扰，而忽略了真正的“行为异常”。

VQ-VAD 的突破点在于，它明确地将焦点锁定在“人”这个最重要的语义主体上（Human-centric），并认为“运动”是区分正常与异常行为更本质、更稳定的特征。它的核心假设是：正常的人类运动模式是有限且可学习的，而异常运动则是对这些模式的偏离。

因此，VQ-VAD 的任务被转化为两个子问题：

1. 如何从视频中提取出干净、鲁棒、以人为中心的运动表示？
2. 如何为这些连续的运动表示建立一个紧凑的“正常模式字典”，并量化其重建误差？

解决了这两个问题，异常检测就变成了计算当前运动模式与“正常字典”的匹配程度（或重建误差）。误差越大，异常的可能性越高。

2. 核心概念拆解：向量量化与运动表示学习

2.1 什么是以人为中心（Human-centric）？

这并不是简单地在画面中检测出人体框。VQ-VAD 论文中强调的“以人为中心”体现在特征层面。它通常利用现成的姿态估计器（如 HRNet、OpenPose）或人体解析模型，提取出人体的关键点序列（骨骼关节点）或密集的形变信息。这些特征直接描述了人的姿态和运动，极大地过滤了背景干扰，让模型专注于行为本身。

2.2 什么是运动表示（Motion Representation）？

对于一段视频片段，模型需要将其编码成一个固定长度的向量，这个向量要能概括这段时间内人体的运动信息。VQ-VAD 通常采用一个编码器-解码器（Encoder-Decoder）结构。

• 编码器（Encoder）：输入是连续多帧的人体关键点序列或裁剪后的人体区域图像块，输出是一个低维的、连续的“运动编码（Motion Code）”向量。这个向量蕴含了运动的语义。
• 解码器（Decoder）：尝试根据这个“运动编码”向量，重建出输入的运动信息（如预测未来帧的姿态）。

在训练阶段，模型通过最小化重建损失，迫使这个“运动编码”向量必须包含足够的信息来还原原始运动。

2.3 向量量化（Vector Quantization, VQ）如何工作？

这是 VQ-VAD 的灵魂。如果没有 VQ，编码器产生的“运动编码”是连续空间中的任意点。我们很难直接定义一个“正常范围”。

VQ 引入了一个可学习的“码本（Codebook）”，你可以把它想象成一个“正常行为词汇表”。这个码本由 K 个向量组成，每个向量代表一种基本的、正常的运动原型。

VQ 的工作流程如下：

1. 编码器为输入视频片段生成一个连续的运动编码向量z_e。
2. 在码本中寻找与z_e最相似（通常使用欧氏距离）的那个向量z_q。
3. 用找到的z_q替换原始的z_e，并将其传递给解码器进行重建。
4. 训练时，通过梯度直通估计器（Straight-Through Estimator）同时更新编码器、解码器和码本。

这个过程带来的巨大好处是：

• 离散化：所有正常的运动模式都被“量化”到码本中有限的几个原型向量上。码本的大小 K 是可控的。
• 异常度量：在推理时，给定一个测试视频片段，编码器产生z_e。模型计算z_e与码本中最接近向量z_q的距离。如果这个距离很大，说明当前运动模式在“正常词汇表”里找不到好的匹配，因此很可能是异常的。这个距离直接作为异常分数。
• 可解释性：码本中的每个向量可以对应一种典型的正常行为模式（例如“行走”、“站立”、“挥手”），尽管模型没有显式的语义标签。

3. 环境准备与依赖安装

我们将使用 PyTorch 来实现一个简化版的 VQ-VAD 核心流程。这个示例侧重于阐明 VQ-VAE（Vector Quantized Variational AutoEncoder）用于运动表示学习的部分，并模拟异常评分。

基础环境：

• 操作系统：Linux (Ubuntu 20.04+) 或 macOS，Windows 也可（需注意部分库的安装）。
• Python：3.8 或 3.9。
• CUDA：11.3 或以上（如果使用 GPU）。CPU 也可运行，但训练较慢。

主要依赖库：

# 创建虚拟环境（可选） conda create -n vqvad python=3.8 conda activate vqvad # 安装核心依赖 pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 # 请根据你的CUDA版本调整 pip install numpy pip install opencv-python pip install matplotlib pip install scikit-learn pip install tqdm pip install einops # 用于方便的张量操作 # 用于人体姿态估计（模拟“以人为中心”的特征提取） pip install mmcv pip install mmpose # 可能需要从源码安装，这里仅作示意。实践中可使用更轻量的库如 `lightweight-human-pose-estimation.pytorch` # 为简化示例，我们将使用合成的人体关键点数据。

项目结构建议：

vqvad_demo/ ├── data/ # 存放数据或生成模拟数据 ├── models/ # 模型定义 │ ├── __init__.py │ ├── encoder.py │ ├── decoder.py │ └── vqvae.py # VQ-VAE 整合模型 ├── utils/ # 工具函数 │ ├── data_loader.py │ └── visualization.py ├── configs/ # 配置文件 │ └── default.yaml ├── train.py # 训练脚本 ├── test.py # 测试与异常评分脚本 └── README.md

4. 核心模型实现：VQ-VAE for Motion

我们首先实现最关键的 VQ-VAE 部分。这里我们假设输入是人体关键点序列（形状为[Batch, T, J, C]，其中 T 是帧数，J 是关键点数量，C 是坐标维度）。

4.1 定义向量量化层（VectorQuantizer）

这是码本学习的关键模块。

# file: models/vector_quantizer.py import torch import torch.nn as nn import torch.nn.functional as F class VectorQuantizer(nn.Module): """ 向量量化层。 输入：编码器输出的连续特征 z_e [B, D, H, W] 或 [B, D, T] (展平后为 [B*N, D]) 输出：量化后的特征 z_q，以及量化损失。 """ def __init__(self, num_embeddings, embedding_dim, commitment_cost=0.25): super(VectorQuantizer, self).__init__() self.embedding_dim = embedding_dim self.num_embeddings = num_embeddings self.commitment_cost = commitment_cost # 初始化码本 self.embedding = nn.Embedding(self.num_embeddings, self.embedding_dim) # 使用均匀分布初始化 self.embedding.weight.data.uniform_(-1/self.num_embeddings, 1/self.num_embeddings) def forward(self, inputs): """ Args: inputs: [B, D, ...] 需要量化的特征 Returns: quantized: 量化后的特征，与 inputs 同形 loss: 量化损失 perplexity: 码本使用分布的熵，用于监控 """ # 将输入变形为 [B*N, D]，其中 N 是空间/时间维度的乘积 original_shape = inputs.shape # e.g., [B, D, T] flat_input = inputs.contiguous().view(-1, self.embedding_dim) # [B*T, D] # 计算输入与码本中所有向量的距离 distances = (torch.sum(flat_input**2, dim=1, keepdim=True) + torch.sum(self.embedding.weight**2, dim=1) - 2 * torch.matmul(flat_input, self.embedding.weight.t())) # [B*T, K] # 找到最接近的码本向量索引 encoding_indices = torch.argmin(distances, dim=1).unsqueeze(1) # [B*T, 1] encodings = torch.zeros(encoding_indices.shape[0], self.num_embeddings, device=inputs.device) encodings.scatter_(1, encoding_indices, 1) # one-hot 编码，[B*T, K] # 量化：用码本向量替换输入 quantized = torch.matmul(encodings, self.embedding.weight) # [B*T, D] quantized = quantized.view(original_shape) # 恢复原始形状 [B, D, T] # 计算损失： commitment loss + codebook loss # 使用 Straight-Through Estimator: 前向传播用 quantized，反向传播将梯度直接拷贝给 inputs # 因此，我们需要定义两个损失来更新码本和编码器 # Loss1: 让码本向量向输入特征靠近 codebook_loss = F.mse_loss(quantized.detach(), inputs) # 只更新码本 # Loss2: 让编码器输出向码本向量靠近（commitment） commitment_loss = F.mse_loss(quantized, inputs.detach()) # 只更新编码器 loss = codebook_loss + self.commitment_cost * commitment_loss # 为了梯度直通，将 quantized 的梯度直接赋给 inputs quantized = inputs + (quantized - inputs).detach() # 计算码本使用情况的困惑度（perplexity），用于监控训练 avg_probs = torch.mean(encodings, dim=0) perplexity = torch.exp(-torch.sum(avg_probs * torch.log(avg_probs + 1e-10))) return quantized, loss, perplexity

4.2 构建运动编码器与解码器

我们构建一个简单的时序卷积网络来处理关键点序列。

# file: models/encoder.py import torch.nn as nn import torch.nn.functional as F class MotionEncoder(nn.Module): """将关键点序列编码为运动特征向量。""" def __init__(self, input_dim, hidden_dim, latent_dim, num_layers=2): super(MotionEncoder, self).__init__() self.input_dim = input_dim # J * C self.hidden_dim = hidden_dim self.latent_dim = latent_dim # 使用一维卷积处理时序数据 self.conv_layers = nn.ModuleList() in_channels = input_dim for i in range(num_layers): self.conv_layers.append( nn.Conv1d(in_channels, hidden_dim, kernel_size=3, padding=1) ) in_channels = hidden_dim self.bn_layers = nn.ModuleList([nn.BatchNorm1d(hidden_dim) for _ in range(num_layers)]) # 输出层，将时序特征池化并映射到 latent_dim self.pool = nn.AdaptiveAvgPool1d(1) # 全局平均池化，得到 [B, hidden_dim, 1] self.fc_mu = nn.Linear(hidden_dim, latent_dim) self.fc_logvar = nn.Linear(hidden_dim, latent_dim) # 用于 VAE 的变分部分，此处简化，VQ-VAE 通常不需要 def forward(self, x): # x: [B, T, J*C] -> 需要转换为 [B, J*C, T] 以适应 Conv1d if x.dim() == 3: x = x.transpose(1, 2) # [B, J*C, T] for conv, bn in zip(self.conv_layers, self.bn_layers): x = F.relu(bn(conv(x))) x = self.pool(x).squeeze(-1) # [B, hidden_dim] z = self.fc_mu(x) # [B, latent_dim] # 在 VQ-VAE 中，我们直接返回 z 作为编码器输出，不进行重参数化采样 return z # file: models/decoder.py class MotionDecoder(nn.Module): """从量化后的运动特征向量重建关键点序列。""" def __init__(self, latent_dim, hidden_dim, output_dim, seq_length, num_layers=2): super(MotionDecoder, self).__init__() self.latent_dim = latent_dim self.seq_length = seq_length self.output_dim = output_dim # 将 latent 向量扩展为初始的时序特征 self.fc_expand = nn.Linear(latent_dim, hidden_dim * seq_length) self.conv_layers = nn.ModuleList() in_channels = hidden_dim for i in range(num_layers): self.conv_layers.append( nn.Conv1d(in_channels, hidden_dim, kernel_size=3, padding=1) ) self.bn_layers = nn.ModuleList([nn.BatchNorm1d(hidden_dim) for _ in range(num_layers)]) # 输出层，映射回关键点维度 self.fc_out = nn.Conv1d(hidden_dim, output_dim, kernel_size=1) def forward(self, z): # z: [B, latent_dim] batch_size = z.shape[0] x = self.fc_expand(z).view(batch_size, -1, self.seq_length) # [B, hidden_dim, T] for conv, bn in zip(self.conv_layers, self.bn_layers): x = F.relu(bn(conv(x))) x = self.fc_out(x) # [B, output_dim, T] x = x.transpose(1, 2) # [B, T, output_dim] return x

4.3 整合 VQ-VAE 模型

将编码器、量化器、解码器组合起来。

# file: models/vqvae.py import torch.nn as nn from .encoder import MotionEncoder from .decoder import MotionDecoder from .vector_quantizer import VectorQuantizer class VQVAE_Motion(nn.Module): """用于运动表示的 VQ-VAE 模型。""" def __init__(self, input_dim, hidden_dim, latent_dim, num_embeddings, embedding_dim, seq_length, commitment_cost=0.25): super(VQVAE_Motion, self).__init__() self.encoder = MotionEncoder(input_dim, hidden_dim, latent_dim) self.decoder = MotionDecoder(latent_dim, hidden_dim, input_dim, seq_length) # 注意：latent_dim 必须等于 embedding_dim，因为量化器作用于编码器的输出 assert latent_dim == embedding_dim, f"latent_dim({latent_dim}) must equal embedding_dim({embedding_dim}) for quantization." self.quantizer = VectorQuantizer(num_embeddings, embedding_dim, commitment_cost) # 记录参数 self.num_embeddings = num_embeddings self.seq_length = seq_length def forward(self, x): # 1. 编码 z_e = self.encoder(x) # [B, latent_dim] # 为适应量化器，增加一个虚拟的“空间”维度 [B, D, 1] z_e_reshaped = z_e.unsqueeze(-1) # [B, D, 1] # 2. 向量量化 z_q, vq_loss, perplexity = self.quantizer(z_e_reshaped) # z_q: [B, D, 1] z_q = z_q.squeeze(-1) # [B, D] # 3. 解码重建 x_recon = self.decoder(z_q) # [B, T, input_dim] return x_recon, vq_loss, perplexity, z_e, z_q def encode_to_indices(self, x): """用于推理：将输入编码并返回最接近的码本索引。""" with torch.no_grad(): z_e = self.encoder(x).unsqueeze(-1) # [B, D, 1] # 这里需要复制量化器内部的最近邻搜索逻辑 flat_input = z_e.contiguous().view(-1, self.quantizer.embedding_dim) distances = (torch.sum(flat_input**2, dim=1, keepdim=True) + torch.sum(self.quantizer.embedding.weight**2, dim=1) - 2 * torch.matmul(flat_input, self.quantizer.embedding.weight.t())) encoding_indices = torch.argmin(distances, dim=1) # [B] return encoding_indices def calculate_anomaly_score(self, x): """计算异常分数：基于量化误差。""" with torch.no_grad(): z_e = self.encoder(x).unsqueeze(-1) # 计算与最近码本向量的距离 flat_input = z_e.contiguous().view(-1, self.quantizer.embedding_dim) distances = (torch.sum(flat_input**2, dim=1, keepdim=True) + torch.sum(self.quantizer.embedding.weight**2, dim=1) - 2 * torch.matmul(flat_input, self.quantizer.embedding.weight.t())) min_distances, _ = torch.min(distances, dim=1) # [B] # 异常分数可以是平均距离或最大距离 anomaly_score = min_distances.mean().item() # 标量 return anomaly_score

5. 训练与验证：在模拟数据上运行

由于真实数据集（如 ShanghaiTech, UCF-Crime）获取和处理复杂，我们生成一个简单的模拟数据集来演示训练流程。

5.1 生成模拟数据

我们模拟两种正常的周期性运动（如行走、挥手）和一些随机噪声作为“异常”。

# file: utils/data_simulator.py import numpy as np import torch from torch.utils.data import Dataset, DataLoader def generate_normal_motion(seq_length, num_joints, motion_type='walk'): """生成一段正常运动的关键点序列。""" t = np.linspace(0, 4*np.pi, seq_length) # 假设每个关节有 (x, y) 坐标 coords = np.zeros((seq_length, num_joints, 2)) for j in range(num_joints): if motion_type == 'walk': # 模拟行走：躯干和四肢的简谐运动 coords[:, j, 0] = 0.1 * j * np.sin(t + j*0.5) # x 方向 coords[:, j, 1] = 0.05 * j * np.cos(t + j*0.3) + j*0.1 # y 方向，加上基线 elif motion_type == 'wave': # 模拟挥手：主要是手臂关节的运动 if j < 2: # 假设前两个关节是手臂 coords[:, j, 0] = 0.2 * np.sin(2*t) coords[:, j, 1] = 0.1 * j else: coords[:, j, 1] = j*0.1 return coords.reshape(seq_length, -1) # 展平为 [T, J*2] def generate_abnormal_motion(seq_length, num_joints): """生成一段异常运动（随机剧烈运动）。""" coords = np.random.randn(seq_length, num_joints, 2) * 0.5 # 大方差噪声 # 添加一个突然的位移模拟摔倒或奔跑 mid = seq_length // 2 coords[mid:, :, 1] -= np.linspace(0, 2, seq_length-mid)[:, np.newaxis] return coords.reshape(seq_length, -1) class SimMotionDataset(Dataset): def __init__(self, num_samples=1000, seq_length=16, num_joints=17, normal_ratio=0.8): self.data = [] self.labels = [] # 0: normal, 1: abnormal self.seq_length = seq_length self.num_joints = num_joints input_dim = num_joints * 2 for i in range(num_samples): if np.random.rand() < normal_ratio: # 生成正常样本 motion_type = np.random.choice(['walk', 'wave']) seq = generate_normal_motion(seq_length, num_joints, motion_type) self.data.append(seq) self.labels.append(0) else: # 生成异常样本 seq = generate_abnormal_motion(seq_length, num_joints) self.data.append(seq) self.labels.append(1) self.data = np.array(self.data, dtype=np.float32) # [N, T, J*2] self.labels = np.array(self.labels, dtype=np.int64) def __len__(self): return len(self.data) def __getitem__(self, idx): return torch.from_numpy(self.data[idx]), torch.tensor(self.labels[idx])

5.2 训练脚本

我们只使用正常数据训练 VQ-VAE。

# file: train.py import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader from models.vqvae import VQVAE_Motion from utils.data_simulator import SimMotionDataset import yaml import os from tqdm import tqdm def train(config): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f"Using device: {device}") # 1. 数据加载 # 注意：训练集只包含正常数据 train_dataset = SimMotionDataset(num_samples=800, seq_length=config['seq_length'], num_joints=config['num_joints'], normal_ratio=1.0) # 100% normal train_loader = DataLoader(train_dataset, batch_size=config['batch_size'], shuffle=True) # 2. 模型初始化 input_dim = config['num_joints'] * 2 model = VQVAE_Motion( input_dim=input_dim, hidden_dim=config['hidden_dim'], latent_dim=config['embedding_dim'], # 必须等于 embedding_dim num_embeddings=config['num_embeddings'], embedding_dim=config['embedding_dim'], seq_length=config['seq_length'], commitment_cost=config['commitment_cost'] ).to(device) # 3. 优化器与损失函数 optimizer = optim.Adam(model.parameters(), lr=config['lr']) reconstruction_loss_fn = nn.MSELoss() # 4. 训练循环 model.train() for epoch in range(config['num_epochs']): total_recon_loss = 0 total_vq_loss = 0 total_perplexity = 0 pbar = tqdm(train_loader, desc=f'Epoch {epoch+1}/{config[\"num_epochs\"]}') for batch_idx, (data, _) in enumerate(pbar): # 忽略标签 data = data.to(device) # [B, T, J*2] optimizer.zero_grad() # 前向传播 recon_batch, vq_loss, perplexity, _, _ = model(data) # 计算重建损失 recon_loss = reconstruction_loss_fn(recon_batch, data) # 总损失 loss = recon_loss + vq_loss # 反向传播 loss.backward() optimizer.step() total_recon_loss += recon_loss.item() total_vq_loss += vq_loss.item() total_perplexity += perplexity.item() pbar.set_postfix({ 'recon_loss': recon_loss.item(), 'vq_loss': vq_loss.item(), 'perplexity': perplexity.item() }) avg_recon = total_recon_loss / len(train_loader) avg_vq = total_vq_loss / len(train_loader) avg_perp = total_perplexity / len(train_loader) print(f"Epoch {epoch+1} | Avg Recon Loss: {avg_recon:.4f} | Avg VQ Loss: {avg_vq:.4f} | Avg Perplexity: {avg_perp:.2f}") # 保存模型 if (epoch + 1) % config['save_interval'] == 0: os.makedirs('checkpoints', exist_ok=True) torch.save({ 'epoch': epoch, 'model_state_dict': model.state_dict(), 'optimizer_state_dict': optimizer.state_dict(), 'avg_recon_loss': avg_recon, }, f'checkpoints/vqvae_epoch_{epoch+1}.pt') print(f"Model saved to checkpoints/vqvae_epoch_{epoch+1}.pt") print("Training finished.") if __name__ == '__main__': # 加载配置 config = { 'seq_length': 16, 'num_joints': 17, 'hidden_dim': 128, 'embedding_dim': 64, 'num_embeddings': 512, # 码本大小 'commitment_cost': 0.25, 'batch_size': 32, 'lr': 1e-3, 'num_epochs': 50, 'save_interval': 10 } train(config)

5.3 异常检测推理脚本

训练完成后，我们用包含正常和异常的数据测试模型，计算异常分数。

# file: test.py import torch from torch.utils.data import DataLoader from models.vqvae import VQVAE_Motion from utils.data_simulator import SimMotionDataset import numpy as np from sklearn.metrics import roc_auc_score import matplotlib.pyplot as plt def evaluate(config, model_path): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # 加载测试集（包含正常和异常） test_dataset = SimMotionDataset(num_samples=200, seq_length=config['seq_length'], num_joints=config['num_joints'], normal_ratio=0.5) # 50% normal test_loader = DataLoader(test_dataset, batch_size=1, shuffle=False) # batch_size=1 方便计算每个样本的分数 # 加载模型 input_dim = config['num_joints'] * 2 model = VQVAE_Motion( input_dim=input_dim, hidden_dim=config['hidden_dim'], latent_dim=config['embedding_dim'], num_embeddings=config['num_embeddings'], embedding_dim=config['embedding_dim'], seq_length=config['seq_length'], commitment_cost=config['commitment_cost'] ).to(device) checkpoint = torch.load(model_path, map_location=device) model.load_state_dict(checkpoint['model_state_dict']) model.eval() print(f"Model loaded from {model_path}") # 计算异常分数和真实标签 anomaly_scores = [] true_labels = [] with torch.no_grad(): for data, label in test_loader: data = data.to(device) score = model.calculate_anomaly_score(data) anomaly_scores.append(score) true_labels.append(label.item()) anomaly_scores = np.array(anomaly_scores) true_labels = np.array(true_labels) # 计算 AUC (Area Under ROC Curve) auc = roc_auc_score(true_labels, anomaly_scores) print(f"Test AUC: {auc:.4f}") # 可视化分数分布 normal_scores = anomaly_scores[true_labels == 0] abnormal_scores = anomaly_scores[true_labels == 1] plt.figure(figsize=(10, 6)) plt.hist(normal_scores, bins=30, alpha=0.7, label='Normal', density=True) plt.hist(abnormal_scores, bins=30, alpha=0.7, label='Abnormal', density=True) plt.xlabel('Anomaly Score (Quantization Distance)') plt.ylabel('Density') plt.title(f'Distribution of Anomaly Scores (AUC = {auc:.3f})') plt.legend() plt.grid(True, alpha=0.3) plt.savefig('anomaly_score_dist.png') plt.show() # 打印一些样本的分数 print("\nSample scores (first 10):") for i in range(min(10, len(anomaly_scores))): print(f" Sample {i}: label={true_labels[i]}, score={anomaly_scores[i]:.4f}") if __name__ == '__main__': config = { 'seq_length': 16, 'num_joints': 17, 'hidden_dim': 128, 'embedding_dim': 64, 'num_embeddings': 512, 'commitment_cost': 0.25, } model_path = 'checkpoints/vqvae_epoch_50.pt' # 请替换为实际训练好的模型路径 evaluate(config, model_path)

6. 运行结果与效果分析

运行train.py和test.py后，你期望看到以下输出和结果：

训练过程输出示例：

Using device: cuda Epoch 1/50: 100%|██████████| 25/25 [00:04<00:00, 5.12it/s, recon_loss=0.124, vq_loss=0.011, perplexity=1.5] Epoch 1 | Avg Recon Loss: 0.1456 | Avg VQ Loss: 0.0123 | Avg Perplexity: 1.78 ... Epoch 50/50: 100%|██████████| 25/25 [00:04<00:00, 5.34it/s, recon_loss=0.012, vq_loss=0.002, perplexity=412.5] Epoch 50 | Avg Recon Loss: 0.0112 | Avg VQ Loss: 0.0018 | Avg Perplexity: 420.3 Training finished.

• 重建损失（Recon Loss）下降：表明模型学会了用码本向量较好地重建正常运动。
• VQ 损失下降：表明码本向量和编码器输出相互靠近。
• 困惑度（Perplexity）上升并接近码本大小（512）：这是一个关键指标。它衡量码本向量的使用均匀程度。值越高（最大为码本大小），说明更多的码本向量被有效利用，模型学到了多样化的正常模式。

测试结果示例：

Model loaded from checkpoints/vqvae_epoch_50.pt Test AUC: 0.9427 Sample scores (first 10): Sample 0: label=0, score=0.0084 Sample 1: label=1, score=0.1567 Sample 2: label=0, score=0.0071 Sample 3: label=0, score=0.0092 Sample 4: label=1, score=0.2310 ...

• AUC 值：这是一个重要的评估指标，越接近 1 越好。0.94 表明模型在模拟数据上能很好地区分正常和异常运动。在真实复杂数据集上，AUC 能达到 0.85 以上通常就是不错的结果。
• 异常分数分布：从示例中可以看出，正常样本（label=0）的分数显著低于异常样本（label=1）。生成的anomaly_score_dist.png图表会直观显示两类分数分布的分离情况。

如何判断成功？

1. 训练损失平稳下降，重建损失和 VQ 损失均收敛到一个较低的值。
2. 困惑度稳步上升，最终稳定在一个较高的水平（例如码本大小的 50%-80%），表明码本被充分利用。
3. 在测试集上，正常和异常样本的异常分数分布有显著差异，AUC 值较高。

7. 常见问题与排查思路

在实际项目中应用 VQ-VAD 或其思想时，你可能会遇到以下问题：

8. 最佳实践与工程建议

要将 VQ-VAD 思想应用于真实项目，以下建议至关重要：

1. 特征工程是关键：

   • 姿态估计器的选择：HRNet、ViTPose精度高但速度慢，MoveNet、Lightweight OpenPose更适合实时场景。务必在目标场景（光照、分辨率、遮挡）下评估其稳定性。
   • 特征后处理：对提取的关键点序列进行平滑滤波（如 Savitzky-Golay 滤波器）以消除抖动噪声。考虑使用相对坐标（以臀部或胸腔为原点）来消除全局平移的影响。
   • 融合外观特征：纯运动特征可能无法区分“挥手”和“招手”（轨迹相似但手势不同）。可以并行使用一个轻量级的 CNN 提取人体区域的外观特征，与运动特征拼接后再进行 VQ 编码。

2. 数据准备与增强：

   • 纯正正常数据：这是半监督异常检测的生命线。必须通过人工或强规则过滤掉任何可疑片段。数据增强（如时序缩放、轻微空间变换）可以增加正常模式的多样性。
   • 序列采样：使用滑动窗口生成训练样本。窗口长度应能覆盖一个完整的动作周期（如一个完整的步态周期）。

3. 模型调优策略：

   • 码本大小K：这是一个关键超参数。K太小，模型欠拟合，无法表达丰富的正常模式；K太大，模型可能过拟合，甚至学会重建一些轻微异常。建议从256或512开始，根据验证集的重建误差和困惑度进行调整。
   • 损失函数权重：重建损失和 VQ 损失的平衡很重要。可以尝试给 VQ 损失加上一个随时间衰减的权重，让模型初期更关注学习好的表示，后期更关注量化。

4. 部署与阈值选择：

   • 在线检测：在实时流中，需要维护一个滑动窗口缓冲区。每收到一帧，更新缓冲区，计算最新窗口的异常分数。为了平滑，可以对分数进行移动平均。
   • 阈值确定：异常阈值通常通过在干净的验证集（仅含正常数据）上计算异常分数的分布来确定。例如，将阈值设为验证集分数分布的mean + 3 * std。绝对不要使用包含异常的数据来调阈值，这会严重破坏无监督的前提。
   • 报警策略：避免单帧或短时波动触发报警。可以采用“连续 N 个窗口分数超过阈值”才触发报警的策略。

5. 局限性认知：

   • 新颖性 vs 异常性：VQ-VAD 检测的是“新颖”或“不常见”的模式，但这不完全是语义上的“异常”。例如，在办公室场景训练的模型，可能会将“跳舞”判定为异常，但这未必是安全事件。需要结合业务逻辑进行后处理。
   • 缓慢演变异常：如果异常行为是缓慢发生的（如逐渐摔倒），其短窗口内的运动模式可能与正常行走差异不大，可能导致漏报。需要考虑多尺度分析。

VQ-VAD 提供了一种优雅且强大的框架，将视频异常检测问题转化为正常运动模式的字典学习与匹配问题。通过向量量化，它获得了离散的、可解释的表示，并自然导出了一个基于距离的异常分数。本文的简化实现揭示了其核心机理，但要应用于工业级场景，还需要在特征提取、模型架构、工程流水线上做大量细致的工作。建议读者在理解本示例的基础上，使用真实数据集（如 ShanghaiTech Campus, UBnormal）进行复现和迭代，并持续关注该领域的最新进展，如结合 Transformer 的时序建模、引入记忆模块等，这些都是提升模型性能的重要方向。