chapter12_part01_text-generation.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "J-WQ_tQxCrEF"
   },
   "source": [
    "이 노트북은 [케라스 창시자에게 배우는 딥러닝 2판](https://tensorflow.blog/kerasdl2/)의 예제 코드를 담고 있습니다.\n",
    "\n",
    "<table align=\"left\">\n",
    "    <tr>\n",
    "        <td>\n",
    "            <a href=\"https://colab.research.google.com/github/rickiepark/deep-learning-with-python-2nd/blob/main/chapter12_part01_text-generation.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>\n",
    "        </td>\n",
    "    </tr>\n",
    "</table>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "THz7hp89CrEI"
   },
   "source": [
    "# 생성 모델을 위한 딥러닝"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "wBnsAPKECrEI"
   },
   "source": [
    "## 텍스트 생성"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "D89JwqPBCrEJ"
   },
   "source": [
    "### 시퀀스 생성을 위한 딥러닝 모델의 간단한 역사"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Z4P5rIr6CrEJ"
   },
   "source": [
    "### 시퀀스 데이터를 어떻게 생성할까?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "zBs8uChTCrEJ"
   },
   "source": [
    "### 샘플링 전략의 중요성"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "pMcqO4_VCrEK"
   },
   "source": [
    "**다른 온도 값을 사용하여 확률 분포의 가중치 바꾸기**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "id": "baD0AHOUCrEK"
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "def reweight_distribution(original_distribution, temperature=0.5):\n",
    "    distribution = np.log(original_distribution) / temperature\n",
    "    distribution = np.exp(distribution)\n",
    "    return distribution / np.sum(distribution)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "9hCKGeqJCrEL"
   },
   "source": [
    "### 케라스를 사용한 텍스트 생성 모델 구현"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "NmTMyS1ZCrEM"
   },
   "source": [
    "#### 데이터 준비"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "IM6Umf4uCrEM"
   },
   "source": [
    "**IMDB 영화 리뷰 데이터셋 다운로드하고 압축 풀기**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "fS_z2t7PCrEN",
    "outputId": "ad33dcf9-6f71-4876-e436-a5b13b500472"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2022-06-01 06:54:18--  https://ai.stanford.edu/~amaas/data/sentiment/aclImdb_v1.tar.gz\n",
      "Resolving ai.stanford.edu (ai.stanford.edu)... 171.64.68.10\n",
      "Connecting to ai.stanford.edu (ai.stanford.edu)|171.64.68.10|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 84125825 (80M) [application/x-gzip]\n",
      "Saving to: ‘aclImdb_v1.tar.gz’\n",
      "\n",
      "aclImdb_v1.tar.gz   100%[===================>]  80.23M  30.8MB/s    in 2.6s    \n",
      "\n",
      "2022-06-01 06:54:21 (30.8 MB/s) - ‘aclImdb_v1.tar.gz’ saved [84125825/84125825]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!wget https://ai.stanford.edu/~amaas/data/sentiment/aclImdb_v1.tar.gz\n",
    "!tar -xf aclImdb_v1.tar.gz"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "TcHe7T_QCrEN"
   },
   "source": [
    "**텍스트 파일(한 파일 = 한 샘플)에서 데이터셋 만들기**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Hb2YjZ86CrEO",
    "outputId": "e229846f-131a-417a-e586-b31d7535b655"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Found 100006 files belonging to 1 classes.\n"
     ]
    }
   ],
   "source": [
    "import tensorflow as tf\n",
    "from tensorflow import keras\n",
    "dataset = keras.utils.text_dataset_from_directory(\n",
    "    directory=\"aclImdb\", label_mode=None, batch_size=256)\n",
    "dataset = dataset.map(lambda x: tf.strings.regex_replace(x, \"<br />\", \" \"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "f-6pyZNQCrEO"
   },
   "source": [
    "**`TextVectorization` 층 준비하기**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "id": "aHTeHqNfCrEO"
   },
   "outputs": [],
   "source": [
    "from tensorflow.keras.layers import TextVectorization\n",
    "\n",
    "sequence_length = 100\n",
    "vocab_size = 15000\n",
    "text_vectorization = TextVectorization(\n",
    "    max_tokens=vocab_size,\n",
    "    output_mode=\"int\",\n",
    "    output_sequence_length=sequence_length,\n",
    ")\n",
    "text_vectorization.adapt(dataset)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "oLhj695ZCrEO"
   },
   "source": [
    "**언어 모델링 데이터셋 만들기**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "id": "pCPsdys6CrEP"
   },
   "outputs": [],
   "source": [
    "def prepare_lm_dataset(text_batch):\n",
    "    vectorized_sequences = text_vectorization(text_batch)\n",
    "    x = vectorized_sequences[:, :-1]\n",
    "    y = vectorized_sequences[:, 1:]\n",
    "    return x, y\n",
    "\n",
    "lm_dataset = dataset.map(prepare_lm_dataset, num_parallel_calls=4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "eZDJxIp9CrEP"
   },
   "source": [
    "#### 트랜스포머 기반의 시퀀스-투-시퀀스 모델"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "id": "O40uNWAWCrEP"
   },
   "outputs": [],
   "source": [
    "import tensorflow as tf\n",
    "from tensorflow.keras import layers\n",
    "\n",
    "class PositionalEmbedding(layers.Layer):\n",
    "    def __init__(self, sequence_length, input_dim, output_dim, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self.token_embeddings = layers.Embedding(\n",
    "            input_dim=input_dim, output_dim=output_dim)\n",
    "        self.position_embeddings = layers.Embedding(\n",
    "            input_dim=sequence_length, output_dim=output_dim)\n",
    "        self.sequence_length = sequence_length\n",
    "        self.input_dim = input_dim\n",
    "        self.output_dim = output_dim\n",
    "\n",
    "    def call(self, inputs):\n",
    "        length = tf.shape(inputs)[-1]\n",
    "        positions = tf.range(start=0, limit=length, delta=1)\n",
    "        embedded_tokens = self.token_embeddings(inputs)\n",
    "        embedded_positions = self.position_embeddings(positions)\n",
    "        return embedded_tokens + embedded_positions\n",
    "\n",
    "    def compute_mask(self, inputs, mask=None):\n",
    "        return tf.math.not_equal(inputs, 0)\n",
    "\n",
    "    def get_config(self):\n",
    "        config = super(PositionalEmbedding, self).get_config()\n",
    "        config.update({\n",
    "            \"output_dim\": self.output_dim,\n",
    "            \"sequence_length\": self.sequence_length,\n",
    "            \"input_dim\": self.input_dim,\n",
    "        })\n",
    "        return config\n",
    "\n",
    "\n",
    "class TransformerDecoder(layers.Layer):\n",
    "    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self.embed_dim = embed_dim\n",
    "        self.dense_dim = dense_dim\n",
    "        self.num_heads = num_heads\n",
    "        self.attention_1 = layers.MultiHeadAttention(\n",
    "          num_heads=num_heads, key_dim=embed_dim)\n",
    "        self.attention_2 = layers.MultiHeadAttention(\n",
    "          num_heads=num_heads, key_dim=embed_dim)\n",
    "        self.dense_proj = keras.Sequential(\n",
    "            [layers.Dense(dense_dim, activation=\"relu\"),\n",
    "             layers.Dense(embed_dim),]\n",
    "        )\n",
    "        self.layernorm_1 = layers.LayerNormalization()\n",
    "        self.layernorm_2 = layers.LayerNormalization()\n",
    "        self.layernorm_3 = layers.LayerNormalization()\n",
    "        self.supports_masking = True\n",
    "\n",
    "    def get_config(self):\n",
    "        config = super(TransformerDecoder, self).get_config()\n",
    "        config.update({\n",
    "            \"embed_dim\": self.embed_dim,\n",
    "            \"num_heads\": self.num_heads,\n",
    "            \"dense_dim\": self.dense_dim,\n",
    "        })\n",
    "        return config\n",
    "\n",
    "    def get_causal_attention_mask(self, inputs):\n",
    "        input_shape = tf.shape(inputs)\n",
    "        batch_size, sequence_length = input_shape[0], input_shape[1]\n",
    "        i = tf.range(sequence_length)[:, tf.newaxis]\n",
    "        j = tf.range(sequence_length)\n",
    "        mask = tf.cast(i >= j, dtype=\"int32\")\n",
    "        mask = tf.reshape(mask, (1, input_shape[1], input_shape[1]))\n",
    "        mult = tf.concat(\n",
    "            [tf.expand_dims(batch_size, -1),\n",
    "             tf.constant([1, 1], dtype=tf.int32)], axis=0)\n",
    "        return tf.tile(mask, mult)\n",
    "\n",
    "    def call(self, inputs, encoder_outputs, mask=None):\n",
    "        causal_mask = self.get_causal_attention_mask(inputs)\n",
    "        if mask is not None:\n",
    "            padding_mask = tf.cast(\n",
    "                mask[:, tf.newaxis, :], dtype=\"int32\")\n",
    "            padding_mask = tf.minimum(padding_mask, causal_mask)\n",
    "        attention_output_1 = self.attention_1(\n",
    "            query=inputs,\n",
    "            value=inputs,\n",
    "            key=inputs,\n",
    "            attention_mask=causal_mask)\n",
    "        attention_output_1 = self.layernorm_1(inputs + attention_output_1)\n",
    "        attention_output_2 = self.attention_2(\n",
    "            query=attention_output_1,\n",
    "            value=encoder_outputs,\n",
    "            key=encoder_outputs,\n",
    "            attention_mask=padding_mask,\n",
    "        )\n",
    "        attention_output_2 = self.layernorm_2(\n",
    "            attention_output_1 + attention_output_2)\n",
    "        proj_output = self.dense_proj(attention_output_2)\n",
    "        return self.layernorm_3(attention_output_2 + proj_output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "_qO94YIGCrEQ"
   },
   "source": [
    "**간단한 트랜스포머 기반 언어 모델**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "id": "LGCHdjUqCrEQ"
   },
   "outputs": [],
   "source": [
    "from tensorflow.keras import layers\n",
    "embed_dim = 256\n",
    "latent_dim = 2048\n",
    "num_heads = 2\n",
    "\n",
    "inputs = keras.Input(shape=(None,), dtype=\"int64\")\n",
    "x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(inputs)\n",
    "x = TransformerDecoder(embed_dim, latent_dim, num_heads)(x, x)\n",
    "outputs = layers.Dense(vocab_size, activation=\"softmax\")(x)\n",
    "model = keras.Model(inputs, outputs)\n",
    "model.compile(loss=\"sparse_categorical_crossentropy\", optimizer=\"rmsprop\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "SXrXEm3ICrEQ"
   },
   "source": [
    "### 가변 온도 샘플링을 사용한 텍스트 생성 콜백"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "zeARJmFICrEQ"
   },
   "source": [
    "**텍스트 생성 콜백**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "id": "RtzXqzI1CrER"
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "tokens_index = dict(enumerate(text_vectorization.get_vocabulary()))\n",
    "\n",
    "def sample_next(predictions, temperature=1.0):\n",
    "    predictions = np.asarray(predictions).astype(\"float64\")\n",
    "    predictions = np.log(predictions) / temperature\n",
    "    exp_preds = np.exp(predictions)\n",
    "    predictions = exp_preds / np.sum(exp_preds)\n",
    "    probas = np.random.multinomial(1, predictions, 1)\n",
    "    return np.argmax(probas)\n",
    "\n",
    "class TextGenerator(keras.callbacks.Callback):\n",
    "    def __init__(self,\n",
    "                 prompt,\n",
    "                 generate_length,\n",
    "                 model_input_length,\n",
    "                 temperatures=(1.,),\n",
    "                 print_freq=1):\n",
    "        self.prompt = prompt\n",
    "        self.generate_length = generate_length\n",
    "        self.model_input_length = model_input_length\n",
    "        self.temperatures = temperatures\n",
    "        self.print_freq = print_freq\n",
    "\n",
    "    def on_epoch_end(self, epoch, logs=None):\n",
    "        if (epoch + 1) % self.print_freq != 0:\n",
    "            return\n",
    "        for temperature in self.temperatures:\n",
    "            print(\"== Generating with temperature\", temperature)\n",
    "            sentence = self.prompt\n",
    "            for i in range(self.generate_length):\n",
    "                tokenized_sentence = text_vectorization([sentence])\n",
    "                predictions = self.model(tokenized_sentence)\n",
    "                next_token = sample_next(predictions[0, i, :])\n",
    "                sampled_token = tokens_index[next_token]\n",
    "                sentence += \" \" + sampled_token\n",
    "            print(sentence)\n",
    "\n",
    "prompt = \"This movie\"\n",
    "text_gen_callback = TextGenerator(\n",
    "    prompt,\n",
    "    generate_length=50,\n",
    "    model_input_length=sequence_length,\n",
    "    temperatures=(0.2, 0.5, 0.7, 1., 1.5))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "IZIt8p-oCrER"
   },
   "source": [
    "**언어 모델 훈련하기**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "PeBx-DacCrER"
   },
   "outputs": [],
   "source": [
    "model.fit(lm_dataset, epochs=200, callbacks=[text_gen_callback])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "7_V8gQCWCrER"
   },
   "source": [
    "### 정리"
   ]
  }
 ],
 "metadata": {
  "accelerator": "GPU",
  "colab": {
   "collapsed_sections": [],
   "name": "chapter12_part01_text-generation.i",
   "provenance": []
  },
  "kernelspec": {
   "display_name": "default:Python",
   "language": "python",
   "name": "conda-env-default-py"
  },
  "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.9.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}