challenging-america-word-gap-prediction/simple_neural_network.ipynb

IMPORTS

import regex as re
import sys
from torchtext.vocab import build_vocab_from_iterator
import lzma
from torch.utils.data import IterableDataset
import itertools
from torch import nn
import torch
import pickle
from torch.utils.data import DataLoader

print(torch.backends.mps.is_available())
print(torch.backends.mps.is_built())

FUNCTIONS

def get_words_from_line(line):
  line = line.rstrip()
  yield '<s>'
  for t in line.split(' '):
      yield t
  yield '</s>'

def get_word_lines_from_file(file_name):
    n = 0
    with lzma.open(file_name, 'r') as fh:
        for line in fh:
            n += 1
            if n % 1000 == 0:
                print(n ,file=sys.stderr)
            yield get_words_from_line(line.decode('utf-8'))

def look_ahead_iterator(gen):
   prev = None
   for item in gen:
      if prev is not None:
         yield (prev, item)
      prev = item

def clean(text):
    text = str(text).lower().replace('-\\\\n', '').replace('\\\\n', ' ').replace('-', '').replace('\'s', ' is').replace('\'re', ' are').replace('\'m', ' am').replace('\'ve', ' have').replace('\'ll', ' will')
    text = re.sub(r'\p{P}', '', text)
    return text

def predict(word, model, vocab):
    try:
        ixs = torch.tensor(vocab.forward([word])).to(device)
    except:
        ixs = torch.tensor(vocab.forward(['<unk>'])).to(device)
        word = '<unk>'
    out = model(ixs)
    top = torch.topk(out[0], 300)
    top_indices = top.indices.tolist()
    top_probs = top.values.tolist()
    top_words = vocab.lookup_tokens(top_indices)
    prob_list = list(zip(top_words, top_probs))
    for index, element in enumerate(prob_list):
        unk = None
        if '<unk>' in element:
            unk = prob_list.pop(index)
            prob_list.append(('', unk[1]))
            break
    if unk is None:
        prob_list[-1] = ('', prob_list[-1][1])
    return ' '.join([f'{x[0]}:{x[1]}' for x in prob_list])

def predicition_for_file(model, vocab, folder, file):
    print('=' * 10, f' do prediction for {folder}/{file} ', '=' * 10)
    with lzma.open(f'{folder}/in.tsv.xz', mode='rt', encoding='utf-8') as f:
        with open(f'{folder}/out.tsv', 'w', encoding='utf-8') as fid:

            for line in f:
                separated = line.split('\t')
                head = clean(separated[6]).split()[-1]
                new_line = predict(head, model, vocab)
                fid.write(new_line + '\n')

CLASSES

class Bigrams(IterableDataset):
  def __init__(self, text_file, vocabulary_size):
      self.vocab = build_vocab_from_iterator(
         get_word_lines_from_file(text_file),
         max_tokens = vocabulary_size,
         specials = ['<unk>'])
      self.vocab.set_default_index(self.vocab['<unk>'])
      self.vocabulary_size = vocabulary_size
      self.text_file = text_file

  def __iter__(self):
     return look_ahead_iterator(
         (self.vocab[t] for t in itertools.chain.from_iterable(get_word_lines_from_file(self.text_file))))

class SimpleBigramNeuralLanguageModel(nn.Module):
  def __init__(self, vocabulary_size, embedding_size):
      super(SimpleBigramNeuralLanguageModel, self).__init__()
      self.model = nn.Sequential(
          nn.Embedding(vocabulary_size, embedding_size),
          nn.Linear(embedding_size, vocabulary_size),
          nn.Softmax()
      )

  def forward(self, x):
      return self.model(x)

PARAMETERS

vocab_size = 30000
embed_size = 1000
batch_size = 5000
device = 'mps'
path_to_training_file = './train/in.tsv.xz'
path_to_model_file = 'model_neural_network.bin'
folder_dev_0, file_dev_0 = 'dev-0', 'in.tsv.xz'
folder_test_a, file_test_a = 'test-A', 'in.tsv.xz'
path_to_vocabulary_file = 'vocabulary_neural_network.pickle'

VOCAB

vocab = build_vocab_from_iterator(
    get_word_lines_from_file(path_to_training_file),
    max_tokens = vocab_size,
    specials = ['<unk>'])

with open(path_to_vocabulary_file, 'wb') as handle:
    pickle.dump(vocab, handle, protocol=pickle.HIGHEST_PROTOCOL)

TRAIN MODEL

train_dataset = Bigrams(path_to_training_file, vocab_size)
model = SimpleBigramNeuralLanguageModel(vocab_size, embed_size).to(device)
data = DataLoader(train_dataset, batch_size=batch_size)
optimizer = torch.optim.Adam(model.parameters())
criterion = torch.nn.NLLLoss()

model.train()
step = 0
for x, y in data:
   x = x.to(device)
   y = y.to(device)
   optimizer.zero_grad()
   ypredicted = model(x)
   loss = criterion(torch.log(ypredicted), y)
   if step % 100 == 0:
      print(step, loss)
   step += 1
   loss.backward()
   optimizer.step()

torch.save(model.state_dict(), path_to_model_file)

LOAD MODEL AND VOCAB

with open(path_to_vocabulary_file, 'rb') as handle:
        vocab = pickle.load(handle)
model = SimpleBigramNeuralLanguageModel(vocab_size, embed_size).to(device)
model.load_state_dict(torch.load(path_to_model_file))
model.eval()

CREATE OUTPUTS FILES

DEV-0

predicition_for_file(model, vocab, folder_dev_0, file_dev_0)

TEST-A

predicition_for_file(model, vocab, folder_test_a, file_test_a)