Adicionando arquivos do projeto

Author: jpmedras

data/leetcode.csv

@@ -0,0 +1,11 @@
+from torch.utils.data import DataLoader
+
+def create_dataloader(dataset, batch_size, type):
+    if type == 'train':
+        shuffle = True
+    else:
+        shuffle = False
+
+    loader = DataLoader(dataset=dataset, batch_size=batch_size, drop_last=True, shuffle=shuffle)
+
+    return loader

workspace/Datasets/dataloader.py

@@ -0,0 +1,13 @@
+from torch.utils.data import Dataset
+from torch import tensor
+
+class ProblemDataset(Dataset):
+    def __init__(self, data, inputs_encoder, labels_encoder):
+        self.inputs = inputs_encoder(data['inputs'])
+        self.labels = labels_encoder(data['labels'])
+            
+    def __len__(self):
+        return len(self.inputs)
+    
+    def __getitem__(self, index):
+        return self.inputs[index], self.labels[index]

workspace/Datasets/dataset.py

@@ -0,0 +1,55 @@
+from torch import tensor, long
+from torch import unsqueeze, cat
+from transformers import BertTokenizer
+
+def define_encoders(max_len):
+    def inputs_encoder(inputs):
+        tokenizer = BertTokenizer.from_pretrained('bert-base-uncased').encode_plus
+        encoded_inputs = []
+        for input in inputs:
+            encoding = tokenizer(
+                text=input,
+                add_special_tokens=True,
+                padding='max_length',
+                truncation='longest_first',
+                max_length=max_len
+            )
+
+            input_ids = encoding['input_ids']
+            attention_mask = encoding['attention_mask']
+            token_type_ids = encoding['token_type_ids']
+
+            encoded_input_ids = tensor(input_ids, dtype=long).unsqueeze(0)
+            encoded_attention_mask = tensor(attention_mask, dtype=long).unsqueeze(0)
+            encoded_token_type_ids = tensor(token_type_ids, dtype=long).unsqueeze(0)
+
+            encoded_input = cat((encoded_input_ids, encoded_attention_mask, encoded_token_type_ids), dim = 0).unsqueeze(0)
+            encoded_inputs.append(encoded_input)
+        
+        try:
+            encoded = cat(encoded_inputs)
+            return encoded
+        except:
+            print('Number of inputs:', len(encoded_inputs))
+            max_dif = max([e.shape[2] for e in encoded_inputs])
+            print('Max shape:', max_dif)
+            for idx, e in enumerate(encoded_inputs):
+                if e.shape[2] != max_len:
+                    print(idx)
+                    print(e.shape[2])
+                    print(inputs[idx])
+                    break
+            return None
+
+    def labels_encoder(labels):
+        encoded_labels = []
+        for label in labels:
+            encoded_labels.append(
+                tensor([label], dtype=long).unsqueeze(0)
+            )
+        
+        encoded = cat(encoded_labels)
+
+        return encoded
+    
+    return inputs_encoder, labels_encoder

workspace/Datasets/encoders.py

@@ -0,0 +1,10 @@
+import pandas as pd
+
+def load_data(data_path):
+    df = pd.read_csv(data_path, encoding='utf-8')
+    data = {
+        'inputs': df.inputs.values.tolist(),
+        'labels': df.labels.values.tolist()
+    }
+
+    return data

workspace/Datasets/load.py

@@ -0,0 +1,9 @@
+import torch
+import random
+import numpy as np
+
+def define_seed(seed=42):
+    torch.manual_seed(seed)
+    random.seed(seed)
+    np.random.seed(seed)
+    #torch.use_deterministic_algorithms(True)

workspace/Datasets/seed.py

@@ -0,0 +1,9 @@
+from torch.utils.data import random_split
+from torch import Generator
+
+def split_data(dataset, lengths, seed=42):
+    assert len(lengths) == 2, "You must define the size of train dataset and test dataset"
+
+    train_set, test_set = random_split(dataset, lengths, Generator().manual_seed(seed))
+
+    return train_set, test_set

workspace/Datasets/split.py

@@ -0,0 +1,123 @@
+from torch import nn, long, argmax, optim, save
+from torch import no_grad
+from transformers import BertModel
+from torch import cuda
+from Datasets.dataloader import create_dataloader
+from loss import calc_loss
+
+import matplotlib.pyplot as plt
+from sklearn.metrics import f1_score, confusion_matrix, ConfusionMatrixDisplay
+
+class BERTModule(nn.Module):
+    def __init__(self, epochs = 10, learning_rate = 1e-05, dropout_p = 0.3):
+        super(BERTModule, self).__init__()
+        self.bert = BertModel.from_pretrained('bert-base-uncased')
+        # for param in self.bert.parameters():
+        #   param.requires_grad = False
+        self.dropout = nn.Dropout(p = dropout_p)
+        self.fc = nn.Linear(768, 3)
+
+        self.epochs = epochs
+        self.learning_rate = learning_rate
+
+    def forward(self, ids, masks, ttis):
+        _, pooled_output = self.bert(ids, attention_mask = masks, token_type_ids = ttis, return_dict = False)
+        output_2 = self.dropout(pooled_output)
+        output = self.fc(output_2)
+        
+        return output
+    
+    def fit(self, train_loader, test_loader):
+        self.device = 'cuda' if cuda.is_available() else 'cpu'
+        self.to(self.device)
+
+        self.train()
+        
+        criterion = nn.CrossEntropyLoss()
+        optimizer = optim.Adam(params = self.parameters(), lr = self.learning_rate)
+        
+        print('Begin training...')
+
+        train_losses = []
+        test_losses = []
+
+        for epoch in range(self.epochs):
+            train_loss = 0.
+
+            for inputs, labels in train_loader:
+                optimizer.zero_grad()
+
+                ids = inputs[:, 0].to(self.device, dtype=long)
+                masks = inputs[:, 1].to(self.device, dtype=long)
+                tti = inputs[:, 2].to(self.device, dtype=long)
+                labels = labels.squeeze().to(self.device, dtype=long)
+
+                assert ids.shape == masks.shape, 'Ids != Masks'
+                assert masks.shape == tti.shape, 'Masks != Ttis'
+                assert ids.shape == tti.shape, 'Ids != Ttis'
+
+                assert ids.shape[0] == labels.shape[0], 'inputs and labels are incompatible'
+
+                outputs = self(ids, masks, tti)
+
+                loss = criterion(outputs, labels)
+
+                loss.backward()
+                optimizer.step()
+
+                train_loss += loss.item()
+    
+            avg_train_loss = train_loss / len(train_loader)
+            avg_test_loss = calc_loss(self, test_loader, criterion, self.device)
+
+            train_losses.append(avg_train_loss)
+            test_losses.append(avg_test_loss)
+
+            print(f'Epoch {epoch + 1}/{self.epochs} Train Loss: {avg_train_loss} Test Loss: {avg_test_loss}')
+
+        print('Ending training...')
+
+        model_name = 'model' + '_' + 'ep' + str(self.epochs) + '_' + 'lr' + str(self.learning_rate) + '.pth'
+        save(self.state_dict(), model_name)
+
+        return train_losses, test_losses
+
+    def evaluate(self, dataloader):
+        self.eval()
+
+        data_labels = []
+        data_outputs = []
+
+        with no_grad():
+            for inputs, labels in dataloader:
+        
+                ids = inputs[:, 0].to(self.device, dtype=long)
+                masks = inputs[:, 1].to(self.device, dtype=long)
+                tti = inputs[:, 2].to(self.device, dtype=long)
+                labels = labels.squeeze().to(self.device, dtype=long)
+
+                assert ids.shape == masks.shape, 'Ids != Masks'
+                assert masks.shape == tti.shape, 'Masks != Ttis'
+                assert ids.shape == tti.shape, 'Ids != Ttis'
+
+                assert ids.shape[0] == labels.shape[0], 'inputs and labels are incompatible'
+
+                outputs = self(ids, masks, tti)
+                outputs = nn.functional.softmax(outputs, dim=1)
+                outputs = argmax(outputs, dim=1)
+
+                data_labels.extend(labels.cpu().detach().numpy().tolist())
+                data_outputs.extend(outputs.cpu().detach().numpy().tolist())
+
+    
+        target_names = ['Easy', 'Medium', 'Hard']
+        macro_f1 = f1_score(data_labels, data_outputs, average='macro')
+        cm = confusion_matrix(data_labels, data_outputs)
+        disp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=['Easy', 'Medium', 'Hard'])
+        print(f'Macro F1: {macro_f1}')
+        disp.plot()
+        plt.show()
+
+
+    def predict():
+        return 1

workspace/Models/model.py

@@ -0,0 +1,22 @@
+import torch
+from torch import long
+
+def calc_loss(model, dataloader, criterion, device):
+
+    with torch.no_grad():
+        total_loss = 0.0
+        for inputs, labels in dataloader:
+            ids = inputs[:, 0].to(device, dtype=long)
+            masks = inputs[:, 1].to(device, dtype=long)
+            tti = inputs[:, 2].to(device, dtype=long)
+            labels = labels.squeeze().to(device, dtype=long)
+
+            outputs = model(ids, masks, tti)
+
+            loss = criterion(outputs, labels)
+            
+            total_loss += loss.item()
+        
+        average_loss = total_loss / len(dataloader)
+        
+        return average_loss

workspace/loss.py

@@ -0,0 +1,11 @@
+import matplotlib.pyplot as plt
+
+def show_loss_evolution(num_epochs, train_losses, test_losses):
+    plt.plot(range(1, num_epochs + 1), train_losses, marker='o', linestyle='-', color='b', label='Train')
+    plt.plot(range(1, num_epochs + 1), test_losses, marker='x', linestyle='-', color='g', label='Test')
+
+    plt.title('Learning Curve')
+    plt.xlabel('Epochs')
+    plt.ylabel('Average Loss')
+    plt.legend()
+    plt.show()