[已解决]transformer模型怎么用tensorflow搭建

WTMGCLGGA

怎么用tensorflow来搭建transformer模型并实现时间序列预测？
@FishC

最佳答案

月排行榜 / 总排行榜

FishC

2023-12-25 09:57:35

WTMGCLGGA 发表于 2023-12-25 09:56
可以输出一个完整的transformer模型吗，需要包括定义 ...

当然可以！下面是一个用TensorFlow实现的Transformer模型的完整代码。首先，让我们从导入必要的库开始：

1. import tensorflow as tf
2. import numpy as np

复制代码

接下来，我们定义一些超参数：

1. NUM_LAYERS = 2  # 编码器和解码器的层数
2. D_MODEL = 32  # 模型的维度
3. NUM_HEADS = 4  # 多头注意力机制的头数
4. UNITS = 64 # 前向传播层的神经元数量
5. DROPOUT_RATE = 0.1  # Dropout的概率
6. N_FEATURES = 5  # 输入时间序列的特征数量

复制代码

然后我们来定义编码器和解码器中的一些子层：多头注意力机制层、前向传播层、位置编码层和掩码层。

1. class MultiHeadAttention(tf.keras.layers.Layer):
2.     def __init__(self, d_model, num_heads):
3.         super(MultiHeadAttention, self).__init__()
4.         self.num_heads = num_heads
5.         self.d_model = d_model
6.         
7.         assert d_model % self.num_heads == 0
8.         
9.         self.depth = d_model // self.num_heads
10.         
11.         self.wq = tf.keras.layers.Dense(d_model)
12.         self.wk = tf.keras.layers.Dense(d_model)
13.         self.wv = tf.keras.layers.Dense(d_model)
14.         
15.         self.dense = tf.keras.layers.Dense(d_model)
16.         
17.     def split_heads(self, x, batch_size):
18.         x = tf.reshape(x, (batch_size, -1, self.num_heads, self.depth))
19.         return tf.transpose(x, perm=[0, 2, 1, 3])
20.    
21.     def call(self, v, k, q, mask):
22.         batch_size = tf.shape(q)[0]
23.         
24.         q = self.wq(q)
25.         k = self.wk(k)
26.         v = self.wv(v)
27.         
28.         q = self.split_heads(q, batch_size)
29.         k = self.split_heads(k, batch_size)
30.         v = self.split_heads(v, batch_size)
31.         
32.         scaled_attention_logits = tf.matmul(q, k, transpose_b=True)  
33.         
34.         scaled_attention_logits /= tf.math.sqrt(tf.cast(self.depth, tf.float32))
35.         
36.         if mask is not None:
37.             scaled_attention_logits += (mask * -1e9)
38.         
39.         attention_weights = tf.nn.softmax(scaled_attention_logits, axis=-1)
40.         
41.         output = tf.matmul(attention_weights, v)
42.         output = tf.transpose(output, perm=[0, 2, 1, 3])
43.         output = tf.reshape(output, (batch_size, -1, self.d_model))
44.         
45.         output = self.dense(output)
46.         
47.         return output, attention_weights
49. class PointWiseFeedForwardNetwork(tf.keras.layers.Layer):
50.     def __init__(self, d_model, num_units):
51.         super(PointWiseFeedForwardNetwork, self).__init__()
52.         self.d_model = d_model
53.         self.num_units = num_units
54.         
55.         self.fc1 = tf.keras.layers.Dense(num_units, activation=tf.nn.relu)
56.         self.fc2 = tf.keras.layers.Dense(d_model)
57.         
58.     def call(self, x):
59.         output = self.fc1(x)
60.         output = self.fc2(output)
61.         return output
62.       
63. class PositionalEncoding(tf.keras.layers.Layer):
64.     def __init__(self, position, d_model):
65.         super(PositionalEncoding, self).__init__()
66.         self.position = position
67.         self.d_model = d_model
68.         
69.         self.pos_encoding = self.positional_encoding(position, d_model)
70.         
71.     def get_angles(self, position, i, d_model):
72.         angles = 1 / tf.pow(10000, (2 * (i // 2)) / tf.cast(d_model, tf.float32))
73.         return position * angles
74.    
75.     def positional_encoding(self, position, d_model):
76.         angle_rads = self.get_angles(
77.             position=tf.range(position, dtype=tf.float32)[:, tf.newaxis],
78.             i=tf.range(d_model, dtype=tf.float32)[tf.newaxis, :],
79.             d_model=d_model)
80.         
81.         sines = tf.math.sin(angle_rads[:, 0::2])
82.         cosines = tf.math.cos(angle_rads[:, 1::2])
83.         
84.         pos_encoding = tf.concat([sines, cosines], axis=-1)
85.         pos_encoding = pos_encoding[tf.newaxis, ...]
86.         
87.         return tf.cast(pos_encoding, tf.float32)
88.    
89.     def call(self, inputs):
90.         return inputs + self.pos_encoding[:, :tf.shape(inputs)[1], :]
91.       
92. def create_padding_mask(seq):
93.     seq = tf.cast(tf.math.equal(seq, 0), tf.float32)
94.     return seq[:, tf.newaxis, tf.newaxis, :]  # 在最后添加额外的维度以符合multi-head attention的要求

复制代码

接下来，我们定义编码器和解码器：

1. class EncoderLayer(tf.keras.layers.Layer):
2.     def __init__(self, d_model, num_heads, units, dropout_rate):
3.         super(EncoderLayer, self).__init__()
4.         self.mha = MultiHeadAttention(d_model, num_heads)
5.         self.ffn = PointWiseFeedForwardNetwork(d_model, units)
6.         
7.         self.layer_norm1 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
8.         self.layer_norm2 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
9.         
10.         self.dropout1 = tf.keras.layers.Dropout(dropout_rate)
11.         self.dropout2 = tf.keras.layers.Dropout(dropout_rate)
12.         
13.     def call(self, x, training, mask):
14.         attn_output, _ = self.mha(x, x, x, mask)  
15.         attn_output = self.dropout1(attn_output, training=training)
16.         out1 = self.layer_norm1(x + attn_output)  
17.         
18.         ffn_output = self.ffn(out1)
19.         ffn_output = self.dropout2(ffn_output, training=training)
20.         out2 = self.layer_norm2(out1 + ffn_output)  
21.         
22.         return out2
24. class DecoderLayer(tf.keras.layers.Layer):
25.     def __init__(self, d_model, num_heads, units, dropout_rate):
26.         super(DecoderLayer, self).__init__()
27.         self.mha1 = MultiHeadAttention(d_model, num_heads)
28.         self.mha2 = MultiHeadAttention(d_model, num_heads)
29.         
30.         self.ffn = PointWiseFeedForwardNetwork(d_model, units)
31.         
32.         self.layer_norm1 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
33.         self.layer_norm2 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
34.         self.layer_norm3 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
35.         
36.         self.dropout1 = tf.keras.layers.Dropout(dropout_rate)
37.         self.dropout2 = tf.keras.layers.Dropout(dropout_rate)
38.         self.dropout3 = tf.keras.layers.Dropout(dropout_rate)
39.         
40.     def call(self, x, enc_output, training,
41.            look_ahead_mask, padding_mask):
42.         attn1, attn_weights_block1 = self.mha1(x, x, x, look_ahead_mask)
43.         attn1 = self.dropout1(attn1, training=training)
44.         out1 = self.layer_norm1(attn1 + x)
45.         
46.         attn2, attn_weights_block2 = self.mha2(
47.             enc_output, enc_output, out1, padding_mask)  
48.         attn2 = self.dropout2(attn2, training=training)
49.         out2 = self.layer_norm2(attn2 + out1)  
50.         
51.         ffn_output = self.ffn(out2)  
52.         ffn_output = self.dropout3(ffn_output, training=training)
53.         out3 = self.layer_norm3(ffn_output + out2)  
54.         
55.         return out3, attn_weights_block1, attn_weights_block2

复制代码

然后是编码器和解码器的堆叠：

1. class Encoder(tf.keras.layers.Layer):
2.     def __init__(self, num_layers, d_model, num_heads, units, dropout_rate,
3.                max_seq_len):
4.         super(Encoder, self).__init__()
5.         self.d_model = d_model
6.         self.num_layers = num_layers
7.         
8.         self.embedding = tf.keras.layers.Dense(d_model)
9.         self.pos_encoding = PositionalEncoding(max_seq_len, d_model)
10.         self.enc_layers = [EncoderLayer(d_model, num_heads, units,
11.                                        dropout_rate) for _ in range(num_layers)]
12.         
13.     def call(self, x, training, mask):
14.         x = self.embedding(x)
15.         x *= tf.math.sqrt(tf.cast(self.d_model, tf.float32))
16.         x = self.pos_encoding(x)
17.         
18.         for i in range(self.num_layers):
19.             x = self.enc_layers[i](x, training, mask)
20.             
21.         return x
23. class Decoder(tf.keras.layers.Layer):
24.     def __init__(self, num_layers, d_model, num_heads, units, dropout_rate,
25.                max_seq_len):
26.         super(Decoder, self).__init__()
27.         self.d_model = d_model
28.         self.num_layers = num_layers
29.         
30.         self.embedding = tf.keras.layers.Dense(d_model)
31.         self.pos_encoding = PositionalEncoding(max_seq_len, d_model)
32.         self.dec_layers = [DecoderLayer(d_model, num_heads, units,
33.                                         dropout_rate) for _ in range(num_layers)]
34.         
35.     def call(self, x, enc_output, training,
36.            look_ahead_mask, padding_mask):
37.         x = self.embedding(x)
38.         x *= tf.math.sqrt(tf.cast(self.d_model, tf.float32))
39.         x = self.pos_encoding(x)
40.         
41.         for i in range(self.num_layers):
42.             x, block1, block2 = self.dec_layers[i](x, enc_output, training,
43.                                                   look_ahead_mask, padding_mask)
44.         
45.         return x

复制代码

最后，我们定义Transformer模型：

1. class Transformer(tf.keras.Model):
2.     def __init__(self, num_layers, d_model, num_heads, units, dropout_rate,
3.                max_seq_len):
4.         super(Transformer, self).__init__()
5.         self.encoder = Encoder(num_layers, d_model, num_heads, units,
6.                                dropout_rate, max_seq_len)
7.         self.decoder = Decoder(num_layers, d_model, num_heads, units,
8.                                dropout_rate, max_seq_len)
9.         self.final_layer = tf.keras.layers.Dense(1)
10.         
11.     def call(self, inp, tar, training, enc_padding_mask,
12.              look_ahead_mask, dec_padding_mask):
13.         enc_output = self.encoder(inp, training, enc_padding_mask)
14.         
15.         dec_output = self.decoder(
16.             tar, enc_output, training, look_ahead_mask, dec_padding_mask)
17.         
18.         final_output = self.final_layer(dec_output)
19.         
20.         return final_output

复制代码

在定义完整的Transformer模型之后，我们可以使用随机生成的时间序列数据进行测试。假设我们有`N`个样本，每个样本包含`t`个时间步和`F`个特征。下面是一个示例：

1. # 生成随机时间序列数据
2. N = 1000
3. t = 10
4. F = 5
5. X = np.random.rand(N, t, F)
6. y = np.random.rand(N, 1)
8. # 划分训练集和测试集
9. train_ratio = 0.8
10. train_size = int(N * train_ratio)
11. X_train, X_test = X[:train_size], X[train_size:]
12. y_train, y_test = y[:train_size], y[train_size:]
14. # 创建掩码
15. train_padding_mask = create_padding_mask(X_train[:, :, 0])
16. train_look_ahead_mask = create_look_ahead_mask(tf.shape(X_train)[1])
17. train_dec_padding_mask = create_padding_mask(X_train[:, :, 0])
19. # 设置超参数
20. num_layers = 2
21. d_model = 32
22. num_heads = 4
23. units = 64
24. dropout_rate = 0.1
25. max_seq_len = t
27. # 创建和训练模型
28. model = Transformer(num_layers, d_model, num_heads, units,
29.                     dropout_rate, max_seq_len)
31. loss_object = tf.keras.losses.MeanSquaredError()
32. optimizer = tf.keras.optimizers.Adam()
34. train_loss = tf.keras.metrics.Mean(name='train_loss')
35. train_accuracy = tf.keras.metrics.MeanSquaredError(name='train_mse')
37. @tf.function
38. def train_step(inp, tar):
39.     tar_inp = tar[:, :-1]
40.     tar_real = tar[:, 1:]
41.    
42.     with tf.GradientTape() as tape:
43.         predictions = model(inp, tar_inp, True,
44.                             train_padding_mask,
45.                             train_look_ahead_mask,
46.                             train_dec_padding_mask)
47.         loss = loss_function(tar_real, predictions)
48.         
49.     gradients = tape.gradient(loss, model.trainable_variables)
50.     optimizer.apply_gradients(zip(gradients, model.trainable_variables))
51.    
52.     train_loss(loss)
53.     train_accuracy(tar_real, predictions)
54.    
55. EPOCHS = 10
56. BATCH_SIZE = 16
58. for epoch in range(EPOCHS):
59.     train_loss.reset_states()
60.     train_accuracy.reset_states()
61.    
62.     for i in range(0, train_size, BATCH_SIZE):
63.         inp = X_train[i:i+BATCH_SIZE]
64.         tar = y_train[i:i+BATCH_SIZE]
65.         train_step(inp, tar)
66.         
67.     print(f'Epoch {epoch+1}, Loss: {train_loss.result()}, MSE: {train_accuracy.result()}')
69. # 进行测试
70. test_padding_mask = create_padding_mask(X_test[:, :, 0])
71. test_look_ahead_mask = create_look_ahead_mask(tf.shape(X_test)[1])
72. test_dec_padding_mask = create_padding_mask(X_test[:, :, 0])
74. predictions = model(X_test, X_test[:, :-1], False,
75.                     test_padding_mask,
76.                     test_look_ahead_mask,
77.                     test_dec_padding_mask)
79. test_loss = loss_object(y_test[:, 1:], predictions)
80. print(f'Test Loss: {test_loss}, MSE: {tf.reduce_mean(tf.square(y_test[:, 1:] - predictions))}')

复制代码

上面的代码演示了如何构建和训练一个Transformer模型，以及如何使用训练模型进行预测。希望对你有所帮助！

以上回复来自 FishC AI 小助理，如未能正确解答您的问题，请继续追问。

跳转到最佳答案楼层

WTMGCLGGA

transformer应该还需要定义位置编码层吧@FishC

WTMGCLGGA

这个是使用pytorch实现的位置编码层，怎么用tensorflow来定义位置编码层呢@FishC

WTMGCLGGA

FishC 发表于 2023-12-25 09:52
嗨！要使用TensorFlow定义位置编码层，我们可以参考一下PyTorch版的代码来实现。

在PyTorch版本的代码中 ...

可以输出一个完整的transformer模型吗，需要包括定义编码器、解码器、多头注意力机制层、前向传播层、位置编码层、掩码层，然后是随机生成一个时间序列来训练并测试，使用前5个时间步的5个特征预测未来一个时间步的1个特征，使用tensorflow进行搭建
@FishC

WTMGCLGGA

FishC 发表于 2023-12-25 09:57
当然可以！下面是一个用TensorFlow实现的Transformer模型的完整代码。首先，让我们从导入必要的库开始：
...

train_look_ahead_mask = create_look_ahead_mask(tf.shape(X_train)[1])
这一行的create_look_ahead_mask还没有定义过吧
@FishC

WTMGCLGGA

FishC 发表于 2023-12-25 10:11
是的，您说得对。在这一行中，create_look_ahead_mask函数还没有定义过。

根据命名约定，这个函数应该是 ...

ValueError                                Traceback (most recent call last)
Cell In[14], line 10
      8     inp = X_train[i:i+BATCH_SIZE]
      9     tar = y_train[i:i+BATCH_SIZE]
---> 10     train_step(inp, tar)
     12 print(f'Epoch {epoch+1}, Loss: {train_loss.result()}, MSE: {train_accuracy.result()}')
运行时出现了这个错误@FishC

WTMGCLGGA

怎么将一个pytorch写的transformer模型转换成tensorflow
@FishC

WTMGCLGGA

class PositionalEncoding(nn.Module):
    def __init__(self,d_model,max_len=5000):
        super(PositionalEncoding,self).__init__()

        pe=torch.zeros(max_len,d_model)
        position=torch.arange(0,max_len,dtype=torch.float).unsqueeze(1)
        div_term=torch.exp(torch.arange(0,d_model,2).float()*(-math.log(10000.0)/d_model))

        pe[:,0::2]=torch.sin(position*div_term)
        pe[:,1::2]=torch.cos(position*div_term)

        pe=pe.unsqueeze(0).transpose(0,1)
        self.register_buffer('pe',pe)
   
    def forward(self,x:torch.Tensor):
        chunk=x.chunk(x.size(-1),dim=2)
        out=torch.Tensor([]).to(x.device)

        for i in range(len(chunk)):
            out=torch.cat((out,chunk[i]+self.pe[:chunk[i].size(0),...]),dim=2)

        return out

def transformer_generate_tgt_mask(length,device):
    mask=torch.tril(torch.ones(length,length,device=device))==1
    mask=(
        mask.float()
        .masked_fill(mask==0,float("-inf"))
        .masked_fill(mask==1,float(0.0))
    )

    return mask

class Transformer(nn.Module):
    def __init__(self,n_encoder_inputs,n_decoder_inputs,Sequence_length,d_model=512,dropout=0.1,num_layer=8):
        super(Transformer,self).__init__()
        
        self.input_pos_embedding=torch.nn.Embedding(500,embedding_dim=d_model)
        self.target_pos_embedding=torch.nn.Embedding(500,embedding_dim=d_model)
        
        encoder_layer=torch.nn.TransformerEncoderLayer(d_model=d_model,nhead=num_layer,dropout=dropout,dim_feedforward=4*d_model)
        decoder_layer=torch.nn.TransformerDecoderLayer(d_model=d_model,nhead=num_layer,dropout=dropout,dim_feedforward=4*d_model)
        
        self.encoder=torch.nn.TransformerEncoder(encoder_layer,num_layers=2)
        self.decoder=torch.nn.TransformerDecoder(decoder_layer,num_layers=4)
        
        self.lstm=nn.LSTM(len_int,len_int,num_layers=1,bidirectional=False)
        self.input_projection=torch.nn.Linear(n_encoder_inputs,d_model)
        self.output_projection=torch.nn.Linear(n_decoder_inputs,d_model)
        
        self.linear=torch.nn.Linear(d_model,1)
        self.ziji_add_linear=torch.nn.Linear(Sequence_length,1)
        self.relu=F.relu
   
    def encode_in(self,src):
        src_start=self.input_projection(src).permute(1,0,2)
        in_sequence_len,batch_size=src_start.size(0),src_start.size(1)
        
        pos_encoder=(torch.arange(0,in_sequence_len,device=src.device).unsqueeze(0).repeat(batch_size,1))
        pos_encoder=self.input_pos_embedding(pos_encoder).permute(1,0,2)

        src=src_start+pos_encoder
        src=self.encoder(src)+src_start
        return src
   
    def decode_out(self,tgt,memory):
        tgt_start=self.output_projection(tgt).permute(1,0,2)
        out_sequence_len,batch_size=tgt_start.size(0),tgt_start.size(1)

        pos_decoder=(torch.arange(0,out_sequence_len,device=tgt.device).unsqueeze(0).repeat(batch_size,1))
        pos_decoder=self.target_pos_embedding(pos_decoder).permute(1,0,2)

        tgt=tgt_start+pos_decoder
        tgt_mask=transformer_generate_tgt_mask(out_sequence_len,tgt.device)

        out=self.decoder(tgt=tgt,memory=memory,tgt_mask=tgt_mask)+tgt_start
        out=out.permute(1,0,2)
        out=self.linear(out)
        return out
        
    def forward(self,src,target_in):
        lstm_out,(h_n,c_n)=self.lstm(src)
        src=self.encode_in(self.relu(lstm_out))
        out=self.decode_out(tgt=target_in,memory=src)
        out=out.squeeze(2)
        out=self.ziji_add_linear(out)
        return out
这是一个pytorch写的transformer模型，能不能将它转换成tensorflow代码
@FishC

WTMGCLGGA

我需要用transformer模型解决一个时间序列预测问题，使用前5个时间步的5个特征来预测未来1个时间步的1个特征，请用tensorflow帮我写一个transformer模型，需要包括定义编码器、解码器、位置编码层、掩码层、多头注意力机制层、前向传播层、各个子层之间的残差连接和标准化层，等transformer结构中需要的层
@FishC