<jupyter_start><jupyter_text>Skipgram Word Vectors with Negative SamplingReplicating Mikolov et al. 2013 with pytorchThanks to [this](https://github.com/DSKSD/DeepNLP-models-Pytorch) repo<jupyter_code>import torch
import torch.nn as nn
import torch.autograd as autograd
import torch.optim as optim
import torch.nn.functional as F
import nltk
import random
import numpy as np
from collections import Counter

random.seed(1024)
torch.manual_seed(1)
flatten = lambda l: [item for sublist in l for item in sublist]

def get_batch(batch_size, train_data):
    random.shuffle(train_data)
    start_index = 0
    end_index = batch_size
    while end_index < len(train_data):
        batch = train_data[start_index:end_index]
        temp = end_index
        end_index = end_index + batch_size
        start_index = temp
        yield batch
    if end_index >= len(train_data):
        batch = train_data[start_index:]
        yield batch

def prepare_sequence(seq, word2idx):
    idxs = list(map(lambda w: word2idx[w] if word2idx.get(w) is not None else word2idx['<UNK>'], seq))
    return(autograd.Variable(torch.tensor(idxs, dtype = torch.long)))

def prepare_word(word, word2idx):
    return(autograd.Variable(torch.tensor([word2idx[word]], dtype = torch.long) if word2idx.get(word) is not None else torch.tensor(word2idx['<UNK>'], dtype = torch.long)))
corpus = list(nltk.corpus.gutenberg.sents('melville-moby_dick.txt'))[:500]
corpus = [[word.lower() for word in sent] for sent in corpus]

word_count = Counter(flatten(corpus))
word_count.most_common()[:10]
MIN_COUNT = 3
exclude = []

for w, c in word_count.items():
    if c < MIN_COUNT:
        exclude.append(w)

vocab = list(set(flatten(corpus)) - set(exclude))
vocab.append('<UNK>')
word2idx = {'<UNK>': 0}
for word in vocab:
    if word2idx.get(word) is None:
        word2idx[word] = len(word2idx)
WINDOW_SIZE = 5
windows = flatten([list(nltk.ngrams(['<DUMMY>'] * WINDOW_SIZE + c + ['<DUMMY>'] * WINDOW_SIZE, WINDOW_SIZE * 2 + 1)) for c in corpus])

windows[:10]
# (Stores center word, context word) for skipgram with window = 5
train_data = []

for window in windows:
    for i in range(WINDOW_SIZE * 2 + 1):
        # Dont include excluded word from vocab
        if window[i] in exclude or window[WINDOW_SIZE] in exclude:
            continue
        # Also dont include center word or <DUMMY> with the center word
        if i == WINDOW_SIZE or window[i] == '<DUMMY>':
            continue
        train_data.append((window[WINDOW_SIZE], window[i]))
X_p = []
y_p = []

# Storing the (center, context) tuple in the form of indices
for pair in train_data:
    X_p.append(prepare_word(pair[0], word2idx).view(1, -1))
    y_p.append(prepare_word(pair[1], word2idx).view(1, -1))
    
train_data = list(zip(X_p, y_p))<jupyter_output><empty_output><jupyter_text>Negative Sampling<jupyter_code>Z = 0.001
num_total_words = sum([c for w, c in word_count.items() if w not in exclude])
unigram_table = []

for word in vocab:
    unigram_table.extend([word] * int(((word_count[word]/num_total_words)**0.75)/Z))

unigram_table[:10]
def negative_sampling(targets, unigram_table, k):
    batch_size = targets.size(0)
    neg_samples = []
    for i in range(batch_size):
        nsample = []
        target_index = targets[i].data.tolist()[0]
        while len(nsample) < k:
            # sample from unigram (^ 0.75) distribution
            neg_word = random.choice(unigram_table)
            if word2idx[neg_word] == target_index:
                continue
            nsample.append(neg_word)
        neg_samples.append(prepare_sequence(nsample, word2idx).view(1, -1))
    return(torch.cat(neg_samples))

negative_sampling(torch.tensor([[174]]), unigram_table, 10)
# 174 = '('
idx2word = {v:k for k,v in word2idx.items()}

test = negative_sampling(torch.tensor([[174]]), unigram_table, 10)
print("Word at 174: '{}' \nNegative Samples: ".format(idx2word.get(174)))
for i in flatten(test.data.tolist()):
    print(idx2word.get(i))<jupyter_output>Word at 174: '(' 
Negative Samples: 
,
letter
these
view
room
did
stern
or
to
sperm<jupyter_text>Defining the Model<jupyter_code>class SGNS(nn.Module):
    
    def __init__(self, vocab_size, ndim):
        super(SGNS, self).__init__()
        self.embedding_v = nn.Embedding(vocab_size, ndim) # Input Vectors
        self.embedding_u = nn.Embedding(vocab_size, ndim) # Output Vectors
        self.logsigmoid = nn.LogSigmoid()
    
        # Xavier Initialization
        initrange = (2.0/(vocab_size + ndim))**5
        self.embedding_v.weight.data.uniform_(-initrange, initrange)
        self.embedding_u.weight.data.uniform_(-0.0, 0.0)
     
    def forward(self, center_words, target_words, neg_words):
        center_vectors = self.embedding_v(center_words)
        target_vectors = self.embedding_u(target_words)
        neg_vectors = -self.embedding_u(neg_words)
        
        positive_score = target_vectors.bmm(center_vectors.transpose(1, 2)).squeeze(2)
        negative_score = torch.sum(neg_vectors.bmm(center_vectors.transpose(1, 2)).squeeze(2), 1).view(neg_vectors.size(0), -1)
        
        loss = self.logsigmoid(positive_score) + self.logsigmoid(negative_score)
        
        return(-torch.mean(loss))
    
    def get_vectors(self, inputs):
        vectors = self.embedding_v(inputs)
        return(vectors)<jupyter_output><empty_output><jupyter_text>Training the Model<jupyter_code>EMBEDDING_SIZE = 30
BATCH_SIZE = 256
EPOCH = 100
NEG = 10

losses = []
model = SGNS(len(word2idx), EMBEDDING_SIZE)
optimizer = optim.Adam(model.parameters(), lr = 0.001)
for epoch in range(EPOCH):
    for i, batch in enumerate(get_batch(BATCH_SIZE, train_data)):
        inputs, targets = zip(*batch)
        
        inputs = torch.cat(inputs)
        targets = torch.cat(targets)
        neg_samples = negative_sampling(targets, unigram_table, NEG)
        model.zero_grad()
        
        loss = model(inputs, targets, neg_samples)
        
        loss.backward()
        optimizer.step()
        
        losses.append(loss.data.tolist())
    if epoch % 10 == 0:
        print("Epoch: %d, mean_loss = %0.2f" % (epoch, np.mean(losses)))
        losses = []
model.get_vectors(prepare_word('man', word2idx))
def nearest_neighbors(target, vocab, k = 10):
    target_vector = model.get_vectors(prepare_word(target, word2idx))
    
    similarities = []
    for i in range(len(vocab)):
        if vocab[i] == target:
            continue
        vector = model.get_vectors(prepare_word(list(vocab)[i], word2idx))
        
        cosine_sim = F.cosine_similarity(target_vector, vector).data.tolist()
        similarities.append([vocab[i], cosine_sim[0]])
    return(sorted(similarities, key = lambda x: x[1], reverse = True)[:k])
        
nearest_neighbors("passenger", vocab, 10)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12511013215859032, 'fraction_numerical': 0.019676945668135097, 'mean_word_length': 3.543695797198132, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 25, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '445484', 'license': 'MIT', 'n_tokens_mistral': 2498, 'n_tokens_neox': 2385, 'n_words': 507, 'path': 'word2vec - Skipgram with Negative Sampling.ipynb', 'repo_name': 'kanishkamisra/pytorch-tutorials', 'size': '6810'}