自然語言處理之HMM模型分詞

漢語中句子以字為單位的，但語義理解仍是以詞為單位，所以也就存在中文分詞問題。主要的技術可以分為：規則分詞、統計分詞以及混合分詞（規則+統計）。

基於規則的分詞是一種機械分詞，主要依賴於維護詞典，在切分時將與劇中的字符串與詞典中的詞進行匹配。主要包括正向最大匹配法、逆向最大匹配法以及雙向最大匹配法。

統計分詞主要思想是將每個詞視作由字組成，如果相連的字在不同文本中出現次數越多，就越可能是一個詞。（隱馬爾可夫【HMM】、條件隨機場【CRF】等）

l  長度為m的字符串確定其概率分布:

P(x1,x2……,xm)=P(x1)P(x1|x1,x2)……P(xm|x1,x2,……,xm-1)

l  用n-gram模型將上式進行簡化，認為其概率僅與其前n-1個詞相關：

P~=P(xi|xi-(n-1),xi-(n-2)……,xi-1)=count(xi-(n-1),xi-(n-2)……,xi-1,xi)/count(xi-(n-1),xi-(n-2)……,xi-1)

 

隱馬爾可夫模型HMM將分詞作為字在字串中的序列標注任務來實現。每個字在構造一個特定詞語時都占據一個確定的構詞結構：B（詞首）、M（詞中）、E（詞尾）和S（單獨成詞）。

l  抽象表示：

W表示輸入的句子，長度n，O表示輸出標簽，則理想是maxP(o1,o2…|w1,w2……)

貝葉斯公式P(o|w)=P(w|o)P(o)/P(w) 因為P(w)是給定輸出為常數，因此要最大化P(w|o)P(o)

P(w|o)= P(w1|o1) P(w2|o3) ……P(w3|o3) 作馬爾可夫假設每個輸出僅與上一個輸出有關

P(o)=P(o1)P(o2|o1)……P(on|o(n-1))

再通過Veterbi算法求最優路徑（如果最終的最優路徑經過某個oi那么從初始結點到oi-1必然也是最優路徑）。

代碼實現如下（訓練語料庫沒有給出）：

 

class HMM(object):
    def __init__(self):
        import os

        # 主要是用於存取算法中間結果，不用每次都訓練模型
        self.model_file = 'hmm_model.pkl'

        # 狀態值集合
        self.state_list = ['B', 'M', 'E', 'S']
        # 參數加載,用於判斷是否需要重新加載model_file
        self.load_para = False

    # 用於加載已計算的中間結果，當需要重新訓練時，需初始化清空結果
    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

    # 計算轉移概率、發射概率以及初始概率
    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] = 0

        def makeLabel(text):
            out_text = []
            if len(text) == 1:
                out_text.append('S')
            else:
                out_text += ['B'] + ['M'] * (len(text) - 2) + ['E']

            return out_text

        init_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

    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 = newpath
            
        if 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:]
#####測試
hmm=HMM()
hmm.train('cutwords\\trainCorpus.txt_utf8.txt')

text='這是一個很棒的方案！'
res=hmm.cut(text)
print(text)
print(str(list(res)))