# PyTorch Imports import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import Dataset, DataLoader # scikit-learn Imports from sklearn.metrics import accuracy_score, confusion_matrix from sklearn.model_selection import train_test_split #Gensim Imports import gensim # Bert improts from transformers import BertForSequenceClassification, BertTokenizer, BertPreTrainedModel, AdamW #Default imports (pandas, numpy, matplotlib, etc.) import pandas as pd import numpy as np import matplotlib.pyplot as plt ## Select Device if torch.cuda.is_available(): DEVICE = torch.device("cuda") else: DEVICE = torch.device("cpu") class SimpleHumorDataset(Dataset): def __init__(self,tokenizer,dataframe,max_length=256): super().__init__() self.tokenizer = tokenizer self.max_length = max_length self.text = dataframe['text'].tolist() self.labels = dataframe['is_humor'].unique().tolist() def __getitem__(self,idx): text = self.text labels = self.labels encoding = self.tokenizer.encode_plus( text, add_special_tokens=True, padding="max_length", trunction = True, return_attention_mask = True, return_token_type_ids = False, max_length=self.max_length, truncation = True, return_tensors = 'pt' ) input_ids = encoding['input_ids'].flatten() attention_mask = encoding['attention_mask'].flatten() return { 'input_ids': torch.as_tensor(input_ids,dtype=torch.long), 'attention_mask':torch.as_tensor(attention_mask,dtype=torch.long), 'labels':torch.as_tensor(self.labels,dtype=torch.float), 'text':text } def __len__(self): return len(self.labels) class CustomBert(nn.Module): def __init__(self,dropout,num_layers=2): super(CustomBert,self).__init__() self.bfsc = BertForSequenceClassification.from_pretrained("google-bert/bert-base-uncased"), self.bert_model = self.bfsc[0] # Add Custom Layers self.dropout = nn.Dropout(dropout), self.dropout = self.dropout[0] self.ln1 = nn.Linear(2,2), self.ln1 = self.ln1[0] self.sm1 = nn.Sigmoid() def forward(self, input_ids, attention_mask): seq_out = self.bert_model(input_ids, attention_mask = attention_mask) x = self.dropout(seq_out.logits) x = self.ln1(x) return self.sm1(x) def training_loop(model,criterion,optimizer,train_loader): model.to(DEVICE) model.train() total_loss = 0 for index, train_batch in enumerate(train_loader): # Set Gradient to Zero optimizer.zero_grad() # Unpack batch values and "push" it to GPU input_ids, att_mask, labels,_ = train_batch.values() input_ids, att_mask, labels = input_ids.to(DEVICE),att_mask.to(DEVICE),labels.to(DEVICE) # Feed Model with Data outputs = model(input_ids, attention_mask=att_mask) # print(f"Output Tensor: {outputs}") loss = criterion(outputs,labels.float()) loss.backward() optimizer.step() total_loss+=loss.item() print(f"Total Loss is {(total_loss/len(train_loader)):.4f}") return (total_loss/len(train_loader)) def eval_loop(model,criterion,validation_loader): model.eval() total_loss = 0.0 total_acc = 0.0 with torch.no_grad(): for val_batch in validation_loader: input_ids, att_mask ,labels,_ = val_batch.values() input_ids, att_mask, labels = input_ids.to(DEVICE),att_mask.to(DEVICE), labels.to(DEVICE) outputs = model(input_ids,attention_mask=att_mask) loss = criterion(outputs,labels) total_loss += loss.item() predictions = torch.argmax(outputs,dim=1) total_acc += (predictions == labels).sum().item() print(f"Total Loss: {total_loss/len(validation_loader)} ### Test Accuracy {total_acc/len(validation_loader)*100}%") def generate_tokens(tokenizer,raw_data): return tokenizer.encode_plus( raw_data, add_special_tokens=True, padding="max_length", return_attention_mask = True, return_token_type_ids = False, max_length=128, truncation = True, return_tensors = 'pt' ) if __name__ == "__main__": torch.manual_seed(501) # Initialize Bert Model with dropout probability and Num End Layers mybert = CustomBert(0.1) print("Bert Initialized") # Set Max Epoch Amount EPOCH = 50 # Read Raw Data from csv and save as DataFrame df = pd.read_csv("./data/hack.csv",encoding="latin1") print("Raw Data read") # Initialize BertTokenizer from Pretrained tokenizer = BertTokenizer.from_pretrained("bert-base-uncased",do_lower_case=True) print("Tokenizer Initialized") #Split DataFrame into Train and Test Sets train,test = train_test_split(df,random_state=501,test_size=.2) print("Splitted Data in Train and Test Sets") # Create Custom Datasets for Train and Test train_data = SimpleHumorDataset(tokenizer,train) test_data = SimpleHumorDataset(tokenizer,test) print("Custom Datasets created") # Initialize Dataloader with Train and Test Sets train_loader = DataLoader(dataset=train_data,batch_size=16,shuffle=True) test_loader = DataLoader(dataset=test_data,batch_size=len(test_data)) print("DataLoaders created") # Set criterion to BCELoss (Binary Cross Entropy) and define Adam Optimizer with model parameters and learning rate criterion_bce = nn.BCELoss() optimizer_adam = optim.Adam(mybert.parameters(), lr = 3e-5) # Set Scheduler for dynamically Learning Rate adjustment scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer_adam) loss_values = [] for epoch in range(EPOCH): print(f"For {epoch+1} the Scores are: ") loss_values.append(training_loop(mybert,optimizer=optimizer_adam,criterion=criterion_bce,train_loader=train_loader)) # bert.eval_loop(criterion=criterion_bce,validation_loader=test_loader) scheduler.step(.1) # Visualize Training Loss plt.plot(loss_values) plt.hlines(np.mean(loss_values),xmin=0,xmax=EPOCH,colors='red',linestyles="dotted",label="Average Loss") plt.title("Training Loss") plt.xlabel("Num Epochs") plt.ylabel("Total Loss of Epoch") plt.show()