Lecture 05 - POS-taggers.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Math L06  - POS taggers.ipynb",
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "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.7.4"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GgPqt4p1y36p"
      },
      "source": [
        "# POS tagging\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "qDDstzhwztQS"
      },
      "source": [
        "import numpy as np\n",
        "import time"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tK3GvoeLSyG5"
      },
      "source": [
        "The following downloads some data and picks it apart to get it on a standard sklearn form. This is a famous data set that you should know about. You can read more on the dataset at https://en.wikipedia.org/wiki/Brown_Corpus."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0SKxeDs2seHV"
      },
      "source": [
        "import nltk\n",
        "nltk.download('brown')\n",
        "from nltk.corpus import brown\n",
        "nltk.download('universal_tagset')\n",
        "\n",
        "n_sentences = len(brown.tagged_sents(tagset='universal'))\n",
        "\n",
        "X = list()\n",
        "y = list()\n",
        "for sentence in brown.tagged_sents(tagset='universal'):\n",
        "  for n, pair in enumerate(sentence):\n",
        "    word, tag = pair\n",
        "    if n == 0:\n",
        "      X.append([word])\n",
        "      y.append([tag])\n",
        "    else:\n",
        "      X[-1].append(word)\n",
        "      y[-1].append(tag)\n",
        "assert len(X) == len(y)\n",
        "\n",
        "print(\"Found %i sentences with %i instances\" % (n_sentences, np.sum([len(x) for x in X])))\n",
        "# print(\"Original data example:\", brown.tagged_sents(tagset='universal')[0])\n",
        "print(\"Processed sentences example:\")\n",
        "print(\"x_0: %s\" % X[0])\n",
        "print(\"y_0: %s\" % y[0])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1ONfHvaW5-Lm"
      },
      "source": [
        "To be able to compare the models, the accuracy and train/test splits are defined here, before the tagger code."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "HBkD_eD95trE"
      },
      "source": [
        "from sklearn.model_selection import train_test_split\n",
        "X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0)\n",
        "\n",
        "def tag_accuracy(y, y_hat):\n",
        "  accuracy = list()\n",
        "  for u, v in zip(y, y_hat):\n",
        "    accuracy.append(np.sum([e1==e2 for e1, e2 in zip(u, v)])/len(u))\n",
        "  return accuracy"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "uIgHbwYDKt2s"
      },
      "source": [
        "print(\"Tag set:\", set([tag for sent in y for tag in sent]))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Or1MIBfNy87Z"
      },
      "source": [
        "## A very naive tagger\n",
        "\n",
        "This is a very naive tagger. It will tag a word with the most common POS tag in the training corpus. However, since getting an accuracy above 90% is not that hard for POS tagging, this can still impress someone not familiar with the problem formulation. If P(NOUN|'walk') < P(VERB|'walk') then the tag will be VERB in every instance."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "hGJnrCQoy9Yp"
      },
      "source": [
        "class NaiveTagger:\n",
        "  def __init__(self):\n",
        "    \"\"\"A very naive POS tagger\"\"\"\n",
        "    pass\n",
        "\n",
        "  def fit(self, X, y):\n",
        "    self._unique_words = list()\n",
        "    for x in X:\n",
        "      self._unique_words.extend(x)\n",
        "    self._unique_words = list(set(self._unique_words))\n",
        "\n",
        "    self._unique_tags = list()\n",
        "    for e in y:\n",
        "      self._unique_tags.extend(e)\n",
        "    self._unique_tags = list(set(self._unique_tags))\n",
        "\n",
        "    self._unique_words_lut = dict()\n",
        "    for i, word in enumerate(self._unique_words):\n",
        "      self._unique_words_lut[word] = i\n",
        "\n",
        "    self._unique_tags_lut = dict()\n",
        "    for i, tag in enumerate(self._unique_tags):\n",
        "      self._unique_tags_lut[tag] = i\n",
        "\n",
        "    self.frequencies = np.zeros((len(self._unique_words), len(self._unique_tags)), dtype=int)\n",
        "\n",
        "    for words, tags in zip(X, y):\n",
        "      for word, tag in zip(words, tags):\n",
        "        self.frequencies[self._unique_words_lut[word], self._unique_tags_lut[tag]] += 1\n",
        "\n",
        "  def predict(self, X):\n",
        "    majority_vote = naive_tagger._unique_tags[np.argmax(np.sum(naive_tagger.frequencies, axis=0))]\n",
        "    ret = list()\n",
        "    for words in X:\n",
        "      tags = list()\n",
        "      for word in words:\n",
        "        if word in self._unique_words_lut:\n",
        "          tags.append(self._unique_tags[np.argmax(self.frequencies[self._unique_words_lut[word], :])])\n",
        "        else:\n",
        "          tags.append(majority_vote)\n",
        "      ret.append(tags)\n",
        "    return ret"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "pbl2BL427NYm"
      },
      "source": [
        "naive_tagger = NaiveTagger()\n",
        "naive_tagger.fit(X_train, y_train)\n",
        "\n",
        "y_hat = naive_tagger.predict(X_test)\n",
        "\n",
        "accuracy = tag_accuracy(y_test, y_hat)\n",
        "print(\"Mean accuracy: %.1f%% [std %.1f%%]\" % (100*np.mean(accuracy), 100*np.std(accuracy)))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "a4cZKd0qiiAn"
      },
      "source": [
        "## Hidden markov model\n",
        "\n",
        "The implementation of the decoding follows the wikipedia description\n",
        "https://en.wikipedia.org/wiki/Viterbi_algorithm . If you want a more in depth description of the inner workings of an HMM, Rabiner's classic tutorial paper is a good start.\n",
        "\n",
        "L. R. Rabiner, \"*A tutorial on hidden Markov models and selected applications in speech recognition,*\" in Proceedings of the IEEE, vol. 77, no. 2, pp. 257-286, Feb. 1989.\n",
        "doi: 10.1109/5.18626\n",
        "\n",
        "\n",
        "For the following model to work, we need to load the cython extension.\n"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "vtkWPq0iVlPz"
      },
      "source": [
        "%load_ext cython"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "BF1FUAk8E0L_"
      },
      "source": [
        "%%cython\n",
        "import numpy as np\n",
        "\n",
        "class POStagger:\n",
        "  def __init__(self):\n",
        "    \"\"\"Simple HMM POS-tagger\"\"\"\n",
        "    pass\n",
        "\n",
        "  def fit(self, X, y):\n",
        "    # Find unique words and tags\n",
        "    unique_words = list()\n",
        "    for words in X:\n",
        "        unique_words.extend(words)\n",
        "    unique_tags = list()\n",
        "    for tags in y:\n",
        "      unique_tags.extend(tags)\n",
        "    self._unique_words = set(unique_words)\n",
        "    self._unique_tags = set(unique_tags)\n",
        "    # Making LUTs for indexing tags and words\n",
        "    self._words_lut = dict()\n",
        "    for n, word in enumerate(self._unique_words):\n",
        "      self._words_lut[word] = n\n",
        "    self._words_lut_inv = [None]*len(self._words_lut)\n",
        "    for word, n in self._words_lut.items():\n",
        "      self._words_lut_inv[n] = word\n",
        "    self._tags_lut = dict()\n",
        "    for n, tag in enumerate(self._unique_tags):\n",
        "      self._tags_lut[tag] = n\n",
        "    self._tags_lut_inv = [None]*len(self._tags_lut)\n",
        "    for tag, n in self._tags_lut.items():\n",
        "      self._tags_lut_inv[n] = tag\n",
        "    assert np.all([e is not None for e in self._words_lut_inv])\n",
        "    assert np.all([e is not None for e in self._tags_lut_inv])\n",
        "    # print(\"Tags:\", self._tags_lut)\n",
        "    # print(\"Tags inverse:\", self._tags_lut_inv)\n",
        "    # print(\"Words:\", repr(self._words_lut))\n",
        "    # print(\"Words inverse:\", repr(self._words_lut_inv))\n",
        "    # Allocate matrices for key statistics (pi, A, B)\n",
        "    self.n_states_ = len(self._tags_lut)\n",
        "    self.n_words_ = len(self._words_lut)\n",
        "    self.pi = np.zeros(self.n_states_)\n",
        "    self.A = np.zeros((self.n_states_, self.n_states_))\n",
        "    self.B = np.zeros((self.n_states_, self.n_words_))\n",
        "    # Set prior\n",
        "    self.pi[:] = 1/self.n_states_\n",
        "    self.A[:] = 1/self.n_states_\n",
        "    self.B[:] = 1/self.n_words_\n",
        "    # Count occurences\n",
        "    for words, tags in zip(X, y):\n",
        "      for n, word, tag in zip(range(len(words)), words, tags):\n",
        "        if n == 0:\n",
        "          # Record start states\n",
        "          self.pi[self._tags_lut[tag]] += 1\n",
        "        else:\n",
        "          # Record tag transitions\n",
        "          self.A[self._tags_lut[last_tag], self._tags_lut[tag]] += 1\n",
        "        # Record word given tag occurences\n",
        "        self.B[self._tags_lut[tag], self._words_lut[word]] += 1\n",
        "        last_tag = tag\n",
        "    # Normalize probabilities\n",
        "    self.pi /= np.sum(self.pi)\n",
        "    for i in range(self.A.shape[0]):\n",
        "      self.A[i, :] /= np.sum(self.A[i, :])\n",
        "    for i in range(self.B.shape[0]):\n",
        "      self.B[i, :] /= np.sum(self.B[i, :])\n",
        "\n",
        "  def predict(self, X):\n",
        "    assert type(X) == list\n",
        "    assert type(X[0]) == list\n",
        "    return [t[0] for t in map(self._predict, X)]\n",
        "\n",
        "  def _predict(self, sentence):\n",
        "    assert type(sentence) == list\n",
        "    cdef int i, j, k\n",
        "    cdef int n_states = self.n_states_\n",
        "    # Forward pass\n",
        "    T1 = np.zeros((n_states, len(sentence)))\n",
        "    cdef double[:,:] T1_memview = T1\n",
        "    T2 = np.zeros(T1.shape, dtype=np.intc)\n",
        "    cdef int[:,:] T2_memview = T2\n",
        "    cdef double[:] pi_memview = self.pi\n",
        "    cdef double[:,:] A_memview = self.A\n",
        "    cdef double[:,:] B_memview = self.B\n",
        "    cdef double p_max, w\n",
        "    B_column = np.zeros(n_states, dtype=np.double)\n",
        "    cdef double[:] B_column_memview = B_column\n",
        "    for j, word in enumerate(sentence):\n",
        "      if word in self._words_lut:\n",
        "        word_idx = self._words_lut[word]\n",
        "        for k in range(n_states):\n",
        "          B_column_memview[k] = B_memview[k, word_idx]\n",
        "      else:\n",
        "        # If word is not in the training data, fall back to uniform probability\n",
        "        w = 1/n_states\n",
        "        for k in range(n_states):\n",
        "          B_column_memview[k] = w\n",
        "      # Special case for the first word in the sentence\n",
        "      if j == 0:\n",
        "        for i in range(n_states):\n",
        "          T1_memview[i, j] = pi_memview[i]*B_column_memview[i]\n",
        "      else:\n",
        "        for i in range(n_states):\n",
        "          p_max = 0\n",
        "          for k in range(n_states):\n",
        "            w = T1_memview[k, j-1]*A_memview[k, i]*B_column_memview[i]\n",
        "            if w > p_max:\n",
        "              p_max = w\n",
        "              T1_memview[i, j] = w\n",
        "              T2_memview[i, j] = k\n",
        "    # Backward pass\n",
        "    traceback = [np.argmax(T1[:, -1])]\n",
        "    for i in range(len(sentence)-1, 0, -1):\n",
        "      traceback.insert(0, T2[traceback[0], i])\n",
        "    # print(output)\n",
        "    assert len(traceback) == len(sentence)\n",
        "    # Translate state numbers to tags and append to output\n",
        "    return [self._tags_lut_inv[i] for i in traceback], T1, T2"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "uuj_Yq-i1Vzb"
      },
      "source": [
        "hmm_tagger = POStagger()\n",
        "\n",
        "t = time.time()\n",
        "hmm_tagger.fit(X_train, y_train)\n",
        "print(\"Training finished after %.1f seconds\" % (time.time()-t))\n",
        "\n",
        "t = time.time()\n",
        "y_hat = hmm_tagger.predict(X_test)\n",
        "t = time.time()-t\n",
        "print(\"Prediction finished after %.1f seconds (%.1f μs/sentence)\" % (t, 1e6*t/len(X_test)))\n",
        "\n",
        "accuracy = tag_accuracy(y_test, y_hat)\n",
        "print(\"Mean accuracy: %.1f%% [std %.1f%%]\" % (100*np.mean(accuracy), 100*np.std(accuracy)))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "s6i7e3j1547K"
      },
      "source": [
        "The following is the code for illustrating the viterbi algorithm used in the slides."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "D2nswP4Q5uKx"
      },
      "source": [
        "sentence = X[-20]\n",
        "# tags = y[-1]\n",
        "forward_pass = None\n",
        "backward_pass = None\n",
        "# def show_trellis(tagger, sentence, tags, forward_pass = None, backward_pass = None):\n",
        "\n",
        "decoded_tags, T1, T2 = hmm_tagger._predict(sentence)\n",
        "if forward_pass is None:\n",
        "  forward_pass = list(range(len(T1[0])))\n",
        "elif type(forward_pass) is int:\n",
        "  forward_pass = list(range(forward_pass))\n",
        "assert type(forward_pass) is list\n",
        "if backward_pass is None:\n",
        "  backward_pass = list(range(len(T1[0])))\n",
        "elif type(backward_pass) is int:\n",
        "  backward_pass = list(range(backward_pass, len(sentence)))\n",
        "assert type(backward_pass) is list\n",
        "\n",
        "import matplotlib.pyplot as plt\n",
        "\n",
        "fig = plt.figure(figsize=(int(len(sentence)*.9), int(hmm_tagger.n_states_*.6)))\n",
        "ax = fig.add_subplot(1, 1, 1)\n",
        "ax.invert_yaxis()\n",
        "\n",
        "# Forward pass lines\n",
        "for j, word in enumerate(sentence):\n",
        "  if j in forward_pass and j > 0:\n",
        "    w = np.asarray([T1[k][j-1] for k in range(hmm_tagger.n_states_)])\n",
        "    w = .9*w/np.max(w) + .1\n",
        "    for i in range(hmm_tagger.n_states_):\n",
        "      for k in range(hmm_tagger.n_states_):\n",
        "        # ax.plot([j, j-1], [i, k], 'r-', linewidth=3, alpha=w[k]*hmm_tagger.A[i][k])\n",
        "        ax.plot([j-1, j], [k, i], 'r-', linewidth=3, alpha=w[k]*hmm_tagger.A[k][i])\n",
        "# Plot the base\n",
        "for i, state in enumerate(hmm_tagger._tags_lut_inv):\n",
        "  ax.text(-.5, i, state, fontsize=14, horizontalalignment='right', verticalalignment='center')\n",
        "  for j, word in enumerate(sentence):\n",
        "    if i == 0:\n",
        "      ax.text(j-.3, -.5, word, size=14, horizontalalignment='left', rotation=45)\n",
        "    ax.scatter(j, i, c='k', s=40, marker='o')\n",
        "    ax.scatter(j, i, c='w', s=30, marker='o')\n",
        "ax.axis('off')\n",
        "# Forward pass dots\n",
        "for j, word in enumerate(sentence):\n",
        "  w = np.asarray([T1[i][j] for i in range(hmm_tagger.n_states_)])\n",
        "  w = w/np.max(w)\n",
        "  if j in forward_pass:\n",
        "    for i, state in enumerate(hmm_tagger._tags_lut_inv):\n",
        "      ax.scatter(j, i, c='r', s=30, marker='o', alpha=w[i])\n",
        "# Backward pass lines\n",
        "i = np.argmax([T1[i][-1] for i in range(hmm_tagger.n_states_)])\n",
        "for j in range(len(sentence)-1, -1, -1):\n",
        "  if j in backward_pass and j > 0:\n",
        "    ax.plot([j-1, j], [T2[i][j], i], 'b--', linewidth=1, alpha=.5)\n",
        "  if j in backward_pass:\n",
        "    ax.text(j, hmm_tagger.n_states_, hmm_tagger._tags_lut_inv[i], fontsize=14, horizontalalignment='center', verticalalignment='top')\n",
        "  if j > 0:\n",
        "    i = T2[i][j]\n",
        "fig.show()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "IOpVa6LwkKHw"
      },
      "source": [
        "test_sentence = \"We went huhnting for sharks .\"\n",
        "\n",
        "def tabbed_print(msg, data):\n",
        "  print(msg, end=\"\\t\")\n",
        "  for d in data:\n",
        "    print(d.ljust(10), end=\"\")\n",
        "  print()\n",
        "\n",
        "data = test_sentence.split()\n",
        "\n",
        "tabbed_print(\"Test sentence:\", data)\n",
        "tabbed_print(\"Naive tagger:\", naive_tagger.predict([data])[0])\n",
        "tabbed_print(\"HMM tagger:\", hmm_tagger.predict([data])[0])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Ic_QndhaoUz-"
      },
      "source": [
        "assert \"Xir\" not in hmm_tagger._words_lut_inv\n",
        "assert \"xir\" not in hmm_tagger._words_lut_inv\n",
        "\n",
        "test_sentence = \"Xir went for a walk .\"\n",
        "\n",
        "data = test_sentence.split()\n",
        "\n",
        "tabbed_print(\"Test sentence:\", data)\n",
        "tabbed_print(\"Naive tagger:\", naive_tagger.predict([data])[0])\n",
        "tabbed_print(\"HMM tagger:\", hmm_tagger.predict([data])[0])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ga-tg4f_vshT"
      },
      "source": [
        "test_sentence = \"colorless green dreams sleep furiously\"\n",
        "\n",
        "data = test_sentence.split()\n",
        "\n",
        "tabbed_print(\"Test sentence:\", data)\n",
        "tabbed_print(\"Naive tagger:\", naive_tagger.predict([data])[0])\n",
        "tabbed_print(\"HMM tagger:\", hmm_tagger.predict([data])[0])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Kxs-k-56Kt26"
      },
      "source": [
        "\n",
        "y_hat = hmm_tagger.predict(X_test)\n",
        "errors = list()\n",
        "for a, b in zip(y_test, y_hat):\n",
        "    errors.append(np.sum([tag1==tag2 for tag1, tag2 in zip(a, b)])/len(a))\n",
        "\n",
        "# Put lower limit on errors to avoid the weird ones\n",
        "errors = [e if e>.8 else 2 for e in errors]\n",
        "idx = np.argmin(errors)\n",
        "\n",
        "data = X_test[idx]\n",
        "\n",
        "tabbed_print(\"Test sentence:\", data)\n",
        "tabbed_print(\"Naive tagger:\", naive_tagger.predict([data])[0])\n",
        "tabbed_print(\"HMM tagger:\", hmm_tagger.predict([data])[0])\n",
        "tabbed_print(\"True tags:    \", y_test[idx])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "VjpDmWLwHxjE"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}