TensorFlow Keras LSTM数据生成器形状不匹配问题求助

问题描述 #

调试TensorFlow Keras LSTM代码时，反复出现两类错误：

• TypeError: generator yielded an element of shape (36, 36, 147) where an element of shape (36, 147) was expected.
• ValueError: Input 0 of layer "sequential" is incompatible with the layer: expected shape=(None, 36, 147), found shape=(36, 147)

原始数据规格：

• X：形状为(418238, 36, 147)的numpy数组（样本数、时间步长、单步特征数）
• y：形状为(418238,)的numpy数组

错误根源 #

1. 变量定义顺序错误：batch_size在创建train_data/val_data之后才定义，导致Dataset的批量处理逻辑使用未定义变量。
2. 批量逻辑重复：数据生成器已按batch_size输出批量数据，后续又调用Dataset.batch()，导致形状多一维。
3. 模型配置错误：batch_input_shape使用未定义的n_batch_size变量；启用stateful=True的LSTM层未在后续层同步设置，导致状态传递失败。

解决方案 #

步骤1：调整变量定义顺序 #

提前定义batch_size等核心参数，避免未定义错误：

batch_size = 36
n_epochs = 10
steps_per_epoch = len(X_train) // batch_size 

步骤2：修正数据生成器与Dataset配置 #

推荐让生成器输出单个样本，由tf.data.Dataset统一处理批量，更符合框架设计习惯：

def data_generator(features, labels):
    data_size = len(features)
    while True:
        for i in range(data_size):
            yield features[i], labels[i]

# 训练数据集：单个样本→批量处理→重复迭代
train_data = tf.data.Dataset.from_generator(
    lambda: data_generator(X_train, y_train),
    output_types=(X.dtype, y.dtype),
    output_shapes=((36, 147), ())  # 单个样本形状：(时间步长, 特征数)，标签为标量
).batch(batch_size, drop_remainder=True).repeat()

# 验证数据集：无需重复迭代
val_data = tf.data.Dataset.from_generator(
    lambda: data_generator(X_val, y_val),
    output_types=(X.dtype, y.dtype),
    output_shapes=((36, 147), ())
).batch(batch_size, drop_remainder=True)

步骤3：修正LSTM模型配置 #

启用stateful=True时，所有LSTM层需同步设置该参数，并在每个epoch后重置状态：

# 自定义回调：epoch结束后重置LSTM状态
class ResetStatesCallback(tf.keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs=None):
        self.model.reset_states()

def build_model(hp):
    model = Sequential()
    # 第一层指定batch_input_shape和stateful=True
    model.add(LSTM(
        units=hp.Int('units1', min_value=50, max_value=750, step=50),
        batch_input_shape=(batch_size, X_train.shape[1], X_train.shape[2]),
        return_sequences=True,
        seed=1,
        stateful=True
    ))
    # 后续LSTM层必须同步设置stateful=True以传递状态
    model.add(LSTM(
        units=hp.Int('units2', min_value=50, max_value=500, step=50),
        return_sequences=True,
        stateful=True
    ))
    model.add(LSTM(
        units=hp.Int('units3', min_value=50, max_value=500, step=50),
        return_sequences=True,
        stateful=True
    ))
    model.add(LSTM(
        units=hp.Int('units4', min_value=50, max_value=500, step=50),
        stateful=True
    ))
    model.add(Dense(1))
    model.compile(
        optimizer=Adam(hp.Choice('learning_rate', values=[1e-1, 1e-2, 1e-3, 1e-4])),
        loss='mean_squared_error'
    )
    return model

步骤4：修正超参数搜索逻辑 #

在调参时加入状态重置回调，保证stateful模型的训练稳定性：

bayesian_opt_tuner = BayesianOptimization(
    build_model,
    seed=1,
    objective='val_loss',
    max_trials=25,
    executions_per_trial=1,
    directory='bayesian_optimization',
    project_name='keras_lstm',
    overwrite=True,
    max_consecutive_failed_trials=1
)

best_hps = bayesian_opt_tuner.search(
    train_data,
    epochs=n_epochs,
    steps_per_epoch=steps_per_epoch,
    validation_data=val_data,
    callbacks=[ResetStatesCallback()],
    verbose=1
)

best_hps = bayesian_opt_tuner.get_best_hyperparameters(num_trials=1)[0]

完整修正代码 #

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
from tensorflow.keras.optimizers import Adam
from kerastuner.tuners import BayesianOptimization
from sklearn.model_selection import train_test_split

# 假设X和y已提前定义，形状分别为(418238, 36, 147)和(418238,)
# X = ...
# y = ...

# 分割数据集
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2)

# 提前定义核心参数
batch_size = 36
n_epochs = 10
steps_per_epoch = len(X_train) // batch_size 

# 修正数据生成器：输出单个样本
def data_generator(features, labels):
    data_size = len(features)
    while True:
        for i in range(data_size):
            yield features[i], labels[i]

# 创建训练数据集
train_data = tf.data.Dataset.from_generator(
    lambda: data_generator(X_train, y_train),
    output_types=(X.dtype, y.dtype),
    output_shapes=((36, 147), ())
).batch(batch_size, drop_remainder=True).repeat()

# 创建验证数据集
val_data = tf.data.Dataset.from_generator(
    lambda: data_generator(X_val, y_val),
    output_types=(X.dtype, y.dtype),
    output_shapes=((36, 147), ())
).batch(batch_size, drop_remainder=True)

# 自定义状态重置回调
class ResetStatesCallback(tf.keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs=None):
        self.model.reset_states()

# 修正模型构建函数
def build_model(hp):
    model = Sequential()
    model.add(LSTM(
        units=hp.Int('units1', min_value=50, max_value=750, step=50),
        batch_input_shape=(batch_size, X_train.shape[1], X_train.shape[2]),
        return_sequences=True,
        seed=1,
        stateful=True
    ))
    model.add(LSTM(
        units=hp.Int('units2', min_value=50, max_value=500, step=50),
        return_sequences=True,
        stateful=True
    ))
    model.add(LSTM(
        units=hp.Int('units3', min_value=50, max_value=500, step=50),
        return_sequences=True,
        stateful=True
    ))
    model.add(LSTM(
        units=hp.Int('units4', min_value=50, max_value=500, step=50),
        stateful=True
    ))
    model.add(Dense(1))
    model.compile(
        optimizer=Adam(hp.Choice('learning_rate', values=[1e-1, 1e-2, 1e-3, 1e-4])),
        loss='mean_squared_error'
    )
    return model

# 贝叶斯优化调参
bayesian_opt_tuner = BayesianOptimization(
    build_model,
    seed=1,
    objective='val_loss',
    max_trials=25,
    executions_per_trial=1,
    directory='bayesian_optimization',
    project_name='keras_lstm',
    overwrite=True,
    max_consecutive_failed_trials=1
)

# 执行调参
best_hps = bayesian_opt_tuner.search(
    train_data,
    epochs=n_epochs,
    steps_per_epoch=steps_per_epoch,
    validation_data=val_data,
    callbacks=[ResetStatesCallback()],
    verbose=1
)

# 获取最优超参数
best_hps = bayesian_opt_tuner.get_best_hyperparameters(num_trials=1)[0]

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