基于Keras构建带预测值反馈的LSTM时序分类模型训练问题

核心思路 #

你需要的是自回归式LSTM模型——训练和推理阶段都用上一步的预测值替代无法实时获取的y_{t-1}，作为当前时间步的输入特征。Keras标准fit方法无法直接处理这种动态反馈逻辑，需要通过自定义训练循环实现。

训练阶段实现方案 #

自定义训练循环（推荐） #

手动编写训练流程，逐时间步记录预测值并反馈到下一个时间步的输入中，完全模拟真实推理时的逻辑。

代码示例

import tensorflow as tf
import numpy as np

# 初始化模型（保留你的基础结构）
model = tf.keras.models.Sequential()
model.add(tf.keras.layers.LSTM(40, activation='tanh', return_sequences=False,
                               input_shape=(x2.shape[1], x2.shape[2]),
                               kernel_initializer=tf.keras.initializers.GlorotUniform(seed=42)))
model.add(tf.keras.layers.Dense(1, activation='sigmoid',
                                kernel_initializer=tf.keras.initializers.GlorotUniform(seed=42)))

# 定义损失、优化器和评估指标
loss_fn = tf.keras.losses.BinaryCrossentropy(from_logits=False)
optimizer = tf.keras.optimizers.Adam()
train_metric = tf.keras.metrics.PrecisionAtRecall(0.85)
val_metric = tf.keras.metrics.PrecisionAtRecall(0.85)

epochs = 160
batch_size = 50
# 假设y_{t-1}是输入特征中的第k列，替换时需对应此索引
k = 0

for epoch in range(epochs):
    print(f"Epoch {epoch+1}/{epochs}")
    epoch_loss = 0.0
    train_metric.reset_state()
    
    # 按批次处理训练数据
    for batch_idx in range(0, len(x2), batch_size):
        x_batch = x2[batch_idx:batch_idx+batch_size].copy()
        y_batch = y2[batch_idx:batch_idx+batch_size]
        
        # 初始时间步用真实y值填充y_{t-1}特征位
        y_prev = y_batch[:, 0:1]
        x_batch[:, 0, k] = tf.squeeze(y_prev, axis=1)
        
        with tf.GradientTape() as tape:
            batch_loss = 0.0
            # 逐时间步迭代，传递预测值
            for t in range(1, x_batch.shape[1]):
                # 输入到当前时间步的序列，预测y_t
                current_input = x_batch[:, :t+1, :]
                y_pred = model(current_input, training=True)
                
                # 计算当前时间步损失并累加
                step_loss = loss_fn(y_batch[:, t:t+1], y_pred)
                batch_loss += step_loss
                train_metric.update_state(y_batch[:, t:t+1], y_pred)
                
                # 将预测值作为下一时间步的y_{t-1}输入
                x_batch[:, t, k] = tf.squeeze(y_pred, axis=1)
        
        # 更新模型权重
        gradients = tape.gradient(batch_loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))
        
        epoch_loss += batch_loss.numpy()
        avg_batch_loss = batch_loss.numpy() / (x_batch.shape[1]-1)
        print(f"Batch {batch_idx//batch_size +1}: Loss={avg_batch_loss:.4f}, Precision@Recall0.85={train_metric.result().numpy():.4f}")
    
    # 验证阶段（同样采用反馈逻辑）
    val_loss = 0.0
    val_metric.reset_state()
    for batch_idx in range(0, len(x2_v), batch_size):
        x_val_batch = x2_v[batch_idx:batch_idx+batch_size].copy()
        y_val_batch = y2_v[batch_idx:batch_idx+batch_size]
        
        y_prev_val = y_val_batch[:, 0:1]
        x_val_batch[:, 0, k] = tf.squeeze(y_prev_val, axis=1)
        
        for t in range(1, x_val_batch.shape[1]):
            current_input_val = x_val_batch[:, :t+1, :]
            y_pred_val = model(current_input_val, training=False)
            
            val_loss += loss_fn(y_val_batch[:, t:t+1], y_pred_val).numpy()
            val_metric.update_state(y_val_batch[:, t:t+1], y_pred_val)
            
            x_val_batch[:, t, k] = tf.squeeze(y_pred_val, axis=1)
    
    avg_epoch_loss = epoch_loss / (len(x2) * (x2.shape[1]-1))
    avg_val_loss = val_loss / (len(x2_v) * (x2_v.shape[1]-1))
    print(f"Epoch Summary: Train Loss={avg_epoch_loss:.4f}, Train Precision@Recall0.85={train_metric.result().numpy():.4f}")
    print(f"Validation Loss={avg_val_loss:.4f}, Validation Precision@Recall0.85={val_metric.result().numpy():.4f}\n")

替代方案：预生成带反馈的训练数据 #

如果不想写自定义循环，可以提前迭代生成训练数据：先让模型用真实y_{t-1}训练几轮，再用模型预测值替换训练数据中的y_{t-1}列，生成带反馈的序列后用标准fit训练。但这种方法效率较低，因为每轮模型更新后都需要重新生成数据。

推理阶段实现 #

推理时按时间步循环，将上一步预测值注入当前输入：

def predict_sequence(model, x_input, y_prev_idx):
    x = x_input.copy()
    predictions = []
    
    # 初始时间步预测
    y_prev = model(x[:, 0:1, :], training=False)
    predictions.append(y_prev.numpy())
    
    # 逐时间步迭代预测
    for t in range(1, x.shape[1]):
        x[:, t, y_prev_idx] = tf.squeeze(y_prev, axis=1)
        y_prev = model(x[:, :t+1, :], training=False)
        predictions.append(y_prev.numpy())
    
    return np.concatenate(predictions, axis=1)

关键注意事项 #

• 务必确认y_{t-1}在输入特征中的索引位置，避免替换错误特征列。
• 训练初期可以先用真实y_{t-1}训练5-10轮，让模型先学习基础模式，再切换到预测值反馈逻辑，避免初期误差累积导致收敛困难。
• 监控验证集指标，若出现误差持续累积，可调整LSTM单元数量、学习率或增加正则化层（如Dropout）。

内容的提问来源于stack exchange，提问作者Virginie Gautier