SemARC：自适应多模态推理——座舱传感器融合的效率革命

论文信息

项目	内容
标题	Efficient Multimodal Inference through Adaptive Acquisition and Sequential Fusion
arXiv	2610.07466
时间	2026年10月5日
作者	Payal Mohapatra 等
领域	cs.LG
链接	https://arxiv.org/abs/2610.07466

核心创新

提出SemARC框架——顺序模态聚合器(SeMA) + 自适应运行时控制器(ARC)。不一次性编码所有模态，而是根据已有证据决定下一个要编码什么模态、何时停止。

一句话总结： 多传感器DMS不需要每帧都跑全部传感器——先跑最便宜的，结果够好就停，不够再跑下一个。

为什么这对座舱DMS重要

# 当前DMS多模态融合的问题
problem = {
    "现状": "每帧编码所有传感器（摄像头+雷达+rPPG+方向盘ECG）",
    "浪费": "90%时间驾驶员状态正常，不需要全部传感器",
    "延迟": "多编码器串行处理导致高延迟",
    "功耗": "持续运行所有传感器功耗高",
    "设备差异": "高端车型4传感器，低端2传感器——模型需适配"
}

# SemARC解决方案
solution = {
    "核心思想": "自适应获取——先跑便宜的模态，够好就停",
    "效果": "macro-F1 +3.2%，GFLOPs -61.4%，延迟 -44.0%",
    "额外收益": "运行时模态缺失时仍能工作（21/24条件最优）"
}

方法详解

1. SemARC架构

import torch
import torch.nn as nn
import numpy as np
from typing import List, Dict

class SeMA(nn.Module):
    """
    Sequential Modality Aggregator (SeMA)
    
    顺序模态聚合器：
    1. 按需选择模态编码
    2. 只运行选中的编码器和融合分支
    3. 固定大小状态更新
    4. 每次获取后即可预测
    """
    
    def __init__(self, modalities: List[str], 
                 d_model: int = 256,
                 n_classes: int = 5):
        super().__init__()
        self.modalities = modalities
        
        # 每个模态的编码器
        self.encoders = nn.ModuleDict({
            mod: self._build_encoder(mod, d_model)
            for mod in modalities
        })
        
        # 融合状态（固定大小）
        self.fusion_state = nn.GRUCell(d_model, d_model)
        
        # 预测头
        self.classifier = nn.Sequential(
            nn.Linear(d_model, d_model // 2),
            nn.GELU(),
            nn.Linear(d_model // 2, n_classes)
        )
    
    def _build_encoder(self, mod: str, d_model: int) -> nn.Module:
        if mod == "camera":
            return nn.Sequential(
                nn.Conv2d(3, 32, 3, 2, 1), nn.GELU(),
                nn.Conv2d(32, 64, 3, 2, 1), nn.GELU(),
                nn.AdaptiveAvgPool2d(1), nn.Flatten(),
                nn.Linear(64, d_model)
            )
        elif mod == "radar":
            return nn.Sequential(
                nn.Linear(50, 128), nn.GELU(),
                nn.Linear(128, d_model)
            )
        elif mod == "rppg":
            return nn.Sequential(
                nn.Linear(10, 64), nn.GELU(),
                nn.Linear(64, d_model)
            )
        elif mod == "steering":
            return nn.Sequential(
                nn.Linear(5, 32), nn.GELU(),
                nn.Linear(32, d_model)
            )
        return nn.Linear(10, d_model)
    
    def forward(self, inputs: Dict[str, torch.Tensor],
                acquisition_order: List[str]) -> List[torch.Tensor]:
        """
        Args:
            inputs: 各模态输入数据
            acquisition_order: 模态获取顺序
        
        Returns:
            predictions: 每次获取后的预测列表
        """
        state = torch.zeros(1, 256, device=next(self.parameters()).device)
        predictions = []
        
        for mod in acquisition_order:
            if mod not in inputs or inputs[mod] is None:
                continue
            
            # 只运行选中的编码器
            x = inputs[mod]
            if mod == "camera":
                feat = self.encoders[mod](x)
            else:
                feat = self.encoders[mod](x)
            
            # 更新融合状态（固定大小，不重新计算之前的）
            state = self.fusion_state(feat.unsqueeze(0), state)
            
            # 每次获取后即可预测
            pred = self.classifier(state)
            predictions.append(pred)
        
        return predictions


class ARC(nn.Module):
    """
    Adaptive Runtime Controller (ARC)
    
    自适应运行时控制器：
    1. 基于当前状态选择下一个模态
    2. 或决定停止（已有证据足够）
    3. 结合边际效用先验 + 残差Q学习
    """
    
    def __init__(self, modalities: List[str], d_model: int = 256):
        super().__init__()
        self.modalities = modalities
        
        # 边际效用先验（每个模态的期望贡献）
        self.marginal_prior = nn.Parameter(torch.ones(len(modalities)))
        
        # Q值估计（残差网络）
        self.q_net = nn.Sequential(
            nn.Linear(d_model + len(modalities), 128),
            nn.GELU(),
            nn.Linear(128, len(modalities) + 1)  # +1 for "stop"
        )
    
    def forward(self, state: torch.Tensor, 
                acquired: List[str]) -> Dict:
        """
        决定下一个模态或停止
        
        Returns:
            action: {"type": "acquire"/"stop", "modality": str}
        """
        # 已获取模态的mask
        acquired_mask = torch.zeros(len(self.modalities))
        for i, mod in enumerate(self.modalities):
            if mod in acquired:
                acquired_mask[i] = 1
        
        # Q值 = 边际先验 + 残差Q
        prior = self.marginal_prior * (1 - acquired_mask)
        q_input = torch.cat([state.squeeze(0), acquired_mask])
        q_values = self.q_net(q_input)
        
        # 结合先验和Q值
        combined = prior + q_values[:-1]  # 模态Q值
        stop_q = q_values[-1]  # 停止Q值
        
        best_mod_idx = combined.argmax().item()
        best_mod_q = combined[best_mod_idx].item()
        
        if stop_q > best_mod_q:
            return {"type": "stop", "reason": "evidence_sufficient"}
        else:
            return {
                "type": "acquire",
                "modality": self.modalities[best_mod_idx],
                "expected_gain": best_mod_q - stop_q
            }


# 完整SemARC系统
class SemARC(nn.Module):
    """完整的SemARC系统"""
    
    def __init__(self, modalities: List[str]):
        super().__init__()
        self.sema = SeMA(modalities)
        self.arc = ARC(modalities)
        self.modalities = modalities
    
    def forward(self, inputs: Dict[str, torch.Tensor]) -> Dict:
        """
        自适应推理流程
        
        1. ARC选择第一个模态
        2. SeMA编码+融合
        3. ARC选择第二个或停止
        4. 重复直到停止
        """
        acquired = []
        state = torch.zeros(1, 256)
        final_pred = None
        total_gflops = 0
        
        # 简化：按固定顺序+提前停止
        order = ["steering", "rppg", "camera", "radar"]
        # 实际ARC会动态选择
        
        for mod in order:
            if mod not in inputs:
                continue
            
            acquired.append(mod)
            
            # 模拟编码
            if mod == "steering":
                total_gflops += 0.01
            elif mod == "rppg":
                total_gflops += 0.05
            elif mod == "camera":
                total_gflops += 2.5
            elif mod == "radar":
                total_gflops += 1.0
            
            # ARC决策：继续还是停？
            decision = self.arc(state, acquired)
            
            if decision["type"] == "stop":
                break
        
        return {
            "acquired_modalities": acquired,
            "total_gflops": total_gflops,
            "stopped_early": len(acquired) < len(self.modalities),
            "savings_vs_full": 1 - total_gflops / 3.56  # 全跑=3.56 GFLOPs
        }


# 测试
if __name__ == "__main__":
    system = SemARC(["camera", "radar", "rppg", "steering"])
    
    # 场景1：正常驾驶（简单）
    inputs_normal = {
        "steering": torch.randn(1, 5),
        "rppg": torch.randn(1, 10),
        "camera": torch.randn(1, 3, 224, 224),
        "radar": torch.randn(1, 50),
    }
    result = system(inputs_normal)
    print("正常驾驶:")
    print(f"  获取模态: {result['acquired_modalities']}")
    print(f"  GFLOPs: {result['total_gflops']:.2f}")
    print(f"  节省: {result['savings_vs_full']*100:.1f}%")
    print(f"  提前停止: {result['stopped_early']}")

2. 实验结果

指标	基线	SemARC	改善
macro-F1	各数据集最优	+3.2%	✅ 更准
总GFLOPs	全量	-61.4%	✅ 更快
GPU延迟	基线	-44.0%	✅ 更快
CPU延迟	基线	-44.0%	✅ 更快
Android INT8	基线最快	-47.2%	✅ 移动端
模态缺失	最优基线	21/24条件最优	✅ 鲁棒

IMS开发启示

1. DMS传感器调度策略

2. 功耗优化估算

策略	平均GFLOPs/帧	功耗(W)	日均成本
全传感器每帧	3.56	~3.5W	100%
SemARC自适应	~1.4 (61%省)	~1.4W	40%
日均节省	-	-2.1W	-60%

3. 不同车型的传感器配置适配

# SemARC天然支持不同传感器配置
vehicle_configs = {
    "高端车型(4传感器)": {
        "sensors": ["camera", "radar", "rppg", "steering"],
        "strategy": "steering→rppg→camera(按需)→radar(罕见)",
        "avg_gflops": 0.5,  # 90%时间只跑前两个
        "accuracy": "最高"
    },
    "中端车型(2传感器)": {
        "sensors": ["camera", "steering"],
        "strategy": "steering→camera(按需)",
        "avg_gflops": 0.3,
        "accuracy": "高"
    },
    "入门车型(1传感器)": {
        "sensors": ["camera"],
        "strategy": "camera always on",
        "avg_gflops": 2.5,
        "accuracy": "中"
    }
}

4. Euro NCAP功能适配

Euro NCAP功能	最少传感器	按需激活	全传感器
疲劳检测	摄像头	+rPPG	+方向盘+雷达
分心检测	摄像头	-	-
CPD	雷达	+摄像头	-
OOP	摄像头	+雷达	+织物传感
安全带误用	摄像头	-	-

测试场景

SA-01 自适应传感器调度测试

前置条件：

• 4传感器系统（摄像头+雷达+rPPG+方向盘）
• SemARC部署于QCS8255

测试步骤：

1. 正常驾驶10分钟 → 记录激活传感器数和GFLOPs
2. 模拟疲劳驾驶10分钟 → 记录激活传感器数
3. 模拟分心驾驶10分钟 → 记录激活传感器数

判定条件：

场景	平均GFLOPs	激活传感器	准确率
正常	<1.0	1-2个	≥95%
疲劳	<2.5	2-3个	≥90%
分心	<3.0	2-3个	≥90%
高风险	≤3.56	全部	≥95%

结论

SemARC为DMS多传感器融合提供了效率范式转移：

1. 自适应获取 — 不每帧跑全部传感器，按需激活
2. 顺序融合 — 先便宜后贵，够好就停
3. 模态缺失鲁棒 — 21/24条件最优，天然适配不同车型
4. 移动端友好 — Android INT8延迟降47%
5. 零精度损失 — 反而+3.2% macro-F1（减少噪声模态干扰）

对IMS的启示： 从”全传感器每帧融合”转向”自适应按需融合”，是DMS边缘部署的效率突破口。

---

论文来源：arXiv:2610.07466 (2026-10-05)