192 lines
7.2 KiB
Python
192 lines
7.2 KiB
Python
# 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
|
|
# Bert imports
|
|
from transformers import BertForSequenceClassification, BertTokenizer, 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'].tolist()
|
|
|
|
def __getitem__(self,idx):
|
|
text = self.text[idx]
|
|
labels = self.labels[idx]
|
|
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(labels,dtype=torch.long),
|
|
'text':text
|
|
}
|
|
|
|
def __len__(self):
|
|
return len(self.labels)
|
|
|
|
|
|
class CustomBert(nn.Module):
|
|
def __init__(self,dropout):
|
|
super(CustomBert,self).__init__()
|
|
|
|
#Bert + Custom Layers (Not a tuple any longer -- idk why)
|
|
self.bfsc = BertForSequenceClassification.from_pretrained("bert-base-uncased")
|
|
self.dropout = nn.Dropout(dropout)
|
|
self.classifier = nn.Linear(2,2)
|
|
self.sm = nn.Sigmoid()
|
|
|
|
def forward(self, input_ids, attention_mask):
|
|
seq_out = self.bfsc(input_ids, attention_mask = attention_mask)
|
|
x = self.dropout(seq_out.logits)
|
|
x = self.classifier(x)
|
|
return self.sm(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)#, labels=labels)
|
|
print(f"{index}::{len(train_loader)} -- Output Tensor: {outputs.shape}, Labels {labels.shape}")
|
|
loss = criterion(outputs,labels)
|
|
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, correct = 0.0, 0.0
|
|
total_loss = 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,1)
|
|
print(outputs.squeeze(0))
|
|
print(f"Prediction: {predictions}. VS Actual {labels}")
|
|
print(predictions == labels)
|
|
total += labels.size(0)
|
|
correct += (predictions == labels).sum().item()
|
|
print(f"Total Loss: {total_loss/len(validation_loader)} ### Test Accuracy {correct/total}%")
|
|
return total_loss/len(validation_loader)
|
|
|
|
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)
|
|
# HYPERPARAMETERS
|
|
# Set Max Epoch Amount
|
|
EPOCH = 5
|
|
# DROPOUT-PROBABILITY
|
|
DROPOUT = 0.1
|
|
# BATCHSIZE
|
|
BATCH_SIZE = 8
|
|
#LEARNING RATE
|
|
LEARNING_RATE = 1e-5
|
|
# Initialize Bert Model with dropout probability and Num End Layers
|
|
mybert = CustomBert(DROPOUT)
|
|
print("Bert Initialized")
|
|
|
|
|
|
# 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")
|
|
# val = []
|
|
# Create Custom Datasets for Train and Test
|
|
train_data = SimpleHumorDataset(tokenizer,train)
|
|
# val_data = SimpleHumorDataset(tokenizer,val)
|
|
test_data = SimpleHumorDataset(tokenizer,test)
|
|
print("Custom Datasets created")
|
|
|
|
|
|
# Initialize Dataloader with Train and Test Sets
|
|
train_loader = DataLoader(dataset=train_data,batch_size=BATCH_SIZE,shuffle=True)
|
|
# val_loader = DataLoader(dataset=val_data,batch_size=BATCH_SIZE,shuffle=True)
|
|
test_loader = DataLoader(dataset=test_data,batch_size=BATCH_SIZE,shuffle=False)
|
|
print("DataLoaders created")
|
|
|
|
# Set criterion to BCELoss (Binary Cross Entropy) and define Adam Optimizer with model parameters and learning rate
|
|
criterion_bce = nn.CrossEntropyLoss()
|
|
optimizer_adamW = optim.AdamW(mybert.parameters(), lr = LEARNING_RATE)
|
|
|
|
# Set Scheduler for dynamically Learning Rate adjustment
|
|
# scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer_adam)
|
|
loss_values = []
|
|
eval_values = []
|
|
for epoch in range(EPOCH):
|
|
print(f"For {epoch+1} the Scores are: ")
|
|
loss_values.append(training_loop(mybert,optimizer=optimizer_adamW,criterion=criterion_bce,train_loader=train_loader))
|
|
eval_values.append(eval_loop(mybert,criterion=criterion_bce,validation_loader=test_loader))
|
|
# scheduler.step(min(eval_values))
|
|
|
|
# Visualize Training Loss
|
|
# plt.plot(loss_values)
|
|
plt.plot(eval_values)
|
|
plt.hlines(np.mean(loss_values),xmin=0,xmax=EPOCH,colors='red',linestyles="dotted",label="Average Loss")
|
|
plt.hlines(np.mean(eval_values),xmin=0,xmax=EPOCH,colors='green',linestyles="dashed",label="Average Val Loss")
|
|
plt.title("Test Loss")
|
|
plt.xlabel("Num Epochs")
|
|
plt.ylabel("Total Loss of Epoch")
|
|
plt.show() |