1. 背景与问题定义1.1 为什么需要 ECG 与 PPG 融合在可穿戴设备和无感监测场景中心电图ECG和光电容积脉搏波PPG是两种最常见的生理信号。ECG 直接反映心脏电活动能够提供准确的 RR 间期、心率变异性HRV等信息PPG 通过光学方式测量外周血容量变化能够间接估计心率、血氧饱和度和脉搏特征。单一模态在实际部署中会遇到明显瓶颈。PPG 对运动伪影非常敏感手部摆动、环境光变化、传感器贴合松动都会导致波形严重失真ECG 虽然形态稳定但在干电极、运动出汗、贴片脱落等场景下同样会出现基线漂移、电极接触噪声和单导联信号丢失。如果系统只在单模态下做预测一旦信号质量下降心率估计、心律分析等关键指标就会快速恶化。CardioFusion-AI 的出发点很直接将 ECG 和 PPG 看作互补的信息源通过融合策略让两路信号在各自受损时互相补偿。比如 PPG 出现运动伪影时ECG 如果仍然可靠系统可以自动降低 PPG 分支的权重反过来ECG 电极脱落时PPG 仍能支撑基本的脉率估计。这种“质量感知”的多模态融合比简单的平均值融合或直接拼接更鲁棒。1.2 信号退化问题具体指什么信号退化不是一个抽象概念在生理信号处理中退化通常表现为以下几类运动伪影传感器与皮肤相对运动带来的低频大幅扰动。基线漂移呼吸、电极移动或放大器温漂导致的缓慢基线变化。高频噪声工频干扰、肌肉电噪声、光学环境光干扰。局部缺失电极脱落、蓝牙丢包、传感器断连导致的数据段缺失。幅值变化不同佩戴位置、肤色、组织厚度带来的幅值差异。模型的鲁棒性就是在这些退化情况下依然保持稳定性能的能力。评估时不能只看正常信号上的精度更要在“信号退化程度由低到高”的谱系上观察性能衰减曲线。1.3 CardioFusion-AI 的核心思想CardioFusion-AI 可以理解为“双分支特征提取 质量门控融合 缺失模态训练”的完整方案。双分支特征提取ECG 和 PPG 分别经过独立编码器提取时序特征不对单模态做过度耦合。质量门控融合每条分支生成一个质量评分融合时根据质量评分动态调整两个分支的贡献程度。缺失模态训练训练时随机屏蔽一个模态让模型学会在部分输入缺失的情况下仍然给出可用预测。这个方案不依赖特定的输入长度或特定传感器因此可以迁移到不同的可穿戴设备平台。2. CardioFusion-AI 整体方案设计2.1 问题建模把问题形式化为一个序列回归任务。输入为两段同步采样信号ecg_signal: 长度 T 的 ECG 波形片段ppg_signal: 长度 T 的 PPG 波形片段目标是预测一组生理指标。为了便于实验本文以“心率估计”作为主任务也就是预测每段信号对应的平均心率。算法框架同样可以扩展到 HRV、呼吸率等指标。补充说明工程中信号可能来自不同采样率比如 ECG 为 250 HzPPG 为 125 Hz需要先重采样到统一频率。本文代码示例假定输入已经预处理为 100 Hz 的等长片段长度 T 1000对应 10 秒窗口。2.2 系统架构整个流程可以分为四个阶段信号预处理与切窗。信号质量评估。双分支特征提取。质量门控融合与预测输出。在训练阶段额外加入噪声增强和随机模态屏蔽模拟信号退化。在部署阶段质量评估结果还可以作为异常告警依据。2.3 关键模块设计模块输入输出职责预处理模块原始 ECG、PPG标准化片段滤波、重采样、归一化质量评估模块单模态片段质量分数估计信号可信度ECG 分支ECG 片段特征向量提取心电特征PPG 分支PPG 片段特征向量提取脉搏特征融合模块两个特征 两个质量分预测结果动态加权整合值得注意的是质量评估模块既可以独立计算也可以由特征编码器预测一个“隐式质量分数”。本文采用“显式 隐式结合”显式质量分来自信号质量指标隐式质量分来自编码器末尾的全连接层。最终融合权重由两者共同决定。3. 环境准备与数据说明3.1 运行环境本文代码以 Python 3.9 和 PyTorch 1.13 为基础。你在运行时可按自己的环境调整版本差异通常不影响整体思路。依赖库建议如下numpy用于数值计算。scipy用于信号滤波和峰值检测。torch用于模型训练。pandas和matplotlib用于数据读取和可视化。安装命令pip install numpy scipy torch pandas matplotlib如果使用 GPU需要确保 CUDA 和 PyTorch 版本匹配。无法使用 GPU 时把代码中的.cuda()相关部分去掉模型也可以在小数据集上跑 CPU 版本只是速度会慢一些。3.2 数据集准备真实生理数据集需要经过伦理审批和脱敏处理。在教程演示中建议用两种方式构建数据公开数据集中截取片段例如多模态生理信号公开数据库中的同步 ECG 与 PPG 信号。用仿真波形生成程序构造带标注的合成数据方便快速验证模型逻辑。本文为了演示训练流程使用以下模拟方案生成一段基础心率信号再叠加可控制噪声生成模拟 ECG 和 PPG。真实项目中请替换为实际数据加载代码。模拟数据结构建议data/ ├── raw/ │ ├── ecg_0001.npy │ ├── ppg_0001.npy │ └── label_0001.npy ├── train.csv ├── val.csv └── test.csv其中每个 npy 文件是一个等长波形数组label 是对应的心率值。3.3 信号预处理流程在输入模型之前需要完成以下步骤。重采样到统一采样率。带通滤波ECG 保留 0.5~45 HzPPG 保留 0.5~10 Hz。去除均值并除以标准差做 z-score 标准化。按固定窗口切分例如每段 10 秒。这里给出一个预处理函数示例# src/preprocess.py import numpy as np from scipy import signal as sp_signal def preprocess_signal(signal_data, original_fs, target_fs100, lowcut0.5, highcut40.0): # 1. 重采样 if original_fs ! target_fs: n_samples int(len(signal_data) * target_fs / original_fs) signal_data sp_signal.resample(signal_data, n_samples) # 2. 带通滤波 nyquist 0.5 * target_fs normal_low lowcut / nyquist normal_high highcut / nyquist b, a sp_signal.butter(4, [normal_low, normal_high], btypeband) signal_data sp_signal.filtfilt(b, a, signal_data) # 3. z-score 标准化 mean np.mean(signal_data) std np.std(signal_data) if std 1e-6: std 1e-6 signal_data (signal_data - mean) / std return signal_data.astype(np.float32)这里使用filtfilt做零相位滤波避免普通滤波带来的相位偏移。注意滤波边界需要根据信号实际频带调整不是所有数据都适合相同的截止频率。4. 核心代码实现4.1 信号质量评估模块信号质量评估的核心是提取一个 0~1 之间的分数分数越高表示信号越可信。常见评估思路使用峰值检测计算相邻峰值的间隔变异系数。计算信号与自身模板的相关性。计算高频噪声能量占比。本文实现一个简单的质量分函数基于峰值间隔的稳定性# src/sqi.py import numpy as np from scipy.signal import find_peaks def compute_sqi(signal, fs100, min_hr30, max_hr220): min_dist int(fs * 60.0 / max_hr) max_dist int(fs * 60.0 / min_hr) peaks, _ find_peaks(signal, distancemin_dist) if len(peaks) 3: return 0.1 intervals np.diff(peaks) / fs * 1000.0 # 转成毫秒 intervals intervals[(intervals min_dist / fs * 1000) (intervals max_dist / fs * 1000)] if len(intervals) 2: return 0.2 cv np.std(intervals) / (np.mean(intervals) 1e-8) # 变异系数越小质量越高 sqi float(np.clip(1.0 - cv * 0.5, 0.0, 1.0)) return sqi这个函数只演示思路。实际项目中还需要加入幅度异常检测、模板相关性、信噪比估计等维度。质量分不能只看峰间间隔否则心律失常信号会被误判为低质量。4.2 单模态编码器ECG 和 PPG 分支共用一个编码器结构但参数彼此独立。编码器采用“一维卷积 GRU 全局池化”的组合既能提取局部波形形态又能捕获长程时序依赖。# src/branch_encoder.py import torch import torch.nn as nn class BranchEncoder(nn.Module): def __init__(self, input_size1, hidden_size64, out_dim64): super().__init__() self.conv1 nn.Conv1d(input_size, 32, kernel_size5, stride1, padding2) self.bn1 nn.BatchNorm1d(32) self.conv2 nn.Conv1d(32, 64, kernel_size5, stride1, padding2) self.bn2 nn.BatchNorm1d(64) self.gru nn.GRU(64, hidden_size, batch_firstTrue, bidirectionalTrue) self.fc nn.Linear(hidden_size * 2, out_dim) def forward(self, x): # x: (batch, 1, time_len) x torch.relu(self.bn1(self.conv1(x))) x torch.relu(self.bn2(self.conv2(x))) # 转成 (batch, time_len, channels) x x.permute(0, 2, 1) x, _ self.gru(x) # 全局平均池化 x x.mean(dim1) return self.fc(x)这里输出一个out_dim维的特征向量。双向 GRU 可以同时看到当前时刻前后信息但模型参数会多一倍。若设备计算力有限可以把bidirectional改成False。4.3 质量门控融合模块融合模块需要完成两件事一是判断两个模态各自贡献多少二是把特征向量整合为最终预测。定义隐式质量权重来自两个分支特征# src/fusion.py import torch import torch.nn as nn class QualityGate(nn.Module): def __init__(self, feature_dim64): super().__init__() self.ecg_gate nn.Sequential( nn.Linear(feature_dim, 32), nn.ReLU(), nn.Linear(32, 1), nn.Sigmoid() ) self.ppg_gate nn.Sequential( nn.Linear(feature_dim, 32), nn.ReLU(), nn.Linear(32, 1), nn.Sigmoid() ) def forward(self, ecg_feat, ppg_feat, ecg_sqi, ppg_sqi): # 隐式权重 implicit_ecg self.ecg_gate(ecg_feat) implicit_ppg self.ppg_gate(ppg_feat) # 显式质量分与隐式权重结合 ecg_weight (0.4 * ecg_sqi 0.6 * implicit_ecg.squeeze(-1)).unsqueeze(-1) ppg_weight (0.4 * ppg_sqi 0.6 * implicit_ppg.squeeze(-1)).unsqueeze(-1) # 归一化保证权重之和为 1 total ecg_weight ppg_weight 1e-8 ecg_weight ecg_weight / total ppg_weight ppg_weight / total return ecg_weight, ppg_weight融合最终特征可以简单写成加权求和def fuse(ecg_feat, ppg_feat, ecg_weight, ppg_weight): fused_feature ecg_feat * ecg_weight ppg_feat * ppg_weight return fused_feature当然更复杂的设计可以使用跨模态注意力让 ECG 特征在融合时关注 PPG 特征中的关键时间点。但对于心率回归任务质量门控加权求和已经足够稳定并且更容易部署。4.4 抗退化训练策略模型要在缺失模态和信号退化条件下保持鲁棒训练策略比模型结构更关键。本文采用三种策略随机模态屏蔽以一定概率把某个分支输入置零同时把对应显式质量分置为 0。信号噪声增强对输入叠加高斯噪声、基线漂移或随机缩放。时间片段挖空随机把某一段连续时间置零模拟传感器瞬间掉线。这些增强策略可以用一个模块统一处理。下面给出噪声增强的部分代码# src/augment.py import numpy as np def apply_degradation(ecg, ppg, noise_level0.1, mask_prob0.3): # 随机模态屏蔽 if np.random.rand() mask_prob: if np.random.rand() 0.5: ecg np.zeros_like(ecg) else: ppg np.zeros_like(ppg) # 添加高斯噪声 ecg ecg noise_level * np.random.randn(*ecg.shape) ppg ppg noise_level * np.random.randn(*ppg.shape) # 随机挖空 1 秒 fs 100 if np.random.rand() 0.2: start np.random.randint(0, len(ecg) - fs) ecg[start:start fs] 0.0 ppg[start:start fs] 0.0 return ecg.astype(np.float32), ppg.astype(np.float32)注意在验证集和测试集上不要做这类随机增强否则评估结果不稳定。可以把mask_prob置为 0单独构造“缺失模态”测试集来评估鲁棒性。4.5 组装完整模型有了分支编码器和质量门控可以组装完整的 CardioFusion-AI 模型。# src/cardiofusion.py import torch import torch.nn as nn from src.branch_encoder import BranchEncoder from src.fusion import QualityGate from src.sqi import compute_sqi class CardioFusion(nn.Module): def __init__(self, hidden_size64, out_dim64, num_task1): super().__init__() self.ecg_encoder BranchEncoder(input_size1, hidden_sizehidden_size, out_dimout_dim) self.ppg_encoder BranchEncoder(input_size1, hidden_sizehidden_size, out_dimout_dim) self.gate QualityGate(feature_dimout_dim) self.regressor nn.Sequential( nn.Linear(out_dim, 32), nn.ReLU(), nn.Linear(32, num_task) ) def forward(self, ecg, ppg, ecg_sqi, ppg_sqi): # ecg: (batch, 1, time_len), ppg: (batch, 1, time_len) ecg_feat self.ecg_encoder(ecg) ppg_feat self.ppg_encoder(ppg) ecg_weight, ppg_weight self.gate(ecg_feat, ppg_feat, ecg_sqi, ppg_sqi) # 质量分低时对应特征也会接近 0 fused_feat ecg_feat * ecg_weight ppg_feat * ppg_weight output self.regressor(fused_feat) return output, ecg_weight, ppg_weight在推理时对每一段信号先用compute_sqi计算显式质量分再输入模型。如果某个模态整段缺失可以直接把输入置零同时把质量分设为 0。5. 训练与评估5.1 损失函数主任务使用均方误差MSE作为监督损失。为了让质量门控产生有意义的权重可以额外增加一个辅助损失用于预测每个模态的真实质量。但实际标注中很难获得每个模态精确的真实质量分所以更常用的做法是主损失只负责最终预测。通过训练数据中的随机屏蔽迫使模型自动学会降低缺失模态的贡献。在验证时观察不同缺失场景下的性能。如果需要对权重做更严格约束也可以加入权重熵正则项避免模型总是偏向某一个模态reg_loss -0.5 * (ecg_weight * torch.log(ecg_weight 1e-8) ppg_weight * torch.log(ppg_weight 1e-8)).mean()这个正则项不是必须的。如果发现融合结果退化成“只用 ECG”或“只用 PPG”再考虑加入。5.2 训练循环下面给出一个简化训练循环。代码省略了数据加载细节核心是演示训练逻辑。import torch import torch.nn as nn def train_one_epoch(model, dataloader, optimizer, device): model.train() total_loss 0.0 criterion nn.MSELoss() for ecg, ppg, label in dataloader: ecg ecg.unsqueeze(1).to(device) ppg ppg.unsqueeze(1).to(device) label label.unsqueeze(1).to(device) # 计算显式质量分 ecg_sqi torch.tensor([compute_sqi(x.cpu().numpy().reshape(-1)) for x in ecg.squeeze(1)]).to(device).unsqueeze(-1) ppg_sqi torch.tensor([compute_sqi(x.cpu().numpy().reshape(-1)) for x in ppg.squeeze(1)]).to(device).unsqueeze(-1) pred, _, _ model(ecg, ppg, ecg_sqi, ppg_sqi) loss criterion(pred, label) optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() return total_loss / max(len(dataloader), 1)这段代码每次循环都用compute_sqi计算质量分效率偏低。工程实现中建议把质量分作为数据预处理结果保存到缓存里训练时直接读取。5.3 评估指标与鲁棒性评估心率回归任务常用指标平均绝对误差MAE。均方根误差RMSE。决定系数R²。除了整体指标还要设计压力测试评估鲁棒性场景说明正常信号ECG 和 PPG 均正常ECG 轻度噪声ECG 叠加高斯噪声PPG 运动伪影PPG 叠加低频基线漂移ECG 缺失ECG 输入全部置零PPG 缺失PPG 输入全部置零双模态噪声两路信号均退化每一类场景单独计算 MAE就能直观看出融合模型在信号退化下是否稳定。比如下面这段简化评估代码def evaluate_by_scenario(model, dataset, device): model.eval() metrics {} for scenario_name, data in dataset.scenarios.items(): preds [] labels [] with torch.no_grad(): for ecg, ppg, label in data: ecg ecg.unsqueeze(0).unsqueeze(1).to(device) ppg ppg.unsqueeze(0).unsqueeze(1).to(device) ecg_sqi torch.tensor([[compute_sqi(ecg.cpu().numpy().squeeze())]]).to(device) ppg_sqi torch.tensor([[compute_sqi(ppg.cpu().numpy().squeeze())]]).to(device) pred, _, _ model(ecg, ppg, ecg_sqi, ppg_sqi) preds.append(pred.item()) labels.append(label.item()) mae np.mean(np.abs(np.array(preds) - np.array(labels))) metrics[scenario_name] mae return metrics评估结果通常会出现一种现象正常信号下融合模型不一定优于表现最好的单模态模型但在单模态缺失或退化场景下明显更鲁棒。这正是融合系统的价值所在。6. 常见问题与排查思路6.1 常见问题表格问题现象常见原因排查与解决思路训练 loss 变成 NaN学习率过大、信号标准化后仍有极大值降低学习率使用梯度裁剪检查输入是否有 inf融合模型不如单模态融合权重没有学到有效信息或单模态分支过拟合先单独训练单模态基线再加入融合层必要时加入权重正则模态屏蔽后预测崩溃模型只在正常数据上训练没有见过缺失输入增加随机模态屏蔽比例让 mask 概率从低到高逐步调整质量分几乎不生效显式质量分和隐式权重计算方式差异过大把显式质量分先缩放到相似范围用线性对齐或 batch norm 处理两路信号时间不对齐数据采集时延或重采样边界不同先用互相关对齐再按相同起点切窗训练慢显存不足序列太长、GRU 双向、batch 太大缩短信号窗口、减少 hidden_size、缩小 batch size6.2 融合权重退化的排查如果训练结束后发现ecg_weight始终接近 1ppg_weight始终接近 0说明模型没有从融合中获益。这种“模态退化”是融合任务里最典型的问题。排查步骤检查两个分支单独训练时是否都能拟合数据。检查质量分输入是否对某些模态产生了偏差。把质量分临时固定为 0.5验证融合结构是否有效。把随机模态屏蔽比例提高强制模型不能只依赖单一模态。6.3 信号质量分不准确的处理compute_sqi只是示例真实信号中可能存在心律失常、早搏等生理性不规律导致峰间间隔变异系数很大但信号本身质量不差。因此更稳健的做法是把多个 SQI 指标组合例如“峰值形态相关性”“信噪比估计”“有效峰比例”。如果模型对某个样本的预测置信度低也可以把预测结果标记为低置信度由上层系统决定是否丢弃或提醒用户。这种做法在医疗监测中尤其重要。7. 最佳实践与工程建议7.1 先建立单模态基线不管是采购新算法还是自己设计融合方案第一步都要先评估单模态模型在“正常信号”和“退化信号”上的表现。只有清楚了单模态的上限和退化特征才能判断融合是否真的有效。推荐用一张表格记录基线场景单模态 ECG (MAE)单模态 PPG (MAE)融合 (MAE)正常信号---ECG 噪声---PPG 噪声---ECG 缺失---PPG 缺失---7.2 用质量分作为先验而不是唯一标准显示质量分可以帮助模型在缺失模态时收敛更快但不能把最终权重完全交给工程规则。原因在于真实信号的质量分数本身可能存在误差。比较稳的做法是把质量分作为“先验特征”输入融合层再由神经网络学习最终权重。7.3 训练时要充分覆盖退化分布模型对信号退化的鲁棒性来自训练数据的退化分布。如果只在干净信号上训练部署时遇到一点噪声就会失效。建议在训练时持续做在线退化增强并每隔若干轮记录一次单模态缺失场景的验证指标。如果项目使用真实临床数据需要额外注意数据必须经过脱敏和授权。退化标注最好由专业人员参与。验证集应按不同受试者切分避免同一个人同时出现在训练和验证集中。7.4 工程部署时控制输入长度和采样率模型在某个窗口长度上训练后部署时不要随意改变窗口长度。窗口长度会影响质量分计算、GRU 时序依赖和特征分布。如果设备端采样率不固定建议在采集模块就统一输出 100 Hz或者把重采样逻辑放在设备端完成而不是集中到云端服务器。这样可以减少网络传输量和延迟。7.5 记录模型版本与退化场景评估结果算法迭代过程中建议每次训练都记录数据版本。预处理参数。训练增强策略。各退化场景的 MAE。模型权重文件的 hash 或路径。生理信号模型的可重复性和可审计性非常关键。记录这些信息不仅便于复盘也是后续发布到真实业务场景前的必要准备。8. 总结与下一步学习路线从 CardioFusion-AI 的实验来看多模态融合的价值不在于把单一指标调到最好而在于让系统在信号退化时依然可用。真正的鲁棒性来自三个层次输入层通过质量评估和缺失标记让模型知道哪些输入不可信。模型层通过双分支独立编码避免两个模态互相干扰。训练层通过模态屏蔽和退化增强让模型提前适应部署环境。如果你准备在真实项目中落地可以先从这个最小框架开始然后逐步改进。下一步可以继续学习的方向跨模态注意力机制例如让 ECG 分支自动关注 PPG 中可靠的峰间隔。自监督预训练用大量无标签信号预训练单模态编码器。在线自适应在设备端根据真实数据分布微调融合权重。异常检测把融合后的重构误差或质量分异常作为传感器故障告警信号。这里最值得记下的一点是不要一开始就把模型设计得很复杂。先跑通 ECG 和 PPG 两路分支、质量门控、缺失模态训练再根据失败场景逐步加深网络。信号退化场景多、质量问题隐蔽稳定可复现的基线比花哨的网络结构更有价值。 ECG与PPG多模态融合：质量门控与抗退化心率估计方案
1. 背景与问题定义1.1 为什么需要 ECG 与 PPG 融合在可穿戴设备和无感监测场景中心电图ECG和光电容积脉搏波PPG是两种最常见的生理信号。ECG 直接反映心脏电活动能够提供准确的 RR 间期、心率变异性HRV等信息PPG 通过光学方式测量外周血容量变化能够间接估计心率、血氧饱和度和脉搏特征。单一模态在实际部署中会遇到明显瓶颈。PPG 对运动伪影非常敏感手部摆动、环境光变化、传感器贴合松动都会导致波形严重失真ECG 虽然形态稳定但在干电极、运动出汗、贴片脱落等场景下同样会出现基线漂移、电极接触噪声和单导联信号丢失。如果系统只在单模态下做预测一旦信号质量下降心率估计、心律分析等关键指标就会快速恶化。CardioFusion-AI 的出发点很直接将 ECG 和 PPG 看作互补的信息源通过融合策略让两路信号在各自受损时互相补偿。比如 PPG 出现运动伪影时ECG 如果仍然可靠系统可以自动降低 PPG 分支的权重反过来ECG 电极脱落时PPG 仍能支撑基本的脉率估计。这种“质量感知”的多模态融合比简单的平均值融合或直接拼接更鲁棒。1.2 信号退化问题具体指什么信号退化不是一个抽象概念在生理信号处理中退化通常表现为以下几类运动伪影传感器与皮肤相对运动带来的低频大幅扰动。基线漂移呼吸、电极移动或放大器温漂导致的缓慢基线变化。高频噪声工频干扰、肌肉电噪声、光学环境光干扰。局部缺失电极脱落、蓝牙丢包、传感器断连导致的数据段缺失。幅值变化不同佩戴位置、肤色、组织厚度带来的幅值差异。模型的鲁棒性就是在这些退化情况下依然保持稳定性能的能力。评估时不能只看正常信号上的精度更要在“信号退化程度由低到高”的谱系上观察性能衰减曲线。1.3 CardioFusion-AI 的核心思想CardioFusion-AI 可以理解为“双分支特征提取 质量门控融合 缺失模态训练”的完整方案。双分支特征提取ECG 和 PPG 分别经过独立编码器提取时序特征不对单模态做过度耦合。质量门控融合每条分支生成一个质量评分融合时根据质量评分动态调整两个分支的贡献程度。缺失模态训练训练时随机屏蔽一个模态让模型学会在部分输入缺失的情况下仍然给出可用预测。这个方案不依赖特定的输入长度或特定传感器因此可以迁移到不同的可穿戴设备平台。2. CardioFusion-AI 整体方案设计2.1 问题建模把问题形式化为一个序列回归任务。输入为两段同步采样信号ecg_signal: 长度 T 的 ECG 波形片段ppg_signal: 长度 T 的 PPG 波形片段目标是预测一组生理指标。为了便于实验本文以“心率估计”作为主任务也就是预测每段信号对应的平均心率。算法框架同样可以扩展到 HRV、呼吸率等指标。补充说明工程中信号可能来自不同采样率比如 ECG 为 250 HzPPG 为 125 Hz需要先重采样到统一频率。本文代码示例假定输入已经预处理为 100 Hz 的等长片段长度 T 1000对应 10 秒窗口。2.2 系统架构整个流程可以分为四个阶段信号预处理与切窗。信号质量评估。双分支特征提取。质量门控融合与预测输出。在训练阶段额外加入噪声增强和随机模态屏蔽模拟信号退化。在部署阶段质量评估结果还可以作为异常告警依据。2.3 关键模块设计模块输入输出职责预处理模块原始 ECG、PPG标准化片段滤波、重采样、归一化质量评估模块单模态片段质量分数估计信号可信度ECG 分支ECG 片段特征向量提取心电特征PPG 分支PPG 片段特征向量提取脉搏特征融合模块两个特征 两个质量分预测结果动态加权整合值得注意的是质量评估模块既可以独立计算也可以由特征编码器预测一个“隐式质量分数”。本文采用“显式 隐式结合”显式质量分来自信号质量指标隐式质量分来自编码器末尾的全连接层。最终融合权重由两者共同决定。3. 环境准备与数据说明3.1 运行环境本文代码以 Python 3.9 和 PyTorch 1.13 为基础。你在运行时可按自己的环境调整版本差异通常不影响整体思路。依赖库建议如下numpy用于数值计算。scipy用于信号滤波和峰值检测。torch用于模型训练。pandas和matplotlib用于数据读取和可视化。安装命令pip install numpy scipy torch pandas matplotlib如果使用 GPU需要确保 CUDA 和 PyTorch 版本匹配。无法使用 GPU 时把代码中的.cuda()相关部分去掉模型也可以在小数据集上跑 CPU 版本只是速度会慢一些。3.2 数据集准备真实生理数据集需要经过伦理审批和脱敏处理。在教程演示中建议用两种方式构建数据公开数据集中截取片段例如多模态生理信号公开数据库中的同步 ECG 与 PPG 信号。用仿真波形生成程序构造带标注的合成数据方便快速验证模型逻辑。本文为了演示训练流程使用以下模拟方案生成一段基础心率信号再叠加可控制噪声生成模拟 ECG 和 PPG。真实项目中请替换为实际数据加载代码。模拟数据结构建议data/ ├── raw/ │ ├── ecg_0001.npy │ ├── ppg_0001.npy │ └── label_0001.npy ├── train.csv ├── val.csv └── test.csv其中每个 npy 文件是一个等长波形数组label 是对应的心率值。3.3 信号预处理流程在输入模型之前需要完成以下步骤。重采样到统一采样率。带通滤波ECG 保留 0.5~45 HzPPG 保留 0.5~10 Hz。去除均值并除以标准差做 z-score 标准化。按固定窗口切分例如每段 10 秒。这里给出一个预处理函数示例# src/preprocess.py import numpy as np from scipy import signal as sp_signal def preprocess_signal(signal_data, original_fs, target_fs100, lowcut0.5, highcut40.0): # 1. 重采样 if original_fs ! target_fs: n_samples int(len(signal_data) * target_fs / original_fs) signal_data sp_signal.resample(signal_data, n_samples) # 2. 带通滤波 nyquist 0.5 * target_fs normal_low lowcut / nyquist normal_high highcut / nyquist b, a sp_signal.butter(4, [normal_low, normal_high], btypeband) signal_data sp_signal.filtfilt(b, a, signal_data) # 3. z-score 标准化 mean np.mean(signal_data) std np.std(signal_data) if std 1e-6: std 1e-6 signal_data (signal_data - mean) / std return signal_data.astype(np.float32)这里使用filtfilt做零相位滤波避免普通滤波带来的相位偏移。注意滤波边界需要根据信号实际频带调整不是所有数据都适合相同的截止频率。4. 核心代码实现4.1 信号质量评估模块信号质量评估的核心是提取一个 0~1 之间的分数分数越高表示信号越可信。常见评估思路使用峰值检测计算相邻峰值的间隔变异系数。计算信号与自身模板的相关性。计算高频噪声能量占比。本文实现一个简单的质量分函数基于峰值间隔的稳定性# src/sqi.py import numpy as np from scipy.signal import find_peaks def compute_sqi(signal, fs100, min_hr30, max_hr220): min_dist int(fs * 60.0 / max_hr) max_dist int(fs * 60.0 / min_hr) peaks, _ find_peaks(signal, distancemin_dist) if len(peaks) 3: return 0.1 intervals np.diff(peaks) / fs * 1000.0 # 转成毫秒 intervals intervals[(intervals min_dist / fs * 1000) (intervals max_dist / fs * 1000)] if len(intervals) 2: return 0.2 cv np.std(intervals) / (np.mean(intervals) 1e-8) # 变异系数越小质量越高 sqi float(np.clip(1.0 - cv * 0.5, 0.0, 1.0)) return sqi这个函数只演示思路。实际项目中还需要加入幅度异常检测、模板相关性、信噪比估计等维度。质量分不能只看峰间间隔否则心律失常信号会被误判为低质量。4.2 单模态编码器ECG 和 PPG 分支共用一个编码器结构但参数彼此独立。编码器采用“一维卷积 GRU 全局池化”的组合既能提取局部波形形态又能捕获长程时序依赖。# src/branch_encoder.py import torch import torch.nn as nn class BranchEncoder(nn.Module): def __init__(self, input_size1, hidden_size64, out_dim64): super().__init__() self.conv1 nn.Conv1d(input_size, 32, kernel_size5, stride1, padding2) self.bn1 nn.BatchNorm1d(32) self.conv2 nn.Conv1d(32, 64, kernel_size5, stride1, padding2) self.bn2 nn.BatchNorm1d(64) self.gru nn.GRU(64, hidden_size, batch_firstTrue, bidirectionalTrue) self.fc nn.Linear(hidden_size * 2, out_dim) def forward(self, x): # x: (batch, 1, time_len) x torch.relu(self.bn1(self.conv1(x))) x torch.relu(self.bn2(self.conv2(x))) # 转成 (batch, time_len, channels) x x.permute(0, 2, 1) x, _ self.gru(x) # 全局平均池化 x x.mean(dim1) return self.fc(x)这里输出一个out_dim维的特征向量。双向 GRU 可以同时看到当前时刻前后信息但模型参数会多一倍。若设备计算力有限可以把bidirectional改成False。4.3 质量门控融合模块融合模块需要完成两件事一是判断两个模态各自贡献多少二是把特征向量整合为最终预测。定义隐式质量权重来自两个分支特征# src/fusion.py import torch import torch.nn as nn class QualityGate(nn.Module): def __init__(self, feature_dim64): super().__init__() self.ecg_gate nn.Sequential( nn.Linear(feature_dim, 32), nn.ReLU(), nn.Linear(32, 1), nn.Sigmoid() ) self.ppg_gate nn.Sequential( nn.Linear(feature_dim, 32), nn.ReLU(), nn.Linear(32, 1), nn.Sigmoid() ) def forward(self, ecg_feat, ppg_feat, ecg_sqi, ppg_sqi): # 隐式权重 implicit_ecg self.ecg_gate(ecg_feat) implicit_ppg self.ppg_gate(ppg_feat) # 显式质量分与隐式权重结合 ecg_weight (0.4 * ecg_sqi 0.6 * implicit_ecg.squeeze(-1)).unsqueeze(-1) ppg_weight (0.4 * ppg_sqi 0.6 * implicit_ppg.squeeze(-1)).unsqueeze(-1) # 归一化保证权重之和为 1 total ecg_weight ppg_weight 1e-8 ecg_weight ecg_weight / total ppg_weight ppg_weight / total return ecg_weight, ppg_weight融合最终特征可以简单写成加权求和def fuse(ecg_feat, ppg_feat, ecg_weight, ppg_weight): fused_feature ecg_feat * ecg_weight ppg_feat * ppg_weight return fused_feature当然更复杂的设计可以使用跨模态注意力让 ECG 特征在融合时关注 PPG 特征中的关键时间点。但对于心率回归任务质量门控加权求和已经足够稳定并且更容易部署。4.4 抗退化训练策略模型要在缺失模态和信号退化条件下保持鲁棒训练策略比模型结构更关键。本文采用三种策略随机模态屏蔽以一定概率把某个分支输入置零同时把对应显式质量分置为 0。信号噪声增强对输入叠加高斯噪声、基线漂移或随机缩放。时间片段挖空随机把某一段连续时间置零模拟传感器瞬间掉线。这些增强策略可以用一个模块统一处理。下面给出噪声增强的部分代码# src/augment.py import numpy as np def apply_degradation(ecg, ppg, noise_level0.1, mask_prob0.3): # 随机模态屏蔽 if np.random.rand() mask_prob: if np.random.rand() 0.5: ecg np.zeros_like(ecg) else: ppg np.zeros_like(ppg) # 添加高斯噪声 ecg ecg noise_level * np.random.randn(*ecg.shape) ppg ppg noise_level * np.random.randn(*ppg.shape) # 随机挖空 1 秒 fs 100 if np.random.rand() 0.2: start np.random.randint(0, len(ecg) - fs) ecg[start:start fs] 0.0 ppg[start:start fs] 0.0 return ecg.astype(np.float32), ppg.astype(np.float32)注意在验证集和测试集上不要做这类随机增强否则评估结果不稳定。可以把mask_prob置为 0单独构造“缺失模态”测试集来评估鲁棒性。4.5 组装完整模型有了分支编码器和质量门控可以组装完整的 CardioFusion-AI 模型。# src/cardiofusion.py import torch import torch.nn as nn from src.branch_encoder import BranchEncoder from src.fusion import QualityGate from src.sqi import compute_sqi class CardioFusion(nn.Module): def __init__(self, hidden_size64, out_dim64, num_task1): super().__init__() self.ecg_encoder BranchEncoder(input_size1, hidden_sizehidden_size, out_dimout_dim) self.ppg_encoder BranchEncoder(input_size1, hidden_sizehidden_size, out_dimout_dim) self.gate QualityGate(feature_dimout_dim) self.regressor nn.Sequential( nn.Linear(out_dim, 32), nn.ReLU(), nn.Linear(32, num_task) ) def forward(self, ecg, ppg, ecg_sqi, ppg_sqi): # ecg: (batch, 1, time_len), ppg: (batch, 1, time_len) ecg_feat self.ecg_encoder(ecg) ppg_feat self.ppg_encoder(ppg) ecg_weight, ppg_weight self.gate(ecg_feat, ppg_feat, ecg_sqi, ppg_sqi) # 质量分低时对应特征也会接近 0 fused_feat ecg_feat * ecg_weight ppg_feat * ppg_weight output self.regressor(fused_feat) return output, ecg_weight, ppg_weight在推理时对每一段信号先用compute_sqi计算显式质量分再输入模型。如果某个模态整段缺失可以直接把输入置零同时把质量分设为 0。5. 训练与评估5.1 损失函数主任务使用均方误差MSE作为监督损失。为了让质量门控产生有意义的权重可以额外增加一个辅助损失用于预测每个模态的真实质量。但实际标注中很难获得每个模态精确的真实质量分所以更常用的做法是主损失只负责最终预测。通过训练数据中的随机屏蔽迫使模型自动学会降低缺失模态的贡献。在验证时观察不同缺失场景下的性能。如果需要对权重做更严格约束也可以加入权重熵正则项避免模型总是偏向某一个模态reg_loss -0.5 * (ecg_weight * torch.log(ecg_weight 1e-8) ppg_weight * torch.log(ppg_weight 1e-8)).mean()这个正则项不是必须的。如果发现融合结果退化成“只用 ECG”或“只用 PPG”再考虑加入。5.2 训练循环下面给出一个简化训练循环。代码省略了数据加载细节核心是演示训练逻辑。import torch import torch.nn as nn def train_one_epoch(model, dataloader, optimizer, device): model.train() total_loss 0.0 criterion nn.MSELoss() for ecg, ppg, label in dataloader: ecg ecg.unsqueeze(1).to(device) ppg ppg.unsqueeze(1).to(device) label label.unsqueeze(1).to(device) # 计算显式质量分 ecg_sqi torch.tensor([compute_sqi(x.cpu().numpy().reshape(-1)) for x in ecg.squeeze(1)]).to(device).unsqueeze(-1) ppg_sqi torch.tensor([compute_sqi(x.cpu().numpy().reshape(-1)) for x in ppg.squeeze(1)]).to(device).unsqueeze(-1) pred, _, _ model(ecg, ppg, ecg_sqi, ppg_sqi) loss criterion(pred, label) optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() return total_loss / max(len(dataloader), 1)这段代码每次循环都用compute_sqi计算质量分效率偏低。工程实现中建议把质量分作为数据预处理结果保存到缓存里训练时直接读取。5.3 评估指标与鲁棒性评估心率回归任务常用指标平均绝对误差MAE。均方根误差RMSE。决定系数R²。除了整体指标还要设计压力测试评估鲁棒性场景说明正常信号ECG 和 PPG 均正常ECG 轻度噪声ECG 叠加高斯噪声PPG 运动伪影PPG 叠加低频基线漂移ECG 缺失ECG 输入全部置零PPG 缺失PPG 输入全部置零双模态噪声两路信号均退化每一类场景单独计算 MAE就能直观看出融合模型在信号退化下是否稳定。比如下面这段简化评估代码def evaluate_by_scenario(model, dataset, device): model.eval() metrics {} for scenario_name, data in dataset.scenarios.items(): preds [] labels [] with torch.no_grad(): for ecg, ppg, label in data: ecg ecg.unsqueeze(0).unsqueeze(1).to(device) ppg ppg.unsqueeze(0).unsqueeze(1).to(device) ecg_sqi torch.tensor([[compute_sqi(ecg.cpu().numpy().squeeze())]]).to(device) ppg_sqi torch.tensor([[compute_sqi(ppg.cpu().numpy().squeeze())]]).to(device) pred, _, _ model(ecg, ppg, ecg_sqi, ppg_sqi) preds.append(pred.item()) labels.append(label.item()) mae np.mean(np.abs(np.array(preds) - np.array(labels))) metrics[scenario_name] mae return metrics评估结果通常会出现一种现象正常信号下融合模型不一定优于表现最好的单模态模型但在单模态缺失或退化场景下明显更鲁棒。这正是融合系统的价值所在。6. 常见问题与排查思路6.1 常见问题表格问题现象常见原因排查与解决思路训练 loss 变成 NaN学习率过大、信号标准化后仍有极大值降低学习率使用梯度裁剪检查输入是否有 inf融合模型不如单模态融合权重没有学到有效信息或单模态分支过拟合先单独训练单模态基线再加入融合层必要时加入权重正则模态屏蔽后预测崩溃模型只在正常数据上训练没有见过缺失输入增加随机模态屏蔽比例让 mask 概率从低到高逐步调整质量分几乎不生效显式质量分和隐式权重计算方式差异过大把显式质量分先缩放到相似范围用线性对齐或 batch norm 处理两路信号时间不对齐数据采集时延或重采样边界不同先用互相关对齐再按相同起点切窗训练慢显存不足序列太长、GRU 双向、batch 太大缩短信号窗口、减少 hidden_size、缩小 batch size6.2 融合权重退化的排查如果训练结束后发现ecg_weight始终接近 1ppg_weight始终接近 0说明模型没有从融合中获益。这种“模态退化”是融合任务里最典型的问题。排查步骤检查两个分支单独训练时是否都能拟合数据。检查质量分输入是否对某些模态产生了偏差。把质量分临时固定为 0.5验证融合结构是否有效。把随机模态屏蔽比例提高强制模型不能只依赖单一模态。6.3 信号质量分不准确的处理compute_sqi只是示例真实信号中可能存在心律失常、早搏等生理性不规律导致峰间间隔变异系数很大但信号本身质量不差。因此更稳健的做法是把多个 SQI 指标组合例如“峰值形态相关性”“信噪比估计”“有效峰比例”。如果模型对某个样本的预测置信度低也可以把预测结果标记为低置信度由上层系统决定是否丢弃或提醒用户。这种做法在医疗监测中尤其重要。7. 最佳实践与工程建议7.1 先建立单模态基线不管是采购新算法还是自己设计融合方案第一步都要先评估单模态模型在“正常信号”和“退化信号”上的表现。只有清楚了单模态的上限和退化特征才能判断融合是否真的有效。推荐用一张表格记录基线场景单模态 ECG (MAE)单模态 PPG (MAE)融合 (MAE)正常信号---ECG 噪声---PPG 噪声---ECG 缺失---PPG 缺失---7.2 用质量分作为先验而不是唯一标准显示质量分可以帮助模型在缺失模态时收敛更快但不能把最终权重完全交给工程规则。原因在于真实信号的质量分数本身可能存在误差。比较稳的做法是把质量分作为“先验特征”输入融合层再由神经网络学习最终权重。7.3 训练时要充分覆盖退化分布模型对信号退化的鲁棒性来自训练数据的退化分布。如果只在干净信号上训练部署时遇到一点噪声就会失效。建议在训练时持续做在线退化增强并每隔若干轮记录一次单模态缺失场景的验证指标。如果项目使用真实临床数据需要额外注意数据必须经过脱敏和授权。退化标注最好由专业人员参与。验证集应按不同受试者切分避免同一个人同时出现在训练和验证集中。7.4 工程部署时控制输入长度和采样率模型在某个窗口长度上训练后部署时不要随意改变窗口长度。窗口长度会影响质量分计算、GRU 时序依赖和特征分布。如果设备端采样率不固定建议在采集模块就统一输出 100 Hz或者把重采样逻辑放在设备端完成而不是集中到云端服务器。这样可以减少网络传输量和延迟。7.5 记录模型版本与退化场景评估结果算法迭代过程中建议每次训练都记录数据版本。预处理参数。训练增强策略。各退化场景的 MAE。模型权重文件的 hash 或路径。生理信号模型的可重复性和可审计性非常关键。记录这些信息不仅便于复盘也是后续发布到真实业务场景前的必要准备。8. 总结与下一步学习路线从 CardioFusion-AI 的实验来看多模态融合的价值不在于把单一指标调到最好而在于让系统在信号退化时依然可用。真正的鲁棒性来自三个层次输入层通过质量评估和缺失标记让模型知道哪些输入不可信。模型层通过双分支独立编码避免两个模态互相干扰。训练层通过模态屏蔽和退化增强让模型提前适应部署环境。如果你准备在真实项目中落地可以先从这个最小框架开始然后逐步改进。下一步可以继续学习的方向跨模态注意力机制例如让 ECG 分支自动关注 PPG 中可靠的峰间隔。自监督预训练用大量无标签信号预训练单模态编码器。在线自适应在设备端根据真实数据分布微调融合权重。异常检测把融合后的重构误差或质量分异常作为传感器故障告警信号。这里最值得记下的一点是不要一开始就把模型设计得很复杂。先跑通 ECG 和 PPG 两路分支、质量门控、缺失模态训练再根据失败场景逐步加深网络。信号退化场景多、质量问题隐蔽稳定可复现的基线比花哨的网络结构更有价值。