Smart Fermenter 数据处理与模型训练详解

本文深入解析 Smart Fermenter 系统的数据处理流程和模型训练机制，涵盖从原始传感器数据到 LSTM/RNN 模型的全过程。

发酵数据特征

发酵过程涉及多种物理化学参数的实时监测：

特征名	说明	单位
`dm_air`	通气流量	L/min
`m_ls_opt_do`	溶解氧浓度	mg/L
`m_ph`	pH 值	-
`m_stirrer`	搅拌速率	RPM
`m_temp`	温度	°C
`dm_o2`	氧气流量	L/min
`dm_spump1-4`	四路补料泵流量	mL/h
`induction`	诱导状态	0/1
`od_600`	光密度(预测目标)	-

数据加载机制

Excel 数据源

def _load_data_from_excel(self):
    nums = self.train_nums if self.train else self.test_nums
    X, Y = [], []
    for n in nums:
        file_path = os.path.join(self.work_dir, folder, "data_clean.xlsx")
        df = pd.read_excel(file_path)
        df = df[self.all_cols]
        X.append(df[self.x_var].values.astype(np.float32))
        Y.append(df[self.y_var].values.astype(np.float32))
    return np.array(X), np.array(Y)

MySQL 数据源

支持从数据库动态加载指定批次的数据：

def _load_data_from_mysql(self):
    batch_ids = self.train_batch_ids if self.train else self.test_batch_ids
    X, Y = [], []
    for batch_id in batch_ids:
        x_data, y_data = utils.load_data_from_db_single(
            batch_id=batch_id,
            x_cols=self.x_var,
            y_cols=self.y_var
        )
        X.append(x_data)
        Y.append(y_data)
    return np.array(X), np.array(Y)

数据预处理流程

1. 标准化处理

def _compute_norm(self):
    flat_X = np.concatenate([np.asarray(x) for x in self.X], axis=0)
    self.x_mean = flat_X.mean(axis=0)
    self.x_std = flat_X.std(axis=0) + 1e-8  # 避免除零

2. 滑动窗口切片

发酵时序数据采用滑动窗口生成训练样本：

原始数据: [x0, x1, x2, x3, x4, x5, x6, x7, ...]
                    │
         ┌──────────┴──────────┐
         ▼                     ▼
    窗口1: [x0,x1,x2,...x19]  窗口2: [x5,x6,x7,...x24]
    标签: x19                  标签: x24

关键参数：

ws=20：窗口大小为 20 个时间步
stride=5：每 5 步滑动一次

def _preprocess(self, X, ws, stride):
    out = []
    for sample in X:
        sample = (sample - self.x_mean) / self.x_std  # 标准化
        # 生成滑动窗口序列
        seqs = [sample[i:i + ws] for i in range(0, len(sample) - ws + 1, stride)]
        out.extend(seqs)
    return np.array(out, dtype=np.float32)

3. 标签处理

标签取窗口最后一个时间步的 OD600 值：

def _preprocess_labels(self, Y, ws, stride):
    out = []
    for y in Y:
        y = (y - self.y_mean) / self.y_std
        # 窗口最后一个时间步的值作为标签
        seqs = [y[i + ws - 1] for i in range(0, len(y) - ws + 1, stride)]
        out.extend(seqs)
    return np.array(out).reshape(-1, 1)

模型架构

LSTM 模型

输入 (batch, seq_len, features)
    │
    ▼
pre_fc: Linear(features → hidden_dim)
    │
    ▼ ReLU
    │
    ▼
LSTM: (hidden_dim → hidden_dim, num_layers=1)
    │
    ▼ ReLU
    │
    ▼
fc: Linear(hidden_dim → 1)
    │
    ▼
输出 (batch, seq_len, 1)

class LSTMPredictor(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, n_layers):
        super().__init__()
        self.pre_fc = nn.Linear(input_dim, hidden_dim)
        self.lstm = nn.LSTM(
            input_size=hidden_dim, 
            hidden_size=hidden_dim, 
            num_layers=n_layers, 
            batch_first=True
        )
        self.fc = nn.Linear(hidden_dim, output_dim)
        self.relu = nn.ReLU()
        
    def forward(self, x, hidden):
        x = self.pre_fc(x)
        x = self.relu(x)
        lstm_out, hidden = self.lstm(x, hidden)
        lstm_out = self.relu(lstm_out)
        out = self.fc(lstm_out)
        return out, hidden

RNN 模型

RNN 结构与 LSTM 类似，但使用基础的 RNN 单元：

class RNNpredictor(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, n_layers):
        super().__init__()
        self.pre_fc = nn.Linear(input_dim, hidden_dim)
        self.rnn = nn.RNN(
            input_size=hidden_dim, 
            hidden_size=hidden_dim, 
            num_layers=n_layers, 
            batch_first=True
        )
        self.fc = nn.Linear(hidden_dim, output_dim)
        self.relu = nn.ReLU()

训练流程

主训练循环

def train(epoch, model, optimiser, trainloader):
    model.train()
    for batch_idx, (input_data, label) in enumerate(trainloader):
        batch_size = input_data.size(0)
        input_data, label = input_data.to(device), label.to(device)
        
        # 初始化 LSTM 隐藏状态
        h = model.init_hidden(batch_size)
        h = tuple([e.data for e in h])
        
        # 前向传播
        output, h = model(input_data.float(), h)
        
        # 计算损失 (MSE)
        loss = mse(output, label.float())
        err = torch.sqrt(mse(output, label.float()))  # RMSE
        
        # 反向传播
        loss.backward()
        optimiser.step()
        optimiser.zero_grad()

超参数配置

参数	默认值	说明
`--batch_size`	256	批大小
`--lr`	0.02	学习率
`--hidden_dim`	49	隐藏层维度
`--num_layers`	1	LSTM 层数
`--num_epochs`	100	训练轮数

训练启动命令

# LSTM 模型训练
python main.py train \
    --run_name lstm_exp_01 \
    --model lstm \
    --batch_size 256 \
    --lr 0.02 \
    --hidden_dim 49 \
    --num_epochs 100 \
    --dataset ./Data5

# 启用 wandb 可视化
python main.py train --run_name experiment_01 --wandb

# 使用 MySQL 数据源
python main.py train \
    --data_source mysql \
    --train_batch_ids "B001,B002,B003" \
    --test_batch_ids "B004"

模型持久化

训练过程中自动保存最佳模型：

1
2
3

if rmse < best:
    utils.save_weights(model, epoch, "logs/" + args.run_name + "/weights_best.tar")
    best = rmse

保存内容包含：

模型权重
训练轮次
优化器状态

数据标准化文件

训练完成后自动生成标准化参数文件：

// norm_file.json
{
    "y_mean": 12.5,
    "y_std": 3.2,
    "X_mean": [0.1, 0.5, ...],
    "X_std": [0.05, 0.1, ...]
}

scaler_new_data.save - 用于推理时对新数据进行标准化

下一步

下篇博客将介绍遗传算法参数优化的实现原理，如何通过进化算法优化发酵过程中的控制参数。

相关阅读：