利用HMM实现分词

HMM模型实现分词的相关代码，请你在最后调用相关函数输出结果。

第一部分 模型训练

#!/usr/bin/env python3
# coding: utf-8
# states: {'B': 0, 'S': 0, 'M': 0, 'E': 0}
# observes: words

class HmmTrain:
    def __init__(self):
        self.line_index = -1    #训练语料的行号
        self.char_set = set()   #训练语料中的字集

    def init(self):  # 初始化字典
        trans_dict = {}  # 隐藏状态转移概率
                            # {'B': {'B': 0.0, 'S': 0.0, 'M': 0.0, 'E': 0.0},
                            #  'S': {'B': 0.0, 'S': 0.0, 'M': 0.0, 'E': 0.0},
                            #  'M': {'B': 0.0, 'S': 0.0, 'M': 0.0, 'E': 0.0},
                            # ...}
        emit_dict = {}  # 发射概率(状态->词语的条件概率),{'B':{ }, 'S': { }, 'M': {}, 'E': { }}
        Count_dict = {}  # 各状态出现次数，用于归一化, {'B': 0, 'S': 0, 'M': 0, 'E': 0}
        start_dict = {}  # 存储状态的初始（首字状态）概率，{'B': 0.0, 'S': 0.0, 'M': 0.0, 'E': 0.0}
        state_list = ['B', 'M', 'E', 'S']  # 状态序列
        
        for state in state_list:
            trans_dict[state] = {}
            for s in state_list:
                trans_dict[state][s] = 0.0
        for state in state_list:
            start_dict[state] = 0.0  #{'B': 0.0, 'S': 0.0, 'M': 0.0, 'E': 0.0}
            emit_dict[state] = {}    #{'B': {}, 'S': {}, 'M': {}, 'E': {}}
            Count_dict[state] = 0    # {'B': 0, 'S': 0, 'M': 0, 'E': 0}
        return trans_dict, emit_dict, start_dict, Count_dict

    #保存模型
    def save_model(self, word_dict, model_path):
        f = open(model_path, 'w')
        f.write(str(word_dict))
        f.close()

    # (词语,状态)=>(word, word_status) => 科威特 ['B', 'M', 'E']
    def get_word_status(self, word):  # 根据词语word，输出词语SBME状态列表
        #S:单字词, B:词的开头, M:词的中间, E:词的末尾
        #能 ['S'], 前往 ['B', 'E'],科威特 ['B', 'M', 'E']
        word_status = []
        if len(word) == 1:
            word_status.append('S')
        elif len(word) == 2:
            word_status = ['B', 'E']
        else:
            M_num = len(word) - 2
            M_list = ['M'] * M_num
            word_status.append('B')
            word_status.extend(M_list)
            word_status.append('E')
        return word_status

    #基于人工标注语料库，学习发射概率emit_dict，初始状态start_dict， 转移概率trans_dict
    def train(self, trainData, transMat, emitMat, startStates):
        trans_dict, emit_dict, start_dict, Count_dict = self.init() #初始化
        for line in open(trainData,encoding='utf-8'):  #逐行(逐句)处理
            self.line_index += 1
            line = line.strip()
            if not line:   #空行
                continue
            char_list = []
            for i in range(len(line)):
                if line[i] == " ":
                    continue
                char_list.append(line[i])   # 字序列
            self.char_set = set(char_list)  # 字集合

            word_list = line.split(" ")  #单词序列
            line_status = []  # 状态序列
            for word in word_list:
                line_status.extend(self.get_word_status(word))  # 一句话对应一行连续的状态

            if len(char_list) != len(line_status): continue  # 字和状态序列不匹配，不处理
            # print(word_list) # ['对外开放', '迈出', '了', '新', '的', '步伐', '。']
            # print(line_status) # ['B', 'M', 'M', 'E', 'B', 'E', 'S', 'S', 'S', 'B', 'E', 'S']
            for i in range(len(line_status)):
                Count_dict[line_status[i]] += 1  # 对应状态次数
                if i == 0:  # 首字状态计数
                    start_dict[line_status[0]] += 1
                else:  # 非首字
                    trans_dict[line_status[i - 1]][line_status[i]] += 1  # 计算转移概率
                    if char_list[i] not in emit_dict[line_status[i]]:   # 统计发射概率
                        emit_dict[line_status[i]][char_list[i]] =1# 0.0
                    else:
                        emit_dict[line_status[i]][char_list[i]] += 1  # 用于计算发射概率
        # 进行归一化
        for key in start_dict:  # 状态的初始概率
            start_dict[key] = start_dict[key] * 1.0 / self.line_index
        for row in trans_dict:  # 状态转移概率
            for col in trans_dict[row]:
                trans_dict[row][col] = trans_dict[row][col] / Count_dict[row]
        for key in emit_dict:  # 发射概率(状态->词语的条件概率)
            for word in emit_dict[key]:
                emit_dict[key][word] = emit_dict[key][word] / Count_dict[key]
        #保存模型
        self.save_model(trans_dict, transMat)
        self.save_model(emit_dict, emitMat)
        self.save_model(start_dict, startStates)
        return #trans_dict, emit_dict, start_dict


trainData = './data/data-hmm/data/train.txt'     #语料放在当前目录的data目录下
transMat = './data/data-hmm/model/prob_trans.model'     #统计模型存储在当前目录的model目录下
emitMat = './data/data-hmm/model/prob_emit.model'
startStates = './data/data-hmm/model/prob_start.model'
trainer = HmmTrain()
trainer.train(trainData, transMat, emitMat, startStates)
result = trainer.get_word_status(trainData)
print(result)

第二部分 模型测试

#!/usr/bin/env python3
# coding: utf-8
# File: hmm_cut.py

class HmmCut:
    def __init__(self):
        trans_path = './data/data-hmm/model/prob_trans.model'
        emit_path = './data/data-hmm/model/prob_emit.model'
        start_path = './data/data-hmm/model/prob_start.model'
        self.prob_trans = self.load_model(trans_path)
        self.prob_emit = self.load_model(emit_path)
        self.prob_start = self.load_model(start_path)

    '''加载模型'''
    def load_model(self, model_path):
        f = open(model_path, 'r')
        a = f.read()
        word_dict = eval(a)
        f.close()
        return word_dict

    '''viterbi算法求解'''
    def viterbi(self, obs, states, start_p, trans_p, emit_p):  # 维特比算法（一种递归算法）
        # 算法的局限在于训练语料要足够大，需要给每个词一个发射概率,.get(obs[0], 0)的用法是如果dict中不存在这个key,则返回0值
        V = [{}]
        path = {}
        for y in states:
            V[0][y] = start_p[y] * emit_p[y].get(obs[0], 0)  # 在位置0，以y状态为末尾的状态序列的最大概率
            path[y] = [y]

        for t in range(1, len(obs)):
            V.append({})
            newpath = {}
            for y in states:
                state_path = ([(V[t - 1][y0] * trans_p[y0].get(y, 0) * emit_p[y].get(obs[t], 0), y0) for y0 in states if V[t - 1][y0] > 0])
                if state_path == []:
                    (prob, state) = (0.0, 'S')
                else:
                    (prob, state) = max(state_path)
                V[t][y] = prob
                newpath[y] = path[state] + [y]

            path = newpath  # 记录状态序列
        (prob, state) = max([(V[len(obs) - 1][y], y) for y in states])  # 在最后一个位置，以y状态为末尾的状态序列的最大概率
        return (prob, path[state])  # 返回概率和状态序列

    # 分词主控函数
    def cut(self, sent):
        prob, pos_list = self.viterbi(sent, ('B', 'M', 'E', 'S'), self.prob_start, self.prob_trans, self.prob_emit)
        seglist = list()
        word = list()
        for index in range(len(pos_list)):
            if pos_list[index] == 'S':
                word.append(sent[index])
                seglist.append(word)
                word = []
            elif pos_list[index] in ['B', 'M']:
                word.append(sent[index])
            elif pos_list[index] == 'E':
                word.append(sent[index])
                seglist.append(word)
                word = []
        seglist = [''.join(tmp) for tmp in seglist]

        return seglist

    #测试
    def test(self):
        sent = '华中农业大学信息学院计算机科学系'
        sent = '目前在自然语言处理技术中，中文处理技术比西文处理技术要落后很大一段距离，许多西文的处理方法中文不能直接采用，就是因为中文必需有分词这道工序。中文分词是其他中文信息处理的基础，搜索引擎只是中文分词的一个应用。'''
        cuter = HmmCut()
        seglist = cuter.cut(sent)
        print(seglist)

第三部分 输出结果：请你建立在读懂以上代码的基础上，调用相关函数输出分词结果（20分）

sent = '独自远离家乡难免总有一点凄凉，每到重阳佳节倍加思念远方的亲人。远远想到兄弟们身佩茱萸登上高处，也会因为少我一人而生遗憾之情。'
model = HmmCut()
seglist = model.cut(sent)
print(seglist)

利用HMM完成NER任务（80分）

HMM模型实现NER任务的相关代码，请你根据要求在相关位置填入代码

import json
import time

import numpy as np
from tqdm import tqdm
from collections import defaultdict
from sklearn import metrics
from itertools import chain

1. 读取json数据 （15分）

请你将data-ner文件夹内的train.json文件按行输入json_data[ ]变量中，并输出json_data[0]

输出样例：

{'text': '浙商银行企业信贷部叶老桂博士则从另一个角度对五道门槛进行了解读。叶老桂认为，对目前国内商业银行而言，',

'label': {'name': {'叶老桂': [[9, 11]]}, 'company': {'浙商银行': [[0, 3]]}}}

path = './data/data-ner/train.json'
with open(path,'r',encoding='utf-8') as f:
    json_data = []
    for line in f:
        json_data.append(json.loads(line))
print(json_data[0])

2.统计实体标签数量（15分）

请你统计json_data内的实体类别数量于n_classes[ ]之中，并输出n_classes

输出样例：['name', 'company', 'game', 'organization', 'movie','address', 'position', 'government', 'scene', 'book']

n_classes = []
for i in range(len(json_data)):
    label = ['O'] * len(json_data[i]['text'])
    for n in json_data[i]['label']:
        if n not in n_classes:
            n_classes.append(n)
print(n_classes)

3.设计tag2idx字典

# 设计tag2idx字典，对每个标签设计两种，如B-name、I-name，并设置其ID值
tag2idx = defaultdict()
tag2idx['O'] = 0
count = 1
for n_class in n_classes:
    tag2idx['B-' + n_class] = count
    count += 1
    tag2idx['I-' + n_class] = count
    count += 1
print(tag2idx)

4.编写数据处理函数（15分）

请你参照NER-crf文件编写数据处理函数data_process( ) 

def data_process(path):
    data = []

    # 读取每一条 JSON 数据
    with open(path, 'r', encoding='utf-8') as fp:
        for line in fp:
            json_data = json.loads(line)
            
            text = json_data['text']
            labels = json_data['label']

            # 初始化标签为'O'
            label = ['O'] * len(text)

            # 处理每个标签类别
            for label_class in labels:
                for entity in labels[label_class]:
                    for start, end in labels[label_class][entity]:
                        label[start] = 'B-' + label_class
                        label[start + 1:end + 1] = ['I-' + label_class] * (end - start)

            # 将文本分解为字符
            text_chars = list(text)

            data.append([text_chars, label])

    return data

# 示例用法
path = './data/data-ner/train.json'
processed_data = data_process(path)
print(processed_data[0])

 5.编写维特比算法函数（20分）

请你参照Q1中维特比算法的相关思想改写代码，适用于本任务中（可以按需求调整其他代码、或以简述代码思想流程方式呈现） 

class HMM_model:
    def __init__(self, tag2idx):
        self.tag2idx = tag2idx  # tag2idx字典
        self.n_tag = len(self.tag2idx)  # 标签个数
        self.n_char = 65535  # 所有字符的Unicode编码个数，包括汉字
        self.epsilon = 1e-100  # 无穷小量，防止归一化时分母为0
        self.idx2tag = dict(zip(self.tag2idx.values(), self.tag2idx.keys()))  # idx2tag字典
        self.A = np.zeros((self.n_tag, self.n_tag))  # 状态转移概率矩阵, shape:(21, 21)
        self.B = np.zeros((self.n_tag, self.n_char))  # 观测概率矩阵, shape:(21, 65535)
        self.pi = np.zeros(self.n_tag)  # 初始隐状态概率,shape：(21,)

    def train(self, train_data):
        print('开始训练数据：')
        for i in tqdm(range(len(train_data))):  # 几组数据
            for j in range(len(train_data[i][0])):  # 每组数据中几个字符
                cur_char = train_data[i][0][j]  # 取出当前字符
                cur_tag = train_data[i][1][j]  # 取出当前标签
                self.B[self.tag2idx[cur_tag]][ord(cur_char)] += 1  # 对B矩阵中标签->字符的位置加一
                if j == 0:
                    # 若是文本段的第一个字符，统计pi矩阵
                    self.pi[self.tag2idx[cur_tag]] += 1
                    continue
                pre_tag = train_data[i][1][j - 1]  # 记录前一个字符的标签
                self.A[self.tag2idx[pre_tag]][self.tag2idx[cur_tag]] += 1  # 对A矩阵中前一个标签->当前标签的位置加一

        # 防止数据下溢,对数据进行对数归一化
        self.A[self.A == 0] = self.epsilon
        self.A = np.log(self.A) - np.log(np.sum(self.A, axis=1, keepdims=True))
        self.B[self.B == 0] = self.epsilon
        self.B = np.log(self.B) - np.log(np.sum(self.B, axis=1, keepdims=True))
        self.pi[self.pi == 0] = self.epsilon
        self.pi = np.log(self.pi) - np.log(np.sum(self.pi))

        # 将A，B，pi矩阵保存到本地
        np.savetxt('./A.txt', self.A)
        np.savetxt('./B.txt', self.B)
        np.savetxt('./pi.txt', self.pi)
        print('训练完毕！')

    # 载入A，B，pi矩阵参数
    def load_paramters(self, A='./A.txt', B='./B.txt', pi='./pi.txt'):
        self.A = np.loadtxt(A)
        self.B = np.loadtxt(B)
        self.pi = np.loadtxt(pi)

   # 修改viterbi方法以使用Viterbi算法解码观测序列
    def viterbi(self, obs):
        # 初始化Viterbi表格
        T = len(obs)
        N = self.n_tag  # 标签个数
        delta = np.zeros((T, N))
        psi = np.zeros((T, N), dtype=int)

        # 初始化时刻 t=0
        for i in range(N):
            delta[0][i] = self.pi[i] + self.B[i][ord(obs[0])]

        # 递推计算
        for t in range(1, T):
            for i in range(N):
                max_delta = float('-inf')
                max_psi = -1
                for j in range(N):
                    score = delta[t - 1][j] + self.A[j][i] + self.B[i][ord(obs[t])]
                    if score > max_delta:
                        max_delta = score
                        max_psi = j
                delta[t][i] = max_delta
                psi[t][i] = max_psi

        # 终止
        max_state = np.argmax(delta[T - 1])
        best_path = [max_state]
        for t in range(T - 1, 0, -1):
            max_state = psi[t][max_state]
            best_path.append(max_state)
        best_path.reverse()

        # 将最佳路径转换为标签序列
        best_tags = [self.idx2tag[i] for i in best_path]

        return best_tags

        
    def predict(self, s):
        results = self.viterbi(s)
        
        for i in range(len(results)):
            print(s[i] )
            print(results[i])

    def valid(self, valid_data):
        y_pred = []
        # 遍历验证集每一条数据，使用维特比算法得到预测序列，并加到列表中
        for i in range(len(valid_data)):
            y_pred.append(self.viterbi(valid_data[i][0]))
        return y_pred

6.读入数据（15分）

请你参照NER-crf文件读入训练与验证数据，并保存在train_data与valid_data之中，并输出训练集与验证集的长度

输出样例：训练集长度：10748

验证集长度：1343

train_path = './data/data-ner/train.json'
valid_path = './data/data-ner/dev.json'
train_data = data_process(train_path)
valid_data = data_process(valid_path)
print('训练集长度:{},验证集长度:{}'.format(len(train_data),len(valid_data)))

7.建立模型

model = HMM_model(tag2idx)
model.train(train_data)
model.load_paramters()
model.predict('浙商银行企业信贷部叶老桂博士则从另一个角度对五道门槛进行了解读。叶老桂认为，对目前国内商业银行而言')
print()
y_pred = model.valid(valid_data)
y_true = [data[1] for data in valid_data]

# 排好标签顺序输入，否则默认按标签出现顺序进行排列
sort_labels = [k for k in tag2idx.keys()]

y_true = list(chain.from_iterable(y_true))
y_pred = list(chain.from_iterable(y_pred))

8.输出结果

# 打印详细分数报告，包括precision(精确率)，recall(召回率)，f1-score(f1分数)，support(个数)，digits=3代表保留3位小数
print(metrics.classification_report(
    y_true, y_pred, labels=sort_labels[1:]
))

NER-crf

# 导入包
import json
# 服务器中未安装sklearn_crfsuite库，此代码仅作参考使用，不可运行
# !pip install sklearn_crfsuite 安装库
import sklearn_crfsuite
from sklearn import metrics
from itertools import chain

import zipfile

# 指定要解压的ZIP文件名
zip_file = 'data.zip'

# 指定解压后的目标文件夹
target_folder = 'data'

# 打开ZIP文件
with zipfile.ZipFile(zip_file, 'r') as zip_ref:
    # 解压所有文件到目标文件夹
    zip_ref.extractall(target_folder)

print(f'{zip_file} 已成功解压到 {target_folder}')

# 创建函数将原始json数据转化成BIO三维标注形式

def data_process(path):
    # 读取每一条json数据放入列表中
    # 由于该json文件含多个数据，不能直接json.loads读取，需使用for循环逐条读取
    json_data = []
    with open(path, 'r', encoding='utf-8') as fp:
        for line in fp:
            json_data.append(json.loads(line))

    # json_data中每一条数据的格式为
    '''
    {'text': '浙商银行企业信贷部叶老桂博士则从另一个角度对五道门槛进行了解读。叶老桂认为，对目前国内商业银行而言，',
     'label': {'name': {'叶老桂': [[9, 11]]}, 'company': {'浙商银行': [[0, 3]]}}}
     '''

    # 将json文件处理成如下格式
    '''
    [['浙', '商', '银', '行', '企', '业', '信', '贷', '部', '叶', '老', '桂', '博', '士', '则', '从', '另', '一', 
    '个', '角', '度', '对', '五', '道', '门', '槛', '进', '行', '了', '解', '读', '。', '叶', '老', '桂', '认', 
    '为', '，', '对', '目', '前', '国', '内', '商', '业', '银', '行', '而', '言', '，'], 
    ['B-company', 'I-company', 'I-company', 'I-company', 'O', 'O', 'O', 'O', 'O', 'B-name', 'I-name', 
    'I-name', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 
    'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O', 'O']]
    '''
    data = []
    # 遍历json_data中每组数据
    for i in range(len(json_data)):
        # 将标签全初始化为'O'
        label = ['O'] * len(json_data[i]['text'])
        # 遍历'label'中几组实体，如样例中'name'和'company'
        for n in json_data[i]['label']:
            # 遍历实体中几组文本，如样例中'name'下的'叶老桂'（有多组文本的情况，样例中只有一组）
            for key in json_data[i]['label'][n]:
                # 遍历文本中几组下标，如样例中[[9, 11]]（有时某个文本在该段中出现两次，则会有两组下标）
                for n_list in range(len(json_data[i]['label'][n][key])):
                    # 记录实体开始下标和结尾下标
                    start = json_data[i]['label'][n][key][n_list][0]
                    end = json_data[i]['label'][n][key][n_list][1]
                    # 将开始下标标签设为'B-' + n，如'B-' + 'name'即'B-name'
                    # 其余下标标签设为'I-' + n
                    label[start] = 'B-' + n
                    label[start + 1: end + 1] = ['I-' + n] * (end - start)

        # 对字符串进行字符级分割
        # 英文文本如'bag'分割成'b'，'a'，'g'三位字符，数字文本如'125'分割成'1'，'2'，'5'三位字符
        texts = []
        for t in json_data[i]['text']:
            texts.append(t)

        # 将文本和标签编成一个列表添加到返回数据中
        data.append([texts, label])
    return data

# 判断字符是否是英文
def is_english(c):
    if ord(c.lower()) >= 97 and ord(c.lower()) <= 122:
        return True
    else:
        return False

# 将文本转换为特征字典
# sklearn-crfsuite的数据输入格式采用字典格式，类似于做特征工程，CRF将这些特征映射成发射概率
def word2features(sent, i):
    # 本代码采用大小为3的滑动窗口构造特征，特征有当前字符、字符是否为数字或英文等，当然可以增大窗口或增加其他特征
    word = sent[i][0]
    features = {
        'bias': 1.0,
        'word': word,
        'word.isdigit()': word.isdigit(),
        'word.is_english()': is_english(word),
    }
    # 该字的前一个字
    if i > 0:
        word = sent[i - 1][0]
        features.update({
            '-1:word': word,
            '-1:word.isdigit()': word.isdigit(),
            '-1:word.is_english()': is_english(word),
        })
    else:
        # 若该字符为序列开头，则增加特征 BOS(begin of sentence)
        features['BOS'] = True
    # 该字的后一个字
    if i < len(sent) - 1:
        word = sent[i + 1][0]
        features.update({
            '+1:word': word,
            '+1:word.isdigit()': word.isdigit(),
            '+1:word.is_english()': is_english(word),
        })
    else:
        # 若该字符为序列结尾，则增加特征 EOS(end of sentence)
        features['EOS'] = True
    return features

# 建立分别将文本与标签转化为对应特征的函数
def sent2features(sent):
    return [word2features(sent, i) for i in range(len(sent))]

def sent2labels(sent):
    return [label for label in sent]
# 读入数据
train = data_process('./data/data-ner/train.json')
valid = data_process('./data/data-ner/dev.json')
print('训练集长度:', len(train))
print('验证集长度:', len(valid))

# 调用函数得到文本与标签特征
X_train = [sent2features(s[0]) for s in train]
y_train = [sent2labels(s[1]) for s in train]
X_dev = [sent2features(s[0]) for s in valid]
y_dev = [sent2labels(s[1]) for s in valid]

# 输出特征样例
'''
序列中的每一个字符处理成如下格式：
{'bias': 1.0,
'word': '商',
'word.isdigit()': False,         # 是否为数字
'word.is_english()': False,      # 是否为英文
'-1:word': '浙',                 # 前一个字 
'-1:word.isdigit()': False,
'-1:word.is_english()': False,
'+1:word': '银',                 # 后一个字
'+1:word.isdigit()': False,
'+1:word.is_english()': False}
'''
print(X_train[0][1])

# algorithm：lbfgs法求解该最优化问题，c1：L1正则系数，c2：L2正则系数，max_iterations：迭代次数，verbose：是否显示训练信息
crf_model = sklearn_crfsuite.CRF(algorithm='lbfgs', c1=0.1, c2=0.1, max_iterations=50,
                                 all_possible_transitions=True, verbose=True)

# 使用异常处理，不会影响训练效果
try:
    crf_model.fit(X_train, y_train)
except:
    pass

# 由于大部分标签都是'O'，故不去关注'O'标签的预测
labels = list(crf_model.classes_)
labels.remove("O")
y_pred = crf_model.predict(X_dev)

y_dev = list(chain.from_iterable(y_dev))
y_pred = list(chain.from_iterable(y_pred))

# 计算F1分数，average可选'micro'，'macro'，'weighted'，处理多类别F1分数的不同计算方法
print('weighted F1 score:', metrics.f1_score(y_dev, y_pred, average='weighted', labels=labels))

# 排好标签顺序输入，否则默认按标签出现顺序进行排列
sorted_labels = sorted(labels, key=lambda name: (name[1:], name[0]))
# 打印详细分数报告，包括precision(精确率)，recall(召回率)，f1-score(f1分数)，support(个数)，digits=3代表保留3位小数
print(metrics.classification_report(y_dev, y_pred, labels=sorted_labels, digits=3))