保姆级教程：用TensorFlow 1.15复现CNN+LSTM睡眠分期模型（附Sleep-EDF/MASS数据集处理）

从零实现基于CNN+LSTM的睡眠分期分析：TensorFlow 1.15实战指南

当你在深夜调试代码时，是否想过计算机也能像人类一样理解睡眠？睡眠分期分析正是将脑电信号（EEG）转化为可解释睡眠阶段的关键技术。不同于大多数教程的理论概述，本文将带你深入工程细节，用TensorFlow 1.15完整实现一个能处理多源数据的混合神经网络模型。我们会从数据集差异处理开始，逐步解决版本兼容、类别不平衡等实际问题，最终让模型在你的本地机器上跑起来。

1. 环境配置与数据准备

1.1 老版本TensorFlow的生存指南

在Python 3.5.4和TensorFlow 1.15.2的环境配置中，最令人头疼的莫过于版本依赖问题。以下是经过验证的安装方案：

# 创建专属虚拟环境 conda create -n tf1.15 python=3.5.4 conda activate tf1.15 # 指定版本安装核心库 pip install tensorflow-gpu==1.15.2 keras==2.2.4 pip install h5py==2.10.0 numpy==1.16.4 scipy==1.2.1

特别注意几个易错点：

• CUDA 10.0和cuDNN 7.6.5是TF 1.15的最佳搭档
• 新版protobuf会导致序列化错误，需强制降级：

  pip install protobuf==3.20.*

1.2 多源EEG数据统一处理框架

Sleep-EDF和MASS数据集存在三个关键差异需要标准化：

数据加载的核心代码结构应包含自适应处理：

def load_eeg(file_path): # 自动检测数据集类型 if 'edf' in file_path.lower(): raw = mne.io.read_raw_edf(file_path, preload=True) annot = mne.read_annotations(file_path.replace('.edf', '.hypnogram')) else: # MASS格式处理 raw, annot = load_mass_h5(file_path) # 统一化处理 raw.resample(100) # 降采样 raw.apply_function(lambda x: (x-x.mean())/x.std()) # Z-score标准化 # 标注转换 stage_mapping = {'W':0, 'N1':1, 'N2':2, 'N3':3, 'R':4} events = [(int(onset*100), 0, stage_mapping[desc]) for onset, _, desc in annot] return raw.get_data(), events

2. 模型架构深度解析

2.1 双分支CNN特征提取器

原始论文的精妙之处在于并行的多尺度卷积设计。我们实现时需要注意三个工程细节：

1. 参数共享机制：两个CNN分支应共享后续层的权重
2. 残差连接：防止深层网络梯度消失
3. 空间注意力：自动聚焦有效EEG频段

def build_cnn_layers(inputs, reuse=False): with tf.variable_scope('feature_extractor', reuse=reuse): # 分支1: 捕捉短时特征 (1秒窗口) branch1 = tf.layers.conv2d(inputs, 64, (30,1), padding='same') branch1 = tf.layers.batch_normalization(branch1) branch1 = tf.nn.leaky_relu(branch1) # 分支2: 捕捉长时特征 (3秒窗口) branch2 = tf.layers.conv2d(inputs, 64, (90,1), padding='same') branch2 = tf.layers.batch_normalization(branch2) branch2 = tf.nn.leaky_relu(branch2) # 特征融合 merged = tf.concat([branch1, branch2], axis=-1) # 加入空间注意力 attention = tf.reduce_mean(merged, axis=[1,2], keepdims=True) attention = tf.layers.dense(attention, units=128, activation='sigmoid') return merged * attention

2.2 双向LSTM时序建模技巧

在处理睡眠阶段的连续转换时，双向LSTM需要特殊配置：

• Peephole连接：增强门控机制对EEG节律的敏感性
• 层归一化：稳定长序列训练过程
• 状态复用：提升小批量训练效果

def build_bilstm(features, seq_length, is_training): # 调整输入维度 [batch*seq_len, ...] -> [batch, seq_len, ...] features = tf.reshape(features, [tf.shape(features)[0]//seq_length, seq_length, features.shape[-1]]) # 实现peephole LSTM单元 def make_cell(): cell = tf.contrib.rnn.LSTMCell( num_units=512, use_peepholes=True, initializer=tf.orthogonal_initializer()) if is_training: cell = tf.contrib.rnn.DropoutWrapper( cell, output_keep_prob=0.8) return tf.contrib.rnn.LayerNormBasicLSTMCell( cell, layer_norm=True) # 双向RNN构建 outputs, _, _ = tf.contrib.rnn.stack_bidirectional_dynamic_rnn( [make_cell() for _ in range(2)], [make_cell() for _ in range(2)], inputs=features, dtype=tf.float32) return outputs

3. 训练策略与调优实战

3.1 两步训练算法实现

针对睡眠数据中阶段分布不均的问题（N1阶段通常<5%），论文提出了分阶段训练策略：

1. 特征学习阶段：冻结LSTM，仅训练CNN部分
2. 序列建模阶段：解冻全部参数，使用加权损失函数

# 自定义加权交叉熵损失 def weighted_loss(logits, labels): class_weights = tf.constant([0.5, 2.0, 0.8, 1.2, 1.0]) # 对应W/N1/N2/N3/REM weights = tf.gather(class_weights, labels) loss = tf.losses.sparse_softmax_cross_entropy( labels=labels, logits=logits, weights=weights) return loss # 分阶段训练操作 def build_train_op(loss, step, lr=1e-4): # 第一阶段只优化CNN参数 phase1_vars = tf.get_collection( tf.GraphKeys.TRAINABLE_VARIABLES, scope='feature_extractor') phase1_op = tf.train.AdamOptimizer(lr).minimize( loss, var_list=phase1_vars) # 第二阶段优化全部参数 phase2_op = tf.train.AdamOptimizer(lr*0.1).minimize(loss) # 根据全局步数选择训练op return tf.cond(step < 10000, lambda: phase1_op, lambda: phase2_op)

3.2 数据增强与正则化技巧

EEG信号的特殊性要求定制化的增强策略：

• 频域增强：随机滤波增强（模拟不同设备特性）
• 时域增强：随机片段缩放（0.9-1.1倍）
• 通道增强：随机噪声注入（SNR>20dB）

def augment_eeg(signal, fs=100): # 时域伸缩 orig_len = signal.shape[0] new_len = int(orig_len * np.random.uniform(0.9, 1.1)) resized = scipy.signal.resample(signal, new_len) # 频域扰动 b, a = scipy.signal.butter(3, [np.random.uniform(0.5,5)/fs*2, np.random.uniform(45,55)/fs*2], 'bandpass') filtered = scipy.signal.filtfilt(b, a, resized) # 添加高斯噪声 noise = np.random.normal(0, 0.05*np.std(filtered), filtered.shape) return filtered + noise

4. 结果可视化与模型部署

4.1 睡眠阶段可视化分析

使用混合矩阵和过渡概率图分析模型表现：

def plot_sleep_stages(true, pred, save_path): # 计算阶段转移概率 trans_matrix = np.zeros((5,5)) for t in range(1, len(true)): trans_matrix[true[t-1], true[t]] += 1 trans_matrix /= trans_matrix.sum(axis=1, keepdims=True) # 绘制双热力图 fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12,5)) sns.heatmap(confusion_matrix(true, pred), annot=True, fmt='d', ax=ax1) sns.heatmap(trans_matrix, annot=True, fmt='.2f', ax=ax2) plt.savefig(save_path)

4.2 模型轻量化与部署

为临床环境部署需要考虑模型压缩：

实际部署时的关键代码：

# 转换到TF-Lite格式 converter = tf.lite.TFLiteConverter.from_saved_model('saved_model') converter.optimizations = [tf.lite.Optimize.DEFAULT] tflite_model = converter.convert() # 加载运行示例 interpreter = tf.lite.Interpreter(model_content=tflite_model) interpreter.allocate_tensors() input_details = interpreter.get_input_details() output_details = interpreter.get_output_details() # 实时推理 def predict(eeg_segment): interpreter.set_tensor( input_details[0]['index'], eeg_segment.astype(np.float32)) interpreter.invoke() return interpreter.get_tensor(output_details[0]['index'])

在模型实际部署后，建议持续监控数据漂移——当新采集的EEG信号与训练数据分布差异超过阈值时触发重新训练。这可以通过计算KL散度或使用专门的概念漂移检测算法来实现。