BERT_U_model_with_zero_shot.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "BTZOIgciGOA7"
      },
      "outputs": [],
      "source": [
        "!pip install transformers\n",
        "!pip install editdistance\n",
        "\n",
        "import editdistance\n",
        "import pandas as pd\n",
        "import transformers\n",
        "import matplotlib.pyplot as plt\n",
        "import torch\n",
        "import torch.cuda\n",
        "import sklearn\n",
        "import numpy as np\n",
        "import random\n",
        "\n",
        "from sklearn.model_selection import train_test_split\n",
        "from sklearn.preprocessing import MultiLabelBinarizer\n",
        "from sklearn.metrics import accuracy_score,f1_score\n",
        "from sklearn.metrics import accuracy_score\n",
        "from transformers import AdamW, get_linear_schedule_with_warmup\n",
        "from sklearn import preprocessing\n",
        "from collections import defaultdict\n",
        "from google.colab import drive\n",
        "from transformers import AutoTokenizer, AutoModelForMaskedLM\n",
        "from transformers import AutoModelForSequenceClassification\n",
        "\n",
        "drive.mount('/content/drive')\n",
        "\n",
        "dataset_mlt = pd.read_csv('/content/drive/MyDrive/Maltese_Unimorph_database_final.txt', sep=\"  \", header=None, error_bad_lines=False)\n",
        "#dataset_mlt = pd.read_csv('/content/drive/MyDrive/mlt')\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])\n",
        "\n",
        "longest_list = max(list_of_lists, key=len)\n",
        "print(longest_list)\n",
        "print(len(longest_list))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "q49Dv0FBxSds"
      },
      "outputs": [],
      "source": [
        "!pip install transformers\n",
        "import transformers\n",
        "from transformers import AutoTokenizer, AutoModelForMaskedLM\n",
        "from transformers import AutoModelForSequenceClassification\n",
        "import matplotlib.pyplot as plt\n",
        "import torch\n",
        "import torch.cuda"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "u5k7H1zAdvaq"
      },
      "outputs": [],
      "source": [
        "# splitting data\n",
        "tokeniser = AutoTokenizer.from_pretrained(\"MLRS/BERTu\",char_level=True) #,char_level=True\n",
        "model = AutoModelForMaskedLM.from_pretrained(\"MLRS/BERTu\")\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",
        "\n",
        "\n",
        "y_train = [sublist if sublist[0] != 'V' else sublist[1:] for sublist in y_train]\n",
        "print(y_train[0:100])\n",
        "print(y_test[0:100])"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4qqAAMoqgugA"
      },
      "outputs": [],
      "source": [
        "import sklearn\n",
        "from sklearn import preprocessing\n",
        "from collections import defaultdict\n",
        "\n",
        "# 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_train = [info + ['<PAD>'] for info in padded_morph_info_train]\n",
        "\n",
        "padded_morph_info_test = padding_grammar(y_test)\n",
        "\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",
        "morph_vocab = defaultdict(lambda: len(morph_vocab)) \n",
        "\n",
        "for i in range(len(all_morph_info[0])):\n",
        "    # extracting the unique values at this position\n",
        "    values = set(info[i] for info in all_morph_info)\n",
        "    # assignign index\n",
        "    for value in values:\n",
        "        morph_vocab[value]\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.shape)\n",
        "print(targets_test.shape)\n",
        "print(indexed_train_x.shape)"
      ]
    },
    {
      "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, 7*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",
        "\n",
        "            output = self.output_layer2(output)\n",
        "            output = self.bn2(output)\n",
        "            output = torch.relu(output)\n",
        "            output = self.drop2(output)\n",
        "            \n",
        "            output = self.output_layer3(output).view(-1, 7, len(morph_vocab))\n",
        "            output = torch.sigmoid(output)\n",
        "        return output\n",
        "\n",
        "bert = transformers.BertForMaskedLM.from_pretrained('MLRS/BERTu')\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": [
        "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, 100+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",
        "      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",
        "    '''\n",
        "        # Cross Entropy Loss\n",
        "          loss_fn = torch.nn.CrossEntropyLoss()\n",
        "          error = loss_fn(output,targets_train[i*batch_size:(i+1)*batch_size])\n",
        "    '''\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): #0.8*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",
      "execution_count": null,
      "metadata": {
        "id": "r4JjnfjilXOW"
      },
      "outputs": [],
      "source": [
        "from collections import defaultdict\n",
        "\n",
        "# Create a dictionary to store the count of tags\n",
        "tag_counts = defaultdict(int)\n",
        "\n",
        "with torch.no_grad():\n",
        "    print('sent', 'prediction')\n",
        "    for i in range(len(indexed_test_x)):\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])\n",
        "\n",
        "            # Count the occurrences of each tag\n",
        "            for tag in predicted_labels:\n",
        "                tag_counts[tag] += 1\n",
        "\n",
        "# Convert the dictionary to a list of tuples (tag, count)\n",
        "tag_counts_list = list(tag_counts.items())\n",
        "\n",
        "# Sort the list based on the count in descending order\n",
        "tag_counts_list.sort(key=lambda x: x[1], reverse=True)\n",
        "\n",
        "# Print the tag and count pairs\n",
        "for tag, count in tag_counts_list:\n",
        "    print(tag, count)\n"
      ]
    },
    {
      "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",
        "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
}