Dapalm | IMS技术前沿

轻量级情感识别 RAFM-SER++——边缘部署的驾驶员情绪监控

论文信息

• 标题： RAFM-SER++: A Lightweight Multimodal Emotion Recognition Framework for Real-Time Behavioral Monitoring in Surveillance Systems
• 作者： Trung Duong 等
• 会议： IEEE AVSS 2026（国际先进视频与信号监控会议）
• arXiv： 2609.07409
• 核心： 非对称残差注意力融合，参数减少 60%，推理 79.60 it/s

核心创新

1. 非对称残差注意力融合 (RAFM)：单向注入语音情感线索到文本表征，非双向交互
2. BYOL 跨模态对齐：无需标签的对比学习目标
3. 注意力引导池化：替代简单均值池化
4. 极致轻量：参数量减少 60%+，推理速度 79.60 it/s

方法详解

1. 设计哲学

方法	融合方式	参数量	推理速度	适用场景
跨模态 Transformer	双向交互	~10M	~20 it/s	云端服务器
MemoCMT	交互+记忆	~8M	~40 it/s	高性能边缘
RAFM-SER++	单向注入	~3M	79.60 it/s	资源受限边缘

2. RAFM 非对称融合

"""
RAFM-SER++: 轻量级多模态情感识别

核心: 非对称残差注意力融合 (RAFM)
- 语音情感 → 文本表征 (单向)
- 不做双向交互，省 60% 参数
- BYOL 跨模态对齐 + 注意力池化

适用: AVSS 监控场景 → IMS 座舱监控
"""

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Dict, Tuple
import numpy as np

class ResidualAttentionFusionModule(nn.Module):
    """
    RAFM: 非对称残差注意力融合
    
    论文核心: 语音情感线索 → 单向注入文本表征
    
    对比:
    - 双向 Transformer: Q,K,V 来自两个模态, O(M²×D) 计算
    - RAFM 单向: 注意力从音频到文本, O(M×N×D) 计算
    
    M=音频序列长度, N=文本序列长度, D=特征维度
    当 M=N=128, D=256: 双向=4M, 单向=2M (省50%)
    """
    def __init__(self, text_dim: int = 768, audio_dim: int = 1024,
                 hidden_dim: int = 256, n_heads: int = 4):
        super().__init__()
        # 维度统一
        self.text_proj = nn.Linear(text_dim, hidden_dim)
        self.audio_proj = nn.Linear(audio_dim, hidden_dim)
        
        # 单向交叉注意力 (音频→文本)
        self.cross_attn = nn.MultiheadAttention(
            hidden_dim, n_heads, 
            kdim=hidden_dim, vdim=hidden_dim,
            batch_first=True
        )
        
        # 残差连接 + LayerNorm
        self.norm1 = nn.LayerNorm(hidden_dim)
        self.norm2 = nn.LayerNorm(hidden_dim)
        
        # FFN
        self.ffn = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim * 2),
            nn.GELU(),
            nn.Linear(hidden_dim * 2, hidden_dim)
        )
        
    def forward(self, text_feat: torch.Tensor,
                audio_feat: torch.Tensor) -> torch.Tensor:
        """
        Args:
            text_feat: 文本特征 (B, T_text, 768)
            audio_feat: 音频特征 (B, T_audio, 1024)
        
        Returns:
            fused: 融合特征 (B, T_text, hidden_dim)
        """
        # 投影
        t = self.text_proj(text_feat)    # (B, T_t, D)
        a = self.audio_proj(audio_feat)  # (B, T_a, D)
        
        # 单向交叉注意力: Q=text, K,V=audio
        attn_out, _ = self.cross_attn(
            query=t,        # 文本查询
            key=a,          # 音频键
            value=a         # 音频值
        )
        
        # 残差 + LayerNorm
        t = self.norm1(t + attn_out)
        
        # FFN
        t = self.norm2(t + self.ffn(t))
        
        return t


class AttentionGuidedPooling(nn.Module):
    """
    注意力引导池化
    
    替代简单均值池化: 用可学习的注意力权重加权
    """
    def __init__(self, hidden_dim: int = 256):
        super().__init__()
        self.attention = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.Tanh(),
            nn.Linear(hidden_dim // 2, 1)
        )
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        Args:
            x: (B, T, D)
        Returns:
            pooled: (B, D)
        """
        # 计算注意力权重
        weights = self.attention(x)       # (B, T, 1)
        weights = F.softmax(weights, dim=1)  # 归一化
        
        # 加权求和
        pooled = (x * weights).sum(dim=1)  # (B, D)
        return pooled


class BYOLCrossModalAlignment(nn.Module):
    """
    BYOL (Bootstrap Your Own Latent) 跨模态对齐
    
    无需负样本的对比学习:
    - 文本和音频表征在同一空间对齐
    - 不需要标签, 可用无标注数据预训练
    """
    def __init__(self, hidden_dim: int = 256):
        super().__init__()
        # 预测器 (BYOL core)
        self.predictor = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim * 2),
            nn.GELU(),
            nn.Linear(hidden_dim * 2, hidden_dim)
        )
        
    def forward(self, text_repr: torch.Tensor,
                audio_repr: torch.Tensor) -> torch.Tensor:
        """
        BYOL loss: ||predict(text) - stop_grad(audio)||²
        
        Args:
            text_repr: (B, D)
            audio_repr: (B, D)
        
        Returns:
            loss: BYOL 对齐损失
        """
        pred = self.predictor(text_repr)
        target = audio_repr.detach()  # stop gradient
        
        loss = F.mse_loss(pred, target)
        return loss


class RAFMSERPlusPlus(nn.Module):
    """
    RAFM-SER++ 完整模型
    
    管道:
    1. 文本编码: 冻结 HuBERT → 投影
    2. 音频编码: 冻结 HuBERT → 投影
    3. RAFM 融合: 音频→文本 单向注入
    4. 注意力池化
    5. 分类 + BYOL 对齐
    """
    def __init__(self, n_classes: int = 7, 
                 text_dim: int = 768, audio_dim: int = 1024,
                 hidden_dim: int = 256):
        super().__init__()
        
        # 冻结特征提取器 (简化)
        self.text_proj = nn.Linear(text_dim, hidden_dim)
        self.audio_proj = nn.Linear(audio_dim, hidden_dim)
        
        # RAFM 融合
        self.rafm = ResidualAttentionFusionModule(
            text_dim=hidden_dim, audio_dim=hidden_dim,
            hidden_dim=hidden_dim
        )
        
        # 注意力池化
        self.pooling = AttentionGuidedPooling(hidden_dim)
        
        # 分类器
        self.classifier = nn.Linear(hidden_dim, n_classes)
        
        # BYOL 对齐
        self.byol = BYOLCrossModalAlignment(hidden_dim)
        
        # 参数统计
        self.total_params = sum(p.numel() for p in self.parameters())
        
    def forward(self, text_feat: torch.Tensor,
                audio_feat: torch.Tensor) -> Dict[str, torch.Tensor]:
        """
        Args:
            text_feat: (B, T_text, 768) HuBERT 文本特征
            audio_feat: (B, T_audio, 1024) HuBERT 音频特征
        
        Returns:
            logits: (B, n_classes)
            byol_loss: BYOL 对齐损失
        """
        # 投影
        t = self.text_proj(text_feat)
        a = self.audio_proj(audio_feat)
        
        # RAFM 融合 (音频→文本 单向)
        fused = self.rafm(t, a)
        
        # 注意力池化
        pooled = self.pooling(fused)
        
        # 分类
        logits = self.classifier(pooled)
        
        # BYOL 对齐 (训练时)
        audio_pooled = self.pooling(a.unsqueeze(1).expand(-1, fused.shape[1], -1))
        byol_loss = self.byol(pooled, audio_pooled)
        
        return {
            'logits': logits,
            'byol_loss': byol_loss,
            'fused_repr': pooled
        }


# IMS 驾驶员情绪边缘部署
class EdgeDriverEmotionMonitor:
    """
    IMS 驾驶员情绪边缘监控系统
    
    基于 RAFM-SER++ 轻量级方案
    
    部署目标: QCS8255 Hexagon NPU
    模型大小: <3MB (量化后)
    推理速度: >50fps
    
    监控场景:
    1. 路怒症 (愤怒+高 arousal)
    2. 疲劳情绪 (悲伤+低 arousal)
    3. 焦虑 (恐惧+中 arousal)
    4. 正常 (愉悦/中性)
    """
    def __init__(self):
        self.model = RAFMSERPlusPlus(n_classes=7, hidden_dim=128)
        self.emotions = ['neutral', 'happy', 'sad', 'angry', 
                         'fearful', 'disgust', 'surprised']
        
        # 风险阈值
        self.risk_thresholds = {
            'angry': 0.7,     # 路怒症
            'sad': 0.6,       # 疲劳情绪
            'fearful': 0.6,   # 焦虑
        }
        
    def monitor_frame(self, audio_feat: np.ndarray,
                      text_feat: np.ndarray = None) -> dict:
        """
        单帧情绪监控
        
        Args:
            audio_feat: 语音特征 (1, T, 1024)
            text_feat: 文本特征 (可选, 转录)
        
        Returns:
            emotion_result: {
                'emotion': str,
                'confidence': float,
                'risk_level': int,
                'action': str
            }
        """
        if text_feat is None:
            # 无文本时用零向量
            text_feat = np.zeros((1, 1, 768), dtype=np.float32)
        
        # 推理
        with torch.no_grad():
            output = self.model(
                torch.from_numpy(text_feat).float(),
                torch.from_numpy(audio_feat).float()
            )
        
        probs = F.softmax(output['logits'], dim=-1)[0]
        
        # 主情绪
        dominant_idx = probs.argmax().item()
        dominant_emo = self.emotions[dominant_idx]
        confidence = probs[dominant_idx].item()
        
        # 风险评估
        risk_level = 0
        action = '正常'
        if dominant_emo in self.risk_thresholds:
            if confidence > self.risk_thresholds[dominant_emo]:
                risk_level = 3 if dominant_emo == 'angry' else 2
                actions = {
                    'angry': '一级警告：路怒症风险',
                    'sad': '二级提醒：疲劳情绪',
                    'fearful': '二级提醒：焦虑状态',
                }
                action = actions[dominant_emo]
        
        return {
            'emotion': dominant_emo,
            'confidence': confidence,
            'risk_level': risk_level,
            'action': action,
            'all_emotions': {e: p.item() for e, p in 
                           zip(self.emotions, probs)}
        }


# 测试
if __name__ == "__main__":
    print("=== RAFM-SER++ 模型测试 ===")
    model = RAFMSERPlusPlus(n_classes=7, hidden_dim=256)
    
    B, T_text, T_audio = 4, 10, 20
    text_feat = torch.randn(B, T_text, 768)
    audio_feat = torch.randn(B, T_audio, 1024)
    
    output = model(text_feat, audio_feat)
    print(f"输入: text={text_feat.shape}, audio={audio_feat.shape}")
    print(f"输出: logits={output['logits'].shape}")
    print(f"      byol_loss={output['byol_loss'].item():.4f}")
    
    # 参数量统计
    total = sum(p.numel() for p in model.parameters())
    trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
    print(f"\n参数量: {total:,} (可训练: {trainable:,})")
    print(f"模型大小估计: {total * 4 / 1024 / 1024:.2f} MB (FP32)")
    print(f"量化后估计: {total * 4 / 1024 / 1024 / 4:.2f} MB (INT8)")
    
    # 边缘部署监控
    monitor = EdgeDriverEmotionMonitor()
    
    # 模拟不同情绪场景
    scenarios = {
        '路怒症': np.random.randn(1, 20, 1024) * 2 + 0.5,
        '疲劳': np.random.randn(1, 20, 1024) * 0.5,
        '正常': np.random.randn(1, 20, 1024),
        '焦虑': np.random.randn(1, 20, 1024) * 1.5 + 0.3,
    }
    
    print(f"\n=== 驾驶员情绪监控测试 ===")
    for name, audio in scenarios.items():
        result = monitor.monitor_frame(audio)
        print(f"\n{name}:")
        print(f"  主情绪: {result['emotion']} ({result['confidence']:.1%})")
        print(f"  风险: {result['risk_level']}")
        print(f"  动作: {result['action']}")
    
    # 性能对比
    print(f"\n=== 性能对比 ===")
    print(f"{'模型':<20} {'参数量':<12} {'速度':<12} {'IEMOCAP':<10} {'ESD'}")
    print(f"{'HuBERT baseline':<20} {'~95M':<12} {'~30 it/s':<12} {'76.5%':<10} {'92.0%'}")
    print(f"{'MemoCMT':<20} {'~8M':<12} {'~40 it/s':<12} {'80.2%':<10} {'94.5%'}")
    print(f"{'RAFM-SER++':<20} {'~3M':<12} {'79.60 it/s':<12} {'81.1%':<10} {'95.4%'}")
    print(f"\n→ 参数减少 60%+, 速度 2x, 精度更高")

3. RAFM 单向融合原理

graph LR
    A[音频特征] -->|K,V| B[交叉注意力]
    T[文本特征] -->|Q| B
    B --> C[残差+LayerNorm]
    C --> D[FFN]
    D --> E[融合表征]
    
    style A fill:#f9f,stroke:#333
    style T fill:#bbf,stroke:#333
    style B fill:#fbb,stroke:#333

4. 性能指标

指标	HuBERT baseline	MemoCMT	RAFM-SER++
参数量	~95M	~8M	~3M (-60%)
推理速度	~30 it/s	~40 it/s	79.60 it/s
IEMOCAP BACC	76.5%	80.2%	81.10%
ESD BACC	92.0%	94.5%	95.39%
模型大小 (INT8)	~24MB	~2MB	~0.75MB

IMS 边缘部署方案

1. QCS8255 部署配置

组件	配置	用途
HuBERT-Base (冻结)	95M 参数, 量化 INT8	特征提取 (文本+音频)
RAFM 融合层	3M 参数, FP16	可训练部分
分类头	0.1M 参数	7 类情绪分类
总内存	~24MB (HuBERT) + 6MB (RAFM)	~30MB
推理速度	79.60 it/s → ~12ms/样本	>50fps 实时

2. 与 ECCV 2026 方案对比

维度	ECCV 2026 (SDT+)	AVSS 2026 (RAFM-SER++)
目标	精度最大化	速度+精度平衡
参数量	~50M	~3M
模态	文本+音频+视觉	文本+音频
融合	双向 Transformer	单向 RAFM
速度	~20 it/s	79.60 it/s
适用	高端车型	入门车型

3. 两级情绪检测架构

层级	模型	硬件	延迟	精度
L1 实时	RAFM-SER++	QCS8255 NPU	12ms	81% BACC
L2 精细	SDT+ (ECCV)	云端/SoC CPU	50ms	90%+ BACC

策略： L1 实时筛选高风险情绪 → L2 精细分类确认

开发启示

1. 单向融合是边缘部署的关键：RAFM 省去双向交互，参数减 60%，精度反而更高
2. BYOL 对齐利用无标注数据：座舱音频/文本可用无标签数据预训练对齐
3. 注意力池化优于均值池化：对情感关键词帧/音段加权
4. 冻结 HuBERT + 轻量头：95M 冻结 + 3M 可训练，微调成本极低
5. 79.60 it/s 满足实时需求：QCS8255 量化后仍可达 >50fps
6. 与 ECCV 方案互补：RAFM 实时筛选 + SDT 精细确认，两级架构
7. 音频是情绪检测的主力模态：语音语调比面部更早反映情绪变化
8. 路怒症检测的首选方案：愤怒情绪在音频中特征最明显

测试场景

EM-02 边缘情绪检测延迟

前置条件：

• QCS8255 已加载 RAFM-SER++ (INT8 量化)
• 麦克风阵列正常工作

测试步骤：

1. 正常驾驶 30s（基线）
2. 播放预录愤怒语音 10s
3. 记录检测延迟和准确率

判定条件：

检测项	通过条件
推理延迟	≤ 20ms
愤怒检测率	≥ 75%
误报率	≤ 10%
恢复延迟	≤ 5s

总结

RAFM-SER++ 为 IMS 边缘情绪检测提供了最优的效率-精度平衡方案：

1. 参数减 60%（8M→3M），推理速度 2x（40→79.60 it/s）
2. 单向 RAFM 融合是轻量化的核心：音频→文本 单向注入
3. BYOL 对齐利用无标注数据，适合座舱数据稀缺场景
4. INT8 量化后 ~0.75MB，可在 QCS8255 上 >50fps 实时运行
5. 与 ECCV SDT+ 互补：RAFM 实时筛选 + SDT 精细确认
6. 音频是情绪检测主力：语调变化比面部更早反映情绪
7. 路怒症检测首选：愤怒在音频中特征最明显