NLP - LSTM Code & Annotations (Paddle)

NLP Learning Notes#02

import sys
import os
import re
import random
import tarfile
from psutil import sensors_battery
import requests
import numpy as np
import paddle
import paddle.nn.functional as F
from paddle.nn import LSTM, Embedding, Dropout, Linear
from torch import dropout

use_gpu = True
paddle.set_device('gpu:0') if use_gpu else paddle.set_device('cpu')

'''
# download IMDB dataset
def download_IMDB ():
    url = "https://dataset.bj.bcebos.com/imdb%2FaclImdb_v1.tar.gz"
    web = requests.get (url)
    corpus = web.content

    with open ('./Practice/NLP/IMDB.tar.gz', 'wb') as f:
        f.write (corpus)
    f.close ()

download_IMDB ()
'''

print ("Start Loading IMDB...")

# in IMDB dataset, pos folder saves comments with positive emotion, neg negative
# the returned dataset contains elements with format (sentence, label) \
# where label = 0 and 1 indicates positive and negative respectively
def load_IMDB (atype): # type = 0: test, type = 1: 
    dataset = []
    for label in ['pos', 'neg']:
        with tarfile.open ('./aclImdb.tar.gz') as f:
            path = 'aclImdb/train/' + label + '/.*\.txt$' if atype == 'train' else 'aclImdb/test/' + label + '/.*\.txt$'
            path_compiler = re.compile (path)
            # f.next () returns a 'tarfile info' type value
            # every time we use f.next (), we would reach next file in the corresponding tar file
            # which means we can use this command to traverse all the files in it
            # when f reaches its end, tf would be None
            tf = f.next ()
            while tf != None:
                if (bool (path_compiler.match (tf.name))):
                    # extractfile returns a io.BufferedReader object
                    # therefore, we need to move on with read () function
                    # then decode with UTF-8 in default
                    content = f.extractfile (tf).read ().decode ()
                    ctype = 1 if label == 'pos' else 0
                    dataset.append ((content, ctype))
                
                tf = f.next ()
    
    return dataset

train_dataset = load_IMDB ('train')
test_dataset = load_IMDB ('test')

print ("End Loading IMDB")
print ("Start Data Preprocessing...")

# divide the sentence into words by spaces
def word_division (corpus):
    dataset = []
    for sentence, label in corpus:
        sentence = sentence.strip ().lower ()
        sentence = sentence.split (' ')

        dataset.append ((sentence, label))
    
    return dataset

train_dataset = word_division (train_dataset)
test_dataset = word_division (test_dataset)

def build_dict (corpus):
    word_freq = dict ()
    for sentence, label in corpus:
        for word in sentence:
            if not word in word_freq:
                word_freq[word] = 0
            word_freq[word] += 1
    
    word_freq = sorted (word_freq.items (), key = lambda dic: dic[1], reverse = True)

    word2id_dict = dict ()
    id_freq = dict ()

    # [oov] means 'out of vocabulary', which indicates those words showing up in test data \
    # but not existing in the training data
    word2id_dict['[oov]'] = 0
    id_freq[0] = 1e10
    # '[blank]' is used for fulfilling those blank positions
    word2id_dict['[blank]'] = 1
    id_freq[1] = 1e10

    for word, freq in word_freq:
        id = len (word2id_dict)
        word2id_dict[word] = id
        id_freq[id] = freq
    
    return id_freq, word2id_dict

id_freq, word2id_dict = build_dict (train_dataset)
vocab_size = len (word2id_dict)


def convert_sentence2id (corpus, word2id_dict):
    dataset = []
    for sentence, label in corpus:
        sentence = [word2id_dict[word] if word in word2id_dict \
                    else word2id_dict['[oov]'] for word in sentence]
        dataset.append ((sentence, label))
    
    return dataset

train_dataset = convert_sentence2id (train_dataset, word2id_dict)
test_dataset = convert_sentence2id (test_dataset, word2id_dict)

print ("End Data Preprocessing")
print ("Start Building Batch...")

# return [batch_size, max_len] tensors
def build_dataloader (word2id_dict, corpus, EPOCH_NUM, batch_size, maxlen, shuffle = True):
    sentence_batch = []
    label_batch = []
    for epoch in range (EPOCH_NUM):
        if shuffle:
            random.shuffle (corpus)
        for sentence, label in corpus:
            sample_sentence = sentence[:min (len (sentence), maxlen)]
            if len (sample_sentence) < maxlen:
                for i in range (maxlen - len (sample_sentence)):
                    sample_sentence.append (word2id_dict['[blank]'])
            
            sentence_batch.append ([[id] for id in sample_sentence])
            label_batch.append ([label])

            if len (sentence_batch) == batch_size:
                yield np.array (sentence_batch).astype ('int64'), \
                      np.array (label_batch).astype ('int64')
                sentence_batch = []
                label_batch = []
        
    if len (sentence_batch) > 0:
        yield np.array (sentence_batch).astype ('int64'), \
              np.array (label_batch).astype ('int64')

batch_size = 128
EPOCH_NUM = 5
maxlen = 128

train_dataloader = build_dataloader (word2id_dict, train_dataset, batch_size = batch_size, \
                                EPOCH_NUM = EPOCH_NUM, maxlen = maxlen, shuffle = True)

print ("End Building Batch")

# in LSTM, hidden states are stored in array 'h', cell states are stored in array 'c'
# 'h' saves those 'short-term memories', and 'c' is able to save those 'long-term memories'
class LSTMClass (paddle.nn.Layer):
    # time steps indicates "how long we remember", e.g., for "ABCDEFG", \
    # if time steps equals to three, and input is 'D', \
    # then what is to be predicted from 'D' will be involved with 'B' & 'C', for time steps equals to three
    def __init__ (self, embedding_size, vocab_size, hidden_size, init_scale = 0.1, class_num = 2, time_steps = 128, \
                    num_layers = 1, dropout_rate = None):
        super (LSTMClass, self).__init__ ()
        self.embedding_size = embedding_size
        self.vocab_size = vocab_size
        self.init_scale = init_scale
        self.hidden_size = hidden_size
        self.class_num = class_num
        self.time_steps = time_steps
        self.num_layers = num_layers
        self.dropout_rate = dropout_rate

        self.embedding = Embedding (num_embeddings = self.vocab_size, embedding_dim = self.embedding_size, sparse = False, \
                                    weight_attr = paddle.ParamAttr (
                                        initializer = paddle.nn.initializer.Uniform (
                                            low = - init_scale, high = init_scale)))
        self.LSTM = LSTM (input_size = self.hidden_size, hidden_size = self.hidden_size, num_layers = self.num_layers)
        self.fc = Linear (in_features = self.hidden_size, out_features = self.class_num)
        self.dropout = Dropout (p = self.dropout_rate)
    
    # dataset indicates input texts with size of [batch_size, maxlen]
    def forward (self, dataset):
        batch_size = dataset.shape[0]

        # we use stop_gradient here cuz it can prevent these values from being updated
        init_hidden = np.zeros ((self.num_layers, batch_size, self.hidden_size), dtype = 'float32')
        init_hidden = paddle.to_tensor (init_hidden)
        init_hidden.stop_gradient = True
        init_cell = np.zeros ((self.num_layers, batch_size, self.hidden_size), dtype = 'float32')
        init_cell = paddle.to_tensor (init_cell)
        init_cell.stop_gradient = True

        x = self.embedding (dataset)
        x = paddle.reshape (x, shape = [batch_size, self.time_steps, self.embedding_size])
        if self.dropout_rate != None and self.dropout_rate > 0:
            x = self.dropout (x)
        
        _, (ret_hidden, _) = self.LSTM (x, (init_hidden, init_cell))
        ret_hidden = ret_hidden[- 1]

        result = self.fc (ret_hidden)

        return result

dropout_rate = 0.2
num_layers = 3
hidden_size = 256
embedding_size = 256
vocab_size = len (id_freq)

model = LSTMClass (embedding_size, vocab_size, hidden_size, num_layers = num_layers, dropout_rate = dropout_rate)
optimizer = paddle.optimizer.Adam (learning_rate = 0.0001, parameters = model.parameters ())

print ("Start Training...")

def train (model):
    model.train ()

    for step, (sentence, label) in enumerate (train_dataloader):
        sentence = paddle.to_tensor (sentence)
        label = paddle.to_tensor (label)

        result = model (sentence)

        loss = F.cross_entropy (result, label)
        loss = paddle.mean (loss)

        loss.backward ()
        optimizer.step ()
        optimizer.clear_grad ()

        if step % 100 == 0:
            print ("step %d, loss %.3f" % (step, loss.numpy ()[0]))

train (model)
print ("End Training")
paddle.save (model.state_dict (), './LSTM.pt')

print ("Start Testing...")

param_dict = paddle.load ('./LSTM.pt')
model.load_dict (param_dict)

def evaluate ():
    model.eval ()

    test_dataloader = build_dataloader (word2id_dict, test_dataset, 1, batch_size, maxlen, False)

    correct, total = 0, 0
    for sentence, label in test_dataloader:
        sentence = paddle.to_tensor (sentence)
        label = paddle.to_tensor (label)

        predict = model (sentence)
        predict = F.softmax (predict)

        total += len (label)
        predict = predict.numpy ()
        for i in range (len (label)):
            if label[i][0] == 1: # shape of label is [batch_size, 1]
                if predict[i][1] > predict[i][0]:
                    correct += 1
            else:
                if predict[i][0] > predict[i][1]:
                    correct += 1
    
    accuracy = correct / total
    print ("Accuracy %.3f:" % accuracy)

evaluate ()

print ("End Testing")