sentiment_analysis_(mp).py

# -*- coding: utf-8 -*-
"""Sentiment Analysis (MP)

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/11IYUIE4aGHeCjMVCezLqM9hFlzzU7jET
"""

!pip install transformers



!pip install fast_ml==3.68

!pip install datasets

!pip install accelerate

!pip install transformers -U

import numpy as np
import pandas as pd
from fast_ml.model_development import train_valid_test_split
from transformers import Trainer, TrainingArguments, AutoConfig, AutoTokenizer, AutoModelForSequenceClassification
import torch
from torch import nn
from torch.nn.functional import softmax
from sklearn.metrics import classification_report
from sklearn.preprocessing import LabelEncoder
import datasets

DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print (f'Device Availble: {DEVICE}')

import os
os.environ['KAGGLE_CONFIG_DIR'] = '/content/'

! kaggle datasets download -d cosmaos98/twitter-and-reddit-sentimental-analysis-dataset

import zipfile

with zipfile.ZipFile('/content/twitter-and-reddit-sentimental-analysis-dataset.zip', 'r') as zip_ref:
    zip_ref.extractall('/content/')

df = pd.read_csv('/content/Reddit_Data.csv')
# df.drop(columns = ['Unnamed: 0'], inplace = True)
df.head()

df1 = pd.read_csv('/content/Twitter_Data.csv')
# df.drop(columns = ['Unnamed: 0'], inplace = True)
df1.head()

le = LabelEncoder()
df1['category'] = le.fit_transform(df1['category'])
df1.head(100)

print(le.classes_)

(train_texts, train_labels,
 val_texts, val_labels,
 test_texts, test_labels) = train_valid_test_split(df1, target = 'category', train_size=0.8, valid_size=0.1, test_size=0.1)

train_texts = train_texts['clean_text'].to_list()
train_labels = train_labels.to_list()
val_texts = val_texts['clean_text'].to_list()
val_labels = val_labels.to_list()
test_texts = test_texts['clean_text'].to_list()
test_labels = test_labels.to_list()

class DataLoader(torch.utils.data.Dataset):
    def __init__(self, sentences=None, labels=None):
        self.sentences = sentences
        self.labels = labels
        self.tokenizer = AutoTokenizer.from_pretrained('distilbert-base-uncased')

    def __getitem__(self, idx):
        encoding = self.tokenizer(self.sentences[idx], padding=True, truncation=True, return_tensors='pt')
        item = {
            'input_ids': encoding['input_ids'].squeeze(),
            'attention_mask': encoding['attention_mask'].squeeze(),
            'labels': torch.tensor(self.labels[idx]) if self.labels else None
        }
        return item

    def __len__(self):
        return len(self.sentences)

train_dataset = DataLoader(train_texts, train_labels)
val_dataset = DataLoader(val_texts, val_labels)
test_dataset = DataLoader(test_texts, test_labels)

print (train_dataset.__getitem__(0))

f1 = datasets.load_metric('f1')
accuracy = datasets.load_metric('accuracy')
precision = datasets.load_metric('precision')
recall = datasets.load_metric('recall')
def compute_metrics(eval_pred):
    metrics_dict = {}
    predictions, labels = eval_pred
    predictions = np.argmax(predictions, axis=1)

    metrics_dict.update(f1.compute(predictions = predictions, references = labels, average = 'macro'))
    metrics_dict.update(accuracy.compute(predictions = predictions, references = labels))
    metrics_dict.update(precision.compute(predictions = predictions, references = labels, average = 'macro'))
    metrics_dict.update(recall.compute(predictions = predictions, references = labels, average = 'macro'))
    return metrics_dict

id2label = {idx: int(label) for idx, label in enumerate(le.classes_)}
label2id = {int(label): idx for idx, label in enumerate(le.classes_)}
config = AutoConfig.from_pretrained('distilbert-base-uncased',
                                    num_labels=4,
                                    id2label=id2label,
                                    label2id=label2id)
model = AutoModelForSequenceClassification.from_config(config)

print (config)

print (model)

# !pip install accelerate -U

# ! pip install transformers[torch]

# !pip install transformers[torch] accelerate

import transformers
import accelerate
print(transformers.__version__)
print(accelerate.__version__)

!pip uninstall transformers -y
!pip uninstall accelerate -y

!pip install transformers==4.11.3
!pip install accelerate==0.20.1

from transformers import TrainingArguments

!pip install transformers[torch]

! pip install accelerate -U

import torch
from transformers import TrainingArguments

# Specify the directory paths for output and logging
output_dir = '/kaggle/working/results'
logging_dir = '/kaggle/working/logs'

# Define the training arguments
training_args = TrainingArguments(
    output_dir=output_dir,
    num_train_epochs=25,
    per_device_train_batch_size=64,
    per_device_eval_batch_size=64,
    warmup_steps=500,
    weight_decay=0.05,
    report_to='none',
    evaluation_strategy='steps',
    logging_dir=logging_dir,
    logging_steps=50
)

training_args = TrainingArguments(
    output_dir='/kaggle/working/results',
    num_train_epochs=25,
    per_device_train_batch_size=64,
    per_device_eval_batch_size=64,
    warmup_steps=500,
    weight_decay=0.05,
    report_to='none',
    evaluation_strategy='steps',
    logging_dir='/kagge/working/logs',
    logging_steps=50)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset,
    compute_metrics=compute_metrics)

trainer.train()

eval_results = trainer.predict(test_dataset)

print (test_results.predictions)

print (test_results.label_ids)

print (test_results.metrics)

label2id_mapper = model.config.id2label
proba = softmax(torch.from_numpy(test_results.predictions))
pred = [label2id_mapper[i] for i in torch.argmax(proba, dim = -1).numpy()]
actual = [label2id_mapper[i] for i in test_results.label_ids]

class_report = classification_report(actual, pred, output_dict = True)
pd.DataFrame(class_report)

trainer.save_model('/kaggle/working/sentiment_model')

import pandas as pd
import numpy as np
from transformers import Trainer, TrainingArguments, AutoConfig, AutoTokenizer, AutoModelForSequenceClassification
import torch
from torch import nn
from torch.nn.functional import softmax

DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print (f'Device Availble: {DEVICE}')

class DataLoader(torch.utils.data.Dataset):
    def __init__(self, sentences=None, labels=None):
        self.sentences = sentences
        self.labels = labels
        self.tokenizer = AutoTokenizer.from_pretrained('distilbert-base-uncased')

        if bool(sentences):
            self.encodings = self.tokenizer(self.sentences,
                                            truncation = True,
                                            padding = True)

    def __getitem__(self, idx):
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}

        if self.labels == None:
            item['labels'] = None
        else:
            item['labels'] = torch.tensor(self.labels[idx])
        return item
    def __len__(self):
        return len(self.sentences)


    def encode(self, x):
        return self.tokenizer(x, return_tensors = 'pt').to(DEVICE)

class SentimentModel():

    def __init__(self, model_path):

        self.model = AutoModelForSequenceClassification.from_pretrained(model_path).to(DEVICE)
        args =  TrainingArguments(output_dir='/kaggle/working/results', per_device_eval_batch_size=64)
        self.batch_model = Trainer(model = self.model, args= args)
        self.single_dataloader = DataLoader()

    def batch_predict_proba(self, x):

        predictions = self.batch_model.predict(DataLoader(x))
        logits = torch.from_numpy(predictions.predictions)

        if DEVICE == 'cpu':
            proba = torch.nn.functional.softmax(logits, dim = 1).detach().numpy()
        else:
            proba = torch.nn.functional.softmax(logits, dim = 1).to('cpu').detach().numpy()
        return proba


    def predict_proba(self, x):

        x = self.single_dataloader.encode(x).to(DEVICE)
        predictions = self.model(**x)
        logits = predictions.logits

        if DEVICE == 'cpu':
            proba = torch.nn.functional.softmax(logits, dim = 1).detach().numpy()
        else:
            proba = torch.nn.functional.softmax(logits, dim = 1).to('cpu').detach().numpy()
        return proba

df = pd.read_csv('/kaggle/input/womens-ecommerce-clothing-reviews/Womens Clothing E-Commerce Reviews.csv')
df.drop(columns = ['Unnamed: 0'], inplace = True)
df_reviews = df.loc[:, ['Review Text', 'Rating']].dropna()
df_reviews['Rating'] = df_reviews['Rating'].apply(lambda x: f'{x} Stars' if x != 1 else f'{x} Star')
df_reviews.head()

batch_sentences = df_reviews.sample(n = 10000, random_state = 1)['Review Text'].to_list()
single_sentence = df_reviews.sample(n = 1, random_state = 1)['Review Text'].to_list()[0]

sentiment_model = SentimentModel('../input/fine-tune-huggingface-sentiment-analysis/sentiment_model')

single_sentence_probas = sentiment_model.predict_proba(single_sentence)
id2label = sentiment_model.model.config.id2label
predicted_class_label = id2label[np.argmax(single_sentence_probas)]
print (predicted_class_label)

batch_sentence_probas = sentiment_model.batch_predict_proba(batch_sentences)
predicted_class_labels = [id2label[i] for i in np.argmax(batch_sentence_probas, axis = -1)]

# Commented out IPython magic to ensure Python compatibility.
# %%time
# for sentence in batch_sentences:
#     single_sentence_probas = sentiment_model.predict_proba(sentence)
# %%time
# batch_sentence_probas = sentiment_model.batch_predict_proba(batch_sentences)