自然语言处理算法中的Transformer

pythonCopy codeimport torch
import torch.nn as nn
import torch.optim as optim
import torchtext
from torchtext.data import Field, TabularDataset, BucketIterator
from torch.nn import TransformerEncoder, TransformerEncoderLayer
# 定义模型
class TransformerClassifier(nn.Module):
    def __init__(self, input_dim, output_dim, hidden_dim, n_layers, n_heads, dropout):
        super(TransformerClassifier, self).__init__()
        self.embedding = nn.Embedding(input_dim, hidden_dim)
        self.encoder_layer = TransformerEncoderLayer(hidden_dim, n_heads, hidden_dim, dropout)
        self.encoder = TransformerEncoder(self.encoder_layer, n_layers)
        self.fc = nn.Linear(hidden_dim, output_dim)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, src):
        embedded = self.embedding(src)
        embedded = embedded * math.sqrt(self.hidden_dim)
        encoded = self.encoder(embedded)
        pooled = torch.mean(encoded, dim=1)
        dropped = self.dropout(pooled)
        output = self.fc(dropped)
        return output
# 定义训练函数
def train(model, iterator, optimizer, criterion):
    model.train()
    epoch_loss = 0
    for batch in iterator:
        optimizer.zero_grad()
        predictions = model(batch.text)
        loss = criterion(predictions, batch.label)
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
    return epoch_loss / len(iterator)
# 定义评估函数
def evaluate(model, iterator, criterion):
    model.eval()
    epoch_loss = 0
    with torch.no_grad():
        for batch in iterator:
            predictions = model(batch.text)
            loss = criterion(predictions, batch.label)
            epoch_loss += loss.item()
    return epoch_loss / len(iterator)
# 数据预处理
TEXT = Field(tokenize='spacy', lower=True)
LABEL = Field(sequential=False, is_target=True)
fields = [('text', TEXT), ('label', LABEL)]
train_data, test_data = TabularDataset.splits(path='data/', train='train.csv', test='test.csv', format='csv', fields=fields)
TEXT.build_vocab(train_data)
LABEL.build_vocab(train_data)
# 划分batch
BATCH_SIZE = 64
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
train_iterator, test_iterator = BucketIterator.splits((train_data, test_data), batch_size=BATCH_SIZE, device=device)
# 初始化模型和优化器
INPUT_DIM = len(TEXT.vocab)
OUTPUT_DIM = len(LABEL.vocab)
HIDDEN_DIM = 256
N_LAYERS = 3
N_HEADS = 8
DROPOUT = 0.2
model = TransformerClassifier(INPUT_DIM, OUTPUT_DIM, HIDDEN_DIM, N_LAYERS, N_HEADS, DROPOUT).to(device)
optimizer = optim.Adam(model.parameters())
criterion = nn.CrossEntropyLoss()
# 训练模型
N_EPOCHS = 10
best_loss = float('inf')
for epoch in range(N_EPOCHS):
    train_loss = train(model, train_iterator, optimizer, criterion)
    test_loss = evaluate(model, test_iterator, criterion)
    if test_loss < best_loss:
        best_loss = test_loss
        torch.save(model.state_dict(), 'best_model.pt')
    print(f'Epoch: {epoch+1}/{N_EPOCHS} | Train Loss: {train_loss:.3f} | Test Loss: {test_loss:.3f}')
# 加载最佳模型并进行预测
model.load_state_dict(torch.load('best_model.pt'))
model.eval()
with torch.no_grad():
    for batch in test_iterator:
        predictions = model(batch.text)
        _, predicted_labels = torch.max(predictions, 1)
        print(predicted_labels)