mBERT_u_model.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "BTZOIgciGOA7"
      },
      "outputs": [],
      "source": [
        "!pip install transformers\n",
        "import transformers\n",
        "import pandas as pd\n",
        "import numpy as np\n",
        "import sklearn\n",
        "import numpy as np\n",
        "import matplotlib.pyplot as plt\n",
        "import torch\n",
        "import torch.cuda\n",
        "\n",
        "from transformers import AutoTokenizer, AutoModelForMaskedLM\n",
        "from transformers import AutoModelForSequenceClassification\n",
        "from transformers import AdamW\n",
        "from sklearn import preprocessing\n",
        "from collections import defaultdict\n",
        "from google.colab import drive\n",
        "\n",
        "drive.mount('/content/drive')\n",
        "dataset_mlt = pd.read_csv('/content/drive/MyDrive/Maltese_Unimorph_database_final.txt', sep=\"  \", header=None, error_bad_lines=False)\n",
        "\n",
        "dataset_copy_mlt=dataset_mlt.copy()\n",
        "list_data_mlt = dataset_copy_mlt.values.tolist()\n",
        "print(list_data_mlt[0:10])\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cu-rK-AQLiC_"
      },
      "outputs": [],
      "source": [
        "split_data_mlt=[]\n",
        "\n",
        "def split(corpus, spit_data):\n",
        "  for sets in corpus:\n",
        "    for x in sets:\n",
        "      parts = x.split(\"\\t\")\n",
        "      spit_data.append(parts)\n",
        "\n",
        "split(list_data_mlt,split_data_mlt)\n",
        "print(split_data_mlt[0:100])\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "OCrv0BppSBqG"
      },
      "outputs": [],
      "source": [
        "# removing duplicates\n",
        "def remove_duplicates(data_array):\n",
        "    unique_data_array = []\n",
        "    second_element_set = set()\n",
        "    for subarray in data_array:\n",
        "        if subarray[1] not in second_element_set:\n",
        "            unique_data_array.append(subarray)\n",
        "            second_element_set.add(subarray[1])\n",
        "    return unique_data_array\n",
        "\n",
        "unique_data_mlt = remove_duplicates(split_data_mlt)\n",
        "print(unique_data_mlt[0:100])\n",
        "\n",
        "#getting the longest word for padding \n",
        "len_longest=0\n",
        "for string in unique_data_mlt:\n",
        "    x=string[1]\n",
        "    if len(x)>len_longest:\n",
        "      len_longest=len(x)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6y1StuJfkAJC"
      },
      "outputs": [],
      "source": [
        "# splitting into words and grammar\n",
        "def split_word_and_grammar(unique_data_array):\n",
        "    last_elements = [subarray[-1] for subarray in unique_data_array]\n",
        "    word_list = [(subarray[0],subarray[1]) for subarray in unique_data_array]\n",
        "    return word_list, last_elements\n",
        "\n",
        "word_list_mlt, last_elements_mlt = split_word_and_grammar(unique_data_mlt)\n",
        "print(last_elements_mlt[0:100])\n",
        "\n",
        "# splitting grammar\n",
        "def splitting_grammar(last_elements):\n",
        "    list_of_lists = []\n",
        "    for info in last_elements:\n",
        "        parts = info.split(';')\n",
        "        num_parts = [int(part) if part.isdigit() else part for part in parts]\n",
        "        list_of_lists.append(num_parts)\n",
        "    return list_of_lists\n",
        "\n",
        "list_of_lists_mlt = splitting_grammar(last_elements_mlt)\n",
        "\n",
        "\n",
        "# adding to lists in alt order\n",
        "split_data = []\n",
        "list_of_lists = []\n",
        "for i in range(len(word_list_mlt)):\n",
        "      # adding mlt word and grammar\n",
        "      array=(word_list_mlt[i][0],) + word_list_mlt[i][1:]\n",
        "      split_data.append(array)\n",
        "      list_of_lists.append(list_of_lists_mlt[i])\n",
        "\n",
        "print(split_data[0:100])\n",
        "print(list_of_lists[0:100])"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "u5k7H1zAdvaq"
      },
      "outputs": [],
      "source": [
        "# splitting data\n",
        "tokeniser = AutoTokenizer.from_pretrained(\"MLRS/mBERTu\",char_level=True) #,char_level=True\n",
        "model = AutoModelForMaskedLM.from_pretrained(\"MLRS/mBERTu\")\n",
        "\n",
        "#tokeniser = AutoTokenizer.from_pretrained(\"bert-base-multilingual-cased\",char_level=True)\n",
        "#model = AutoModelForMaskedLM.from_pretrained(\"bert-base-multilingual-cased\")\n",
        "\n",
        "from sklearn.model_selection import train_test_split\n",
        "device = 'cuda:0' if torch.cuda.is_available() else 'cpu'\n",
        "\n",
        "X = split_data\n",
        "y = list_of_lists\n",
        "\n",
        "X_train, X_rem, y_train, y_rem = train_test_split(X,y, train_size=0.8)\n",
        "X_valid, X_test, y_valid, y_test = train_test_split(X_rem,y_rem, test_size=0.2)\n",
        "print(X_train[0:100])\n",
        "print(y_train[0:100])\n",
        "print(len(X_train))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4qqAAMoqgugA"
      },
      "outputs": [],
      "source": [
        "# tokenize the training set\n",
        "token_indexes = tokeniser(X_train, return_tensors='pt', padding=True, truncation=True, is_split_into_words=False, max_length=512)\n",
        "indexed_train_x = token_indexes['input_ids'].to(device).squeeze()\n",
        "mask_train_x = token_indexes['attention_mask'].to(device).squeeze()\n",
        "\n",
        "# tokenize the test set\n",
        "new_tuples = [((\"\",) + t[1:]) for t in X_test]\n",
        "token_indexes_test = tokeniser(new_tuples, return_tensors='pt', padding=True, truncation=True, is_split_into_words=False, max_length=512)\n",
        "indexed_test_x = token_indexes_test['input_ids'].to(device).squeeze()\n",
        "mask_test_x = token_indexes_test['attention_mask'].to(device).squeeze()\n",
        "\n",
        "# padding w/0 so gramm info is all the same length\n",
        "def padding_grammar(Y):\n",
        "    max_len = max(len(info) for info in Y)\n",
        "    padded_array = [info + ['<PAD>'] * (max_len - len(info)) for info in Y]\n",
        "    return padded_array\n",
        "padded_morph_info_train = padding_grammar(y_train)\n",
        "padded_morph_info_test = padding_grammar(y_test)\n",
        "print(padded_morph_info_train[0:10])\n",
        "\n",
        "all_morph_info = padded_morph_info_train + padded_morph_info_test\n",
        "# creating a dictionary for mapping \n",
        "# Creating a dictionary for mapping\n",
        "morph_vocab = defaultdict(lambda: len(morph_vocab))\n",
        "morph_vocab['<PAD>'] = 0  # Assigning index 0 to 'pad'\n",
        "\n",
        "for info in all_morph_info:\n",
        "    # Extracting the unique values at this position\n",
        "    values = set(i for i in info)\n",
        "    # Assigning index\n",
        "    for value in values:\n",
        "        if value not in morph_vocab and value != '<PAD>':\n",
        "            morph_vocab[value] = len(morph_vocab)\n",
        "\n",
        "print(morph_vocab)\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "YA9KsSI6EFGH"
      },
      "outputs": [],
      "source": [
        "# map the morphological information to indexes using the vocabulary\n",
        "def get_targets_one_hot(padded_info):\n",
        "  targets = [[morph_vocab[value] for value in info] for info in padded_info]\n",
        "  targets = [[morph_vocab['<PAD>'] if value == 0 else value for value in info] for info in targets]\n",
        "  targets=torch.tensor(targets).to(device)\n",
        "  \n",
        "  # one-hot encode the targets\n",
        "  num_classes = len(morph_vocab)\n",
        "  targets = torch.nn.functional.one_hot(targets, num_classes=num_classes).to(torch.float32)\n",
        "  targets = torch.tensor(targets, requires_grad=True).to(device)\n",
        "  return targets\n",
        "\n",
        "targets_train = get_targets_one_hot(padded_morph_info_train)\n",
        "targets_test = get_targets_one_hot(padded_morph_info_test)\n",
        "\n",
        "print(targets_train[0:100])\n",
        "print(indexed_train_x[0:100])"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "yNnANdFhh_nG"
      },
      "outputs": [],
      "source": [
        "class Model(torch.nn.Module):\n",
        "    def __init__(self, bert):\n",
        "        super().__init__()\n",
        "        self.bert = bert\n",
        "        self.output_layer1 = torch.nn.Linear(768, 512)\n",
        "        self.bn1 = torch.nn.BatchNorm1d(512)  # Add batch normalization for faster training and better generalization\n",
        "        self.drop1 = torch.nn.Dropout(p=0.3)\n",
        "        self.output_layer2 = torch.nn.Linear(512, 256)  # Add another fully connected layer for improved feature extraction\n",
        "        self.bn2 = torch.nn.BatchNorm1d(256)\n",
        "        self.drop2 = torch.nn.Dropout(p=0.1)\n",
        "        self.output_layer3 = torch.nn.Linear(256, 10*len(morph_vocab))\n",
        "        \n",
        "    def forward(self, x, mask):\n",
        "        with torch.cuda.amp.autocast():\n",
        "            vecs = self.bert(x, attention_mask=mask, output_hidden_states=True).hidden_states[8][:, 0, :] \n",
        "            output = self.output_layer1(vecs)\n",
        "            output = self.bn1(output)\n",
        "            output = torch.relu(output)\n",
        "            output = self.drop1(output)\n",
        "            output = self.output_layer2(output)\n",
        "            output = self.bn2(output)\n",
        "            output = torch.relu(output)\n",
        "            output = self.drop2(output)\n",
        "            output = self.output_layer3(output).view(-1, 10, len(morph_vocab))\n",
        "            output = torch.sigmoid(output)\n",
        "        return output\n",
        "\n",
        "bert = transformers.BertForMaskedLM.from_pretrained('MLRS/mBERTu')\n",
        "#bert = transformers.BertForMaskedLM.from_pretrained('bert-base-multilingual-cased')\n",
        "model = Model(bert)\n",
        "model.to(device)\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cTfxLzrQLsaf"
      },
      "outputs": [],
      "source": [
        "'''\n",
        "def grammatical_accuracy(output, target):\n",
        "\n",
        "    output = output.detach().cpu().numpy()\n",
        "    target = target.detach().cpu().numpy()\n",
        "\n",
        "    # find index of the predicted and target labels\n",
        "    output_idx = np.argmax(output, axis=1)\n",
        "    target_idx = np.argmax(target, axis=1)  \n",
        "\n",
        "    # distance between predicted and targer\n",
        "    distance = np.abs(output_idx - target_idx)\n",
        "\n",
        "    correct = np.sum(distance == 0) + np.sum(distance == 1)\n",
        "    total = distance.size\n",
        "    accuracy = float(correct) / total\n",
        "\n",
        "    return accuracy\n",
        "'''\n",
        "\n",
        "def grammatical_accuracy(output, target):\n",
        "    output = output.detach().cpu().numpy()\n",
        "    target = target.detach().cpu().numpy()\n",
        "\n",
        "    output = np.argmax(output, axis=1)\n",
        "    target = np.argmax(target, axis=1)\n",
        "    accuracy = np.mean(output == target) * 100\n",
        "    return accuracy\n",
        "\n",
        "accuracy_count=[]\n",
        "def get_acc(matches):\n",
        "    if len(matches) == 0:\n",
        "        return 0\n",
        "    accuracy_count.append(sum(matches)/len(matches))\n",
        "    return sum(matches)/len(matches)\n",
        "#return round(100*sum(preds)/len(preds),3)\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "dQTIkiB-AEkT"
      },
      "outputs": [],
      "source": [
        "batch_size=64\n",
        "\n",
        "#bert 32,1e-5,2,150\n",
        "#bertu 62,1e-5,8,300\n",
        "optimizer = AdamW(model.parameters(), lr=1e-5, weight_decay=0.1)\n",
        "\n",
        "print('step', 'error', 'accuracy')\n",
        "train_errors = []\n",
        "all_true=[]\n",
        "all_pred=[]\n",
        "accumulation_steps = 4\n",
        "for step in range(1, 2000+1):\n",
        "    optimizer.zero_grad()\n",
        "    model.train(True)\n",
        "    #.CrossEntropyLoss()\n",
        "    loss_fn = torch.nn.CrossEntropyLoss()\n",
        "    all_outputs = []\n",
        "    all_targets = []\n",
        "    for i in range(accumulation_steps):\n",
        "     # print(targets_train.shape)\n",
        "      output = model(indexed_train_x[i*batch_size:(i+1)*batch_size], mask_train_x[i*batch_size:(i+1)*batch_size])\n",
        "     \n",
        "      # MSELoss   \n",
        "      error = loss_fn(output, targets_train[i*batch_size:(i+1)*batch_size])\n",
        "      error = error/accumulation_steps\n",
        "      error.backward()\n",
        "\n",
        "      all_outputs.append(output.cpu().detach().numpy())\n",
        "      all_targets.append(targets_train[i*batch_size:(i+1)*batch_size].cpu().detach().numpy())\n",
        "\n",
        "    optimizer.step()\n",
        "    model.train(False)\n",
        "\n",
        "    train_errors.append(error.detach().tolist())\n",
        "\n",
        "    train_accs = []\n",
        "    \n",
        "    # one_hot_output = np.round(output.cpu().detach()).numpy().astype(int)\n",
        "    \n",
        "    \n",
        "    for x in range(0,batch_size): # every vector in the output\n",
        "      targets = targets_train[i*batch_size:(i+1)*batch_size].cpu().detach()\n",
        "      targets = targets[x].argmax(dim=1).tolist()\n",
        "      predicted_labels = output.cpu().detach()[x].argmax(dim=1).tolist()\n",
        "      results=0\n",
        "\n",
        "      all_true.append(targets)\n",
        "      all_pred.append(predicted_labels)\n",
        "\n",
        "     # print('t',targets)\n",
        "     # print('p',predicted_labels)\n",
        "      for vecs1 in predicted_labels: # every vector in output vector i\n",
        "        for vecs2 in targets: # every vector in target vector i\n",
        "            if vecs1==vecs2:\n",
        "                results=results+1\n",
        "           \n",
        "      if results>=len(predicted_labels): \n",
        "        train_accs.append(1)\n",
        "      else:\n",
        "        train_accs.append(0)  \n",
        "    accuracy=get_acc(train_accs)\n",
        "\n",
        "    if step % 10 == 0:\n",
        "        print(step, train_errors[-1], accuracy)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "LEMOwfjL5-ms"
      },
      "outputs": [],
      "source": [
        "from sklearn.preprocessing import MultiLabelBinarizer\n",
        "from sklearn.metrics import accuracy_score,f1_score, precision_score, recall_score\n",
        "!pip install editdistance\n",
        "import editdistance\n",
        "\n",
        "m = MultiLabelBinarizer().fit(all_true)\n",
        "\n",
        "f1_weighted=f1_score(m.transform(all_true),\n",
        "         m.transform(all_pred),\n",
        "         average='weighted')\n",
        "\n",
        "f1_macro=f1_score(m.transform(all_true),\n",
        "         m.transform(all_pred),\n",
        "         average='macro')\n",
        "\n",
        "print(\"F1 score weighted:\", f1_weighted)\n",
        "print(\"F1 score macro:\", f1_macro)\n",
        "\n",
        "precision = precision_score(m.transform(all_true), m.transform(all_pred), average='micro')\n",
        "recall = recall_score(m.transform(all_true), m.transform(all_pred), average='micro')\n",
        "print(\"Precision:\", precision)\n",
        "print(\"Recall:\", recall)\n",
        "\n",
        "\n",
        "f1_micro = f1_score(m.transform(all_true),  m.transform(all_pred), average='micro')\n",
        "print(\"F1 score micro:\", f1_micro)\n",
        "\n",
        "all_true_flat = [label for sublist in all_true for label in sublist]\n",
        "all_pred_flat = [label for sublist in all_pred for label in sublist]\n",
        "\n",
        "\n",
        "# problem with lev distance, uses order, order doesnt matter a lot here\n",
        "reverse_dict = {v: k for k, v in morph_vocab.items()} \n",
        "distances = []\n",
        "for i in range(len(all_true)):\n",
        "    predicted_labels = [reverse_dict[j] for j in all_pred[i]]\n",
        "    true_labels = [reverse_dict[j] for j in all_true[i]]\n",
        "    distance = editdistance.eval(predicted_labels, true_labels)\n",
        "    distances.append(distance)\n",
        "\n",
        "average_distance = sum(distances) / len(distances)\n",
        "print(\"Levenshtein distances:\", average_distance)\n",
        "\n",
        "avg= sum(accuracy_count)/len(accuracy_count)\n",
        "print(\"the average is: \", avg)"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "second_values = [v for k, v in morph_vocab.items()]\n",
        "print(second_values)"
      ],
      "metadata": {
        "id": "5K_mLXTbAuZZ"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "m = MultiLabelBinarizer().fit(all_true)\n",
        "\n",
        "# get the list of all possible morphosyntactic tags in the dataset\n",
        "labels =second_values\n",
        "print(labels)\n",
        "\n",
        "# initialize dictionaries to store the scores for each label\n",
        "precision_dict = {}\n",
        "recall_dict = {}\n",
        "f1_dict = {}\n",
        "\n",
        "for label in labels:\n",
        "    # transform the true and predicted labels to binary format for the current label\n",
        "    true_labels_label = [1 if label in sublist else 0 for sublist in all_true]\n",
        "    pred_labels_label = [1 if label in sublist else 0 for sublist in all_pred]\n",
        "    \n",
        "    # calculate precision, recall, and F1 score for the current label\n",
        "    precision_dict[label] = precision_score(true_labels_label, pred_labels_label)\n",
        "    recall_dict[label] = recall_score(true_labels_label, pred_labels_label)\n",
        "    f1_dict[label] = f1_score(true_labels_label, pred_labels_label)\n",
        "\n",
        "reverse_dict = {v: k for k, v in morph_vocab.items()} \n",
        "# print the scores for each label\n",
        "for label in labels:\n",
        "    print(\"Label:\", reverse_dict[label])\n",
        "    print(\"Precision:\", precision_dict[label])\n",
        "    print(\"Recall:\", recall_dict[label])\n",
        "    print(\"F1 score:\", f1_dict[label])"
      ],
      "metadata": {
        "id": "xjaRLYIQAuyY"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from prettytable import PrettyTable\n",
        "\n",
        "# create a table object with column names\n",
        "table = PrettyTable(['Label', 'Precision', 'Recall', 'F1 Score'])\n",
        "\n",
        "# add rows to the table\n",
        "for label in labels:\n",
        "    table.add_row([reverse_dict[label], precision_dict[label], recall_dict[label], f1_dict[label]])\n",
        "\n",
        "# print the table\n",
        "print(table)"
      ],
      "metadata": {
        "id": "P8boxUUPAu-H"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "r4JjnfjilXOW"
      },
      "outputs": [],
      "source": [
        "with torch.no_grad():\n",
        "    print('sent', 'prediction')\n",
        "    for i in range(len(indexed_test_x) // batch_size):\n",
        "        outputs = model(indexed_test_x[i*batch_size:(i+1)*batch_size], mask_test_x[i*batch_size:(i+1)*batch_size])\n",
        "\n",
        "        # reversing the dict to get the grammar from the index\n",
        "        reverse_dict = {v: k for k, v in morph_vocab.items()} \n",
        "\n",
        "        for j, x in enumerate(outputs):\n",
        "            predicted_labels = [reverse_dict[i] for i in x.argmax(dim=1).tolist()]\n",
        "\n",
        "            # changing from char to word\n",
        "            input_word = ''.join(tokeniser.decode(indexed_test_x[i*batch_size:(i+1)*batch_size][j]).split())\n",
        "\n",
        "            print(input_word, predicted_labels)\n",
        "            print(\"actual\")\n",
        "            print(y_test[i*batch_size:(i+1)*batch_size][j])"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "S1WWbWYuAFE5"
      },
      "outputs": [],
      "source": [
        "(fig, ax) = plt.subplots(1, 1)\n",
        "ax.set_xlabel('step')\n",
        "ax.set_ylabel('$E$')\n",
        "ax.plot(range(1, len(train_errors) + 1), train_errors, color='blue', linestyle='-', linewidth=3)\n",
        "ax.grid()\n",
        "\n",
        "print('   ')\n",
        "(fig, ax) = plt.subplots(1, 1)\n",
        "ax.set_xlabel('step')\n",
        "ax.set_ylabel('$A$')\n",
        "ax.plot(range(1, len(accuracy_count) + 1), accuracy_count, color='blue', linestyle='-', linewidth=3)\n",
        "ax.grid()"
      ]
    }
  ],
  "metadata": {
    "accelerator": "GPU",
    "colab": {
      "provenance": []
    },
    "gpuClass": "standard",
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}