Skip-gram负采样原理及代码实现（pytorch）_python负采样

 1、skip-gram

        绝大多数的词向量学习模型本质都是建立在词和其上下文之间的联系。比如我们常见的CBOW模型，是根据上下窗口中的词的合集作为输入去预测目标词，即，其中 其中的k为超参数文本窗口大小。而skip-gram模型在此基础上做了进一步的简化，使用中的每一个词作为独立的上下文对目标词进行预测。因此skip-gram模型可以用于建立词与词之间的共现关系，即，其中。

2、负采样 

         当普通skip-gram的任务所训练的词表过大时，假设有1000000个单词，那在网络最后的分类输出层，softmax的表达式为：

        词表过大时，分母的计算量会变的非常大，有没有一种方法可以避免这种和词表大小相同的分类任务呢，这时候负采样就隆重登场了，负采样的任务目标：

        最大化当前词与其上下文的共现概率，最小化当前词和非共现的词的共现概率。

        我们用D来表示是否共现，1代表是，0代表否，则当前词c与w的共现概率表示为： 

        

        不共现概率表示为：

        其中是w和c对应的词向量，也是skip-gram网络的参数，会随着训练的进行不断更新，所以可以说词向量是skip-gram目标任务的副产物。

        损失函数为：

        其中K为超参数，即正负样本的比例，一个正样本对应K个负样本，一般来说大型训练集K的取值为2-5，小型训练集K的取值为5-20。

        现在我们有了负采样的概率、损失函数，还剩最后一个问题，负样本如何采取，我们用表示为我们的负采样概率，表示一元分词（unigram）的概率，所谓一元分词就是根据训练语料中每个词出现的次数得到的概率，则我们希望负采样是是根据来得到的，即  ，这个n是我们人为选定的一个超参数，我们可以很容易的发现，当：

        n = 0 的时候，采样方式就变成了随机采样，这显然不符合我们的采样需求。

        而当 n = 1的时候，采样方式就变成了完全依据词频进行采样，这样的话一些高频的词语比如is、are等被采样的频率都会特别高，也不符合我们的采样需求。

        所以我们需要找一个0和1之间的n作为负采样超参数，目前被证明的一种有效的负采样所对应的n为0.75。（不要问为什么不是0.74、0.76，这都是前人的智慧，反正知道n取0.75效果很好就完事了）。

3、pytorch的实现

         我用的数据是NLTK上面的reuters数据。

3.1 构建词表

from collections import defaultdict
 
class Vocab:
    def __init__(self, tokens = None):
        self.idx_to_token = list()
        self.token_to_idx = dict()
        
        if tokens is not None:
            if "<unk>" not in tokens:
                tokens.append("<unk>")
            for token in tokens:
                self.idx_to_token.append(token)
                self.token_to_idx[token] = len(self.idx_to_token) - 1
            # 单独取出未知词的下标
            self.unk = self.token_to_idx["<unk>"]
            
    @classmethod
    def build(cls, text, min_freq = 1, reserved_tokens = None):
        token_freqs = defaultdict(int)
        for sentence in text:
            for token in sentence:
                token_freqs[token] += 1
        uniq_tokens = ["<unk>"] + (reserved_tokens if reserved_tokens else [])
        uniq_tokens += [token for token, freq in token_freqs.items() if freq >= min_freq and token != "<unk>"]
        
        return cls(uniq_tokens)
        
    def __len__(self):
        return len(self.idx_to_token)
    def __getitem__(self, token):
        return self.token_to_idx.get(token, self.unk)
    def covert_token_to_idx(self, tokens):
        # 查找一系列输入词对应的索引
        return [self.token_to_idx[token] for token in tokens]
    def covert_idx_to_token(self, indices):
        # 查找一系列索引值对应的输入
        return [self.idx_to_token[index] for index in indices]

其中的@classmethod是一个装饰器，可以写一个方法，在类初始化前先进行一系列的预处理，具体的用法可以参考(1条消息) python @classmethod_俭任G的博客-CSDN博客

        Vocab输入是一个二维数组，每一维都是一个句子，每个句子中包含若干个词，build函数用于给每个句子添加一些标记，比如说<BOS>表示句子的开头，<unk>表示未知词或低频词，其中min_freq参数表示最低词频，词频小于min_freq的单词默认为低频词，不参与词表的映射构建，减少词表的大小和后期训练的计算开销。   

        下面进行reuters数据集的词表构建，返回整个词表映射完成的数据corpus和Vocab的实例化对象reuters。

BOS_TOKEN = "<BOS>" # 句首标记
EOS_TOKEN = "<EOS>" # 句尾标记
PAD_TOKEN = "<PAD>" # 填充标记
 
def load_reuters():
    from nltk.corpus import reuters
    text = reuters.sents()
    # 把所有词语进行小写处理，也是降低词表的一种方法
    text = [[token.lower() for token in sentence] for sentence in text]
    vocab = Vocab.build(text, reserved_tokens=[BOS_TOKEN,EOS_TOKEN,PAD_TOKEN])
    corpus = [vocab.covert_token_to_idx(sentence) for sentence in text]
    return corpus, vocab

 3.2 skip-gram模型构建

 先导入一些需要用到的库

from torch.utils.data import Dataset,DataLoader
import torch
from torch import nn,optim
from tqdm import tqdm
from torch.nn.utils.rnn import pad_sequence
import torch.nn.functional as F

进行skip-gram模型的数据库构建

# 模型输入（w,context,neg_context); 
class SkipGramDataset(Dataset):
    def __init__(self, corpus, vocab, context_size, n_negatives, ns_dist):
        self.data = []
        self.bos = vocab[BOS_TOKEN]
        self.eos = vocab[EOS_TOKEN]
        self.pad = vocab[PAD_TOKEN]
        for sentence in tqdm(corpus, desc = 'Dataset Construction'):
            sentence = [self.bos] + sentence + [self.eos]
            for i in range(1,len(sentence)-1):
                w = sentence[i]
                # 确定上下文左右边界，不够的地方用pad填充
                left_index = max(0,i-context_size)
                right_index = min(len(sentence),i + context_size)
                context = sentence[left_index:i] + sentence[i+1:right_index+1]
                context += [self.pad] * (context_size * 2 - len(context))
                self.data.append((w, context))
        # 正样本和负样本的比例，比如对于一个w有正样本4（context）个,则负采样20(context*n_negative)个
        self.n_negatives = n_negatives
        # 负采样的分布
        self.ns_dist = ns_dist
    def __len__(self):
        return len(self.data)
    def __getitem__(self,index):
        return self.data[index]
    def collate_fn(self, batch_datas):
        words = torch.tensor([batch[0] for batch in batch_datas], dtype = torch.long)# （batch_size）
        contexts = torch.tensor([batch[1] for batch in batch_datas], dtype = torch.long)# (batch_size,context_size)
        batch_size, context_size = contexts.shape
        neg_contexts = []
        for i in range(batch_size):
            # 保证负样本中不包含当前样本中的context，index_fill的三个参数分别表示：
            # 第一个0表示在第一个维度进行填充，原本ns_dist也就是一维的
            # 第二个context[i]表示一个句子的所有词的词表下标
            # 第三个.0表示把第二个参数所有的词表下标对应的获取概率设为0.0
            ns_dist = self.ns_dist.index_fill(0,contexts[i],.0)
            # torch.multinomial,作用是按照给的概率随机提取数组的下标
            # 第一个参数是和目标数组等大的概率数组，里面可以是小数也可以是整数
            # 第二个参数是随机取下标的数量
            # 第三个参数是取得下标是否放回，就随机取的下标是否可以重复, True就是可以重复
            neg_contexts.append(torch.multinomial(ns_dist, context_size * self.n_negatives, replacement = True))
        # 把neg_contexts 沿着维度0重新组合起来
        neg_contexts = torch.stack(neg_contexts, dim = 0)# (batch_size,context_size * n_negatives)
        return words, contexts, neg_contexts

进行skip-gram的模型构建，其实就是两层embedding层，一个用于目标词，一个用于上下文和词库的词。

# 模型中w和c分别用不同的embedding，便于训练，最后会进行参数的合并
class SkipGramModule(nn.Module):
    def __init__(self, vocab_size, embedding_dim):
        super().__init__()
        self.w_embedding = nn.Embedding(vocab_size, embedding_dim)
        self.c_embedding = nn.Embedding(vocab_size, embedding_dim)
    def forward_w(self, words):
        w_embeds = self.w_embedding(words)
        return w_embeds
    def forward_c(self, contexts):
        c_embeds = self.c_embedding(contexts)

 获取训练语料中每个词出现的频率

# 编写训练语料中的每个词的出现频率
# unigram 一元分词，把句子分成一个一个的汉字
# bigram 二元分词，把句子从头到尾每两个字组成一个词语
# trigram 三元分词，把句子从头到尾每三个字组成一个词语.
def get_unigram_distribution(corpus,vocab_size):
    token_count = torch.tensor([0]*vocab_size)
    total_count = 0
    for sentence in corpus:
        total_count += len(sentence)
        for token in sentence:
            token_count[token] += 1
    unigram_dist = torch.div(token_count.float(), total_count)
    return unigram_dist

3.3 模型的训练

先把超参数都给列出来

embedding_size = 128
# hidden_dim = 256
batch_size = 32
num_epoch = 10
context_size = 3
n_negatives = 5
device = 'cuda:0' if torch.cuda.is_available() else 'cpu'

训练前数据库的构建、模型优化器的实例化

corpus, vocab = load_reuters()
unigram_dist = get_unigram_distribution(corpus, len(vocab))
negative_sample_dist = unigram_dist ** 0.75
negative_sample_dist /= negative_sample_dist.sum()
 
dataset = SkipGramDataset(corpus, vocab, context_size = context_size, n_negatives = n_negatives, ns_dist = negative_sample_dist)
dataloader = DataLoader(dataset,batch_size = batch_size, shuffle = True, collate_fn = dataset.collate_fn)
 
# criterion = nn.NLLLoss()
model = SkipGramModule(len(vocab), embedding_size).to(device)
optimizer = optim.Adam(model.parameters(),lr = 0.001)
 

开始训练，其中有个neg()函数，作用是对一个tensor对象内的所有值取反，和取负效果一样，但逼格高点hhhh。

model.train()
for epoch in range(num_epoch):
    total_loss = 0
    for batch in tqdm(dataloader,desc = f"Training Epoch {epoch}"):
        words, contexts, neg_contexts = [x.to(device) for x in batch]
        # 取得关键词、上下文、负样本的向量表示
        word_embeds = model.w_embedding(words).unsqueeze(dim=2)#(batch_size, embedding_size, 1)
        context_embeds = model.c_embedding(contexts)#(batch_size, context_size, embedding_size)
        neg_context_embeds = model.c_embedding(neg_contexts)#(batch_size, context_size * n_negatives, embedding_size)
        # 计算损失值
        context_loss = F.logsigmoid(torch.matmul(context_embeds, word_embeds).squeeze(dim=2))#（batch_size,context_size)
        context_loss = context_loss.mean(dim=1)#(batch_size)
        neg_context_loss = F.logsigmoid(torch.matmul(neg_context_embeds, word_embeds).squeeze(dim=2).neg())#（batch_size,context_size)
        neg_context_loss = neg_context_loss.mean(dim=1)#(batch_size)
        loss = -(context_loss+neg_context_loss).mean()
 
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    print(f"loss:{total_loss:.2f}")

最后把两个embedding参数进行合并

combined_embeds = model.w_embeddings.weight + model.c_embeddings.weight

其中combined_embeds就是我们训练得到的词向量。

训练结果，用3070笔记本花了1小时...