Python自然語言處理實戰（3）：中文分詞技術
3.1、中文分詞簡介
在英文中，單詞本身就是“詞”的表達，一篇英文文章就是“單詞”加分隔符（空格）來表示的，而在漢語中，詞以字為基本單位的，但是一篇文章的語義表達卻仍然是以詞來劃分的。
自中文自動分詞被提出以來，歷經將近30年的探索，提出了很多方法，可主要歸納為“規則分詞”、“統計分詞”和“混合分詞”這三個主要流派。
3.2、規則分詞
基於規則的分詞是一種機械分詞方法，主要是通過維護詞典，在切分語句時，將語句的每個字串與詞表中的詞進行逐一匹配，找到則切分，否則不予切分。按照匹配切分的方式，主要有正向最大匹配法、逆向最大匹配法以及雙向最大匹配法三種方法。
正向最大匹配法
逆向最大匹配（Reverse Maxinum Match Method，RMM法）的基本原理和MM法相同，不同的是分詞切分的方向與MM法相反。
雙向最大匹配法（Bi-direction Matching method）是將正向最大匹配法得到的分詞結果和逆向最大匹配法得到的結構進行比較，然後按照最大匹配原則，選取詞數切分最少的作為結構。據SumM.S.和Benjamin K.T.（1995）的研究表明，中文中90.0%的句子兩種切分方法得到的結果不一樣，但其中必有一個是正確的（歧義檢測成功），只有不到1.0%的句子，使用正向最大匹配法和逆向最大匹配法的切分雖重合卻是錯的，或者正向最大匹配法和逆向最大匹配法切分不同但兩個都不對（歧義檢測失敗）。這正是雙向最大匹配法在實用中文資訊處理系統中得以廣泛使用的原因。
# 逆向最大匹配 class IMM(object): def __init__(self, dic_path): self.dictionary = set(); self.maximum = 0 # 讀取詞典 with open(dic_path, 'r', encoding="utf8") as f: for line in f: line = line.strip() if not line: continue self.dictionary.add(line) self.maximum = max(self.maximum, len(line)) def cut(self, text): result = [] index = len(text) while index > 0: word = None for size in range(self.maximum, 0, -1): if index - size < 0: continue piece = text[(index - size):index] if piece in self.dictionary: word = piece result.append(word) index -= size break if word is None: index -= 1 return result[::-1] def main(): text = "南京市長江大橋" tokenizer = IMM('./data/imm_dic.utf8') print(tokenizer.cut(text))
3.3 統計分詞
其主要思想是把每個詞看做是由詞的最小單位的各個字組成的，如果相連的字在不同的文字中出現的次數越多，就證明這相連的字很可能就是一個詞。一般要做如下兩步操作：
1）建立統計語言模型：為長度為m的字串確定其概率分佈P(w1, w2, ... , wm)。當文字過長時，右部從第三項起的每一項計算難度都很大。為了解決該問題，有人提出n元模型（n-gram model）降低該計算難度。所謂n元模型就是在估算條件概率時，忽略距離大於等於n的上文詞的影響，因此P(wi | w1, w2, ... , wi-1）的計算可簡化為：P(wi | w1, w2, ... , wi-1) ~= P(wi | wi-(n-1), ... , wi-1)。
顯然當n>=2時，該模型是可以保留一定的詞序資訊的，而且n越大，保留的詞序資訊越豐富，但計算成本也呈指數級增長。一般使用頻率計數的比例來計算n元條件概率：
P(wi | wi-(n-1) , ... , wi-1) = count(wi-(n-1), ... , wi-1, wi) / count(wi-(n-1), ... , wi-1 )
由於會出現分子分母為零的情況，一般在n元模型中需要配合相應的平滑演算法解決，如拉普拉斯平滑演算法等。
2）對句子進行單詞劃分，然後對劃分結果進行概率計算，獲得概率最大的分詞方式。這裡就用到了統計學習演算法，如隱含馬爾可夫（HMM）、條件隨機場（CRF）等。
HMM是將分詞作為字在字串中的序列標註任務來實現的。其基本思路是：每個字在構造一個特定的詞語時，都佔據著一個確定的構詞位置（即詞位）。現規定每個字最多隻有四個構詞位置，即B（詞首）、M（詞中）、E（詞尾）和S（單獨成詞）。
max = max P(o1o2...on| r1r2...rn)
假設每個字的輸出僅僅與當前字有關，就能得到：P(o1o2...on| r1r2...rn)=P(o1|r1)P(o2|r2)...P(on|rn)，但該方法完全沒有考慮上下文，且會出現不合理的情況。
HMM就是用來解決該問題的一種方法。P(o | r) = P(o,r)/P(r) = P(r|o)P(o)/P(r)，其中P(r)為常數，因此求最大化P(r|o)P(o)。
在HMM中，求解max(P(r|o)P(o))的常用方法是Veterbi演算法。它是一種動態規劃方法，核心思想是：如果最終的最優路徑經過某個oi，那麼從初始節點到oi-1點的路徑必然也是一個最優路徑---因此每個節點oi只會影響前後兩個P(oi-1 | oi) 和 P(oi | oi+1)。
# -*- coding: utf-8 -*-
class HMM(object):
def __init__(self):
pass
def try_load_model(self, trained):
pass
def train(self, path):
pass
def viterbi(self, text, states, start_p, trans_p, emit_p):
pass
def cut(self, text):
pass
class HMM(object):
def __init__(self):
import os
# 主要是用於存取演算法中間結果，不用每次都訓練模型
self.model_file = "./data/hmm_model.pkl"
self.state_list = ['B', 'M', 'E', 'S']
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
# 採用人民日報的分詞語料，通過統計，得到HMM所需的初始概率、轉移概率以及發射概率
def train(self, path):
self.try_load_model(False)
Count_dic = {} #求p(o)
# 初始化引數
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="utf-8") 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))
#print(word_list)
#print(line_state)
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):
print(start_p)
#print(trans_p)
#print(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 = {}
print(text[t])
#檢驗訓練的發射概率矩陣中是否有該字
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
print(next)
if next < len(text):
yield text[next:]
hmm = HMM()
hmm.train('./data/trainCorpus.txt_utf8')
text = '這是一個非常棒的方案！'
res = hmm.cut(text)
print(text)
print(str(list(res))) 這是一個非常棒的方案！
{'M': 0.0, 'S': 0.41798844132394497, 'E': 0.0, 'B': 0.5820149148537713}
是
一
個
非
常
棒
的
方
案
！
0
2
2
4
4
6
7
8
8
10
11
['這是', '一個', '非常', '棒', '的', '方案', '！']3.5 中文分詞工具---jieba
jieba分詞結合了基於規則和基於統計這兩類方法。首先基於字首詞典進行詞圖掃描，字首詞典是指詞典中的詞按照字首包含的順序排列，可以快速構建包含全部可能分詞結果的有向無環圖，這個圖中包含多條分詞路徑，有向是指全部的路徑都始於第一個字、止於最後一個字，無環是指節點之間不構成閉環。基於標註語料，使用動態規劃的方法可以找出最大概率路徑，並將其作為最終的分詞結果。對於未登入詞，jieba使用了基於漢字成詞的HMM模型，採用了Viterbi演算法進行推導。
實戰之高頻詞提取：高頻詞一般是指文件中出現頻率較高且非無用的詞語，其一定程度上代表了文件的焦點所在。針對單篇文件，可以作為一種關鍵詞來看。對於如新聞這樣的多篇文件，可以將其作為熱詞，發現輿論焦點。需要去掉標點符號和停用詞。
下面對搜狗實驗室的新聞資料，進行高頻詞的提取
# read data
def get_content(path):
with open(path, 'r', encoding='gbk', errors='ignore') as f:
content = ''
for l in f:
l = l.strip()
content += l
return content
def get_TF(words, topK=10):
tf_dic = {}
for w in words:
tf_dic[w] = tf_dic.get(w, 0) + 1
return sorted(tf_dic.items(), key=lambda x: x[1], reverse=True)[:topK]
def stop_words(path):
with open(path) as f:
return [l.strip() for l in f]
def main():
import glob
import random
import jieba
files = glob.glob('./data/news/C000013/*.txt')
corpus = [get_content(x) for x in files]
sample_inx = random.randint(0, len(corpus))
#split_words = list(jieba.cut(corpus[sample_inx])) #停用詞
split_words = [x for x in jieba.cut(corpus[sample_inx]) if x not in stop_words('./data/stop_words.utf8')]
print('yangben 1: ' + corpus[sample_inx])
print('/ '.join(split_words))
print('topK(10): ' + str(get_TF(split_words)))topK(10): [('前列腺', 34), ('食品', 7), ('做', 7), ('男人', 6), ('排尿', 6), ('充血', 5), ('引起', 5), ('前列腺癌', 5), ('導致', 5), ('壓力', 4)]有時需要定製自己的領域詞典，用以提升分詞的效果。
jieba.load_userdict('./data/user_dict.utf8')
要求格式一般為：詞語、詞頻（可省略）、詞性（可省略），用空格隔開，順序不可顛倒，需為utf8編碼