7-语言模型

什么是语言模型

• 通俗来讲
  语言模型评价一句话是否“合理”或“是人话”
• 数学上讲
  P(今天天气不错) > P(今错不天天气)
  语言模型用于计算文本的成句概率

语言模型的用途

语音识别、手写识别、输入法。

N-gram语言模型

• 平滑问题
  
• 平滑方法-回退
  
• 平滑方法-加一平滑
  
• 平滑方法-低频词转为
  
• 平滑方法-插值
  

语言模型的评价指标

• 困惑度 perplexity
  一般使用合理的目标文本来计算PPL，<若PPL值低，则说明成句概率高>，也就说明由此语言模型来判断，该句子的合理性高，这样是一个好的语言模型
  

神经网络语言模型优势

向量化表示语义信息优于字符统计，泛化性更好
输入长度不影响模型大小，长距离建模优势
Softmax带来的自带平滑
对下游任务的适配更加方便（作为预训练基座）

• 语言模型应用
  1.话者分离：根据说话内容判断说话人。常用于语言识别系统中，判断录音对话中角色。如客服对话录音，判断坐席或客户
  2.文本纠错：错误可能是同音字或形近字等。对每一个字建立一个混淆字集合。计算整句话成句概率。用混淆字集合中的词替代原句中的字，重新计算概率。
  3.数字归一化：将数字部分依照其格式替换为<阿拉伯数字><汉字数字><汉字连读>等token。使用带token文本训练语言模型。对于新输入的文本，同样使用正则表达式找到数字部分，之后分别带入各个token，使用语言模型计算概率。选取概率最高的token为最终数字格式。
  4.文本打标：可以理解为一种粗粒度的分词。可以依照类似方式，处理分词、文本加标点、文本段落切分等任务。分词或切分段落只需要一种token；打标点时，可以用多种分隔token，代表不同标点。

• 总结
  1.语言模型的核心能力是计算成句概率，依赖这一能力，可以完成大量不同类型的NLP任务。
  2.基于统计的语言模型和基于神经网络的语言模型各有使用的场景，大体上讲，基于统计的模型优势在于解码速度，而神经网络的模型通常效果更好。
  3.单纯通过PPL评价语言模型是有局限的，通过下游任务效果进行整体评价更好。
  4.深入的理解一种算法，有助于发现更多的应用方式。
  5.看似简单（甚至错误）的假设，也能带来有意义的结果，事实上，这是简化问题的常见方式。

NgramLanguageModel

import math
from collections import defaultdictclass NgramLanguageModel:def __init__(self, corpus=None, n=3):self.n = nself.sep = "_"     # 用来分割两个词，没有实际含义，只要是字典里不存在的符号都可以self.sos = "<sos>"    #start of sentence，句子开始的标识符self.eos = "<eos>"    #end of sentence，句子结束的标识符self.unk_prob = 1e-5  #给unk分配一个比较小的概率值，避免集外词概率为0self.fix_backoff_prob = 0.4  #使用固定的回退概率self.ngram_count_dict = dict((x + 1, defaultdict(int)) for x in range(n))self.ngram_count_prob_dict = dict((x + 1, defaultdict(int)) for x in range(n))self.ngram_count(corpus)self.calc_ngram_prob()#将文本切分成词或字或tokendef sentence_segment(self, sentence):return sentence.split()#return jieba.lcut(sentence)#统计ngram的数量def ngram_count(self, corpus):for sentence in corpus:word_lists = self.sentence_segment(sentence)word_lists = [self.sos] + word_lists + [self.eos]  #前后补充开始符和结尾符for window_size in range(1, self.n + 1):           #按不同窗长扫描文本for index, word in enumerate(word_lists):#取到末尾时窗口长度会小于指定的gram，跳过那几个if len(word_lists[index:index + window_size]) != window_size:continue#用分隔符连接word形成一个ngram用于存储ngram = self.sep.join(word_lists[index:index + window_size])self.ngram_count_dict[window_size][ngram] += 1#计算总词数，后续用于计算一阶ngram概率self.ngram_count_dict[0] = sum(self.ngram_count_dict[1].values())return#计算ngram概率def calc_ngram_prob(self):for window_size in range(1, self.n + 1):for ngram, count in self.ngram_count_dict[window_size].items():if window_size > 1:ngram_splits = ngram.split(self.sep)              #ngram        :a b cngram_prefix = self.sep.join(ngram_splits[:-1])   #ngram_prefix :a bngram_prefix_count = self.ngram_count_dict[window_size - 1][ngram_prefix] #Count(a,b)else:ngram_prefix_count = self.ngram_count_dict[0]     #count(total word)# word = ngram_splits[-1]# self.ngram_count_prob_dict[word + "|" + ngram_prefix] = count / ngram_prefix_countself.ngram_count_prob_dict[window_size][ngram] = count / ngram_prefix_countreturn#获取ngram概率，其中用到了回退平滑，回退概率采取固定值def get_ngram_prob(self, ngram):n = len(ngram.split(self.sep))if ngram in self.ngram_count_prob_dict[n]:#尝试直接取出概率return self.ngram_count_prob_dict[n][ngram]elif n == 1:#一阶gram查找不到，说明是集外词，不做回退return self.unk_probelse:#高于一阶的可以回退ngram = self.sep.join(ngram.split(self.sep)[1:])return self.fix_backoff_prob * self.get_ngram_prob(ngram)#回退法预测句子概率def calc_sentence_ppl(self, sentence):word_list = self.sentence_segment(sentence)word_list = [self.sos] + word_list + [self.eos]sentence_prob = 0for index, word in enumerate(word_list):ngram = self.sep.join(word_list[max(0, index - self.n + 1):index + 1])prob = self.get_ngram_prob(ngram)# print(ngram, prob)sentence_prob += math.log(prob)return 2 ** (sentence_prob * (-1 / len(word_list)))if __name__ == "__main__":corpus = open("sample.txt", encoding="utf8").readlines()lm = NgramLanguageModel(corpus, 3)print("词总数:", lm.ngram_count_dict[0])print(lm.ngram_count_prob_dict)print(lm.calc_sentence_ppl("a c b e f d"))

NNLM

#coding:utf8
import torch
import torch.nn as nn
import numpy as np
import math
import random
import os"""
基于pytorch的rnn语言模型
"""
class LanguageModel(nn.Module):def __init__(self, input_dim, vocab):super(LanguageModel, self).__init__()self.embedding = nn.Embedding(len(vocab) + 1, input_dim)self.layer = nn.RNN(input_dim, input_dim, num_layers=2, batch_first=True)self.classify = nn.Linear(input_dim, len(vocab) + 1)self.dropout = nn.Dropout(0.1)self.loss = nn.functional.cross_entropy#当输入真实标签，返回loss值；无真实标签，返回预测值def forward(self, x, y=None):x = self.embedding(x)  #output shape:(batch_size, sen_len, input_dim)x, _ = self.layer(x)      #output shape:(batch_size, sen_len, input_dim)x = x[:, -1, :]        #output shape:(batch_size, input_dim)x = self.dropout(x)y_pred = self.classify(x)   #output shape:(batch_size, vocab_size)if y is not None:return self.loss(y_pred, y) #[1*vocab_size] []else:return torch.softmax(y_pred, dim=-1)#读取语料获得字符集
#输出一份
def build_vocab_from_corpus(path):vocab = set()with open(path, encoding="utf8") as f:for index, char in enumerate(f.read()):vocab.add(char)vocab.add("<UNK>") #增加一个unk token用来处理未登录词writer = open("vocab.txt", "w", encoding="utf8")for char in sorted(vocab):writer.write(char + "\n")return vocab#加载字表
def build_vocab(vocab_path):vocab = {}with open(vocab_path, encoding="utf8") as f:for index, line in enumerate(f):char = line[:-1]        #去掉结尾换行符vocab[char] = index + 1 #留出0位给pad tokenvocab["\n"] = 1return vocab#加载语料
def load_corpus(path):return open(path, encoding="utf8").read()#随机生成一个样本
#从文本中截取随机窗口，前n个字作为输入，最后一个字作为输出
def build_sample(vocab, window_size, corpus):start = random.randint(0, len(corpus) - 1 - window_size)end = start + window_sizewindow = corpus[start:end]target = corpus[end]# print(window, target)x = [vocab.get(word, vocab["<UNK>"]) for word in window]   #将字转换成序号y = vocab[target]return x, y#建立数据集
#batch_size 输入需要的样本数量。需要多少生成多少
#vocab 词表
#window_size 样本长度
#corpus 语料字符串
def build_dataset(batch_size, vocab, window_size, corpus):dataset_x = []dataset_y = []for i in range(batch_size):x, y = build_sample(vocab, window_size, corpus)dataset_x.append(x)dataset_y.append(y)return torch.LongTensor(dataset_x), torch.LongTensor(dataset_y)#建立模型
def build_model(vocab, char_dim):model = LanguageModel(char_dim, vocab)return model#计算文本ppl
def calc_perplexity(sentence, model, vocab, window_size):prob = 0model.eval()with torch.no_grad():for i in range(1, len(sentence)):start = max(0, i - window_size)window = sentence[start:i]x = [vocab.get(char, vocab["<UNK>"]) for char in window]x = torch.LongTensor([x])target = sentence[i]target_index = vocab.get(target, vocab["<UNK>"])if torch.cuda.is_available():x = x.cuda()pred_prob_distribute = model(x)[0]target_prob = pred_prob_distribute[target_index]prob += math.log(target_prob, 10)return 2 ** (prob * ( -1 / len(sentence)))def train(corpus_path, save_weight=True):epoch_num = 10        #训练轮数batch_size = 128       #每次训练样本个数train_sample = 10000   #每轮训练总共训练的样本总数char_dim = 128        #每个字的维度window_size = 6       #样本文本长度vocab = build_vocab("vocab.txt")       #建立字表corpus = load_corpus(corpus_path)     #加载语料model = build_model(vocab, char_dim)    #建立模型if torch.cuda.is_available():model = model.cuda()optim = torch.optim.Adam(model.parameters(), lr=0.001)   #建立优化器for epoch in range(epoch_num):model.train()watch_loss = []for batch in range(int(train_sample / batch_size)):x, y = build_dataset(batch_size, vocab, window_size, corpus) #构建一组训练样本if torch.cuda.is_available():x, y = x.cuda(), y.cuda()optim.zero_grad()    #梯度归零loss = model(x, y)   #计算losswatch_loss.append(loss.item())loss.backward()      #计算梯度optim.step()         #更新权重print("=========\n第%d轮平均loss:%f" % (epoch + 1, np.mean(watch_loss)))if not save_weight:returnelse:base_name = os.path.basename(corpus_path).replace("txt", "pth")model_path = os.path.join("model", base_name)torch.save(model.state_dict(), model_path)return#训练corpus文件夹下的所有语料，根据文件名将训练后的模型放到model文件夹
def train_all():for path in os.listdir("corpus"):corpus_path = os.path.join("corpus", path)train(corpus_path)if __name__ == "__main__":train_all()

---