import math import pandas as pd import torch from torch import nn from d2l import torch as d2l class TransformerEncoder(d2l.Encoder): """Transformer编码器""" def __init__(self, vocab_size, key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, num_layers, dropout, use_bias=False, **kwargs): super(TransformerEncoder, self).__init__(**kwargs) self.num_hiddens = num_hiddens self.embedding = nn.Embedding(vocab_size, num_hiddens) self.pos_encoding = d2l.PositionalEncoding(num_hiddens, dropout) self.blks = nn.Sequential() for i in range(num_layers): self.blks.add_module("block"+str(i), EncoderBlock(key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, dropout, use_bias)) def forward(self, X, valid_lens, *args): # 因为位置编码值在-1和1之间， # 因此嵌入值乘以嵌入维度的平方根进行缩放， # 然后再与位置编码相加。 X = self.pos_encoding(self.embedding(X) * math.sqrt(self.num_hiddens)) self.attention_weights = [None] * len(self.blks) for i, blk in enumerate(self.blks): X = blk(X, valid_lens) self.attention_weights[ i] = blk.attention.attention.attention_weights return X X = torch.ones((2, 100, 24)) valid_lens = torch.tensor([3, 2]) encoder_blk = EncoderBlock(24, 24, 24, 24, [100, 24], 24, 48, 8, 0.5) encoder_blk.eval() encoder_blk(X, valid_lens).shape torch.Size([2, 100, 24])

这段代码定义了一个Transformer编码器（TransformerEncoder）的类，它继承自d2l.Encoder。该编码器包含了嵌入层（Embedding）、位置编码层（PositionalEncoding）和多个EncoderBlock组成的序列。

在初始化方法中，首先创建了一个Embedding实例self.embedding，用于将输入的词索引映射为词向量。然后创建了一个PositionalEncoding实例self.pos_encoding，用于对词向量进行位置编码。接下来创建了一个包含多个EncoderBlock的序列self.blks，每个EncoderBlock具有指定的参数。

在前向传播方法中，首先将输入张量X通过嵌入层和位置编码层进行处理。嵌入层将词索引转换为词向量，并与位置编码相加。然后，依次对序列中的每个EncoderBlock进行前向传播，并记录每个Attention层的注意力权重。最后返回处理后的张量X和注意力权重列表。

在代码的最后，创建了一个大小为(2, 100, 24)的张量X和有效长度valid_lens，表示输入和有效长度。然后创建了一个EncoderBlock的实例encoder_blk，并对其进行了评估（eval()）。将张量X和有效长度valid_lens输入到encoder_blk中，并打印出输出张量的形状。

结果是一个大小为(2, 100, 24)的张量，表示经过Transformer编码器处理后的输出张量的形状与输入张量相同。

相关问题

import math import pandas as pd import torch from torch import nn from d2l import torch as d2l class EncoderBlock(nn.Module): """Transformer编码器块""" def __init__(self, key_size, query_size, value_size, num_hiddens,norm_shape, ffn_num_input, ffn_ num_hiddens, num_heads,dropout, use_bias=False, **kwargs): super(EncoderBlock, self).__init__(**kwargs) self.attention = d2l.MultiHeadAttention( key_size, query_size, value_size, num_hiddens, num_heads, dropout, use_bias) self.addnorm1 = AddNorm(norm_shape, dropout) self.ffn = PositionWiseFFN( ffn_num_input, ffn_num_hiddens, num_hiddens) self.addnorm2 = AddNorm(norm_shape, dropout) def forward(self, X, valid_lens): Y = self.addnorm1(X, self.attention(X, X, X, valid_lens)) return self.addnorm2(Y, self.ffn(Y)) X = torch.ones((2, 100, 24)) valid_lens = torch.tensor([3, 2]) encoder_blk = EncoderBlock(24, 24, 24, 24, [100, 24], 24, 48, 8, 0.5) encoder_blk.eval() encoder_blk(X, valid_lens).shape torch.Size([2, 100, 24])

这段代码定义了一个Transformer编码器块（EncoderBlock）的类。该编码器块包含了多头注意力（MultiHeadAttention）、残差连接与层规范化（AddNorm）和基于位置的前馈网络（PositionWiseFFN）。

在初始化方法中，首先创建了一个多头注意力的实例self.attention，然后创建了两个AddNorm实例self.addnorm1和self.addnorm2，分别用于在注意力和前馈网络之后进行残差连接与层规范化。最后创建了一个PositionWiseFFN实例self.ffn。

在前向传播方法中，输入张量X和有效长度valid_lens被输入到多头注意力中进行自注意力计算，并通过残差连接与层规范化进行处理。然后将处理后的张量输入到基于位置的前馈网络中，再次通过残差连接与层规范化进行处理。最后返回处理后的张量Y。

在代码的最后，创建了一个EncoderBlock的实例encoder_blk，并对其进行了评估（eval()）。然后，将一个大小为(2, 100, 24)的张量X和有效长度valid_lens输入到encoder_blk中，并打印出输出张量的形状。

结果是一个大小为(2, 100, 24)的张量，表示经过Transformer编码器块后的输出张量的形状与输入张量相同。

使用transformer实现光伏预测项目

光伏预测是一个时间序列预测问题，可以使用Transformer模型进行建模和预测。下面是一个简单的光伏预测示例代码：

import torch
import torch.nn as nn
import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error

# 加载数据
data = pd.read_csv('pv_data.csv', index_col=0)

# 数据预处理
scaler = MinMaxScaler()
data_scaled = scaler.fit_transform(data.values)

# 划分训练集和测试集
train_size = int(len(data_scaled) * 0.8)
train_data = data_scaled[:train_size]
test_data = data_scaled[train_size:]

# 定义超参数
input_size = 1
hidden_size = 16
num_layers = 2
dropout = 0.2
lr = 0.001
num_epochs = 100

# 定义Transformer模型
class TransformerModel(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, dropout):
        super(TransformerModel, self).__init__()
        self.model_type = 'Transformer'
        self.pos_encoder = PositionalEncoding(hidden_size, dropout)
        encoder_layers = nn.TransformerEncoderLayer(hidden_size, nhead=8)
        self.transformer_encoder = nn.TransformerEncoder(encoder_layers, num_layers)
        self.encoder = nn.Linear(input_size, hidden_size)
        self.decoder = nn.Linear(hidden_size, 1)

    def forward(self, src):
        src = self.encoder(src) * np.sqrt(self.hidden_size)
        src = self.pos_encoder(src)
        output = self.transformer_encoder(src)
        output = self.decoder(output)
        return output

# 定义位置编码器
class PositionalEncoding(nn.Module):
    def __init__(self, d_model, dropout=0.1, max_len=5000):
        super(PositionalEncoding, self).__init__()
        self.dropout = nn.Dropout(p=dropout)

        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):
        x = x + self.pe[:x.size(0), :]
        return self.dropout(x)

# 训练模型
model = TransformerModel(input_size, hidden_size, num_layers, dropout)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=lr)

for epoch in range(num_epochs):
    model.train()
    train_loss = 0.0
    for i in range(input_size, len(train_data)):
        x = train_data[i - input_size:i, :]
        y = train_data[i, :]
        x = torch.from_numpy(x).float().unsqueeze(0)
        y = torch.from_numpy(y).float()
        optimizer.zero_grad()
        output = model(x)
        loss = criterion(output, y)
        loss.backward()
        optimizer.step()
        train_loss += loss.item()

    train_loss /= (len(train_data) - input_size)
    print('Epoch [{}/{}], Train Loss: {:.4f}'.format(epoch + 1, num_epochs, train_loss))

# 测试模型
model.eval()
test_loss = 0.0
predictions = []
with torch.no_grad():
    for i in range(input_size, len(test_data)):
        x = test_data[i - input_size:i, :]
        y = test_data[i, :]
        x = torch.from_numpy(x).float().unsqueeze(0)
        y = torch.from_numpy(y).float()
        output = model(x)
        loss = criterion(output, y)
        test_loss += loss.item()
        predictions.append(output.detach().numpy().squeeze())

test_loss /= (len(test_data) - input_size)
rmse = np.sqrt(mean_squared_error(test_data[input_size:], predictions))
print('Test Loss: {:.4f}, RMSE: {:.4f}'.format(test_loss, rmse))

在这个示例中，我们使用 `pandas` 库加载了一个光伏发电量数据集，并使用 `sklearn` 库中的 `MinMaxScaler` 对数据进行预处理，将数据缩放到0到1的范围内。然后，我们将数据划分为训练集和测试集，并定义了一些超参数，包括输入大小、隐藏层大小、层数、dropout、学习率和训练轮数。

接下来，我们定义了一个Transformer模型，包括位置编码器和Transformer编码器。在训练模型时，我们使用均方误差作为损失函数，使用Adam优化器更新模型参数。在测试模型时，我们计算了测试集上的均方误差和均方根误差。

需要注意的是，这个示例只是一个简单的光伏预测示例，实际应用需要根据具体情况进行调整和优化。