用隐马尔科夫模型实现中文分词 NLP

隐马尔科夫模型通常用来解决序列标注问题，因此可以将分词问题转化为一个序列标注问题来进行建模。
例如可以对中文句子中得每个字做以下标注：B表示一个词开头的第一个字，E表示一个词结尾的最后一个字，M表示一个词中间的字，S表示一个单字词，那么隐藏状态空间就是{B, E, M, S}。同时对隐藏状态的转移概率可以给出一些先验知识：B和M后面只能是M或者E，S和E后面只能是B或者S。而每个字就是模型中得观测状态，取值空间就是语料中得所有中文字。
首先定义定义模型：

class HMM(object): def __init__(self): passdef try_load_model(self, trained): passdef train(self, path): passdef viterbi(self, text, states, start_p, trans_p, emit_p): passdef cut(self, text): pass

1. __init__ 函数
__init__ 主要是初始化一些全局信息和成员变量，例如状态集合[‘B’, ‘M’, ‘E’, ‘S’]，以及存取概率计算的中间文件hmm_model.pkl。

def __init__(self): import os # 主要是用于存取算法中间结果，不用每次都训练模型 self.model_file = './data/hmm_model.pkl' # 状态值集合 self.state_list = ['B', 'M', 'E', 'S'] # 参数加载,用于判断是否需要重新加载model_file self.load_para = False

2. try_load_model 函数
trained 参数用于判断是否加载中间文件结果。如果加载中间文件，则不用训练语料，可以直接进行分词调用。否则，try_load_model函数初始化初始概率、状态转移概率、以及发射概率等信息。

# 用于加载已计算的中间结果，当需要重新训练时，需初始化清空结果 def try_load_model(self, trained): if trained: import pickle with open(self.model_file, 'rb') as f: self.A_dic = pickle.load(f) self.B_dic = pickle.load(f) self.Pi_dic = pickle.load(f) self.load_para = True else: # 状态转移概率（状态->状态的条件概率） self.A_dic = {} # 发射概率（状态->词语的条件概率） self.B_dic = {} # 状态的初始概率 self.Pi_dic = {} self.load_para = False

3. train 函数

# 计算转移概率、发射概率以及初始概率 def train(self, path): # 重置几个概率矩阵 self.try_load_model(False) # 统计状态出现次数，求p(o) Count_dic = {}# 初始化参数 def init_parameters(): for state in self.state_list: self.A_dic[state] = {s: 0.0 for s in self.state_list} self.Pi_dic[state] = 0.0 self.B_dic[state] = {} Count_dic[state] = 0def makeLabel(text): out_text = [] if len(text) == 1: out_text.append('S') else: out_text += ['B'] + ['M'] * (len(text) - 2) + ['E'] return out_textinit_parameters() line_num = -1 # 观察者集合，主要是字以及标点等 words = set() with open(path, encoding='utf8') as f: for line in f: line_num += 1 line = line.strip() if not line: continue word_list = [i for i in line if i != ' '] words |= set(word_list)# 更新字的集合 linelist = line.split() line_state = [] for w in linelist: line_state.extend(makeLabel(w)) assert len(word_list) == len(line_state) for k, v in enumerate(line_state): Count_dic[v] += 1 if k == 0: self.Pi_dic[v] += 1# 每个句子的第一个字的状态，用于计算初始状态概率 else: self.A_dic[line_state[k - 1]][v] += 1# 计算转移概率 self.B_dic[line_state[k]][word_list[k]] = \ self.B_dic[line_state[k]].get(word_list[k], 0) + 1.0# 计算发射概率self.Pi_dic = {k: v * 1.0 / line_num for k, v in self.Pi_dic.items()} self.A_dic = {k: {k1: v1 / Count_dic[k] for k1, v1 in v.items()} for k, v in self.A_dic.items()} #加1平滑 self.B_dic = {k: {k1: (v1 + 1) / Count_dic[k] for k1, v1 in v.items()} for k, v in self.B_dic.items()} #序列化 import pickle with open(self.model_file, 'wb') as f: pickle.dump(self.A_dic, f) pickle.dump(self.B_dic, f) pickle.dump(self.Pi_dic, f)return self

train 函数主要用于通过给定的分词语料进行训练。语料的格式为每行一句话，没歌词以空格分隔。train 函数主要就是通过对语料的统计，得到HMM所需要的初始概率、转移概率以及发射概率（观测状态概率）。
4. cut 函数

def viterbi(self, text, states, start_p, trans_p, emit_p): V = [{}] path = {} for y in states: V[0][y] = start_p[y] * emit_p[y].get(text[0], 0) path[y] = [y] for t in range(1, len(text)): V.append({}) newpath = {} #检验训练的发射概率矩阵中是否有该字 neverSeen = text[t] not in emit_p['S'].keys() and \ text[t] not in emit_p['M'].keys() and \ text[t] not in emit_p['E'].keys() and \ text[t] not in emit_p['B'].keys() for y in states: emitP = emit_p[y].get(text[t], 0) if not neverSeen else 1.0 #设置未知字单独成词 (prob, state) = max( [(V[t - 1][y0] * trans_p[y0].get(y, 0) * emitP, y0) for y0 in states if V[t - 1][y0] > 0]) V[t][y] = prob newpath[y] = path[state] + [y] path = newpathif emit_p['M'].get(text[-1], 0)> emit_p['S'].get(text[-1], 0): (prob, state) = max([(V[len(text) - 1][y], y) for y in ('E','M')]) else: (prob, state) = max([(V[len(text) - 1][y], y) for y in states])return (prob, path[state])def cut(self, text): import os if not self.load_para: self.try_load_model(os.path.exists(self.model_file)) prob, pos_list = self.viterbi(text, self.state_list, self.Pi_dic, self.A_dic, self.B_dic) begin, next = 0, 0 for i, char in enumerate(text): pos = pos_list[i] if pos == 'B': begin = i elif pos == 'E': yield text[begin: i+1] next = i+1 elif pos == 'S': yield char next = i+1 if next < len(text): yield text[next:]

cut 函数用于实际的分词操作，通过加载训练好的中间文件hmm_model.pkl，然后调用 veterbi 函数来完成。Veterbi算法在前面的《隐马尔科夫模型的三个基本问题-最可能隐藏状态序列求解》以及提到了，主要是求最大概率的路径。
【用隐马尔科夫模型实现中文分词】5. 分词测试