标签：text self 矩阵 state hmm 马尔可夫 dict Pi 分词

本文摘要 · 理论来源：【统计自然语言处理】第七章 自动分词；【统计学习方法】第十章 隐马尔可夫模型
· 代码目的：手写HMM进行中文分词
作者：CSDN 征途黯然.

一、数据集

  数据集的形式如下：

新	B
华	M
社	E
北	B
京	E
二	B
月	E
十	B
二	M
日	E
电	S

中	B
国	E
十	B
四	E
……

  数据集已经标注好了四种状态（B、M、E、S），每个句子之间用换行分割。

  获取本数据集或者代码工程，可以关注公众号‘三黄工作室’回复‘中文分词’。

二、代码介绍

  1. 定义一个类，HmmModel。

  2. 类中定义属性，分词状态self.STATE，状态转移矩阵self.A_dict，发射矩阵self.B_dict，初始矩阵self.Pi_dict。

  3. 类中函数load，先加载序列化的中间模型，如果中间模型数据不存在，则去加载语料库，重新训练，训练好了之后，把中间数据序列化保存成.pkl文件。
  这里的中间模型，指的是状态转移矩阵self.A_dict，发射矩阵self.B_dict，初始矩阵self.Pi_dict这3个矩阵的数据。
  把中间数据序列化保存成.pkl文件，需要调用第4步的save方法。

  4. 类中函数save，保存状态转移矩阵self.A_dict，发射矩阵self.B_dict，初始矩阵self.Pi_dict这3个矩阵的数据到.pkl文件。

  5. 类中函数viterbi，维特比算法。根据输入的句子text，进行索引，返回最优的状态序列。

  6. 类中函数cut，把维特比算法中返回最优的状态序列进行识别切分。

三、代码

import pickle


class HmmModel:

    def __init__(self):
        # 分词状态
        self.STATE = {'B', 'M', 'E', 'S'}
        # 状态转移矩阵
        self.A_dict = {}
        # 发射矩阵
        self.B_dict = {}
        # 初始矩阵
        self.Pi_dict = {}

    # 加载数据 先加载模型数据，没有就读取语料库重新训练
    def load(self, model_file='../dataset/hmm/model.pkl', train_file='../dataset/hmm/train.txt'):

        # 加载模型数据
        try:
            with open(model_file, 'rb') as f:
                self.A_dict = pickle.load(f)
                self.B_dict = pickle.load(f)
                self.Pi_dict = pickle.load(f)
                return
        except FileNotFoundError:
            pass

        # 统计状态出现次数 方便求发射矩阵
        Count_dict = {}
        # 存放初始语料所有数据
        data = []
        # 存放初始语料中的一个句子
        sentence = []

        # 初始化模型参数
        def init_params():
            for state in self.STATE:
                self.A_dict[state] = {s: 0.0 for s in self.STATE}
                self.Pi_dict[state] = 0.0
                self.B_dict[state] = {}
                Count_dict[state] = 0

        init_params()

        # 读取语料库
        with open(train_file, encoding='utf8') as f:
            # 每句按元组存在data中
            for line in f:
                line = line.strip()
                word_list = [i for i in line if i != '\t']
                if not line:
                    data.append(sentence)
                    sentence = []
                else:
                    sentence.append((word_list[0], word_list[1]))

            # 统计次数
            for s in data:
                for k, v in enumerate(s):
                    Count_dict[v[1]] += 1
                    if k == 0:
                        self.Pi_dict[v[1]] += 1  # 每个句子的第一个字的状态，用于计算初始状态概率
                    else:
                        self.A_dict[s[k - 1][1]][v[1]] += 1  # 计算转移概率
                        self.B_dict[s[k][1]][v[0]] = self.B_dict[s[k][1]].get(v[0], 0) + 1.0  # 计算发射概率

            # 计算频率
            self.Pi_dict = {k: v * 1.0 / len(data) for k, v in self.Pi_dict.items()}
            self.A_dict = {k: {k1: v1 / Count_dict[k] for k1, v1 in v.items()} for k, v in self.A_dict.items()}
            # 加1平滑
            self.B_dict = {k: {k1: (v1 + 1) / Count_dict[k] for k1, v1 in v.items()} for k, v in self.B_dict.items()}

            # 把中间模型数据保存下来
            self.save()

    # 保存中间模型数据
    def save(self, model_file='../dataset/hmm/model.pkl'):
        # 序列化
        import pickle
        with open(model_file, 'wb') as f:
            pickle.dump(self.A_dict, f)
            pickle.dump(self.B_dict, f)
            pickle.dump(self.Pi_dict, f)

    # 维特比算法
    def viterbi(self, text):
        # 加载数据
        self.load()
        # 赋别名
        states, start_p, trans_p, emit_p = self.STATE, self.Pi_dict, self.A_dict, self.B_dict
        # 初始化顶点集、路径集
        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:
                # 生词值为1，发射矩阵一行内词找不到为0(发射矩阵有4行)
                emitP = emit_p[y].get(text[t], 0) if not neverSeen else 1.0  # 设置未知字单独成词

                # 在当前状态为y下，计算前一个时刻的四种状态的代价乘积，取max
                (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):
        prob, pos_list = self.viterbi(text)
        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 = HmmModel()
text = '人类社会前进的航船就要驶入21世纪的新航程。'
res = hmm.cut(text)
print(str(list(res)))

  测试结果:

['人类', '社会', '前进', '的', '航船', '就', '要', '驶入', '21', '世纪', '的', '新', '航程', '。']

获取本项目的源代码

如果需要本组件的源代码，请扫描关注我的公众号，回复“中文分词”即可。

标签：text,self,矩阵,state,hmm,马尔可夫,dict,Pi,分词
来源： https://blog.csdn.net/qq_43592352/article/details/120541578

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。