<jupyter_start><jupyter_text>1. Importing libraries and loading Data  1.1 Installing necessary libraries<jupyter_code>"""
%pip install emoji
%pip install tensorflow
%pip install transformers
%pip install pandas
%pip install sklearn
%pip install matplotlib
%pip install seaborn
"""

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
import emoji
import re
import string
from transformers import TFBertModel, BertTokenizerFast, BertConfig
import tensorflow as tf
from keras.layers import Input, Dropout, Dense, BatchNormalization
from keras.models import Model
from sklearn.metrics import f1_score, recall_score, precision_score
from tensorflow.keras.utils import plot_model
from keras.initializers import TruncatedNormal
import keras.backend as K<jupyter_output><empty_output><jupyter_text>1.2 Plot Defaults<jupyter_code>colors = sns.color_palette('rocket_r')
sns.set_palette('rocket_r')<jupyter_output><empty_output><jupyter_text>1.3 Helper Functions<jupyter_code>def idx2class(idx_list):
    """
    This function converts a list of class indices to a list of class labels.

    Parameters
    ----------
    idx_list : list
        List of class indices.
    
    Returns
    -------
    class_list : list
        List of class labels.
    """
    arr = []
    for i in idx_list:
        arr.append(labels[int(i)])
    return arr

def EmotionMapping(list_of_emotions):
    list = []  
    for i in list_of_emotions:
        if i in ekman_map['anger']:
            list.append('anger')
        if i in ekman_map['disgust']:
            list.append('disgust')
        if i in ekman_map['fear']:
            list.append('fear')
        if i in ekman_map['joy']:
            list.append('joy')
        if i in ekman_map['sadness']:
            list.append('sadness')
        if i in ekman_map['surprise']:
            list.append('surprise')
        if i == 'neutral':
            list.append('neutral')      
    return list

def SentimentMapping(list_of_emotions):
    list = []  
    for i in list_of_emotions:
        if i in sentiment_map['positive']:
            list.append('positive')
        if i in sentiment_map['negative']:
            list.append('negative')
        if i in sentiment_map['ambiguous']:
            list.append('ambiguous')
    return list<jupyter_output><empty_output><jupyter_text>1.4 Loading data<jupyter_code>train_url = 'https://github.com/google-research/google-research/raw/master/goemotions/data/train.tsv'
valid_url = 'https://github.com/google-research/google-research/raw/master/goemotions/data/dev.tsv'
test_url = 'https://github.com/google-research/google-research/raw/master/goemotions/data/test.tsv'
train_df = pd.read_csv(train_url, sep='\t', encoding='utf-8',
                       names=['text', 'emotion', 'annotator'], header=None)
valid_df = pd.read_csv(valid_url, sep='\t', encoding='utf-8',
                       names=['text', 'emotion', 'annotator'], header=None)
test_df = pd.read_csv(test_url, sep='\t', encoding='utf-8',
                      names=['text', 'emotion', 'annotator'], header=None)
train_df.head(2)
train_df.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 43410 entries, 0 to 43409
Data columns (total 3 columns):
 #   Column     Non-Null Count  Dtype 
---  ------     --------------  ----- 
 0   text       43410 non-null  object
 1   emotion    43410 non-null  object
 2   annotator  43410 non-null  object
dtypes: object(3)
memory usage: 1017.5+ KB<jupyter_text>1.5 Preprocessing Column 2 "annotator" is unnecessary, so we can drop it.<jupyter_code>train_df.drop('annotator', axis=1, inplace=True)
valid_df.drop('annotator', axis=1, inplace=True)
test_df.drop('annotator', axis=1, inplace=True)<jupyter_output><empty_output><jupyter_text>Dictionaries for mapping emotions to indices and vice versa. The variable `ekman_map` is used to map 27 emotions to 7 emotions. This is done to reduce the number of classes.The 27 emotions can also be mapped to the 3 emotions using the `sentiment_map` dictionary for sentiment analysis tasks.<jupyter_code>labels = {
    0: 'admiration',
    1: 'amusement',
    2: 'anger',
    3: 'annoyance',
    4: 'approval',
    5: 'caring',
    6: 'confusion',
    7: 'curiosity',
    8: 'desire',
    9: 'disappointment',
    10: 'disapproval',
    11: 'disgust',
    12: 'embarrassment',
    13: 'excitement',
    14: 'fear',
    15: 'gratitude',
    16: 'grief',
    17: 'joy',
    18: 'love',
    19: 'nervousness',
    20: 'optimism',
    21: 'pride',
    22: 'realization',
    23: 'relief',
    24: 'remorse',
    25: 'sadness',
    26: 'surprise',
    27: 'neutral'
}

ekman_map = {
    'anger': ['anger', 'annoyance', 'disapproval'],
    'disgust': ['disgust'],
    'fear': ['fear', 'nervousness'],
    'joy': ['joy', 'amusement', 'approval', 'excitement', 'gratitude',  'love', 'optimism', 'relief', 'pride', 'admiration', 'desire', 'caring'],
    'sadness': ['sadness', 'disappointment', 'embarrassment', 'grief',  'remorse'],
    'surprise': ['surprise', 'realization', 'confusion', 'curiosity'],
    'neutral': ['neutral']
}

sentiment_map = {
    "positive": ["amusement", "excitement", "joy", "love", "desire", "optimism", "caring", "pride", "admiration", "gratitude", "relief", "approval"],
    "negative": ["fear", "nervousness", "remorse", "embarrassment", "disappointment", "sadness", "grief", "disgust", "anger", "annoyance", "disapproval"],
    "ambiguous": ["realization", "surprise", "curiosity", "confusion", "neutral"]
}<jupyter_output><empty_output><jupyter_text>First, let's extract all emotions from the each example and store them in a list.<jupyter_code>train_df['list of emotions'] = train_df['emotion'].apply(lambda x: x.split(','))
test_df['list of emotions'] = test_df['emotion'].apply(lambda x: x.split(','))
valid_df['list of emotions'] = valid_df['emotion'].apply(lambda x: x.split(','))<jupyter_output><empty_output><jupyter_text>We can then apply index to class mapping to get the class labels for each row<jupyter_code>train_df['emotion'] = train_df['list of emotions'].apply(lambda x: idx2class(x))
test_df['emotion'] = test_df['list of emotions'].apply(lambda x: idx2class(x))
valid_df['emotion'] = valid_df['list of emotions'].apply(lambda x: idx2class(x))<jupyter_output><empty_output><jupyter_text>Finally, we can reduce the number of classes to 7 by using the EmotionMapping function.<jupyter_code>train_df['ekman_emotion'] = train_df['emotion'].apply(lambda x: EmotionMapping(x))
test_df['ekman_emotion'] = test_df['emotion'].apply(lambda x: EmotionMapping(x))
valid_df['ekman_emotion'] = valid_df['emotion'].apply(lambda x: EmotionMapping(x))
train_df.head(10)
def clean_text(text):
    """
    This function cleans the text in the dataframe and returns a list of cleaned text.
    text: a string

    return: modified initial string
    """
    # Removing Emojis
    text = emoji.demojize(text)  # remove emojis
    text = str(text).lower()  # text to lower case
    text = re.sub(r'[%s]' % re.escape(string.punctuation), ' ', text)  # remove punctuation
    return text<jupyter_output><empty_output><jupyter_text>One hot encoding of emotions<jupyter_code>for i in ekman_map:
    train_df[i] = train_df['ekman_emotion'].apply(lambda x: 1 if i in x else 0)
    test_df[i] = test_df['ekman_emotion'].apply(lambda x: 1 if i in x else 0)
    valid_df[i] = valid_df['ekman_emotion'].apply(lambda x: 1 if i in x else 0)
train_df.head(10)<jupyter_output><empty_output><jupyter_text>1.6 Visualization Bar plot of distribution of emotions<jupyter_code>labels_summary = train_df.iloc[:, 4:].sum()
labels_summary.sort_values(ascending=False, inplace=True)

fig = plt.figure(figsize=(12, 8))
sns.barplot(x=labels_summary.index,
            y=labels_summary.values, palette='rocket_r')
plt.xticks(rotation=45)
plt.ylabel('Frequency')
plt.show()<jupyter_output><empty_output><jupyter_text>Number of emotions in each sample<jupyter_code>train_df['n_emotions'] = train_df.iloc[:, 4:].apply(lambda x: x.sum(), axis=1)

fig = plt.figure(figsize=(12, 8))
sns.countplot(x='n_emotions', data=train_df, palette='rocket_r')
plt.xticks(rotation=45)
plt.title('Number of emotions per sample')
plt.ylabel('Frequency')
plt.xlabel('Number of emotions')
plt.show()<jupyter_output><empty_output><jupyter_text>Distribution of text length in the train set<jupyter_code>full_text = pd.concat([train_df['text'], valid_df['text'], test_df['text']])
lengths = full_text.apply(lambda x: len(x.split()))
fig = plt.figure(figsize=(12, 10))
sns.displot(lengths, kde=True, rug=False, color=colors[5])
plt.title('Distribution of Text Lengths')
plt.xlabel('Text Length')
plt.ylabel('Frequency')
plt.xlim(0, 40)
plt.show()<jupyter_output><empty_output><jupyter_text>2. Model  2.1 Base model config  Computing max length of samples`max_length` variable is used to limit the length of the input text that is fed to the model. The sequence will be padded with the `` token if the length of the sequence is less than `max_length` and the sequence will be truncated if the length of the sequence is more than `max_length`. This is done to ensure that the model can handle any size of input text.<jupyter_code>full_text = pd.concat([train_df['text'], valid_df['text'], test_df['text']])
max_length = full_text.apply(lambda x: len(x.split())).max()
max_length<jupyter_output><empty_output><jupyter_text>I am going to use Google's BERT base model which contains 110M parameters.<jupyter_code>model_name = 'bert-base-uncased'
config = BertConfig.from_pretrained(model_name, output_hidden_states=False)
tokenizer = BertTokenizerFast.from_pretrained(pretrained_model_name_or_path = model_name, config = config)
transformer_model = TFBertModel.from_pretrained(model_name, config = config)<jupyter_output>Some layers from the model checkpoint at bert-base-uncased were not used when initializing TFBertModel: ['mlm___cls', 'nsp___cls']
- This IS expected if you are initializing TFBertModel from the checkpoint of a model trained on another task or with another architecture (e.g. initializing a BertForSequenceClassification model from a BertForPreTraining model).
- This IS NOT expected if you are initializing TFBertModel from the checkpoint of a model that you expect to be exactly identical (initializing a BertForSequenceClassification model from a BertForSequenceClassification model).
All the layers of TFBertModel were initialized from the model checkpoint at bert-base-uncased.
If your task is similar to the task the model of the checkpoint was trained on, you can already use TFBertModel for predictions without further training.<jupyter_text>2.2 Model architecture model takes three inputs that result from tokenization:- `input_ids`: indices of input sequence tokens in the vocabulary- `token_type_ids`: Segment token indices to indicate first and second portions of the inputs. 0 for sentence A and 1 for sentence B- `attention mask`: Mask to avoid performing attention on padding token indices. 0 for masked and 1 for not maskedI have a sigmoided output layer in the model because it is more appropriate than a softmax layer. This is because I are trying to predict the probability of each label and not the label itself.<jupyter_code>def my_model(n_labels):

  # Load the MainLayer
  bert = transformer_model.layers[0]

  ## INPUTS
  input_ids = Input(shape=(max_length,), name='input_ids', dtype='int32')
  attention_mask = Input(shape=(max_length,), name='attention_mask', dtype='int32')
  token_type_ids = Input(shape=(max_length,), name='token_type_ids', dtype='int32')
  inputs = {'input_ids': input_ids, 'attention_mask': attention_mask, 'token_type_ids': token_type_ids}

  ## LAYERS
  bert_model = bert(inputs)[1]
  dropout = Dropout(config.hidden_dropout_prob, name='pooled_output')
  pooled_output = dropout(bert_model, training=False)

  ## OUTPUT
  emotion = Dense(units=n_labels, activation='sigmoid', kernel_initializer=TruncatedNormal(stddev=config.initializer_range), name='emotion')(pooled_output)
  outputs = emotion

  model = Model(inputs=inputs, outputs=outputs, name='BERT_Emotion_Classifier')

  return model
model = my_model(len(ekman_map))
model.summary()
plot_model(model, show_shapes=True, dpi=300)<jupyter_output><empty_output><jupyter_text>2.3 Data tokenization<jupyter_code>## Train 
x_train = train_df['text']
y_train = train_df.loc[:, ekman_map.keys()].values

train_tokenized = tokenizer(
    text = list(x_train),
    add_special_tokens = True,
    max_length = max_length,
    padding = 'max_length',
    truncation = True,
    return_tensors = 'tf',
    return_attention_mask = True,
    return_token_type_ids = True
)

## Test
x_test = test_df['text']
y_test = test_df.loc[:, ekman_map.keys()].values

test_tokenized = tokenizer(
    text = list(x_test),
    add_special_tokens = True,
    max_length = max_length,
    padding = 'max_length',
    truncation = True,
    return_tensors = 'tf',
    return_attention_mask = True,
    return_token_type_ids = True
)

## Validation
x_valid = valid_df['text']
y_valid = valid_df.loc[:, ekman_map.keys()].values

valid_tokenized = tokenizer(
    text = list(x_valid),
    add_special_tokens = True,
    max_length = max_length,
    padding = 'max_length',
    truncation = True,
    return_tensors = 'tf',
    return_attention_mask = True,
    return_token_type_ids = True
)<jupyter_output><empty_output><jupyter_text>2.4 Creating BERT compatible inputs<jupyter_code>tf_train = {'input_ids': train_tokenized['input_ids'], 'attention_mask': train_tokenized['attention_mask'], 'token_type_ids': train_tokenized['token_type_ids']}
tf_test = {'input_ids': test_tokenized['input_ids'], 'attention_mask': test_tokenized['attention_mask'], 'token_type_ids': test_tokenized['token_type_ids']}
tf_valid = {'input_ids': valid_tokenized['input_ids'], 'attention_mask': valid_tokenized['attention_mask'], 'token_type_ids': valid_tokenized['token_type_ids']}
train = tf.data.Dataset.from_tensor_slices((tf_train, y_train)).batch(80)
valid = tf.data.Dataset.from_tensor_slices((tf_valid, y_valid)).batch(80)
test = tf.data.Dataset.from_tensor_slices((tf_test, y_test)).batch(80)
learning_rate = tf.keras.optimizers.schedules.ExponentialDecay(
    initial_learning_rate=5e-5,
    decay_rate=0.7,
    decay_steps=340,
    staircase=True)

optimizer = tf.keras.optimizers.Adam(learning_rate=learning_rate)

loss = tf.keras.losses.BinaryCrossentropy(from_logits=False)
K.clear_session()<jupyter_output><empty_output><jupyter_text>Prior experiments with BERT showed that the model starts to overfit after ~2 epochs and Tanh performed significantly worse than sigmoid.<jupyter_code>model.compile(optimizer=optimizer, loss=loss, metrics=['accuracy'])
history = model.fit(train, epochs=2, validation_data=valid)
model.save_weights('./models/sigmoid_bert.h5')<jupyter_output><empty_output><jupyter_text>3. Evaluation When dealing with unbalanced data, it is essential to mini-batch train the model instead of training it on all the data. This helps to prevent the model from overfitting the minority class. It is also essential to be thoughtful about what metric is being used for model evaluation. When dealing with unbalanced data, accuracy is not a good metric, as the model can predict the majority class every time and still have high accuracy. Instead, it is crucial to use the precision/recall or the F1 score, as these metrics consider false positives and false negatives.<jupyter_code>model = my_model(len(ekman_map))
model.load_weights('./models/sigmoid_bert.h5')
THRESHOLD = 0.83

y_pred = model.predict(test)
probabilities = y_pred

probabilities = pd.DataFrame(probabilities, columns=ekman_map.keys())
probabilities.index = x_test
probabilities.reset_index(inplace=True)
probabilities.head(10)
y_pred = np.where(y_pred > THRESHOLD, 1, 0)

recall = []
f1 = []
precision = []
emotions = ekman_map.keys()

for i in range(len(emotions)):
    f1.append(f1_score(y_test[:, i], y_pred[:, i], average='macro'))
    precision.append(precision_score(y_test[:, i], y_pred[:, i], average='macro'))

results = pd.DataFrame({'precision': precision, 'f1': f1})
results.index = emotions

means = {'precision': np.mean(precision), 'f1': np.mean(f1)}
means = pd.DataFrame(means, index=['mean'])

pd.concat([results, means], axis=0)<jupyter_output><empty_output><jupyter_text>3.1 Optimization Finding the best value of Threshold. I chose f1-score as the main metric because it is more robust than precision and recall alone.<jupyter_code>best_threshold = 0
best_f1 = 0
pred = model.predict(test)

for threshold in np.arange(0.30, 0.99, 0.01):
    preds = np.where(pred > threshold, 1, 0)

    f1 = f1_score(y_test, preds, average='macro', zero_division=0)

    if f1 > best_f1:
        best_threshold = threshold
        best_f1 = f1
    else:
        continue
print(f'Best threshold: {best_threshold}\nBest f1: {best_f1}')
THRESHOLD = 0.39<jupyter_output><empty_output><jupyter_text>4. Make Predictions<jupyter_code>def pred(text, model, THRESHOLD):

    text = [clean_text(text) for text in text]
    
    tokenized = tokenizer(
        text = text,
        add_special_tokens = True,
        max_length = max_length,
        padding = 'max_length',
        truncation = True,
        return_tensors = 'tf',
        return_attention_mask = True,
        return_token_type_ids = True
    )

    tf_test = {'input_ids': tokenized['input_ids'], 'attention_mask': tokenized['attention_mask'], 'token_type_ids': tokenized['token_type_ids']}

    pred = model.predict(tf_test)

    probabilities = pred
    probabilities = pd.DataFrame(probabilities, columns=ekman_map.keys())
    probabilities.index = text
    probabilities.reset_index(inplace=True)

    pred = np.where(pred > THRESHOLD, 1, 0)

    pred = pd.DataFrame(pred, columns=ekman_map.keys())
    pred['emotion'] = pred.iloc[:, 1:].idxmax(axis=1)
    pred.drop(columns=emotions, inplace=True)
    pred.index = text
    pred.reset_index(inplace=True)

    return pred, probabilities
result, probabilities = pred(['A Ukrainian woman who escaped Russias assault on Mariupol says troops were targeting apartment buildings as if they were playing a computer game', 'I often go to parks to walk and destress and enjoy nature', 'How can this be', 'This is the worst muffin ive ever had'], model, THRESHOLD)
result
probabilities<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12391600763566948, 'fraction_numerical': 0.013144259612762476, 'mean_word_length': 4.920568291895383, 'pattern_counts': {'":': 3, '<': 88, '<?xml version=': 0, '>': 92, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '22', 'dir': 'jupyter-structured-clean-dedup', 'id': '149700', 'license': 'MIT', 'n_tokens_mistral': 6255, 'n_tokens_neox': 6001, 'n_words': 1564, 'path': 'notebooks/Textual_emotion_detection.ipynb', 'repo_name': 'vineetver/Textual-Emotion-Detection', 'size': '18335'}