Motion Classification Code Explanation - CBAM + CNN + Bi-LSTM

Last updated on 17 Feb 2026

我来非常详细地讲解这个代码。这是一个时序传感器数据分类的端到端深度学习pipeline，用于SmartPouch项目（看起来是可穿戴设备或传感器数据分析）。

整体架构概览

这个pipeline做了以下事情：

加载6通道传感器数据（加速度计+陀螺仪）
用滑动窗口切分时序数据
训练 CBAM + CNN + Bi-LSTM 混合模型
使用加权交叉熵处理类别不平衡

第一部分：配置参数（Lines 19-31）

CSV_PATH = "dataset/prepared_data_for_training.csv"
WINDOW_LENGTH = 5000        # 每个窗口5000个时间步
WINDOW_STEP   = 2500        # 50%重叠滑动
MIN_LABEL_RATIO = 0.8       # 窗口内80%样本必须是同一类别

关键设计决策：

5000时间步：如果采样率是500Hz，这是10秒数据
50%重叠：增加训练样本，提高模型鲁棒性
MIN_LABEL_RATIO=0.8：过滤掉"混杂"窗口（动作转换期间的数据）

💡 这会让招聘经理给你100k+ offer吗？ 部分会。这展示了你理解时序数据预处理的核心挑战（窗口大小vs上下文，标签噪声处理）。但要提升到更高层级：你需要能解释为什么选择这些参数（实验对比？领域知识？），以及如何自动调优这些超参数。

第二部分：CBAM注意力机制（Lines 38-103）

2.1 Channel Attention（通道注意力）

def channel_attention(inputs, reduction_ratio=2):
    avg_pool = tf.reduce_mean(inputs, axis=1)  # 全局平均池化
    max_pool = tf.reduce_max(inputs, axis=1)   # 全局最大池化
    
    # 共享MLP
    hidden_units = max(channel_dim // reduction_ratio, 1)
    shared_dense_1 = layers.Dense(hidden_units, activation='relu')
    shared_dense_2 = layers.Dense(channel_dim)
    
    # 计算注意力权重
    scale = tf.nn.sigmoid(avg_out + max_out)
    return inputs * scale

工作原理：

对时间维度做全局池化（avg和max），得到(B, C)
通过共享MLP（降维→升维）学习通道间关系
用Sigmoid生成每个通道的重要性权重
加权原始输入

实际意义：

对于传感器数据，某些通道（如xG）可能比其他通道更重要
模型自动学习"走路时主要看yG，跑步时主要看zG"

2.2 Temporal Attention（时间注意力）

def temporal_attention(inputs, kernel_size=7):
    avg_pool = tf.reduce_mean(inputs, axis=2, keepdims=True)  # (B,T,1)
    max_pool = tf.reduce_max(inputs, axis=2, keepdims=True)
    
    concat = tf.concat([avg_pool, max_pool], axis=-1)  # (B,T,2)
    conv = layers.Conv1D(filters=1, kernel_size=7, activation="sigmoid")(concat)
    
    return inputs * conv

工作原理：

对通道维度做池化，得到每个时间步的"重要性信号"
用1D卷积（kernel=7）捕获局部时序模式
生成每个时间步的权重

实际意义：

突出"关键时刻"（如动作开始/结束的瞬间）
抑制噪声或过渡阶段

💡 更高层级思维： CBAM是2018年的经典模块，但你需要能回答：为什么不用Transformer的self-attention？（计算复杂度O(T²) vs O(T)）是否尝试过SE-Net、ECA-Net等轻量化注意力？如何量化注意力的有效性？（可视化attention maps，做消融实验）

第三部分：CNN模块（Lines 109-134）

def conv_block(x, filters, kernel_size, dilation_rate=1, dropout_rate=0.0):
    x = layers.Conv1D(filters, kernel_size, padding="same", dilation_rate=dilation_rate)(x)
    x = layers.BatchNormalization()(x)
    x = layers.Activation("relu")(x)
    if dropout_rate > 0.0:
        x = layers.Dropout(dropout_rate)(x)
    return x

标准CNN Block结构：Conv → BN → ReLU → Dropout

关键参数：

dilation_rate：空洞卷积，扩大感受野而不增加参数
- 普通卷积 kernel=3 → 感受野=3
- dilated卷积 kernel=3, dilation=2 → 感受野=5

第四部分：完整模型架构（Lines 140-224）

def build_smartpouch_model(window_length, num_channels, num_classes):
    # 输入: (B, 5000, 6)
    
    # 1. CBAM注意力
    x = cbam_block(inputs)  # (B, 5000, 6)
    
    # 2. CNN特征提取
    x = conv_block(x, filters=64, kernel_size=7)    # (B, 5000, 64)
    x = conv_block(x, filters=128, kernel_size=5)   # (B, 5000, 128)
    x = conv_block(x, filters=128, kernel_size=3, dilation_rate=2)  # 空洞卷积
    
    # 3. 时间降采样
    x = layers.MaxPooling1D(pool_size=4)(x)  # (B, 1250, 128)
    
    # 4. Bi-LSTM捕获长程依赖
    x = layers.Bidirectional(LSTM(128))(x)  # (B, 1250, 256)
    x = layers.Bidirectional(LSTM(128))(x)  # (B, 1250, 256)
    
    # 5. 全局池化
    avg = GlobalAveragePooling1D()(x)  # (B, 256)
    max = GlobalMaxPooling1D()(x)      # (B, 256)
    x = Concatenate()([avg, max])      # (B, 512)
    
    # 6. 分类头
    x = Dense(128, activation="relu")(x)
    outputs = Dense(num_classes, activation="softmax")(x)

架构设计逻辑

为什么这样设计？

CBAM在前：先做注意力筛选，减少后续模块的噪声
CNN提取局部特征：卷积核从7→5→3，逐步聚焦细节
空洞卷积：不增加参数但扩大感受野（捕获更长时间跨度的模式）
MaxPooling降维：5000→1250，减少LSTM计算量
Bi-LSTM：双向处理，捕获"过去+未来"的上下文
双池化concat：avg保留整体信息，max保留峰值特征

💡 高级思考：为什么不用Transformer？（数据量可能不够大，LSTM更sample-efficient）如何优化推理速度？（知识蒸馏、量化、ONNX转换）如何部署到边缘设备？（TensorFlow Lite、模型剪枝）

第五部分：数据处理Pipeline（Lines 259-377）

5.1 滑动窗口构建

def build_windows(df, window_length=5000, window_step=2500, min_label_ratio=0.8):
    signals = df[["aG", "bG", "cG", "xG", "yG", "zG"]].values  # (N, 6)
    labels = df["label_id"].values  # (N,)
    
    start = 0
    while start + window_length <= N:
        end = start + window_length
        
        window_signal = signals[start:end]  # (5000, 6)
        window_labels = labels[start:end]   # (5000,)
        
        # 多数投票
        counts = np.bincount(window_labels)
        maj_label = counts.argmax()
        maj_ratio = counts[maj_label] / window_length
        
        if maj_ratio >= 0.8:  # 只保留"纯净"窗口
            X_list.append(window_signal)
            y_list.append(maj_label)
        
        start += window_step  # 滑动2500步

关键决策：

多数投票：窗口标签 = 出现最多的原始标签
MIN_LABEL_RATIO：过滤掉标签混乱的窗口（避免标签噪声）

示例：

原始序列: [走路, 走路, 走路, ..., 跑步, 跑步, ...]
窗口1 (0-5000): 4800个"走路" + 200个"跑步" → 96%走路 → 保留，标签=走路
窗口2 (过渡期): 2500个"走路" + 2500个"跑步" → 50%/50% → 丢弃！

5.2 标准化

mean = X_train.mean(axis=(0, 1), keepdims=True)  # (1, 1, 6)
std = X_train.std(axis=(0, 1), keepdims=True)

X_train_norm = (X_train - mean) / std
X_val_norm = (X_val - mean) / std

Per-channel标准化：

每个传感器通道独立标准化（aG、bG等有不同量纲）
只用训练集统计量，避免数据泄露

第六部分：加权交叉熵（Lines 231-246）

def compute_class_weights_from_labels(y, num_classes):
    counts = np.bincount(y)  # 每类样本数
    total = counts.sum()
    weights = total / (num_classes * counts)  # 逆频率加权
    
    # 示例输出:
    # Class 0 (走路): 10000样本 → weight = 0.5
    # Class 1 (跑步): 1000样本  → weight = 5.0
    return {i: w for i, w in enumerate(weights)}

作用：

自动平衡类别不平衡问题
少数类的loss权重更高，模型更关注少数类

训练时应用：

model.fit(
    train_ds,
    class_weight=class_weight,  # Keras自动加权loss
    ...
)

第七部分：训练流程（Lines 384-469）

# 1. 数据准备
df = load_flat_dataframe(CSV_PATH)
df, label_to_id, id_to_label, num_classes = encode_labels(df)

# 2. 滑动窗口
X, y = build_windows(df)

# 3. Train/Val划分 + 标准化
X_train, X_val, y_train, y_val = train_val_split_and_normalize(X, y)

# 4. tf.data pipeline（高效数据加载）
train_ds = tf.data.Dataset.from_tensor_slices((X_train, y_train))
train_ds = train_ds.shuffle(10000).batch(64).prefetch(tf.data.AUTOTUNE)

# 5. 构建模型
model = build_smartpouch_model(window_length=5000, num_channels=6, num_classes=num_classes)

# 6. 编译（Adam优化器 + 稀疏交叉熵）
model.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
    loss="sparse_categorical_crossentropy",  # 标签是整数
    metrics=["accuracy"]
)

# 7. Callbacks
callbacks = [
    ModelCheckpoint(monitor="val_loss", save_best_only=True),  # 保存最优模型
    EarlyStopping(patience=8, restore_best_weights=True),      # 防止过拟合
    ReduceLROnPlateau(factor=0.5, patience=3)                  # 学习率衰减
]

# 8. 训练
history = model.fit(
    train_ds,
    epochs=50,
    validation_data=val_ds,
    class_weight=class_weight,  # 加权loss
    callbacks=callbacks
)

💼 这段代码对你求职的价值

✅ 展示的技能

端到端ML工程能力
- 数据预处理 → 模型设计 → 训练 → 保存部署
深度学习架构设计
- 注意力机制（CBAM）
- 混合架构（CNN+LSTM）
- 时序数据处理
实际工程问题解决
- 类别不平衡（加权loss）
- 标签噪声（窗口过滤）
- 过拟合控制（Dropout、EarlyStopping）

⚠️ 缺失的"100k+层级"技能

要从"能写代码"到"平台架构师"，你需要补充：

可扩展性 🚀
- 这代码只能在单机跑，如何用Ray/Spark分布式处理TB级数据？
- 如何用MLflow/Weights&Biases追踪实验？
生产化 🏭
- 如何用TFServing/TorchServe部署模型？
- 如何监控模型性能漂移（Data Drift）？
- 如何做A/B测试？
系统设计 🏗️
- 如何设计实时推理pipeline（Kafka + 模型服务）？
- 如何处理冷启动问题？
- 如何设计特征存储（Feature Store）？
业务思维 💰
- 这个模型的ROI是什么？节省多少人力？
- 如何向非技术管理层解释技术选型？

🎯 你的行动计划

短期（1-2个月）

扩展这个项目：
- 加hyperparameter tuning（Optuna/Ray Tune）
- 添加模型可解释性（SHAP/GradCAM）
- 写详细的README + 技术博客
拿ML Professional证书：
- 对100k+ offer 有帮助但不是决定性的
- 更重要的是证书学习过程中掌握的生产级ML知识

中期（3-6个月）

构建端到端ML平台项目：
- 用Databricks/AWS Sagemaker搭建完整pipeline
- 包含：数据摄入 → 特征工程 → 训练 → 部署 → 监控
学习系统设计：
- 阅读"Designing Machine Learning Systems"（Chip Huyen）
- 练习ML system design面试题

关键建议

💡 从"AI工程师"到"平台架构师"的思维转变：

AI工程师思维	平台架构师思维
"这个模型accuracy 95%"	"这个模型在生产环境throughput多少？latency p99是多少？"
"我用LSTM做时序预测"	"为什么不用Prophet/N-BEATS？trade-off是什么？"
"模型训练完了"	"如何自动化再训练？如何监控性能退化？"
"我会用TensorFlow"	"我能设计一个支持多框架的serving层"

针对Frankfurt市场：

Databricks在欧洲有强劲增长，你的证书很契合
重点关注：Deutsche Börse、SAP、Lufthansa的AI平台岗位
突出你的双语能力（中英德）和跨文化团队经验

你当前的代码展示了solid ML engineering能力，价值约€70-80k。要冲击€100k+，需要展示platform/架构级的思考和经验。继续努力！🚀