import copy import torch from torch import nn from torch import optim import torchtext from torchtext import data from torchtext import datasets from gensim.models import Word2Vec

Load data splits

train_data, val_data, test_data = datasets.SST.splits(TEXT, LABEL)

Train Word2Vec model

sentences = [example.text for example in train_data] word2vec_model = Word2Vec(sentences, size=embedding_dim, min_count=1)

Convert Word2Vec model to embeddings

embedding_matrix = torch.zeros((vocab_size, embedding_dim)) for i, word in enumerate(TEXT.vocab.itos): if word in word2vec_model.wv.vocab: embedding_matrix[i] = torch.tensor(word2vec_model.wv[word])

Update embedding layer weights with pre-trained embeddings

model.embedding.weight.data.copy_(embedding_matrix)

hyperparameters

vocab_size = len(TEXT.vocab) label_size = len(LABEL.vocab) padding_idx = TEXT.vocab.stoi[''] embedding_dim = 128 hidden_dim = 128

build iterators

train_iter, val_iter, test_iter = data.BucketIterator.splits( (train_data, val_data, test_data), batch_size=32)

use_cuda = torch.cuda.is_available() device = torch.device("cuda" if use_cuda else "cpu")

Training function

def train(model, train_loader, optimizer, criterion):
model.train()
total_loss = 0.0
total_correct = 0
for batch in train_loader:
text, labels = batch.text.to(device), batch.label.to(device)
optimizer.zero_grad()
logits = model(text) loss = criterion(logits, labels)

    loss.backward()  
    optimizer.step()  
    
    total_loss += loss.item() * text.size(0)  
    preds = logits.argmax(dim=1)  
    total_correct += (preds == labels).sum().item()  
avg_loss = total_loss / len(train_loader.dataset)  
accuracy = total_correct / len(train_loader.dataset)  
return avg_loss, accuracy  

def evaluate(model, iterator, criterion): epoch_loss = 0 epoch_acc = 0

model.eval()

with torch.no_grad():
    for batch in iterator:
        text, labels = batch.text.to(device), batch.label.to(device)  
        predictions = model(text)

        loss = criterion(predictions, batch.label)
        acc = accuracy(predictions, batch.label)

        epoch_loss += loss.item()
        epoch_acc += acc.item()

return epoch_loss / len(iterator), epoch_acc / len(iterator)

def accuracy(predictions, labels): _, predicted_labels = torch.max(predictions, 1) correct = (predicted_labels == labels).float() accuracy = correct.sum() / len(correct) return accuracy

class RNNClassifier(nn.Module):
def init(self, vocab_size, embedding_dim, hidden_dim, label_size, padding_idx):
super(RNNClassifier, self).init()
self.vocab_size = vocab_size
self.embedding_dim = embedding_dim
self.hidden_dim = hidden_dim
self.label_size = label_size
self.num_layers = 2
self.dropout_num = 0.5 self.dropout = nn.Dropout(0.5) self.bidirectional=False

    # Embedding Layer  
    self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=padding_idx)  
    self.embedding_dropout = nn.Dropout(self.dropout_num)

    # LSTM Layer  
    self.lstm = nn.LSTM(embedding_dim, hidden_dim, num_layers=self.num_layers, batch_first=True,bidirectional=self.bidirectional)  
    self.lstm_dropout = nn.Dropout(self.dropout_num)

    # Fully Connected Layer  
    self.fc = nn.Linear(hidden_dim, label_size)  
    
    self.fc_dropout = nn.Dropout(self.dropout_num)
    self.softmax = nn.LogSoftmax(dim=1)

def zero_state(self, batch_size):  
    hidden = torch.zeros(self.num_layers, batch_size,  self.hidden_dim)  
    cell = torch.zeros(self.num_layers, batch_size,  self.hidden_dim) 
    return hidden, cell

def forward(self, text):  
    emb = self.embedding(text)
    emb = self.embedding_dropout(emb)
    emb = torch.mean(emb, dim=1)

    h0, c0 = self.zero_state(text.size(0))
    output, (hn, cn) = self.lstm(emb.unsqueeze(1), (h0, c0))
    output = self.lstm_dropout(output)

    output = torch.mean(output, dim=1)
    output = self.fc(output)
    output = self.fc_dropout(output)
    return output

model = RNNClassifier(vocab_size, embedding_dim, hidden_dim, label_size,padding_idx).to(device)
optimizer = optim.Adam(model.parameters(), lr=0.001)
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=1, gamma=0.8) criterion = nn.CrossEntropyLoss() num_epochs = 10

train_losses = [] val_losses = [] train_accs = [] val_accs = []

def model_running(model,train_iter, val_iter,optimizer, criterion): best_val_acc = 0.0 for epoch in range(1,num_epochs+1): train_acc = 0.0 train_loss = 0.0 val_acc = 0.0 val_loss = 0.0

    train_loss, train_acc = train(model, train_iter, optimizer, criterion)
    val_loss, val_acc = evaluate(model, val_iter, criterion)
    
    print('Epoch [{}/{}], Train Loss: {:.4f}, Val Loss: {:.4f}, Train Acc: {:.4f}, Val Acc: {:.4f}' 
      .format(epoch, num_epochs, train_loss, val_loss, train_acc, val_acc))
    
    train_losses.append(train_loss)
    val_losses.append(val_loss)
    train_accs.append(train_acc)
    val_accs.append(val_acc)
    
if val_acc > best_val_acc:  
    best_val_acc = val_acc  
    best_model = model.state_dict()  
    
return best_model

best_model = model_running(model,train_iter, val_iter,optimizer, criterion) model.load_state_dict(best_model)
test_loss, test_acc = evaluate(model, test_iter, criterion)
print(f"Test loss: {test_loss:.4f}, Test accuracy: {test_acc:.4f}")