HG-TransDFD：超图 Transformer 异常检测疲劳识别新范式

一、论文信息

标题： Rethinking driver fatigue detection as anomaly identification: A hypergraph-transformer approach
期刊： Information Processing & Management (ScienceDirect), 2025
DOI： 10.1016/j.ipm.2025.103494
发表时间： 2025年12月

---

二、疲劳检测的范式转变

2.1 传统分类 vs 异常检测

范式	定义	优势	局限
分类范式	将疲劳分为固定类别	直观、可解释	类别边界模糊
异常检测范式	检测偏离正常的行为	无需疲劳样本、适应个体差异	需定义”正常”基线

2.2 为什么选择异常检测？

疲劳定义的模糊性：

• 不同人的疲劳表现差异大
• 疲劳程度是连续变量，非离散类别
• 收集标注疲劳样本成本高

异常检测优势：

• 仅需”正常驾驶”数据
• 自动适应个体差异
• 无需人工标注疲劳等级

---

三、超图理论基础

3.1 图 vs 超图

graph LR
    subgraph 图
        A1[节点] --- B1[边]
        B1 --- C1[节点]
    end
    
    subgraph 超图
        A2[节点] --- B2[超边]
        C2[节点] --- B2
        D2[节点] --- B2
    end

关键区别：

• 图边：连接 2 个节点
• 超边：连接 ≥2 个节点（建模高阶关系）

3.2 疲劳检测中的超图建模

超边设计：

• 时间超边：连续时间帧的关联
• 空间超边：面部关键点的空间关系
• 语义超边：同一疲劳特征的表现组合

---

四、HG-TransDFD 架构详解

4.1 整体框架

graph TB
    A[视频输入] --> B[关键点提取]
    B --> C[超图构建]
    
    C --> D[超图卷积]
    D --> E[Transformer 编码]
    
    E --> F[正常行为建模]
    F --> G{异常检测}
    
    G -->|偏离阈值| H[疲劳警告]
    G -->|正常范围| I[继续监控]

4.2 超图卷积实现

import torch
import torch.nn as nn
import torch.nn.functional as F

class HypergraphConvolution(nn.Module):
    """
    超图卷积层
    
    公式: X' = D_v^(-1/2) H W D_e^(-1) H^T D_v^(-1/2) X W_0
    """
    
    def __init__(self, in_features, out_features):
        super().__init__()
        
        self.linear = nn.Linear(in_features, out_features)
    
    def forward(self, x, H, W=None):
        """
        Args:
            x: 节点特征, shape=(B, N, D)
            H: 关联矩阵, shape=(B, N, E)
                N: 节点数
                E: 超边数
            W: 超边权重, shape=(B, E), 可选
        
        Returns:
            x': 更新后的节点特征, shape=(B, N, D')
        """
        # 1. 计算度矩阵
        # 节点度 D_v: 每个节点连接的超边数
        D_v = H.sum(dim=2, keepdim=True).clamp(min=1)  # (B, N, 1)
        D_v_inv_sqrt = torch.pow(D_v, -0.5)
        
        # 超边度 D_e: 每条超边包含的节点数
        D_e = H.sum(dim=1, keepdim=True).clamp(min=1)  # (B, 1, E)
        D_e_inv = torch.pow(D_e, -1)
        
        # 2. 归一化
        H_norm = D_v_inv_sqrt * H  # (B, N, E)
        
        if W is not None:
            H_norm = H_norm * W.unsqueeze(1) * D_e_inv  # (B, N, E)
        else:
            H_norm = H_norm * D_e_inv
        
        # 3. 超图卷积: X' = H W D_e^-1 H^T D_v^-1/2 X W_0
        # Step 1: H^T X
        x = torch.bmm(H.transpose(1, 2), x)  # (B, E, N)^T @ (B, N, D) -> (B, E, D)
        
        # Step 2: D_e^-1 H^T X
        x = x * D_e_inv.transpose(1, 2)  # (B, E, D)
        
        # Step 3: H D_v^-1/2 ... 
        x = torch.bmm(H_norm, x)  # (B, N, E) @ (B, E, D) -> (B, N, D)
        
        # Step 4: Linear
        x = self.linear(x)
        
        return x


class HGTransBlock(nn.Module):
    """
    超图-Transformer 联合模块
    """
    
    def __init__(self, num_nodes, embed_dim, num_heads=8, num_hyperedges=32):
        super().__init__()
        
        # 1. 超图卷积
        self.hg_conv = HypergraphConvolution(embed_dim, embed_dim)
        
        # 2. Transformer 编码
        self.transformer = nn.TransformerEncoderLayer(
            d_model=embed_dim,
            nhead=num_heads,
            dim_feedforward=embed_dim * 4,
            dropout=0.1,
            batch_first=True
        )
        
        # 3. 归一化
        self.norm1 = nn.LayerNorm(embed_dim)
        self.norm2 = nn.LayerNorm(embed_dim)
    
    def forward(self, x, H):
        """
        Args:
            x: 节点特征, shape=(B, N, D)
            H: 关联矩阵, shape=(B, N, E)
        
        Returns:
            x': 更新后的特征, shape=(B, N, D)
        """
        # 1. 超图卷积（捕捉高阶关系）
        x_hg = self.hg_conv(x, H)
        
        # 2. Transformer（捕捉全局依赖）
        x_trans = self.transformer(x)
        
        # 3. 融合
        x = self.norm1(x + x_hg + x_trans)
        
        return x


class HGTransDFD(nn.Module):
    """
    HG-TransDFD 疲劳异常检测模型
    """
    
    def __init__(self, num_keypoints=68, embed_dim=256, num_frames=16, num_hyperedges=32):
        super().__init__()
        
        self.num_keypoints = num_keypoints
        self.num_frames = num_frames
        
        # 1. 关键点嵌入
        self.keypoint_embed = nn.Linear(2, embed_dim)  # 2D 坐标
        
        # 2. 时序位置编码
        self.temporal_pos = nn.Parameter(
            torch.randn(1, num_frames, 1, embed_dim)
        )
        
        # 3. HG-Trans 块
        self.blocks = nn.ModuleList([
            HGTransBlock(num_keypoints, embed_dim, num_heads=8)
            for _ in range(4)
        ])
        
        # 4. 正常行为建模（重构）
        self.decoder = nn.Sequential(
            nn.Linear(embed_dim, embed_dim),
            nn.ReLU(),
            nn.Linear(embed_dim, 2)  # 重构 2D 坐标
        )
        
        # 5. 异常检测头
        self.anomaly_head = nn.Sequential(
            nn.Linear(embed_dim, embed_dim // 2),
            nn.ReLU(),
            nn.Linear(embed_dim // 2, 1),
            nn.Sigmoid()
        )
    
    def build_hypergraph(self, keypoint_seq):
        """
        构建 关联矩阵
        
        Args:
            keypoint_seq: 关键点序列, shape=(B, T, N, 2)
        
        Returns:
            H: 关联矩阵, shape=(B, N*T, E)
        """
        B, T, N, _ = keypoint_seq.shape
        
        # 简化：基于空间邻接构建超边
        H = torch.zeros(B, N * T, 32)
        
        # 超边 1-8: 面部区域（眉毛、眼睛、鼻子、嘴巴）
        # 超边 9-16: 时间窗口（连续 2 帧）
        # ...（实现细节略）
        
        # 归一化
        H = H / H.sum(dim=1, keepdim=True).clamp(min=1)
        
        return H
    
    def forward(self, keypoint_seq):
        """
        Args:
            keypoint_seq: 关键点序列, shape=(B, T, N, 2)
        
        Returns:
            anomaly_score: 异常分数, shape=(B, 1)
            reconstructed: 重构的关键点, shape=(B, T, N, 2)
        """
        B, T, N, _ = keypoint_seq.shape
        
        # 1. 关键点嵌入
        x = self.keypoint_embed(keypoint_seq)  # (B, T, N, D)
        
        # 2. 添加位置编码
        x = x + self.temporal_pos
        
        # 3. Flatten
        x = x.view(B, T * N, -1)  # (B, T*N, D)
        
        # 4. 构建超图
        H = self.build_hypergraph(keypoint_seq)
        
        # 5. HG-Trans 编码
        for block in self.blocks:
            x = block(x, H)
        
        # 6. 重构（正常行为建模）
        reconstructed = self.decoder(x)  # (B, T*N, 2)
        reconstructed = reconstructed.view(B, T, N, 2)
        
        # 7. 异常检测（基于重构误差）
        # 高误差 = 异常（疲劳）
        recon_error = (keypoint_seq - reconstructed).pow(2).mean(dim=[2, 3])  # (B, T)
        anomaly_score = recon_error.mean(dim=1)  # (B,)
        
        # 归一化到 [0, 1]
        anomaly_score = torch.sigmoid(anomaly_score)
        
        return anomaly_score, reconstructed


# 测试代码
if __name__ == "__main__":
    model = HGTransDFD(
        num_keypoints=68,
        embed_dim=256,
        num_frames=16,
        num_hyperedges=32
    )
    
    # 模拟输入（68 个关键点，16 帧）
    keypoint_seq = torch.randn(2, 16, 68, 2)
    
    # 前向传播
    anomaly_score, reconstructed = model(keypoint_seq)
    
    print(f"输入形状: {keypoint_seq.shape}")
    print(f"异常分数: {anomaly_score}")
    print(f"重构形状: {reconstructed.shape}")

---

四、训练策略：重构自编码器

4.1 正常行为建模

核心思想：

• 仅用正常驾驶数据训练
• 模型学习重构正常行为
• 疲劳时重构误差增大

def train_hgtransdfd(model, train_loader, num_epochs=50):
    """
    训练 HG-TransDFD
    
    仅使用正常驾驶数据
    """
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
    
    for epoch in range(num_epochs):
        model.train()
        total_loss = 0.0
        
        for batch in train_loader:
            keypoint_seq = batch['keypoints']  # 正常驾驶数据
            
            optimizer.zero_grad()
            
            # 前向传播
            anomaly_score, reconstructed = model(keypoint_seq)
            
            # 重构损失（仅正常数据）
            recon_loss = F.mse_loss(reconstructed, keypoint_seq)
            
            # 反向传播
            recon_loss.backward()
            optimizer.step()
            
            total_loss += recon_loss.item()
        
        avg_loss = total_loss / len(train_loader)
        print(f"Epoch {epoch+1}/{num_epochs}, Loss: {avg_loss:.4f}")
    
    return model


def detect_fatigue(model, keypoint_seq, threshold=0.5):
    """
    疲劳检测
    
    Args:
        model: 训练好的模型
        keypoint_seq: 测试关键点序列
        threshold: 异常阈值
    
    Returns:
        is_fatigue: 是否疲劳
        anomaly_score: 异常分数
    """
    model.eval()
    
    with torch.no_grad():
        anomaly_score, _ = model(keypoint_seq)
    
    is_fatigue = (anomaly_score > threshold).item()
    
    return is_fatigue, anomaly_score.item()

---

五、实验结果

5.1 性能对比

方法	AUC	F1-score	检测延迟
SVM（分类）	0.82	0.78	5.2s
AutoEncoder	0.89	0.85	3.8s
GAN-Anomaly	0.91	0.88	4.5s
HG-TransDFD	0.96	0.94	2.1s

5.2 个体适应性

受试者	固定阈值	自适应阈值
Subject 1	85.3%	94.2%
Subject 2	88.7%	96.1%
Subject 3	82.5%	93.8%

---

六、IMS 集成方案

6.1 在线学习流程

graph LR
    A[新驾驶员] --> B[初始化模型]
    B --> C[正常驾驶数据收集]
    C --> D[在线微调]
    D --> E[个性化阈值]
    
    E --> F[实时检测]
    F --> G{异常分数}
    
    G -->|> 阈值| H[疲劳警告]
    G -->|<= 阈值| I[继续监控]
    
    I --> J[更新基线]
    J --> E

6.2 开发检查清单

数据准备：

• 收集正常驾驶数据（≥30 分钟/人）
• 提取面部关键点（68 点）
• 构建超图关联矩阵

模型训练：

• 训练重构自编码器
• 验证重构误差分布
• 设置异常阈值

在线部署：

• 实现关键点实时提取
• 计算重构误差
• 动态调整阈值

---

七、参考资源

1. 论文原文： https://doi.org/10.1016/j.ipm.2025.103494
2. 超图理论： https://arxiv.org/abs/2005.04853
3. 异常检测综述： https://arxiv.org/abs/1901.03407

---

八、总结

HG-TransDFD 实现0.96 AUC 异常检测疲劳识别，关键创新：

1. 异常检测范式 - 无需疲劳标注数据
2. 超图建模 - 捕捉高阶时空关系
3. 个体适应性 - 在线学习个性化基线

IMS 开发建议：

• 采用重构误差作为异常指标
• 收集驾驶员正常驾驶数据建立基线
• 动态调整阈值适应个体差异

---

本文基于 Information Processing & Management 2025 论文深度解读。