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"}

# 将 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]

初始化函数（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

# 定义分词器类 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

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

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

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

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]))

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]))

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]))

3.3 关键词提取:

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

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

import genius

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