4.HMM和CRF的使用和应用

1.生成式模型和判别式模型

• 生成式模型和判别式模型都用于有监督学习，有监督学习的任务就是从数据中学习一个模型（也叫分类器），应用这一模型，对给定的输入 X 预测相应的输出 Y。这个模型的一般形式为：决策函数 Y=f(X) 或者条件概率分布 P(Y|X)。

简单从贝叶斯定理说起，若记 P(A)、P(B) 分别表示事件 A 和事件 B 发生的概率，则 P(A|B) 表示事件 B 发生的情况下事件 A 发生的概率；P(AB)表示事件 A 和事件 B 同时发生的概率。

（1）生成式模型：估计的是联合概率分布，P(Y, X)=P(Y|X)*P(X)，由联合概率密度分布 P(X,Y)，然后求出条件概率分布 P(Y|X) 作为预测的模型，即生成模型公式为：P(Y|X)= P(X,Y)/ P(X)。基本思想是首先建立样本的联合概率密度模型 P(X,Y)，然后再得到后验概率 P(Y|X)，再利用它进行分类，其主要关心的是给定输入 X 产生输出 Y 的生成关系。

（2）判别式模型：估计的是条件概率分布， P(Y|X)，是给定观测变量 X 和目标变量 Y 的条件模型。由数据直接学习决策函数 Y=f(X) 或者条件概率分布 P(Y|X) 作为预测的模型，其主要关心的是对于给定的输入 X，应该预测什么样的输出 Y。

HMM 使用隐含变量生成可观测状态，其生成概率有标注集统计得到，是一个生成模型。其他常见的生成式模型有：Gaussian、 Naive Bayes、Mixtures of multinomials 等。

而 CRF 就像一个反向的隐马尔可夫模型（HMM），通过可观测状态判别隐含变量，其概率亦通过标注集统计得来，是一个判别模型。其他常见的判别式模型有：K 近邻法、感知机、决策树、逻辑斯谛回归模型、最大熵模型、支持向量机、提升方法等。

2. HMM模型

HMM 模型是由一个“五元组”组成的集合：

• StatusSet：状态值集合，状态值集合为 (B, M, E, S)，其中 B 为词的首个字，M 为词中间的字，E 为词语中最后一个字，S 为单个字，B、M、E、S 每个状态代表的是该字在词语中的位置。
  举个例子，对“中国的人工智能发展进入高潮阶段”，分词可以标注为：“中B国E的S人B工E智B能E发B展E进B入E高B潮E阶B段E”，最后的分词结果为：[‘中国’, ‘的’, ‘人工’, ‘智能’, ‘发展’, ‘进入’, ‘高潮’, ‘阶段’]。

• ObservedSet：观察值集合，观察值集合就是所有语料的汉字，甚至包括标点符号所组成的集合。

• TransProbMatrix：转移概率矩阵，状态转移概率矩阵的含义就是从状态 X 转移到状态 Y 的概率，是一个4×4的矩阵，即 {B,E,M,S}×{B,E,M,S}。

• EmitProbMatrix：发射概率矩阵，发射概率矩阵的每个元素都是一个条件概率，代表 P(Observed[i]|Status[j]) 概率。

• InitStatus：初始状态分布，初始状态概率分布表示句子的第一个字属于 {B,E,M,S} 这四种状态的概率。

将 HMM 应用在分词上，要解决的问题是：参数（ObservedSet、TransProbMatrix、EmitRobMatrix、InitStatus）已知的情况下，求解状态值序列。

解决这个问题的最有名的方法是 Viterbi 算法。Viterbi 算法

2.1 基于HMM训练中文分词器

# 定义 HMM 中的状态，初始化概率，以及中文停顿词：
import pickle
import json

STATES = {'B', 'M', 'E', 'S'}
EPS = 0.0001
#定义停顿标点
seg_stop_words = {" ","，","。","“","”",'“', "？", "！", "：", "《", "》", "、", "；", "·", "‘ ", "’", "──", ",", ".", "?", "!", "`", "~", "@", "#", "$", "%", "^", "&", "*", "(", ")", "-", "_", "+", "=", "[", "]", "{", "}", '"', "'", "<", ">", "\\", "|" "\r", "\n","\t"}
1
2
3
4
5
6
7
8

# 将 HMM 模型封装为独立的类 HMM_Model，下面先给出类的结构定义：
class HMM_Model:
    def __init__(self):
        self.trans_mat = {}  
        self.emit_mat = {} 
        self.init_vec = {}  
        self.state_count = {} 
        self.states = {}
        self.inited = False

    #初始化    
    def setup(self):
        for state in self.states:
            # build trans_mat
            self.trans_mat[state] = {}
            for target in self.states:
                self.trans_mat[state][target] = 0.0
            self.emit_mat[state] = {}
            self.init_vec[state] = 0
            self.state_count[state] = 0
        self.inited = True
        
     #模型保存   
    def save(self, filename="hmm.json", code='json'):
        fw = open(filename, 'w', encoding='utf-8')
        data = {
            "trans_mat": self.trans_mat,
            "emit_mat": self.emit_mat,
            "init_vec": self.init_vec,
            "state_count": self.state_count
        }
        if code == "json":
            txt = json.dumps(data)
            txt = txt.encode('utf-8').decode('unicode-escape')
            fw.write(txt)
        elif code == "pickle":
            pickle.dump(data, fw)
        fw.close()
        
    #模型加载
    def load(self, filename='hmm.json', code='json'):
        fr = open(filename, 'r', encoding='utf-8')
        if code == "json":
            txt = fr.read()
            model = json.loads(txt)
        elif code == "pickle":
            model = pickle.load(fr)
        self.trans_mat = model["trans_mat"]
        self.emit_mat = model["emit_mat"]
        self.init_vec = model["init_vec"]
        self.state_count = model["state_count"]
        self.inited = True
        fr.close()

    #模型训练
    def do_train(self, observes, states):
        if not self.inited:
            self.setup()
        for i in range(len(states)):
            if i == 0:
                self.init_vec[states[0]] += 1
                self.state_count[states[0]] += 1
            else:
                self.trans_mat[states[i - 1]][states[i]] += 1
                self.state_count[states[i]] += 1
                if observes[i] not in self.emit_mat[states[i]]:
                    self.emit_mat[states[i]][observes[i]] = 1
                else:
                    self.emit_mat[states[i]][observes[i]] += 1
    
    #HMM计算
    def get_prob(self):
        init_vec = {}
        trans_mat = {}
        emit_mat = {}
        default = max(self.state_count.values())  

        for key in self.init_vec:
            if self.state_count[key] != 0:
                init_vec[key] = float(self.init_vec[key]) / self.state_count[key]
            else:
                init_vec[key] = float(self.init_vec[key]) / default

        for key1 in self.trans_mat:
            trans_mat[key1] = {}
            for key2 in self.trans_mat[key1]:
                if self.state_count[key1] != 0:
                    trans_mat[key1][key2] = float(self.trans_mat[key1][key2]) / self.state_count[key1]
                else:
                    trans_mat[key1][key2] = float(self.trans_mat[key1][key2]) / default

        for key1 in self.emit_mat:
            emit_mat[key1] = {}
            for key2 in self.emit_mat[key1]:
                if self.state_count[key1] != 0:
                    emit_mat[key1][key2] = float(self.emit_mat[key1][key2]) / self.state_count[key1]
                else:
                    emit_mat[key1][key2] = float(self.emit_mat[key1][key2]) / default
        return init_vec, trans_mat, emit_mat
    
    #模型预测
    def do_predict(self, sequence):
        tab = [{}]
        path = {}
        init_vec, trans_mat, emit_mat = self.get_prob()

        # 初始化
        for state in self.states:
            tab[0][state] = init_vec[state] * emit_mat[state].get(sequence[0], EPS)
            path[state] = [state]

        # 创建动态搜索表
        for t in range(1, len(sequence)):
            tab.append({})
            new_path = {}
            for state1 in self.states:
                items = []
                for state2 in self.states:
                    if tab[t - 1][state2] == 0:
                        continue
                    prob = tab[t - 1][state2] * trans_mat[state2].get(state1, EPS) * emit_mat[state1].get(sequence[t], EPS)
                    items.append((prob, state2))
                best = max(items)  
                tab[t][state1] = best[0]
                new_path[state1] = path[best[1]] + [state1]
            path = new_path

        # 搜索最有路径
        prob, state = max([(tab[len(sequence) - 1][state], state) for state in self.states])
        return path[state]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

初始化函数（Init）的解释：

• trans_mat：状态转移矩阵，trans_mat[state1][state2] 表示训练集中由 state1 转移到 state2 的次数。
• emit_mat：观测矩阵，emit_mat[state][char] 表示训练集中单字 char 被标注为 state 的次数。
• init_vec：初始状态分布向量，init_vec[state] 表示状态 state 在训练集中出现的次数。
• state_count：状态统计向量，state_count[state]表示状态 state 出现的次数。
• word_set：词集合，包含所有单词。

函数解释：

• save()，用来保存训练好的模型，filename 指定模型名称，默认模型名称为 hmm.json，这里提供两种格式的保存类型，JSON 或者 pickle 格式，通过参数 code 来决定，code 的值为 code=‘json’ 或者 code = ‘pickle’，默认为 code=‘json’。
• load()，与第三个 save() 方法对应，用来加载模型，filename 指定模型名称，默认模型名称为 hmm.json，这里提供两种格式的保存类型，JSON 或者 pickle 格式，通过参数 code 来决定，code 的值为 code=‘json’ 或者 code = ‘pickle’，默认为 code=‘json’
• do_train()，用来训练模型，因为使用的标注数据集， 因此可以使用更简单的监督学习算法，训练函数输入观测序列和状态序列进行训练， 依次更新各矩阵数据。类中维护的模型参数均为频数而非频率， 这样的设计使得模型可以进行在线训练，使得模型随时都可以接受新的训练数据继续训练，不会丢失前次训练的结果。
• get_prob()，在进行预测前，需将数据结构的频数转换为频率.
• do_predict()，预测采用 Viterbi 算法求得最优路径.

# 对输入的训练语料中的每个词进行标注，因为训练数据是空格隔开的，可以进行转态标注，该方法用在训练数据的标注

def get_tags(src):
    tags = []
    if len(src) == 1:
        tags = ['S']
    elif len(src) == 2:
        tags = ['B', 'E']
    else:
        m_num = len(src) - 2
        tags.append('B')
        tags.extend(['M'] * m_num)
        tags.append('E')
    return tags

# 根据预测得到的标注序列将输入的句子分割为词语列表，也就是预测得到的状态序列，解析成一个 list 列表进行返回

def cut_sent(src, tags):
    word_list = []
    start = -1
    started = False

    if len(tags) != len(src):
        return None

    if tags[-1] not in {'S', 'E'}:
        if tags[-2] in {'S', 'E'}:
            tags[-1] = 'S'  
        else:
            tags[-1] = 'E'  

    for i in range(len(tags)):
        if tags[i] == 'S':
            if started:
                started = False
                word_list.append(src[start:i])  
            word_list.append(src[i])
        elif tags[i] == 'B':
            if started:
                word_list.append(src[start:i])  
            start = i
            started = True
        elif tags[i] == 'E':
            started = False
            word = src[start:i+1]
            word_list.append(word)
        elif tags[i] == 'M':
            continue
    return word_list


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

# 定义分词器类 HMMSoyoger，继承 HMM_Model 类并实现中文分词器训练、分词功能
class HMMSoyoger(HMM_Model):
    def __init__(self, *args, **kwargs):
        super(HMMSoyoger, self).__init__(*args, **kwargs)
        self.states = STATES
        self.data = None

    #加载训练数据
    def read_txt(self, filename):
        self.data = open(filename, 'r', encoding="utf-8")

    #模型训练函数: train()，根据单词生成观测序列和状态序列，并通过父类的 do_train() 方法进行训练
    def train(self):
        if not self.inited:
                self.setup()
        for line in self.data:
            line = line.strip()
            if not line:
                continue

           #观测序列
            observes = []
            for i in range(len(line)):
                if line[i] == " ":
                    continue
                observes.append(line[i])

            #状态序列
            words = line.split(" ")  

            states = []
            for word in words:
                if word in seg_stop_words:
                    continue
                states.extend(get_tags(word))
            #开始训练
            if(len(observes) >= len(states)):
                self.do_train(observes, states)
            else:
                pass

    #模型分词预测: lcut()，模型训练好之后，通过该方法进行分词测试
    def lcut(self, sentence):
        try:
            tags = self.do_predict(sentence)
            return cut_sent(sentence, tags)
        except:
            return sentence
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

# 实例化 HMMSoyoger 类，然后通过 read_txt() 方法加载语料，再通过 train() 进行在线训练
soyoger = HMMSoyoger()
soyoger.read_txt("syj_trainCorpus_utf8.txt")
soyoger.train()
1
2
3
4

print(soyoger.lcut("中国的人工智能发展进入高潮阶段。"))
soyoger.lcut("中文自然语言处理是人工智能技术的一个重要分支。")

# 想取得更好的结果，可自行制备更大更丰富的训练数据集。
1
2
3
4

3. 基于CRF进行中文分词

目前分词语料出自人民日报1998年1月份，所以对于新闻类文章分词较为准确。
CRF分词效果很大程度上依赖于训练语料的类别以及覆盖度，若解决语料问题分词和标注效果还有很大的提升空间。

Genius 是一个基于 CRF 的开源中文分词工具，采用了 Wapiti 做训练与序列标注，支持 Python。

3.1 分词 genius.seg_text() 函数接受5个参数，其中 text 是必填参数：

• text 第一个参数为需要分词的字。
• use_break 代表对分词结构进行打断处理，默认值 True。
• use_combine 代表是否使用字典进行词合并，默认值 False。
• use_tagging 代表是否进行词性标注，默认值 True。
• use_pinyin_segment 代表是否对拼音进行分词处理，默认值 True。

Algorithm：

• 采用trie树进行合并词典查找；
• 基于wapiti实现条件随机场分词；
• 可以通过genius.loader.ResourceLoader来重载默认的字典；

import genius
text = "中文自然语言处理是人工智能技术的一个重要分支。"
seg_list = genius.seg_text(
    text,
    use_combine=True,
    use_pinyin_segment=True,
    use_tagging=True,
    use_break=True
)
# print(' '.join([word.text for word in seg_list]))
1
2
3
4
5
6
7
8
9
10

import genius
# text = u"""昨天,我和施瓦布先生一起与部分企业家进行了交流,大家对中国经济当前、未来发展的态势、走势都十分关心。"""
text = "中文自然语言处理是人工智能技术的一个重要分支。"
seg_list = genius.seg_text(
    text,
    use_combine=True,
    use_pinyin_segment=True,
    use_tagging=True,
    use_break=True
)
print('\n'.join(['%s\t%s' % (word.text, word.tagging) for word in seg_list]))
1
2
3
4
5
6
7
8
9
10
11

3.2 面向索引分词

genius.seg_keywords 方法专门为搜索引擎索引准备，保留歧义分割，其中text是必填参数。

• text第一个参数为需要分词的字符
• use_break代表对分词结构进行打断处理，默认值True
• use_tagging代表是否进行词性标注，默认值False
• use_pinyin_segment代表是否对拼音进行分词处理，默认值False
• 由于合并操作与此方法有意义上的冲突，此方法并不提供合并功能；并且如果采用此方法做索引时候，检索时不推荐genius.seg_text使用use_combine=True参数。

import genius

seg_list = genius.seg_keywords(u'中文自然语言处理是人工智能技术的一个重要分支')
print('\n'.join([word.text for word in seg_list]))
1
2
3
4

3.3 关键词提取:

genius.tag_extract方法专门为提取tag关键字准备，其中text是必填参数。
1

• text第一个参数为需要分词的字符
• use_break代表对分词结构进行打断处理，默认值True
• use_combine代表是否使用字典进行词合并，默认值False
• use_pinyin_segment代表是否对拼音进行分词处理，默认值False

import genius

tag_list = genius.extract_tag(u'中文自然语言处理是人工智能技术的一个重要分支。')
print('\n'.join(tag_list))
1
2
3
4