<jupyter_start><jupyter_text>Initializing the seed<jupyter_code>seed = 0

torch.manual_seed(seed)<jupyter_output><empty_output><jupyter_text>Loading the dataset<jupyter_code>train_data, test_data = datasets.load_dataset('imdb', split=['train', 'test'])<jupyter_output><empty_output><jupyter_text>Initializing  a tokenizer<jupyter_code>tokenizer = torchtext.data.utils.get_tokenizer('basic_english')<jupyter_output><empty_output><jupyter_text>Function for tokenizing the data<jupyter_code>def tokenize_data(example, tokenizer, max_length):
    tokens = tokenizer(example['text'])[:max_length]
    length = len(tokens)
    return {'tokens': tokens, 'length': length}<jupyter_output><empty_output><jupyter_text>Tokenizing the dataset (also creates a padding if there words are less than max length)<jupyter_code>max_length = 256

train_data = train_data.map(tokenize_data, fn_kwargs={'tokenizer': tokenizer, 'max_length': max_length})
test_data = test_data.map(tokenize_data, fn_kwargs={'tokenizer': tokenizer, 'max_length': max_length})<jupyter_output><empty_output><jupyter_text>Creating a validation dataset (as the imdb dataset only has train and test set)<jupyter_code>valid_size= 0.1

train_valid_data = train_data.train_test_split(test_size=valid_size)
train_data = train_valid_data['train']
valid_data = train_valid_data['test']<jupyter_output><empty_output><jupyter_text>Including words that occur minimum 5 times<jupyter_code>min_freq = 5
special_tokens = ['<unk>', '<pad>']

vocab = torchtext.vocab.build_vocab_from_iterator(train_data['tokens'],
                                                  min_freq=min_freq,
                                                  specials=special_tokens)
unk_index = vocab['<unk>']
pad_index = vocab['<pad>']

vocab.set_default_index(unk_index)<jupyter_output><empty_output><jupyter_text>Function for converting the tokens into numbers according to their index<jupyter_code>def numericalize_data(example, vocab):
    ids = [vocab[token] for token in example['tokens']]
    return {'ids': ids}
train_data = train_data.map(numericalize_data, fn_kwargs={'vocab': vocab})
valid_data = valid_data.map(numericalize_data, fn_kwargs={'vocab': vocab})
test_data = test_data.map(numericalize_data, fn_kwargs={'vocab': vocab})<jupyter_output><empty_output><jupyter_text>Converting the data to pytorch tensors<jupyter_code>train_data = train_data.with_format(type='torch', columns=['ids', 'label', 'length'])
valid_data = valid_data.with_format(type='torch', columns=['ids', 'label', 'length'])
test_data = test_data.with_format(type='torch', columns=['ids', 'label', 'length'])

train_data[5]<jupyter_output><empty_output><jupyter_text>Class for LSTM model<jupyter_code>class LSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, n_layers, bidirectional,
                 dropout_rate, pad_index):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_index)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, n_layers, bidirectional=bidirectional,
                            dropout=dropout_rate, batch_first=True)
        self.fc = nn.Linear(hidden_dim * 2 if bidirectional else hidden_dim, output_dim)
        self.dropout = nn.Dropout(dropout_rate)
        
    def forward(self, ids, length):
        # ids = [batch size, seq len]
        # length = [batch size]
        embedded = self.dropout(self.embedding(ids))
        # embedded = [batch size, seq len, embedding dim]
        packed_embedded = nn.utils.rnn.pack_padded_sequence(embedded, length, batch_first=True, 
                                                            enforce_sorted=False)
        packed_output, (hidden, cell) = self.lstm(packed_embedded)
        # hidden = [n layers * n directions, batch size, hidden dim]
        # cell = [n layers * n directions, batch size, hidden dim]
        output, output_length = nn.utils.rnn.pad_packed_sequence(packed_output)
        # output = [batch size, seq len, hidden dim * n directions]
        if self.lstm.bidirectional:
            hidden = self.dropout(torch.cat([hidden[-1], hidden[-2]], dim=-1))
            # hidden = [batch size, hidden dim * 2]
        else:
            hidden = self.dropout(hidden[-1])
            # hidden = [batch size, hidden dim]
        prediction = self.fc(hidden)
        # prediction = [batch size, output dim]
        return prediction<jupyter_output><empty_output><jupyter_text>Setting the dimensions<jupyter_code>vocab_size = len(vocab)
embedding_dim = 300
hidden_dim = 300
output_dim = len(train_data.unique('label'))
n_layers = 2
bidirectional = True
dropout_rate = 0.5

model = LSTM(vocab_size, embedding_dim, hidden_dim, output_dim, n_layers, bidirectional, dropout_rate, 
             pad_index)<jupyter_output><empty_output><jupyter_text>Counting the number of parameters<jupyter_code>def count_parameters(model):
    return sum(p.numel() for p in model.parameters() if p.requires_grad)

print(f'The model has {count_parameters(model):,} trainable parameters')<jupyter_output><empty_output><jupyter_text>Initialzing the parameters(weights and biases) for the layers using methods like Xavier initialization .....<jupyter_code>def initialize_weights(m):
    if isinstance(m, nn.Linear):
        nn.init.xavier_normal_(m.weight)
        nn.init.zeros_(m.bias)
    elif isinstance(m, nn.LSTM):
        for name, param in m.named_parameters():
            if 'bias' in name:
                nn.init.zeros_(param)
            elif 'weight' in name:
                nn.init.orthogonal_(param)<jupyter_output><empty_output><jupyter_text>Applying the initial parameter values<jupyter_code>model.apply(initialize_weights)<jupyter_output><empty_output><jupyter_text>Downloading pre-trained word embeddings<jupyter_code>vectors = torchtext.vocab.FastText()
pretrained_embedding = vectors.get_vecs_by_tokens(vocab.get_itos())<jupyter_output><empty_output><jupyter_text>Initialing the parmeterts of embedding layer using pre-trained embeddings<jupyter_code>model.embedding.weight.data = pretrained_embedding<jupyter_output><empty_output><jupyter_text>Initializing learning rate and optimizer<jupyter_code>lr = 5e-4

optimizer = optim.Adam(model.parameters(), lr=lr)<jupyter_output><empty_output><jupyter_text>The loss function we'll use is the cross-entropy loss<jupyter_code>criterion = nn.CrossEntropyLoss()<jupyter_output><empty_output><jupyter_text>Checking whether GPU is available<jupyter_code>device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')<jupyter_output><empty_output><jupyter_text>Deploying model and lass layer on GPU<jupyter_code>model = model.to(device)
criterion = criterion.to(device)<jupyter_output><empty_output><jupyter_text>Function to stack sequences<jupyter_code>def collate(batch, pad_index):
    batch_ids = [i['ids'] for i in batch]
    batch_ids = nn.utils.rnn.pad_sequence(batch_ids, padding_value=pad_index, batch_first=True)
    batch_length = [i['length'] for i in batch]
    batch_length = torch.stack(batch_length)
    batch_label = [i['label'] for i in batch]
    batch_label = torch.stack(batch_label)
    batch = {'ids': batch_ids,
             'length': batch_length,
             'label': batch_label}
    return batch<jupyter_output><empty_output><jupyter_text>Creating dataloaders to put batches on GPU<jupyter_code>batch_size = 64

collate = functools.partial(collate, pad_index=pad_index)

train_dataloader = torch.utils.data.DataLoader(train_data, 
                                               batch_size=batch_size, 
                                               collate_fn=collate, 
                                               shuffle=True)

valid_dataloader = torch.utils.data.DataLoader(valid_data, batch_size=batch_size, collate_fn=collate)
test_dataloader = torch.utils.data.DataLoader(test_data, batch_size=batch_size, collate_fn=collate)<jupyter_output><empty_output><jupyter_text>Function for Training<jupyter_code>def train(dataloader, model, criterion, optimizer, device):

    model.train()
    epoch_losses = []
    epoch_accs = []

    for batch in tqdm.tqdm(dataloader, desc='training...', file=sys.stdout):
        ids = batch['ids'].to(device)
        length = batch['length']
        label = batch['label'].to(device)
        prediction = model(ids, length)
        loss = criterion(prediction, label)
        accuracy = get_accuracy(prediction, label)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        epoch_losses.append(loss.item())
        epoch_accs.append(accuracy.item())

    return epoch_losses, epoch_accs<jupyter_output><empty_output><jupyter_text>Function for evaluating the model<jupyter_code>def evaluate(dataloader, model, criterion, device):
    
    model.eval()
    epoch_losses = []
    epoch_accs = []

    with torch.no_grad():
        for batch in tqdm.tqdm(dataloader, desc='evaluating...', file=sys.stdout):
            ids = batch['ids'].to(device)
            length = batch['length']
            label = batch['label'].to(device)
            prediction = model(ids, length)
            loss = criterion(prediction, label)
            accuracy = get_accuracy(prediction, label)
            epoch_losses.append(loss.item())
            epoch_accs.append(accuracy.item())

    return epoch_losses, epoch_accs<jupyter_output><empty_output><jupyter_text>Function to get Accuracy<jupyter_code>def get_accuracy(prediction, label):
    batch_size, _ = prediction.shape
    predicted_classes = prediction.argmax(dim=-1)
    correct_predictions = predicted_classes.eq(label).sum()
    accuracy = correct_predictions / batch_size
    return accuracy
n_epochs = 10
best_valid_loss = float('inf')

train_losses = []
train_accs = []
valid_losses = []
valid_accs = []

for epoch in range(n_epochs):

    train_loss, train_acc = train(train_dataloader, model, criterion, optimizer, device)
    valid_loss, valid_acc = evaluate(valid_dataloader, model, criterion, device)

    train_losses.extend(train_loss)
    train_accs.extend(train_acc)
    valid_losses.extend(valid_loss)
    valid_accs.extend(valid_acc)
    
    epoch_train_loss = np.mean(train_loss)
    epoch_train_acc = np.mean(train_acc)
    epoch_valid_loss = np.mean(valid_loss)
    epoch_valid_acc = np.mean(valid_acc)
    
    if epoch_valid_loss < best_valid_loss:
        best_valid_loss = epoch_valid_loss
        torch.save(model.state_dict(), 'lstm.pt')
    
    print(f'epoch: {epoch+1}')
    print(f'train_loss: {epoch_train_loss:.3f}, train_acc: {epoch_train_acc:.3f}')
    print(f'valid_loss: {epoch_valid_loss:.3f}, valid_acc: {epoch_valid_acc:.3f}')
fig = plt.figure(figsize=(10,6))
ax = fig.add_subplot(1,1,1)
ax.plot(train_losses, label='train loss')
ax.plot(valid_losses, label='valid loss')
plt.legend()
ax.set_xlabel('updates')
ax.set_ylabel('loss');

fig = plt.figure(figsize=(10,6))
ax = fig.add_subplot(1,1,1)
ax.plot(train_accs, label='train accuracy')
ax.plot(valid_accs, label='valid accuracy')
plt.legend()
ax.set_xlabel('updates')
ax.set_ylabel('accuracy');

test_loss, test_acc = evaluate(test_dataloader, model, criterion, device)

epoch_test_loss = np.mean(test_loss)
epoch_test_acc = np.mean(test_acc)

print(f'test_loss: {epoch_test_loss:.3f}, test_acc: {epoch_test_acc:.3f}')
def predict_sentiment(text, model, tokenizer, vocab, device):
    tokens = tokenizer(text)
    ids = [vocab[t] for t in tokens]
    length = torch.LongTensor([len(ids)])
    tensor = torch.LongTensor(ids).unsqueeze(dim=0).to(device)
    prediction = model(tensor, length).squeeze(dim=0)
    probability = torch.softmax(prediction, dim=-1)
    predicted_class = prediction.argmax(dim=-1).item()
    predicted_probability = probability[predicted_class].item()
    return predicted_class, predicted_probability
text = "This film is terrible!"

predict_sentiment(text, model, tokenizer, vocab, device)

text = "This film is great!"

predict_sentiment(text, model, tokenizer, vocab, device)
text = "This film is not terrible, it's great!"

predict_sentiment(text, model, tokenizer, vocab, device)

text = "This film is not great, it's terrible!"

predict_sentiment(text, model, tokenizer, vocab, device)<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup