Transformer-LSTM混合模型在时序回归中的完整流程研究

引言与背景

深度学习中的长期依赖建模一直是时序预测的核心问题。长短期记忆网络（LSTM）作为一种循环神经网络，因其特殊的门控结构能够有效捕捉序列的历史信息，并在时序预测中表现出色；然而LSTM的序列计算方式限制了其并行处理能力和对全局上下文信息的感知。相比之下，Transformer模型通过自注意力机制可以并行处理数据并捕捉全局依赖，对复杂时序数据（例如季节性、周期性强的数据）具有天然优势。因此，将两者结合可以兼顾短期局部模式和长期全局关联：LSTM负责提取连续时序中的细节信息，Transformer补充全局依赖权重，从而提高预测精度和泛化性能。

此类混合模型适用于多种时序回归场景，例如金融市场（股票、指数预测）、能耗预测、工业过程监控等。已有研究表明，在复杂工程系统如地下钻井进度预测、可再生能源管理等领域，LSTM-Transformer混合模型往往比单一模型具有更高精度和实时性。例如，Cao等人的研究指出，将LSTM的记忆单元与Transformer的自注意力机制相结合，可以在多个任务上获得明显提升。基于这些背景，本研究将详细介绍Transformer-LSTM混合模型在时序回归任务中的从头到尾的完整流程，包括数据处理、模型设计、训练优化、评估、部署等环节。

数据准备

对时序回归任务而言，数据准备是至关重要的一步。合成数据常用于验证模型框架的有效性，例如生成带有趋势和周期的正弦波序列或随机漫步序列进行实验；真实数据则需要进行清洗与预处理，包括去除缺失值、异常值处理，以及归一化或标准化等操作，使数据尺度一致。通常采用滑动窗口的方法将时序数据转为监督学习格式：设定一个固定长度的窗口，将窗口内的历史$W$步数据作为输入特征，用于预测接下来$H$步或1步的目标值。例如，图1示例了两个变量的多变量时序数据：使用过去若干时刻的两个特征（Var1、Var2）同时作为模型输入，预测未来值。下图即为示例，其中蓝色线为Var1随时间的变化，绿色线为Var2的变化，两者同时输入模型进行联合预测。

图1. 示例多变量时序数据（蓝色Var1、绿色Var2）随着时间的变化。多特征时序常常需要对每个时间步的多维观测进行组合，作为模型输入。

在具体实现中，滑动窗口的数据集构造可以使用numpy或TensorFlow等工具实现。下面给出一个示例代码，演示如何生成合成时序并构建滑动窗口数据集：

import numpy as np
import tensorflow as tf

# 生成合成时序数据（两个特征）
t = np.arange(0, 1000, 0.5)
series1 = np.sin(0.02 * t) + 0.5 * np.random.randn(len(t))  # 特征1：正弦 + 噪声
series2 = np.cos(0.01 * t) + 0.5 * np.random.randn(len(t))  # 特征2：余弦 + 噪声
data = np.stack([series1, series2], axis=1)  # 形状 (len(t), 2)

# 数据归一化（保持尺度一致）
data_mean = data.mean(axis=0)
data_std = data.std(axis=0)
data = (data - data_mean) / data_std

# 滑动窗口参数
input_width = 20   # 输入窗口长度
label_width = 1    # 预测步长
shift = 1          # 步幅为1，即下一个时刻

# 使用 tf.keras.utils.timeseries_dataset_from_array 构造数据集
dataset = tf.keras.utils.timeseries_dataset_from_array(
    data=data,
    targets=data[:, 0],       # 假设预测第一个特征（可根据任务不同选择）
    sequence_length=input_width,
    sequence_stride=1,
    sampling_rate=1,
    batch_size=32,
    shuffle=True
)
# 划分训练/验证集
total_samples = len(data) - input_width - label_width + 1
train_size = int(total_samples * 0.8)
train_dataset = dataset.take(train_size // 32)
val_dataset = dataset.skip(train_size // 32)

上述代码中，我们生成了两个特征的合成时序数据，并使用timeseries_dataset_from_array将其转换为滑动窗口数据集，每个样本由过去input_width步的数据（形状为(batch, input_width, 2)）映射到下一步第1个特征的值。对真实数据，同样需要根据时间戳做排序，然后进行类似的窗口划分。此外，可选地加入时间特征（如小时、星期等周期信息）或进行差分处理等特征工程，以提高模型对周期性和趋势的敏感度。

模型结构详解

本研究所用的混合模型结构主要由两部分组成：LSTM层用于捕捉短期序列依赖，Transformer编码器（自注意力层）用于建模全局依赖。在该设计中，输入首先经由若干层LSTM编码为时序特征表示，然后送入Transformer模块进行全局上下文编码，最后通过全连接层输出回归结果。LSTM层的记忆单元能够保留历史信息，而Transformer的自注意力机制可以对序列中不同位置进行加权，互补了LSTM对长序列建模的局限。这种串联结构有效结合了两者的优势，例如LSTM擅长捕捉序列模式，Transformer擅长抓取宏观信息，有利于提高预测的准确性。

在TensorFlow/Keras中，可以使用tf.keras.layers.LSTM和tf.keras.layers.MultiHeadAttention来实现上述混合结构。下面给出一个示例模型结构的代码片段：

from tensorflow.keras import layers, Model

input_width = 20  # 输入窗口长度
n_features = 2    # 特征数量
inputs = layers.Input(shape=(input_width, n_features))

# LSTM编码层
x = layers.LSTM(64, return_sequences=True)(inputs)  # 输出序列用于自注意力
# 可选添加dropout防止过拟合
x = layers.Dropout(0.2)(x)

# Transformer自注意力模块（单层示例）
# 1. 多头自注意力
attn_output = layers.MultiHeadAttention(num_heads=4, key_dim=16)(x, x)
# 2. 残差连接与Layer Norm
x = layers.Add()([x, attn_output])
x = layers.LayerNormalization()(x)
# 3. 前馈网络
ff = layers.Dense(64, activation='relu')(x)
ff = layers.Dense(x.shape[-1])(ff)
x = layers.Add()([x, ff])
x = layers.LayerNormalization()(x)

# 聚合输出（取最后一个时间步）并回归输出
x = layers.Flatten()(x)
outputs = layers.Dense(1)(x)  # 回归预测值
model = Model(inputs, outputs)
model.summary()

在这个示例中，先构建了一个LSTM层（return_sequences=True），输出为与输入长度相同的序列，以便后续进行自注意力计算。然后通过多头注意力层（MultiHeadAttention）实现Transformer编码，添加了残差连接和LayerNormalization保持稳定。最后将序列输出展平，接全连接层预测目标值。该结构图如图2所示：

图2. LSTM-Transformer混合模型结构示意图（输入序列->LSTM编码->Transformer编码->全连接输出）。

模型结构还可以有多种变体，例如将Transformer模块置于LSTM之前，或并行设计多个分支后融合输出，也可在后续加入额外的LSTM或卷积层强化特征提取。但无论结构如何变化，核心思想都是让LSTM和Transformer协同工作，补充彼此的长短期依赖建模能力。

训练与调优

训练阶段首先需要设置损失函数和优化器。对于回归任务，常用均方误差（MSE）作为损失函数，优化器可选用Adam或AdamW等高级优化算法。Adam优化器通过对梯度一阶、二阶矩估计自适应调整学习率，通常能带来稳定快速的收敛。例如，编译模型时可写：

model.compile(loss='mse', optimizer='adam', metrics=['mae'])

超参数设置方面，主要包括学习率、批量大小、LSTM单元数、Transformer头数、注意力维度、层数等。参考经验，常见的超参数有：嵌入（隐藏）维度、注意力头数、Transformer层数、学习率、批量大小等。例如，可以先尝试学习率$10^{-3}$、批量大小32、LSTM隐藏层64、注意力头数4、Transformer深度12层。然后根据验证效果调整：过大的学习率可能使训练不稳定，过小则收敛缓慢；批量大小受显存限制，一般32128之间为佳。

正则化与早停也是重要手段。可在LSTM和全连接层之间加Dropout层来减少过拟合；例如在上述代码中已经加入了Dropout(0.2)。另外，可以使用tf.keras.callbacks.EarlyStopping监控验证集损失，当损失在若干轮内不再下降时提前终止训练，以避免过拟合。可选的学习率调度策略包括ReduceLROnPlateau（当验证性能停滞时衰减学习率）或Transformer常用的学习率预热（warmup）后衰减策略。

示例训练代码，其中包含超参数、早停等设置：

from tensorflow.keras.callbacks import EarlyStopping

# 假设已有 train_dataset, val_dataset
early_stop = EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True)
history = model.fit(
    train_dataset,
    validation_data=val_dataset,
    epochs=50,
    callbacks=[early_stop]
)

通过fit训练，模型将自动在训练集和验证集上更新权重，并保存训练过程中的损失值曲线用于后续分析。以示例代码为基准，可以视实际情况改变超参数，并通过多次实验观察模型性能变化，采用网格搜索、随机搜索等方法调优。

性能评估

训练结束后，通过多种指标和可视化来评估模型性能。常用的回归指标包括平均绝对误差（MAE）和均方根误差（RMSE），它们分别衡量预测值与真实值偏差的平均大小和均方根大小。通常，指标值越低代表预测越准确。我们可以使用scikit-learn等库来计算这些指标：

import numpy as np
from sklearn.metrics import mean_absolute_error, mean_squared_error

# 假设 test_X, test_y 为测试集输入输出
y_pred = model.predict(test_X)
mae = mean_absolute_error(test_y, y_pred)
rmse = np.sqrt(mean_squared_error(test_y, y_pred))
print(f"测试集 MAE: {mae:.4f}, RMSE: {rmse:.4f}")

除了数值指标，还应绘制训练过程中的损失曲线和预测结果对比图。训练曲线（损失随迭代变化）可以反映是否出现了过拟合或欠拟合；预测对比图（将模型输出与真实序列叠加或做散点图）则直观展示预测精度。图3中展示了模型在测试集上的预测趋势示例，我们可以将预测值与真实值一起绘制，观察二者吻合程度。

图3. 模型预测效果示例图（图中显示某时序数据的真实值与预测值变化曲线）。

在实际评估中，还应与基线模型做对比，例如纯LSTM模型、传统ARIMA模型或其它简单方法，比较MAE/RMSE等指标差异。如果混合模型显著优于基线，则说明Transformer的引入有效改善了预测效果。已有研究表明，与单一网络相比，LSTM-Transformer组合通常能取得更好的效果。最后，可以对预测误差进行误差分析：检查哪些时刻误差较大，是异常值引起还是模型欠拟合等，为后续改进提供参考。

部署实践

训练完成后，需要将模型用于生产环境的预测服务中。首先将训练好的模型导出：对于Keras模型，可使用model.save将其保存为TensorFlow SavedModel格式，以便于TensorFlow Serving部署。例如：

# 导出模型用于部署
model.save('saved_model/my_model', save_format='tf')

上面代码会在saved_model/my_model/目录生成SavedModel所需的assets/, variables/, saved_model.pb等文件。此后，可以选择两种常见的服务方式：

• TensorFlow Serving：使用官方的TFS服务器，将导出的模型加载为REST或gRPC接口服务。可以通过Docker快速部署，并通过HTTP请求实现高并发预测。例如执行命令 tensorflow_model_server --model_base_path=saved_model --rest_api_port=8501 即可启动服务，随后客户端可发送POST请求获取预测结果。
• 自定义REST API：使用Flask、FastAPI等Python框架封装模型推理接口。示例代码如下：

from flask import Flask, request, jsonify
import tensorflow as tf

app = Flask(__name__)
# 加载SavedModel
loaded_model = tf.keras.models.load_model('saved_model/my_model')

@app.route('/predict', methods=['POST'])
def predict():
    data = request.json['input_data']  # 假设客户端以JSON格式发送输入数据
    # 转换为numpy数组并进行预测
    x = np.array(data)
    pred = loaded_model.predict(x)
    return jsonify(pred.tolist())

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

部署后需要进行推理效率测试：衡量模型在目标硬件上的响应时间和吞吐量。这可以通过测试固定大小的输入批次预测耗时来完成（可用time.time()或timeit测量），或者使用加载监控（如TFS自带的监控端点）评估并发性能。对于延迟敏感的场景，还可考虑模型量化或TensorRT优化来加速推理。

总结与扩展建议

本研究详细梳理了Transformer-LSTM混合模型在时序回归任务中的从数据准备到部署的全流程。此类混合模型充分结合了LSTM的记忆能力和Transformer的全局建模能力，已在多个领域展现出优越性能。总结如下：

• 优势：混合结构能同时捕捉局部和全局时序特征，提高了复杂数据上的预测精度；Transformer的并行化计算也加快了训练速度。
• 改进方向：可尝试更深的网络（多层LSTM/Transformer）、不同的融合方式（如并行分支）、更丰富的输入特征（如外部时间标签、统计量），或结合卷积层处理局部模式等。
• 迁移性：由于时序模式在金融、气象、医疗、工业等领域普遍存在，此模型结构具有较好的迁移潜力。在具体应用时可根据领域特点调整模型架构和超参数。
• 未来研究：可以探索在线学习和增量更新（使模型随新数据持续优化）、注意力机制改进（如引入稀疏注意力、层次化注意力）、结合图神经网络处理时空关系等方向，不断提升模型的适用性和性能。

总之，Transformer-LSTM混合模型为时序回归提供了一种高效而灵活的方案，本文的完整流程与经验分享可为相关场景的建模与实践提供参考和借鉴。

具体代码流程

1. 合成与真实数据生成及滑窗处理
2. 数据归一化与训练/验证集切分
3. 混合模型定义：串联 LSTM 与 Transformer 编码器
4. 训练策略：MSE 损失、Adam 优化、EarlyStopping、学习率调度
5. Keras Tuner 自动化超参数搜索
6. 性能评估：MAE、RMSE 计算与可视化
7. 模型导出与 Flask REST API 部署

以下代码一气呵成，可直接运行并根据实际场景替换数据部分。

---

环境依赖安装

pip install tensorflow keras-tuner scikit-learn matplotlib flask

---

1. 数据生成与预处理

import numpy as np
import tensorflow as tf
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split

# 1.1 合成时序数据
np.random.seed(42)
t = np.arange(0, 1000, 0.5)
series1 = np.sin(0.02 * t) + 0.5 * np.random.randn(len(t))
series2 = np.cos(0.01 * t) + 0.5 * np.random.randn(len(t))
data = np.stack([series1, series2], axis=1)  # (2000, 2)

# 1.2 归一化
scaler_X = MinMaxScaler(); scaler_y = MinMaxScaler()
data_norm = scaler_X.fit_transform(data)  # :contentReference[oaicite:0]{index=0}
target = series1.reshape(-1, 1)
target_norm = scaler_y.fit_transform(target)  # :contentReference[oaicite:1]{index=1}

# 1.3 滑动窗口转换
input_width = 20
ds = tf.keras.utils.timeseries_dataset_from_array(
    data=data_norm,
    targets=target_norm,
    sequence_length=input_width,
    sequence_stride=1,
    batch_size=32,
    shuffle=True
)  # :contentReference[oaicite:2]{index=2}

# 1.4 划分训练/验证集
total = len(data_norm) - input_width + 1
train_size = int(total * 0.8)
train_ds = ds.take(train_size // 32)
val_ds = ds.skip(train_size // 32)

---

2. 模型构建

from tensorflow.keras import layers, Model

def build_transformer_lstm(input_shape, lstm_units=64, heads=4, dim=32, num_layers=2, dropout=0.2):
    inputs = layers.Input(shape=input_shape)
    # 2.1 LSTM 编码
    x = layers.LSTM(lstm_units, return_sequences=True)(inputs)  # :contentReference[oaicite:3]{index=3}
    x = layers.Dropout(dropout)(x)

    # 2.2 Transformer 编码器堆叠
    for _ in range(num_layers):
        attn = layers.MultiHeadAttention(num_heads=heads, key_dim=dim)(x, x)  # :contentReference[oaicite:4]{index=4}
        x = layers.Add()([x, attn])
        x = layers.LayerNormalization()(x)  # :contentReference[oaicite:5]{index=5}
        # 前馈
        ffn = layers.Dense(lstm_units, activation='relu')(x)
        ffn = layers.Dense(x.shape[-1])(ffn)
        x = layers.Add()([x, ffn])
        x = layers.LayerNormalization()(x)

    # 2.3 回归输出
    x = layers.GlobalAveragePooling1D()(x)
    x = layers.Dense(64, activation='relu')(x)
    x = layers.Dropout(dropout)(x)
    outputs = layers.Dense(1)(x)
    return Model(inputs, outputs)

input_shape = (input_width, data.shape[-1])
model = build_transformer_lstm(input_shape)
model.compile(optimizer='adam', loss='mse', metrics=['mae'])  # :contentReference[oaicite:6]{index=6}
model.summary()

---

3. 训练策略与超参数调优

from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau
import keras_tuner as kt

# 3.1 回调
early_stop = EarlyStopping(monitor='val_loss', patience=5, restore_best_weights=True)  # :contentReference[oaicite:7]{index=7}
reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.5, patience=3)  # :contentReference[oaicite:8]{index=8}

# 3.2 Keras Tuner 自动调参
def tuner_model(hp):
    hp_lstm = hp.Int('lstm_units', 32, 128, step=32)
    hp_heads = hp.Int('heads', 2, 8, step=2)
    hp_layers = hp.Int('layers', 1, 3)
    hp_lr = hp.Float('lr', 1e-4, 1e-2, sampling='log')

    m = build_transformer_lstm(input_shape, lstm_units=hp_lstm, heads=hp_heads,
                               num_layers=hp_layers)
    m.compile(
        optimizer=tf.keras.optimizers.Adam(hp_lr),
        loss='mse', metrics=['mae']
    )  # :contentReference[oaicite:9]{index=9}
    return m

tuner = kt.Hyperband(
    tuner_model,
    objective='val_loss',
    max_epochs=20,
    factor=3,
    directory='tuner_dir',
    project_name='ts_transformer_lstm'
)
tuner.search(train_ds, validation_data=val_ds, callbacks=[early_stop, reduce_lr])
best_hp = tuner.get_best_hyperparameters(1)[0]
print(best_hp.values)

---

4. 最终训练与评估

# 4.1 根据最佳超参构建模型
opt_model = build_transformer_lstm(
    input_shape,
    lstm_units=best_hp.get('lstm_units'),
    heads=best_hp.get('heads'),
    num_layers=best_hp.get('layers')
)
opt_model.compile(
    optimizer=tf.keras.optimizers.Adam(best_hp.get('lr')),
    loss='mse', metrics=['mae']
)

# 4.2 训练
history = opt_model.fit(
    train_ds,
    validation_data=val_ds,
    epochs=50,
    callbacks=[early_stop, reduce_lr]
)

# 4.3 性能评估
import matplotlib.pyplot as plt
from sklearn.metrics import mean_absolute_error, mean_squared_error

# 预测
X_test, y_test = next(iter(val_ds))
y_pred = opt_model.predict(X_test)
mae = mean_absolute_error(y_test, y_pred)
rmse = np.sqrt(mean_squared_error(y_test, y_pred))
print(f"MAE: {mae:.4f}, RMSE: {rmse:.4f}")  # :contentReference[oaicite:10]{index=10}

# 4.4 可视化
plt.figure(figsize=(10,4))
plt.plot(y_test[:100], label='True')
plt.plot(y_pred[:100], label='Pred')
plt.legend(); plt.title('Prediction vs True (first 100 samples)')
plt.show()

---

5. 模型导出与部署

# 5.1 保存模型
opt_model.save('saved_model/ts_transformer_lstm')

# 5.2 Flask REST API 示例
from flask import Flask, request, jsonify
import tensorflow as tf

app = Flask(__name__)
m = tf.keras.models.load_model('saved_model/ts_transformer_lstm')

@app.route('/predict', methods=['POST'])
def predict():
    data = np.array(request.json['instances'])
    preds = m.predict(data).flatten().tolist()
    return jsonify({'predictions': preds})

if __name__=='__main__':
    app.run(host='0.0.0.0', port=5000)

---

通过上面给出的完整代码，即可在本地复现Transformer‑LSTM 混合模型的全流程，从数据处理到部署，涵盖早停、学习率调整及自动超参搜索，帮助你在时序回归场景中获得更优性能。