#!/usr/bin/env python # -*- coding:utf-8 -*- # @FileName :tf_transformer.py # @Time :2025/5/08 14:03 # @Author :David # @Company: shenyang JY from tensorflow.keras.layers import Input, Dense, LSTM, concatenate, Conv1D, Conv2D, MaxPooling1D, Reshape, Flatten, LayerNormalization, Dropout from tensorflow.keras.models import Model, load_model from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping, TensorBoard, ReduceLROnPlateau from tensorflow.keras import optimizers, regularizers from models_processing.model_tf.losses import region_loss import numpy as np from common.database_dml_koi import * from models_processing.model_tf.settings import set_deterministic from threading import Lock import argparse model_lock = Lock() set_deterministic(42) class TransformerHandler(object): def __init__(self, logger, args): self.logger = logger self.opt = argparse.Namespace(**args) self.model = None self.model_params = None def get_model(self, args): """ 单例模式+线程锁,防止在异步加载时引发线程安全 """ try: with model_lock: loss = region_loss(self.opt) self.model, self.model_params = get_keras_model_from_mongo(args, {type(loss).__name__: loss}) except Exception as e: self.logger.info("加载模型权重失败:{}".format(e.args)) @staticmethod def get_keras_model(opt, time_series=1, lstm_type=1): loss = region_loss(opt) l1_reg = regularizers.l1(opt.Model['lambda_value_1']) l2_reg = regularizers.l2(opt.Model['lambda_value_2']) nwp_input = Input(shape=(opt.Model['time_step']*time_series, opt.Model['input_size']), name='nwp') con1 = Conv1D(filters=64, kernel_size=5, strides=1, padding='valid', activation='relu', kernel_regularizer=l2_reg)(nwp_input) con1_p = MaxPooling1D(pool_size=5, strides=1, padding='valid', data_format='channels_last')(con1) nwp_lstm = LSTM(units=opt.Model['hidden_size'], return_sequences=False, kernel_regularizer=l2_reg)(con1_p) if lstm_type == 2: output = Dense(opt.Model['time_step'], name='cdq_output')(nwp_lstm) else: output = Dense(opt.Model['time_step']*time_series, name='cdq_output')(nwp_lstm) model = Model(nwp_input, output) adam = optimizers.Adam(learning_rate=0.001, beta_1=0.9, beta_2=0.999, epsilon=1e-7, amsgrad=True) model.compile(loss=loss, optimizer=adam) return model @staticmethod def get_transformer_model(opt, time_series=1): time_steps = 48 input_features = 21 output_steps = 16 hidden_size = opt.Model.get('hidden_size', 64) num_heads = opt.Model.get('num_heads', 4) ff_dim = opt.Model.get('ff_dim', 128) l2_reg = regularizers.l2(opt.Model.get('lambda_value_2', 0.0)) nwp_input = Input(shape=(opt.Model['time_step'] * time_series, opt.Model['input_size']), name='nwp') # 输入嵌入 x = Conv1D(hidden_size, 1, kernel_regularizer=l2_reg)(nwp_input) # Transformer编码器层 for _ in range(opt.Model.get('num_layers', 2)): # 多头自注意力 x = tf.keras.layers.MultiHeadAttention( num_heads=num_heads, key_dim=hidden_size, kernel_regularizer=l2_reg )(x, x) x = LayerNormalization()(x) x = tf.keras.layers.Dropout(0.1)(x) # 前馈网络 x = tf.keras.layers.Dense(ff_dim, activation='relu', kernel_regularizer=l2_reg)(x) x = tf.keras.layers.Dense(hidden_size, kernel_regularizer=l2_reg)(x) x = LayerNormalization()(x) x = tf.keras.layers.Dropout(0.1)(x) # 提取中间时间步 start_idx = (time_steps - output_steps) // 2 x = x[:, start_idx:start_idx + output_steps, :] # 输出层 output = Dense(output_steps, name='cdq_output')(x[:, -1, :]) # 或者使用所有时间步 model = Model(nwp_input, output) # 编译模型 adam = optimizers.Adam( learning_rate=opt.Model.get('learning_rate', 0.001), beta_1=0.9, beta_2=0.999, epsilon=1e-7, amsgrad=True ) loss = region_loss(opt) model.compile(loss=loss, optimizer=adam) return model def train_init(self): try: # 进行加强训练,支持修模 loss = region_loss(self.opt) base_train_model, self.model_params = get_keras_model_from_mongo(vars(self.opt), {type(loss).__name__: loss}) base_train_model.summary() self.logger.info("已加载加强训练基础模型") return base_train_model except Exception as e: self.logger.info("加载加强训练模型权重失败:{}".format(e.args)) return False def training(self, model, train_and_valid_data): model.summary() train_x, train_y, valid_x, valid_y = train_and_valid_data early_stop = EarlyStopping(monitor='val_loss', patience=self.opt.Model['patience'], mode='auto') history = model.fit(train_x, train_y, batch_size=self.opt.Model['batch_size'], epochs=self.opt.Model['epoch'], verbose=2, validation_data=(valid_x, valid_y), callbacks=[early_stop], shuffle=False) loss = np.round(history.history['loss'], decimals=5) val_loss = np.round(history.history['val_loss'], decimals=5) self.logger.info("-----模型训练经过{}轮迭代-----".format(len(loss))) self.logger.info("训练集损失函数为:{}".format(loss)) self.logger.info("验证集损失函数为:{}".format(val_loss)) return model def predict(self, test_x, batch_size=1): result = self.model.predict(test_x, batch_size=batch_size) self.logger.info("执行预测方法") return result if __name__ == "__main__": run_code = 0