KanggoChatbot.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "gpuType": "T4"
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    },
    "accelerator": "GPU"
  },
  "cells": [
    {
      "cell_type": "code",
      "source": [
        "!pip install konlpy"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "1n6vYJOpH3DC",
        "outputId": "bc8db2ca-11b8-4acf-ec2c-ed94cb979580"
      },
      "execution_count": 45,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting konlpy\n",
            "  Downloading konlpy-0.6.0-py2.py3-none-any.whl (19.4 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m19.4/19.4 MB\u001b[0m \u001b[31m70.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting JPype1>=0.7.0 (from konlpy)\n",
            "  Downloading JPype1-1.4.1-cp310-cp310-manylinux_2_12_x86_64.manylinux2010_x86_64.whl (465 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m465.3/465.3 kB\u001b[0m \u001b[31m52.9 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: lxml>=4.1.0 in /usr/local/lib/python3.10/dist-packages (from konlpy) (4.9.3)\n",
            "Requirement already satisfied: numpy>=1.6 in /usr/local/lib/python3.10/dist-packages (from konlpy) (1.23.5)\n",
            "Requirement already satisfied: packaging in /usr/local/lib/python3.10/dist-packages (from JPype1>=0.7.0->konlpy) (23.2)\n",
            "Installing collected packages: JPype1, konlpy\n",
            "Successfully installed JPype1-1.4.1 konlpy-0.6.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import re\n",
        "\n",
        "import pandas as pd\n",
        "import torch\n",
        "from torch.utils.data import Dataset\n",
        "from konlpy.tag import Okt\n",
        "from tqdm import tqdm\n",
        "\n",
        "PAD_TOKEN = 0\n",
        "SOS_TOKEN = 1\n",
        "EOS_TOKEN = 2\n",
        "\n",
        "\n",
        "class WordVocab():\n",
        "    def __init__(self):\n",
        "        self.word2index = {\n",
        "            '<PAD>': PAD_TOKEN,\n",
        "            '<SOS>': SOS_TOKEN,\n",
        "            '<EOS>': EOS_TOKEN,\n",
        "        }\n",
        "        self.word2count = {}\n",
        "        self.index2word = {\n",
        "            PAD_TOKEN: '<PAD>',\n",
        "            SOS_TOKEN: '<SOS>',\n",
        "            EOS_TOKEN: '<EOS>'\n",
        "        }\n",
        "\n",
        "        self.n_words = 3  # PAD, SOS, EOS 포함\n",
        "\n",
        "    def add_sentence(self, sentence):\n",
        "        for word in sentence.split(' '):\n",
        "            self.add_word(word)\n",
        "\n",
        "    def add_word(self, word):\n",
        "        if word not in self.word2index:\n",
        "            self.word2index[word] = self.n_words\n",
        "            self.word2count[word] = 1\n",
        "            self.index2word[self.n_words] = word\n",
        "            self.n_words += 1\n",
        "        else:\n",
        "            self.word2count[word] += 1\n",
        "\n",
        "\n",
        "class TextDataset(Dataset):\n",
        "    def __init__(self, filename, min_length=2, max_length=32):\n",
        "        super(TextDataset, self).__init__()\n",
        "\n",
        "        # TOKEN 정의\n",
        "        self.PAD_TOKEN = 0  # Padding 토큰\n",
        "        self.SOS_TOKEN = 1  # SOS 토큰\n",
        "        self.EOS_TOKEN = 2  # EOS 토큰\n",
        "\n",
        "        self.tagger = Okt()   # 형태소 분석기\n",
        "        self.max_length = max_length  # 한 문장의 최대 길이 지정\n",
        "\n",
        "        # CSV 데이터 로드\n",
        "        df = pd.read_csv(filename)\n",
        "\n",
        "        # 한글 정규화\n",
        "        korean_pattern = r'[^ ?,.!A-Za-z0-9가-힣+]'\n",
        "        self.normalizer = re.compile(korean_pattern)\n",
        "\n",
        "        # src: 질의, tgt: 답변\n",
        "        src_clean = []\n",
        "        tgt_clean = []\n",
        "\n",
        "        # 단어 사전 생성\n",
        "        wordvocab = WordVocab()\n",
        "\n",
        "        for _, row in tqdm(df.iterrows(), total=len(df)):\n",
        "            src = row[\"question\"]\n",
        "            tgt = row[\"answer\"]\n",
        "\n",
        "            # 한글 전처리\n",
        "            src = self.clean_text(src)\n",
        "            tgt = self.clean_text(tgt)\n",
        "\n",
        "            if len(src.split()) > min_length and len(tgt.split()) > min_length:\n",
        "                # 최소 길이를 넘어가는 문장의 단어만 추가\n",
        "                wordvocab.add_sentence(src)\n",
        "                wordvocab.add_sentence(tgt)\n",
        "                src_clean.append(src)\n",
        "                tgt_clean.append(tgt)\n",
        "\n",
        "        self.srcs = src_clean\n",
        "        self.tgts = tgt_clean\n",
        "        self.wordvocab = wordvocab\n",
        "\n",
        "    def normalize(self, sentence):\n",
        "        # 정규표현식에 따른 한글 정규화\n",
        "        return self.normalizer.sub(\"\", sentence)\n",
        "\n",
        "    def clean_text(self, sentence):\n",
        "        # 한글 정규화\n",
        "        sentence = self.normalize(sentence)\n",
        "        # 형태소 처리\n",
        "        sentence = self.tagger.morphs(sentence)\n",
        "        sentence = ' '.join(sentence)\n",
        "        sentence = sentence.lower()\n",
        "        return sentence\n",
        "\n",
        "    def texts_to_sequences(self, sentence):\n",
        "        # 문장 -> 시퀀스로 변환\n",
        "        return [self.wordvocab.word2index[w] for w in sentence.split()]\n",
        "\n",
        "    def pad_sequence(self, sentence_tokens):\n",
        "        # 문장의 맨 끝 토큰은 제거\n",
        "        sentence_tokens = sentence_tokens[:(self.max_length - 1)]\n",
        "        token_length = len(sentence_tokens)\n",
        "\n",
        "        # 문장의 맨 끝부분에 <EOS> 토큰 추가\n",
        "        sentence_tokens.append(self.EOS_TOKEN)\n",
        "\n",
        "        for i in range(token_length, (self.max_length - 1)):\n",
        "            # 나머지 빈 곳에 <PAD> 토큰 추가\n",
        "            sentence_tokens.append(self.PAD_TOKEN)\n",
        "        return sentence_tokens\n",
        "\n",
        "    def __getitem__(self, idx):\n",
        "        inputs = self.srcs[idx]\n",
        "        inputs_sequences = self.texts_to_sequences(inputs)\n",
        "        inputs_padded = self.pad_sequence(inputs_sequences)\n",
        "\n",
        "        outputs = self.tgts[idx]\n",
        "        outputs_sequences = self.texts_to_sequences(outputs)\n",
        "        outputs_padded = self.pad_sequence(outputs_sequences)\n",
        "\n",
        "        return torch.tensor(inputs_padded), torch.tensor(outputs_padded)\n",
        "\n",
        "    def __len__(self):\n",
        "        return len(self.srcs)"
      ],
      "metadata": {
        "id": "9B42SKHDufUs"
      },
      "execution_count": 11,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import random\n",
        "import torch\n",
        "from torch import nn, optim\n",
        "import torch.nn.functional as F\n",
        "\n",
        "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
        "\n",
        "\n",
        "class Encoder(nn.Module):\n",
        "    def __init__(self, num_vocabs, hidden_size, embedding_dim, num_layers):\n",
        "        super(Encoder, self).__init__()\n",
        "\n",
        "        # 단어 사전의 개수 지정\n",
        "        self.num_vocabs = num_vocabs\n",
        "        # 임베딩 레이어 정의 (number of vocabs, embedding dimension)\n",
        "        self.embedding = nn.Embedding(num_vocabs, embedding_dim)\n",
        "        # GRU (embedding dimension)\n",
        "        self.gru = nn.GRU(embedding_dim,\n",
        "                          hidden_size,\n",
        "                          num_layers=num_layers,\n",
        "                          bidirectional=False)\n",
        "\n",
        "    def forward(self, x):\n",
        "        x = self.embedding(x).permute(1, 0, 2)\n",
        "        output, hidden = self.gru(x)\n",
        "        return output, hidden\n",
        "\n",
        "\n",
        "class Decoder(nn.Module):\n",
        "    def __init__(self, num_vocabs, hidden_size, embedding_dim, num_layers=1, dropout=0.2):\n",
        "        super(Decoder, self).__init__()\n",
        "        # 단어사전 개수\n",
        "        self.num_vocabs = num_vocabs\n",
        "        self.embedding = nn.Embedding(num_vocabs, embedding_dim)\n",
        "        self.dropout = nn.Dropout(dropout)\n",
        "        self.gru = nn.GRU(embedding_dim,\n",
        "                          hidden_size,\n",
        "                          num_layers=num_layers,\n",
        "                          bidirectional=False)\n",
        "\n",
        "        # 최종 출력은 단어사전의 개수\n",
        "        self.fc = nn.Linear(hidden_size, num_vocabs)\n",
        "\n",
        "    def forward(self, x, hidden_state):\n",
        "        x = x.unsqueeze(0)  # (1, batch_size) 로 변환\n",
        "        embedded = F.relu(self.embedding(x))\n",
        "        embedded = self.dropout(embedded)\n",
        "        output, hidden = self.gru(embedded, hidden_state)\n",
        "        output = self.fc(output.squeeze(0))  # (sequence_length, batch_size, hidden_size(32) x bidirectional(1))\n",
        "        return output, hidden\n",
        "\n",
        "\n",
        "class Seq2Seq(nn.Module):\n",
        "    def __init__(self, encoder, decoder, device):\n",
        "        super(Seq2Seq, self).__init__()\n",
        "        self.encoder = encoder\n",
        "        self.decoder = decoder\n",
        "        self.device = device\n",
        "\n",
        "    def forward(self, inputs, outputs, teacher_forcing_ratio=0.5):\n",
        "        # inputs : (batch_size, sequence_length)\n",
        "        # outputs: (batch_size, sequence_length)\n",
        "\n",
        "        batch_size, output_length = outputs.shape\n",
        "        output_num_vocabs = self.decoder.num_vocabs\n",
        "\n",
        "        # 리턴할 예측된 outputs를 저장할 임시 변수\n",
        "        # (sequence_length, batch_size, num_vocabs)\n",
        "        predicted_outputs = torch.zeros(output_length, batch_size, output_num_vocabs).to(self.device)\n",
        "\n",
        "        # 인코더에 입력 데이터 주입, encoder_output은 버리고 hidden_state 만 살립니다.\n",
        "        # 여기서 hidden_state가 디코더에 주입할 context vector 입니다.\n",
        "        # (Bidirectional(1) x number of layers(1), batch_size, hidden_size)\n",
        "        _, decoder_hidden = self.encoder(inputs)\n",
        "\n",
        "        # (batch_size) shape의 SOS TOKEN으로 채워진 디코더 입력 생성\n",
        "        decoder_input = torch.full((batch_size,), SOS_TOKEN, device=self.device)\n",
        "\n",
        "        # 순회하면서 출력 단어를 생성합니다.\n",
        "        # 0번째는 SOS TOKEN이 위치하므로, 1번째 인덱스부터 순회합니다.\n",
        "        for t in range(0, output_length):\n",
        "            # decoder_input : 디코더 입력 (batch_size) 형태의 SOS TOKEN로 채워진 입력\n",
        "            # decoder_output: (batch_size, num_vocabs)\n",
        "            # decoder_hidden: (Bidirectional(1) x number of layers(1), batch_size, hidden_size), context vector와 동일 shape\n",
        "            decoder_output, decoder_hidden = self.decoder(decoder_input, decoder_hidden)\n",
        "\n",
        "            # t번째 단어에 디코더의 output 저장\n",
        "            predicted_outputs[t] = decoder_output\n",
        "\n",
        "            # teacher forcing 적용 여부 확률로 결정\n",
        "            # teacher forcing 이란: 정답치를 다음 RNN Cell의 입력으로 넣어주는 경우. 수렴속도가 빠를 수 있으나, 불안정할 수 있음\n",
        "            teacher_force = random.random() < teacher_forcing_ratio\n",
        "\n",
        "            # top1 단어 토큰 예측\n",
        "            top1 = decoder_output.argmax(1)\n",
        "\n",
        "            # teacher forcing 인 경우 ground truth 값을, 그렇지 않은 경우, 예측 값을 다음 input으로 지정\n",
        "            decoder_input = outputs[:, t] if teacher_force else top1\n",
        "\n",
        "        return predicted_outputs.permute(1, 0, 2)  # (batch_size, sequence_length, num_vocabs)로 변경\n",
        "\n",
        "\n",
        "dataset = TextDataset(\"ChatbotData.csv\")\n",
        "\n",
        "# Encoder 정의\n",
        "encoder = Encoder(num_vocabs=dataset.wordvocab.n_words,\n",
        "                  hidden_size=32,\n",
        "                  embedding_dim=64,\n",
        "                  num_layers=1)\n",
        "# Decoder 정의\n",
        "decoder = Decoder(num_vocabs=dataset.wordvocab.n_words,\n",
        "                  hidden_size=32,\n",
        "                  embedding_dim=64,\n",
        "                  num_layers=1)\n",
        "# Seq2Seq 정의\n",
        "seq2seq = Seq2Seq(encoder, decoder, 'cpu')\n",
        "\n",
        "device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
        "\n",
        "NUM_VOCABS = dataset.wordvocab.n_words\n",
        "HIDDEN_SIZE = 128\n",
        "EMBEDDIMG_DIM = 64\n",
        "\n",
        "print(f'num_vocabs: {NUM_VOCABS}\\n======================')\n",
        "\n",
        "# Encoder 정의\n",
        "encoder = Encoder(num_vocabs=NUM_VOCABS,\n",
        "                  hidden_size=HIDDEN_SIZE,\n",
        "                  embedding_dim=EMBEDDIMG_DIM,\n",
        "                  num_layers=1)\n",
        "# Decoder 정의\n",
        "decoder = Decoder(num_vocabs=NUM_VOCABS,\n",
        "                  hidden_size=HIDDEN_SIZE,\n",
        "                  embedding_dim=EMBEDDIMG_DIM,\n",
        "                  num_layers=1)\n",
        "\n",
        "# Seq2Seq 생성\n",
        "# encoder, decoder를 device 모두 지정\n",
        "model = Seq2Seq(encoder.to(device), decoder.to(device), device)\n",
        "print(model)\n",
        "\n",
        "LR = 1e-3\n",
        "optimizer = optim.Adam(model.parameters(), lr=LR)\n",
        "loss_fn = nn.CrossEntropyLoss(ignore_index=PAD_TOKEN)\n",
        "\n",
        "data_loader = torch.utils.data.DataLoader(\n",
        "    dataset, batch_size=32, shuffle=True,\n",
        ")"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "68ZCseKA6zLF",
        "outputId": "2c909d09-1cc8-418c-b0c9-f5dc7a7a85b7"
      },
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "100%|██████████| 11823/11823 [00:28<00:00, 413.08it/s]\n"
          ]
        },
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "num_vocabs: 11964\n",
            "======================\n",
            "Seq2Seq(\n",
            "  (encoder): Encoder(\n",
            "    (embedding): Embedding(11964, 64)\n",
            "    (gru): GRU(64, 128)\n",
            "  )\n",
            "  (decoder): Decoder(\n",
            "    (embedding): Embedding(11964, 64)\n",
            "    (dropout): Dropout(p=0.2, inplace=False)\n",
            "    (gru): GRU(64, 128)\n",
            "    (fc): Linear(in_features=128, out_features=11964, bias=True)\n",
            "  )\n",
            ")\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "model.train()\n",
        "for epoch in tqdm(range(10)):\n",
        "  for x, y in data_loader:\n",
        "    x, y = x.to(device), y.to(device)\n",
        "\n",
        "    optimizer.zero_grad()\n",
        "\n",
        "    # output: (batch_size, sequence_length, num_vocabs)\n",
        "    output = model(x, y)\n",
        "    output_dim = output.size(2)\n",
        "\n",
        "    # 1번 index 부터 슬라이싱한 이유는 0번 index가 SOS TOKEN 이기 때문\n",
        "    # (batch_size*sequence_length, num_vocabs) 로 변경\n",
        "    output = output.reshape(-1, output_dim)\n",
        "\n",
        "    # (batch_size*sequence_length) 로 변경\n",
        "    y = y.view(-1)\n",
        "\n",
        "    # Loss 계산\n",
        "    loss = loss_fn(output, y)\n",
        "    loss.backward()\n",
        "    optimizer.step()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "_mzhFl7F9y17",
        "outputId": "657dff75-3531-474e-ab64-6db1ad7c3d2a"
      },
      "execution_count": 13,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "100%|██████████| 10/10 [04:12<00:00, 25.28s/it]\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "model.eval()\n",
        "\n",
        "\n",
        "input_sentence = input('Q: ')\n",
        "\n",
        "input_sentence = dataset.clean_text(input_sentence)\n",
        "input_sentence = dataset.texts_to_sequences(input_sentence)\n",
        "input_sentence = dataset.pad_sequence(input_sentence)\n",
        "input_sentence = torch.tensor(input_sentence).unsqueeze(0).to(device)\n",
        "\n",
        "# 입력 문장에 대한 예측된 출력 문장 생성\n",
        "output = model(input_sentence, input_sentence, teacher_forcing_ratio=0.0)\n",
        "\n",
        "# 출력 문장의 첫번째 단어는 SOS_TOKEN 이므로 제외\n",
        "output = output.argmax(dim=2)[0, 1:].cpu().numpy()\n",
        "\n",
        "# 예측된 토큰을 문장으로 변환\n",
        "output_sentence = ' '.join([\n",
        "    dataset.wordvocab.index2word[i] for i in output if i != EOS_TOKEN])\n",
        "\n",
        "print(f'A: {output_sentence}')"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "5YQazO0k687N",
        "outputId": "8d5d8fc8-c97a-4f79-e35d-62dbfadfef42"
      },
      "execution_count": 19,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Q: 감기 걸린 것 같아\n",
            "A: 생각 해보세요 .\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "HX-pUP7T9a3b"
      },
      "execution_count": 16,
      "outputs": []
    }
  ]
}