4_Hingbert.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6112e66a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.preprocessing import LabelEncoder\n",
    "from sklearn.metrics import accuracy_score, classification_report, f1_score\n",
    "import torch\n",
    "from transformers import AutoTokenizer, AutoModelForSequenceClassification\n",
    "from torch.utils.data import DataLoader, TensorDataset\n",
    "\n",
    "\n",
    "\n",
    "# Encode labels\n",
    "label_encoder = LabelEncoder()\n",
    "df['emotion_encoded'] = label_encoder.fit_transform(df['emotion'])\n",
    "val_df['emotion_encoded'] = label_encoder.transform(val_df['emotion'])\n",
    "\n",
    "# Load Hinglish RoBERTa tokenizer and model for sequence classification\n",
    "tokenizer = AutoTokenizer.from_pretrained(\"l3cube-pune/hing-bert\")\n",
    "model = AutoModelForSequenceClassification.from_pretrained(\"l3cube-pune/hing-bert\", num_labels=8)  # Set num_labels to the number of emotion classes\n",
    "\n",
    "# Tokenize and prepare data\n",
    "def tokenize_data(data, tokenizer, max_length=128):\n",
    "    tokenized_data = tokenizer(data['utterance'].tolist(), truncation=True, padding='max_length', max_length=max_length, return_tensors='pt')\n",
    "    return tokenized_data\n",
    "\n",
    "train_tokenized = tokenize_data(df, tokenizer)\n",
    "val_tokenized = tokenize_data(val_df, tokenizer)\n",
    "\n",
    "# Convert labels to PyTorch tensors\n",
    "y_train = torch.tensor(df['emotion_encoded'].values)\n",
    "y_val = torch.tensor(val_df['emotion_encoded'].values)\n",
    "\n",
    "# Create DataLoader for training and validation\n",
    "train_dataset = TensorDataset(train_tokenized['input_ids'], train_tokenized['attention_mask'], y_train)\n",
    "val_dataset = TensorDataset(val_tokenized['input_ids'], val_tokenized['attention_mask'], y_val)\n",
    "\n",
    "train_dataloader = DataLoader(train_dataset, batch_size=64, shuffle=True)\n",
    "val_dataloader = DataLoader(val_dataset, batch_size=64, shuffle=False)\n",
    "\n",
    "# Define training loop\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "model.to(device)\n",
    "\n",
    "optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5)\n",
    "criterion = torch.nn.CrossEntropyLoss()\n",
    "\n",
    "num_epochs = 5\n",
    "\n",
    "for epoch in range(num_epochs):\n",
    "    model.train()\n",
    "    for batch in train_dataloader:\n",
    "        input_ids, attention_mask, labels = batch\n",
    "        input_ids, attention_mask, labels = input_ids.to(device), attention_mask.to(device), labels.to(device)\n",
    "\n",
    "        optimizer.zero_grad()\n",
    "        outputs = model(input_ids, attention_mask=attention_mask, labels=labels)\n",
    "        loss = outputs.loss\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "\n",
    "    # Validation\n",
    "    model.eval()\n",
    "    val_predictions = []\n",
    "    val_true_labels = []\n",
    "\n",
    "    with torch.no_grad():\n",
    "        for batch in val_dataloader:\n",
    "            input_ids, attention_mask, labels = batch\n",
    "            input_ids, attention_mask, labels = input_ids.to(device), attention_mask.to(device), labels.to(device)\n",
    "\n",
    "            outputs = model(input_ids, attention_mask=attention_mask)\n",
    "            logits = outputs.logits\n",
    "            predictions = torch.argmax(logits, dim=1).cpu().numpy()\n",
    "\n",
    "            val_predictions.extend(predictions)\n",
    "            val_true_labels.extend(labels.cpu().numpy())\n",
    "\n",
    "    val_accuracy = accuracy_score(val_true_labels, val_predictions)\n",
    "    val_report = classification_report(val_true_labels, val_predictions)\n",
    "    val_weighted_f1 = f1_score(val_true_labels, val_predictions, average='weighted')\n",
    "\n",
    "    print(f\"Epoch {epoch + 1}/{num_epochs}\")\n",
    "    print(\"Validation Accuracy:\", val_accuracy)\n",
    "    print(\"Validation Classification Report:\\n\", val_report)\n",
    "    print(\"Validation Weighted F1 Score:\", val_weighted_f1)\n",
    "    print(\"-\" * 50)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5cb9505d",
   "metadata": {},
   "source": [
    "Hingbert with LSTM"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5bddc1b7",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.preprocessing import LabelEncoder\n",
    "from sklearn.metrics import accuracy_score, classification_report, f1_score\n",
    "import torch\n",
    "from transformers import AutoTokenizer, AutoModel\n",
    "from torch.utils.data import DataLoader, TensorDataset\n",
    "from transformers import AdamW, get_linear_schedule_with_warmup\n",
    "from torch.nn import LSTM, Dropout\n",
    "\n",
    "\n",
    "\n",
    "# Encode labels\n",
    "label_encoder = LabelEncoder()\n",
    "df['emotion_encoded'] = label_encoder.fit_transform(df['emotion'])\n",
    "val_df['emotion_encoded'] = label_encoder.transform(val_df['emotion'])\n",
    "\n",
    "# Load Hinglish BERT tokenizer and model\n",
    "tokenizer = AutoTokenizer.from_pretrained(\"l3cube-pune/hing-bert\")\n",
    "base_model = AutoModel.from_pretrained(\"l3cube-pune/hing-bert\")\n",
    "\n",
    "# Define a custom classifier with an additional LSTM layer\n",
    "class CustomClassifier(torch.nn.Module):\n",
    "    def __init__(self, base_model, num_labels):\n",
    "        super(CustomClassifier, self).__init__()\n",
    "        self.bert = base_model\n",
    "        self.lstm = LSTM(input_size=768, hidden_size=128, batch_first=True, bidirectional=True)\n",
    "        self.dropout = Dropout(0.5)  # Experiment with dropout rate\n",
    "        self.classifier = torch.nn.Linear(256, num_labels)\n",
    "\n",
    "    def forward(self, input_ids, attention_mask):\n",
    "        outputs = self.bert(input_ids, attention_mask=attention_mask)\n",
    "        last_hidden_state = outputs.last_hidden_state\n",
    "        lstm_output, _ = self.lstm(last_hidden_state)\n",
    "        avg_pooled = lstm_output.mean(dim=1)\n",
    "        dropout_output = self.dropout(avg_pooled)\n",
    "        logits = self.classifier(dropout_output)\n",
    "        return logits\n",
    "\n",
    "# Create an instance of the custom classifier\n",
    "model = CustomClassifier(base_model, num_labels=len(label_encoder.classes_))\n",
    "\n",
    "# Tokenize and prepare data\n",
    "def tokenize_data(data, tokenizer, max_length=128):\n",
    "    tokenized_data = tokenizer(data['utterance'].tolist(), truncation=True, padding='max_length', max_length=max_length, return_tensors='pt')\n",
    "    return tokenized_data\n",
    "\n",
    "train_tokenized = tokenize_data(df, tokenizer)\n",
    "val_tokenized = tokenize_data(val_df, tokenizer)\n",
    "\n",
    "# Convert labels to PyTorch tensors\n",
    "y_train = torch.tensor(df['emotion_encoded'].values)\n",
    "y_val = torch.tensor(val_df['emotion_encoded'].values)\n",
    "\n",
    "# Create DataLoader for training and validation\n",
    "train_dataset = TensorDataset(train_tokenized['input_ids'], train_tokenized['attention_mask'], y_train)\n",
    "val_dataset = TensorDataset(val_tokenized['input_ids'], val_tokenized['attention_mask'], y_val)\n",
    "\n",
    "train_dataloader = DataLoader(train_dataset, batch_size=64, shuffle=True)\n",
    "val_dataloader = DataLoader(val_dataset, batch_size=64, shuffle=False)\n",
    "\n",
    "# Define training loop\n",
    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
    "model.to(device)\n",
    "\n",
    "optimizer = AdamW(model.parameters(), lr=1e-5)\n",
    "criterion = torch.nn.CrossEntropyLoss()\n",
    "\n",
    "# Use a learning rate scheduler\n",
    "total_steps = len(train_dataloader) * num_epochs\n",
    "scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps=0, num_training_steps=total_steps)\n",
    "\n",
    "# Define early stopping parameters\n",
    "best_val_accuracy = 0.0\n",
    "patience = 3\n",
    "early_stopping_counter = 0\n",
    "\n",
    "for epoch in range(num_epochs):\n",
    "    model.train()\n",
    "    for batch in train_dataloader:\n",
    "        input_ids, attention_mask, labels = batch\n",
    "        input_ids, attention_mask, labels = input_ids.to(device), attention_mask.to(device), labels.to(device)\n",
    "\n",
    "        optimizer.zero_grad()\n",
    "        outputs = model(input_ids, attention_mask=attention_mask)\n",
    "        loss = criterion(outputs, labels)\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        scheduler.step()\n",
    "\n",
    "    # Validation\n",
    "    model.eval()\n",
    "    val_predictions = []\n",
    "    val_true_labels = []\n",
    "\n",
    "    with torch.no_grad():\n",
    "        for batch in val_dataloader:\n",
    "            input_ids, attention_mask, labels = batch\n",
    "            input_ids, attention_mask, labels = input_ids.to(device), attention_mask.to(device), labels.to(device)\n",
    "\n",
    "            outputs = model(input_ids, attention_mask=attention_mask)\n",
    "            logits = outputs\n",
    "            predictions = torch.argmax(logits, dim=1).cpu().numpy()\n",
    "\n",
    "            val_predictions.extend(predictions)\n",
    "            val_true_labels.extend(labels.cpu().numpy())\n",
    "\n",
    "    val_accuracy = accuracy_score(val_true_labels, val_predictions)\n",
    "    val_report = classification_report(val_true_labels, val_predictions)\n",
    "    val_weighted_f1 = f1_score(val_true_labels, val_predictions, average='weighted')\n",
    "\n",
    "    print(f\"Epoch {epoch + 1}/{num_epochs}\")\n",
    "    print(\"Validation Accuracy:\", val_accuracy)\n",
    "    print(\"Validation Classification Report:\\n\", val_report)\n",
    "    print(\"Validation Weighted F1 Score:\", val_weighted_f1)\n",
    "    print(\"-\" * 50)\n",
    "\n",
    "    # Save the model if validation accuracy improves\n",
    "    if val_accuracy > best_val_accuracy:\n",
    "        best_val_accuracy = val_accuracy\n",
    "        early_stopping_counter = 0\n",
    "    else:\n",
    "        early_stopping_counter += 1\n",
    "\n",
    "    # Early stopping\n",
    "    if early_stopping_counter >= patience:\n",
    "        print(\"Early stopping. No improvement in validation accuracy.\")\n",
    "        break\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a8a4c474",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}