06_01_lstm_text_train.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# LSTM Training"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import re\n",
    "from IPython.display import clear_output\n",
    "\n",
    "from tensorflow.keras.layers import Dense, LSTM, Input, Embedding, Dropout\n",
    "from tensorflow.keras.utils import to_categorical\n",
    "from tensorflow.keras.models import Model, load_model\n",
    "from tensorflow.keras.optimizers import Adam, RMSprop\n",
    "from tensorflow.keras.preprocessing.sequence import pad_sequences\n",
    "from tensorflow.keras.preprocessing.text import Tokenizer\n",
    "from tensorflow.keras.callbacks import LambdaCallback"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "load_saved_model = False\n",
    "train_model = True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "token_type = 'word'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#load in the text and perform some cleanup\n",
    "\n",
    "seq_length = 20\n",
    "\n",
    "filename = \"./data/aesop/data.txt\"\n",
    "\n",
    "with open(filename, encoding='utf-8-sig') as f:\n",
    "    text = f.read()\n",
    "    \n",
    "    \n",
    "#removing text before and after the main stories\n",
    "start = text.find(\"THE FOX AND THE GRAPES\\n\\n\\n\")\n",
    "end = text.find(\"ILLUSTRATIONS\\n\\n\\n[\")\n",
    "text = text[start:end]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "start_story = '| ' * seq_length\n",
    "    \n",
    "text = start_story + text\n",
    "text = text.lower()\n",
    "text = text.replace('\\n\\n\\n\\n\\n', start_story)\n",
    "text = text.replace('\\n', ' ')\n",
    "text = re.sub('  +', '. ', text).strip()\n",
    "text = text.replace('..', '.')\n",
    "\n",
    "text = re.sub('([!\"#$%&()*+,-./:;<=>?@[\\]^_`{|}~])', r' \\1 ', text)\n",
    "text = re.sub('\\s{2,}', ' ', text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "if token_type == 'word':\n",
    "    tokenizer = Tokenizer(char_level = False, filters = '')\n",
    "else:\n",
    "    tokenizer = Tokenizer(char_level = True, filters = '', lower = False)\n",
    "    \n",
    "    \n",
    "tokenizer.fit_on_texts([text])\n",
    "\n",
    "total_words = len(tokenizer.word_index) + 1\n",
    "\n",
    "token_list = tokenizer.texts_to_sequences([text])[0]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total_words"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "print(tokenizer.word_index)\n",
    "print(token_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_sequences(token_list, step):\n",
    "    \n",
    "    X = []\n",
    "    y = []\n",
    "\n",
    "    for i in range(0, len(token_list) - seq_length, step):\n",
    "        X.append(token_list[i: i + seq_length])\n",
    "        y.append(token_list[i + seq_length])\n",
    "    \n",
    "\n",
    "    y = to_categorical(y, num_classes = total_words)\n",
    "    \n",
    "    num_seq = len(X)\n",
    "    print('Number of sequences:', num_seq, \"\\n\")\n",
    "    \n",
    "    return X, y, num_seq\n",
    "\n",
    "step = 1\n",
    "seq_length = 20\n",
    "\n",
    "X, y, num_seq = generate_sequences(token_list, step)\n",
    "\n",
    "X = np.array(X)\n",
    "y = np.array(y)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "X.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "y.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define the LSTM model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if load_saved_model:\n",
    "    # model = load_model('./saved_models/lstm_aesop_1.h5')\n",
    "    model = load_model('./saved_models/aesop_dropout_100.h5')\n",
    "\n",
    "else:\n",
    "\n",
    "    n_units = 256\n",
    "    embedding_size = 100\n",
    "\n",
    "    text_in = Input(shape = (None,))\n",
    "    embedding = Embedding(total_words, embedding_size)\n",
    "    x = embedding(text_in)\n",
    "    x = LSTM(n_units)(x)\n",
    "    # x = Dropout(0.2)(x)\n",
    "    text_out = Dense(total_words, activation = 'softmax')(x)\n",
    "\n",
    "    model = Model(text_in, text_out)\n",
    "\n",
    "    opti = RMSprop(lr = 0.001)\n",
    "    model.compile(loss='categorical_crossentropy', optimizer=opti)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def sample_with_temp(preds, temperature=1.0):\n",
    "    # helper function to sample an index from a probability array\n",
    "    preds = np.asarray(preds).astype('float64')\n",
    "    preds = np.log(preds) / temperature\n",
    "    exp_preds = np.exp(preds)\n",
    "    preds = exp_preds / np.sum(exp_preds)\n",
    "    probas = np.random.multinomial(1, preds, 1)\n",
    "    return np.argmax(probas)\n",
    "\n",
    "\n",
    "\n",
    "def generate_text(seed_text, next_words, model, max_sequence_len, temp):\n",
    "    output_text = seed_text\n",
    "    \n",
    "    seed_text = start_story + seed_text\n",
    "    \n",
    "    for _ in range(next_words):\n",
    "        token_list = tokenizer.texts_to_sequences([seed_text])[0]\n",
    "        token_list = token_list[-max_sequence_len:]\n",
    "        token_list = np.reshape(token_list, (1, max_sequence_len))\n",
    "        \n",
    "        probs = model.predict(token_list, verbose=0)[0]\n",
    "        y_class = sample_with_temp(probs, temperature = temp)\n",
    "        \n",
    "        if y_class == 0:\n",
    "            output_word = ''\n",
    "        else:\n",
    "            output_word = tokenizer.index_word[y_class]\n",
    "            \n",
    "        if output_word == \"|\":\n",
    "            break\n",
    "            \n",
    "        if token_type == 'word':\n",
    "            output_text += output_word + ' '\n",
    "            seed_text += output_word + ' '\n",
    "        else:\n",
    "            output_text += output_word + ' '\n",
    "            seed_text += output_word + ' '\n",
    "            \n",
    "            \n",
    "    return output_text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def on_epoch_end(epoch, logs):\n",
    "    seed_text = \"\"\n",
    "    gen_words = 500\n",
    "\n",
    "    print('Temp 0.2')\n",
    "    print (generate_text(seed_text, gen_words, model, seq_length, temp = 0.2))\n",
    "    print('Temp 0.33')\n",
    "    print (generate_text(seed_text, gen_words, model, seq_length, temp = 0.33))\n",
    "    print('Temp 0.5')\n",
    "    print (generate_text(seed_text, gen_words, model, seq_length, temp = 0.5))\n",
    "    print('Temp 1.0')\n",
    "    print (generate_text(seed_text, gen_words, model, seq_length, temp = 1))\n",
    "\n",
    "    \n",
    "    \n",
    "if train_model:\n",
    "    epochs = 1000\n",
    "    batch_size = 32\n",
    "    num_batches = int(len(X) / batch_size)\n",
    "    callback = LambdaCallback(on_epoch_end=on_epoch_end)\n",
    "    model.fit(X, y, epochs=epochs, batch_size=batch_size, callbacks = [callback], shuffle = True)\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "seed_text = \"the frog and the snake . \"\n",
    "gen_words = 500\n",
    "temp = 0.1\n",
    "\n",
    "print (generate_text(seed_text, gen_words, model, seq_length, temp))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_human_led_text(model, max_sequence_len):\n",
    "    \n",
    "    output_text = ''\n",
    "    seed_text = start_story\n",
    "    \n",
    "    while 1:\n",
    "        token_list = tokenizer.texts_to_sequences([seed_text])[0]\n",
    "        token_list = token_list[-max_sequence_len:]\n",
    "        token_list = np.reshape(token_list, (1, max_sequence_len))\n",
    "        \n",
    "        probs = model.predict(token_list, verbose=0)[0]\n",
    "\n",
    "        top_10_idx = np.flip(np.argsort(probs)[-10:])\n",
    "        top_10_probs = [probs[x] for x in top_10_idx]\n",
    "        top_10_words = tokenizer.sequences_to_texts([[x] for x in top_10_idx])\n",
    "        \n",
    "        for prob, word in zip(top_10_probs, top_10_words):\n",
    "            print('{:<6.1%} : {}'.format(prob, word))\n",
    "\n",
    "        chosen_word = input()\n",
    "                \n",
    "        if chosen_word == '|':\n",
    "            break\n",
    "            \n",
    "        \n",
    "        seed_text += chosen_word + ' '\n",
    "        output_text += chosen_word + ' '\n",
    "        \n",
    "        clear_output()\n",
    "\n",
    "        print (output_text)\n",
    "            \n",
    "    \n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "generate_human_led_text(model, 20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# model.save('./saved_models/aesop_no_dropout_100.h5')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "gdl_code",
   "language": "python",
   "name": "gdl_code"
  },
  "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}