使用BiLSTM神经网络+PyTorch实现汉语分词模型的训练

本次实验源码及数据集已上传到Github，有需要自行下载。

第一部分：实验分析与设计

一、实验内容描述

此次实验主要是为了深入比较和评估不同中文分词方法的性能，以便于更全面地理解它们的优点和局限性。在此次实验中我将使用两种主要方法来实现中文分词：一种是基于词典的正向匹配算法，另一种是基于神经网络的双层双向长短时记忆网络（LSTM）模型。

方法一：基于词典的正向匹配算法

这种方法比较简单，在这种方法中，我们将利用一个包含大部分常用中文词汇的词典。然后，使用正向匹配算法，将待分词的文本与词典中的词汇逐一匹配。匹配成功的部分将被输出作为分词结果。这种方法的优势在于其简单性和速度，但它可能无法处理未知词汇或歧义情况。相较于神经网络非常容易实现，也不需要麻烦的数据预处理，还不需要修改数据格式，主要算法就是字符串匹配。

方法二：基于神经网络的双层双向LSTM模型

在这个方法中，我们将使用pyTorch构建一个神经网络来实现中文词语分词算法。首先，我们将准备一个中规模的中文语料文件，作为训练数据集。我们将使用PyTorch框架构建一个双层双向LSTM模型，该模型能够学习如何分词。在训练过程中，模型将学习词汇和上下文之间的关系，以便更准确地分词。

数据集

为了评估两种方法的性能，我们将使用以下数据集：

语料文件：一个包含大量中文文本的语料文件，用于神经网络的训练。该语料文件将包括各种文本类型和难度级别的文本。其中使用空格分开每一个词语，如下其中一句所示：

1	1. 迈向充满希望的新世纪 —— 一九九八年新年讲话（附图片１张）

测试数据：一个用于评估两种分词方法的测试数据集，包括中文文本。这些测试数据将用于比较方法的性能。其中数据为一句句话，但是没有使用空格进行分割。

测试数据结果：测试数据对应的结果，每一行都和测试数据相对应，不过每个词语之间有空格间隔。测试数据和测试数据结果如下：

1 2	1. 测试数据：共同创造美好的新世纪——二○○一年新年贺词 2. 测试数据结果：共同创造美好的新世纪 —— 二○○一年新年贺词

数据预处理：准备词典、对语料文件进行分词和标记化，以及创建神经网络模型的输入数据。

模型训练：使用语料文件进行神经网络模型的训练。模型将学习如何分词。

模型评估：使用测试数据集来评估两种分词方法的性能，包括准确率、召回率、F1分数等指标。

结果分析：比较基于词典的正向匹配算法和基于神经网络的方法的性能，讨论它们的优势和不足之处。

实验验证：重复实验，以确保结果的稳定性和一致性。

通过这个实验，我们希望得出关于不同中文分词方法的性能比较，并为选择最适合特定任务的分词方法提供有力依据。

二、实验内容与设计思想

在第一种方法中，首先，我们使用简单的正向匹配算法，对于字符串中的每个字符串进行匹配。从文本的开头开始，每次匹配最长的词语，直到文本被分完为止。如果当前位置没有匹配到任何词语，则将当前位置的字符作为一个单独的词语。

具体来说，该算法的实现过程如下：

初始化一个空的词语列表tokens，以及文本的长度text_len和词典中最长词语的长度max_word_len。
从文本的开头开始，依次匹配最长的词语。具体地，从max_word_len开始，依次尝试匹配长度为max_word_len、max_word_len-1、…、1的词语，直到匹配到一个词语为止。
如果匹配到了一个词语，则将该词语添加到tokens列表中，并将当前位置i移动到词语的末尾。
如果没有匹配到任何词语，则将当前位置的字符作为一个单独的词语，并将当前位置i移动到下一个位置。
重复步骤2到步骤4，直到文本被分完为止。

该算法的时间复杂度为O(n^2)，其中n为文本的长度。在实际应用中，该算法的效率较低，但是实现简单，可以作为其他分词算法的基础。

第二种方法主要需要使用pytorch，所以比较麻烦，首先我们需要对于所有句子进行预处理，由于模型无法直接输入文字，所以我们得将其进行编码，编码这里我选择的是每个字出现的频率，按照从小到大排序进行编码，这样一方面可以实现我们的编码功能，另一方面还有一部分字频特征，比较直接。通过这种方式将其转换为数字列表后，再将结果进行处理，通过上网查阅资料可知，在这种模型我们的结果需要使用编码进行标识是否是一个词语或者单独的字。编码规则如下：

1. target_dict = {'B': 0, # begin
2.                  'M': 1, # middle
3.                  'E': 2, # end
4.                  'S': 3, # single
5.                  'N': 4} # null

编码成功后，我们的每个句子转换为了数字类型，除此之外，如果有需要我们还需要将其转换为one-Hot编码，不过在torch模型中不是很需要，所以这里我们就不转换了。

数据预处理后，我们就需要创建模型了，为了模型更加贴合我们的实验，我选择了双层双向LSTM。双向LSTM（Bidirectional LSTM）是一种常用的循环神经网络模型，它可以同时考虑输入序列的前向和后向信息，从而更好地捕捉序列中的上下文信息。在分词模型中，双向LSTM可以很好地处理中文分词中的歧义问题，提高分词的准确性。

具体来说，双向LSTM可以将输入序列分别从前向后和从后向前进行处理，得到两个输出序列。这两个输出序列可以分别表示输入序列中当前位置之前和之后的上下文信息。然后，这两个输出序列可以被合并起来，得到一个综合的上下文表示，用于进行下一步的分类或预测。

在分词模型中，双向LSTM可以很好地处理中文分词中的歧义问题。例如，在中文分词中，一个汉字可能既可以作为一个词语的开始，也可以作为另一个词语的中间部分。这种歧义问题可以通过双向LSTM来解决，因为双向LSTM可以同时考虑当前位置之前和之后的上下文信息，从而更好地判断当前位置的标记。

然后，我选择了交叉熵作为损失函数，并且定义了优化器，然后将数据上传到GPU上进行运算，现在整个项目就比较清晰了。

三、实验使用环境

系统：Windows 11专业版 22H2 22621.2361
软件：Visual Studio Code
环境：Anaconda3(base):Python 3.9.13
PyTorch 2.0.1+cu118
硬件：LEGION R9000P 2021H
CPU:R7-5800H
GPU:RTX 3060 Laptop

第二部分：实验调试与结果分析

一、调试过程

首先我们实现第一种正向最大匹配分词实现汉字分词，我们的词典中有若干词汇，并且按照首字母顺序进行排序，如下图所示：

dirctory.txt

我们所需要实现的功能就是，遍历一串句子的每一个字，使用这个字及后面的若干字组成临时词汇，在列表中搜索对应的词汇，如果有这个词汇，分词，并进行下一句话的分词，如果没有，单独成词，并继续遍历下一个词汇。实现代码如下：

# 定义正向最大匹配分词函数
def forward_max_match(text):
    tokens = []
    text_len = len(text)
    max_word_len = max(len(word) for word in dictionary_data)

    i = 0
    while i < text_len:
        matched = False
        for j in range(max_word_len, 0, -1):
            if i + j <= text_len:
                word = text[i:i + j]
                if word in dictionary_data:
                    tokens.append(word)
                    i += j
                    matched = True
                    break
        if not matched:
            tokens.append(text[i])
            i += 1

    return tokens

其中的directory_data就是我们的字典文件，经过split函数，将其每个空格隔开的每个单词分别放入到列表中便于查询。我们测试一条句子，得到以下结果：

正向匹配结果

我们可以看见，这个准确率还不错，除了“—”和“二○○一年”没有出现在字典中，其他的分的效果挺好的。然后计算一下准确率，F1值，召回率参数，这一步是个比较难的点，由于是文字的原因，我们无法很好的对比他是否分割完毕或者在这里分割。为了统一标准，我们选择了，对比词汇是否正确划分出来，并且使用正确划分的词汇数量进行评判。我们写了如下代码循环进行测试：


# 准确率,召回率, F1值
自己分的词语总数 = 0
正确结果的词语总数 = 0
正确的词语总数 = 0

data = test_data[:100]
# print('data[:10]:', list(data)[:5])

# 循环处理每行文本
for i in range(len(data)):

    # 获取测试数据和测试结果
    test_data_line = data[i]
    result_data_line = test_result[i]

    # 使用正向最大匹配算法分词
    segmented_text = forward_max_match(test_data_line)

    # 计算匹配的分词数量
    自己分的词语总数 += len(segmented_text)
    正确结果的词语总数 += len(result_data_line)

    j = -1
    # 计算正确的分词数量
    for seg_word in segmented_text:
        j += 1
        if seg_word in result_data_line:
            # 如果分词结果在正确结果中,则，查找所有匹配的位置并append到locations中
            locations = find_same_word_locations(result_data_line, seg_word)
            # 遍历locations中的每个位置，如果分词结果在正确结果中，则正确的词语总数加1
            for loc in locations:
                # 计算对比两个在字符串中的位置是否相同
                if get_word_location(segmented_text, j) == get_word_location(result_data_line, loc):
                    # 如果相同，直接加1
                    正确的词语总数 += 1
    
    # 查看分词计算进度
    print(f"分词计算进度：{i + 1}/{len(data)}\t准确率：{正确的词语总数 / 自己分的词语总数}\t", end='\r')


# 计算准确率
准确率 = 正确的词语总数 / 自己分的词语总数
召回率 = 正确的词语总数 / 正确结果的词语总数
F1值 = 2 * 准确率 * 召回率 / (准确率 + 召回率)

print(f"\n正确分词数量：{正确的词语总数}")
print(f"总分词数量：{自己分的词语总数}")
print(f"正确结果的词语总数：{正确结果的词语总数}")

print(f"准确率：{准确率:.2%}")
print(f"召回率：{召回率:.2%}")
print(f"F1值：{F1值:.2%}")

最终我们得到相关数据：

正确分词数量：4983
总分词数量：5655
正确结果的词语总数：5414
准确率：88.12%
召回率：92.04%
F1值：90.04%

第二种方法我们选择使用PyTorch实现神经网络进行分词，我们首先定义一下思路我们想要实现功能，那么必须要向神经网络中传输数据，我们的数据应该是什么形状的呢？由于是句子，首先句子的长度可能会有较大变化，其次是维度，我们每个句子一行行堆砌为列表，最终的结果应该也会很大。数量多没问题，但是如果数据变长度，我们能很好地是西安功能吗？首先我写了一个BP神经网络简单实现，发现最终效果很差，所以，为了模型更好的拟合，我选择最大长度为32，大于32的句子删掉，小于32的句子末尾补零，当然我们按照所有的标点符号进行划分，确保大部分数据都能派上用场。理论可行，实践开始：

获取训练数据：我们的语料文件是每个词语之间用空格分隔开，但是我们输入数据中不应该有这些空格，这些空格应该是以标签的形式输入，所以我们首先需要将空格去掉，然后按照标点符号进行断句，经过统计，无标点符号的句子长度几乎所有都不会超过32，这样就不会浪费数据了。处理代码如下：

def build_train_data(file_path):   #file_path = 'data/train.txt'
    with open(file_path,'r',encoding='utf-8') as f:
        lines = f.read()
    
    # 将读取到的文本按照标点符号和换行符进行切分，得到一个列表
    lines = re.split(r"[，。！？、（）—《》…；“”\n]",lines)

    # 去掉列表中的空字符串及空格
    lines = [line.strip() for line in lines if line.strip()]

    # 去掉每个句子中的空格
    lines = [line.replace(" ","") for line in lines]
    
    # 去掉所有长度大于max_len的句子
    lines = [line for line in lines if len(line)<=max_len]

    phrase_expel = lines

    print("phrase_expel:",phrase_expel[:10])
    print("phrase_expel len:",len(phrase_expel))

    with open('data/generate_pkl/train_data.pkl', 'wb') as f: #把这个处理后的文件当作训练数据
        pkl.dump(phrase_expel, f)   #把文件写成pkl格式

其中处理句子长度并在末尾添加零的代码写到后面，因为后面还会使用到这个数据。

下面就是我们的标签数据了，我们需要想办法将其中不同的分词保存成一种数据，经过上网查询，我选择了使用对应字母来表示每个字的位置，通过转换，我们获取到其中几条数据的输出：

训练数据分句结果

模型无法识别汉字，所以我们需要给汉字编码，这里我选择使用字频字典进行编码。统计字频后保存字频文件以便于后面测试时调用，另外我们需要空缺出0的编码，以进行后续32个字符的补齐。

def build_vocab_dict(file_path):  #'data/train_data.pkl'
    vocab_dic = {}
    with open(file_path, 'rb') as f:
        z = pkl.load(f)
        for line in z:
            for hang in line:  #统计词频，按照词多到少排列
                vocab_dic[hang] = vocab_dic.get(hang, 0) + 1
        vocab_dic_sorted = sorted(vocab_dic.items(), key=lambda x: x[1], reverse=True)

    # 按照词频排序后，构建词典，词频越高，索引越小，并且下标从1开始
    vocab_dic2 = {}
    for i, j in enumerate(vocab_dic_sorted):
        vocab_dic2[j[0]] = i + 1

    # 展示前10个词
    print("vocab_dic2:",list(vocab_dic2.items())[:10])
    print("vocab_dic2 len:",len(vocab_dic2))

    with open('data/generate_pkl/vocab.pkl', 'wb') as f:
        pkl.dump(vocab_dic2, f)

下面我们只需要给训练数据，也就是那一串串字符搞成数字，再在末尾补0，即可达到要求。

下面我们开始定义模型类和相关参数，考虑到是一个分词模型，我们选择双向LSTM实现，为了效果达到最好，我尝试选择一个双层双向LSTM进行训练。模型定义如下：

class BiLSTM_Model(nn.Module):
    def __init__(self, vocab_size, embedding_size, hidden_size):
        super(BiLSTM_Model, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_size)
        self.bilstm1 = nn.LSTM(embedding_size, hidden_size, bidirectional=True, batch_first=True)
        self.dropout1 = nn.Dropout(0.6)
        self.bilstm2 = nn.LSTM(hidden_size * 2, hidden_size, bidirectional=True, batch_first=True)
        self.dropout2 = nn.Dropout(0.6)
        self.fc = nn.Linear(hidden_size * 2, 5)
        
    def forward(self, x):
        x = self.embedding(x)
        x, _ = self.bilstm1(x)
        x = self.dropout1(x)
        x, _ = self.bilstm2(x)
        x = self.dropout2(x)
        x = self.fc(x)
        return x

这个模型中，我们可以看到，我们添加了两个双向LSTM，并在其中穿插Dropout层防止过拟合，并且在最终添加了一个fc全连接层，将最终结果转化为分类结果。然后我们只需要定义模型并训练，即可得到我们的结果。

注意训练时我们需要选择GPU进行计算，先定义模型和模型所使用的损失函数优化器，然后将模型和数据送到GPU即可：

model = BiLSTM_Model(voc_size + 1, config.embed_dim, config.hidden_dim)
print(model)

# 定义损失函数和优化器
loss_function = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=config.learning_rate)

通过这部分运行代码的结果我们可以看见我们的模型大致内容。

模型参数

然后将数据上传到GPU上，注意在上传时我们应该选择自己的设备，如果设备不支持GPU，就可以直接进行下一步了，如果支持进行这一步骤并选择设备id以提高速度。

# 将数据和模型上传到GPU进行计算
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
print(device)
model.to(device)
train_data_list = [i.to(device) for i in train_data_list]
target_list = [i.to(device) for i in target_list]

下面我们就可以进行训练过程了。训练过程我们每个epoch分开进行训练，训练结束后输出损失再进行下一步

# 开始训练
for epoch in range(config.epoch):
    print("Epoch:", epoch + 1)
    for i in tqdm(range(len(train_data_list))):
        model.zero_grad()
        input_data = train_data_list[i].view(1, -1)
        target = target_list[i].view(-1)
        output = model(input_data)
        loss = loss_function(output.view(-1, 5), target)
        loss.backward()
        optimizer.step()
    print("Loss:", loss.item())
    # # 打印这个epoch的准确率
    # with torch.no_grad():
    #     right_num = 0
    #     all_num = 0
    #     for i in tqdm(range(len(train_data_list))):
    #         input_data = train_data_list[i].view(1, -1)
    #         target = target_list[i].view(-1)
    #         output = model(input_data)
    #         output = torch.argmax(output.view(-1, 5), dim=1)
    #         for j in range(len(output)):
    #             if output[j] == target[j]:
    #                 right_num += 1
    #             all_num += 1
    #     print("Accuracy:", right_num / all_num)

这里我们可以看见很多的系数，如config.Learning_rate，config.epoch等等，这里我们是定义了一个参数类以保存各种参数，如下：

class Config(object):
    # 参数设置类，包含一些相关参数
    def __init__(self):
        self.vocab = pkl.load(open('data/generate_pkl/vocab.pkl', 'rb'))  # 读取词表
        self.train_data = pkl.load(open('data/generate_pkl/train_data.pkl', 'rb'))  # 读取训练数据
        self.target = pkl.load(open('data/generate_pkl/target.pkl', 'rb'))  # 读取标签

        self.learning_rate = 0.0015  # 学习率
        self.epoch = 4  # epoch次数
        self.dropout = 0.6 # dropout
        self.max_len = 32

        self.output_size = 4
        self.embed_dim = 128
        self.hidden_dim = 64
        self.hout1 = 32
        self.hout2 = 64

        self.num_layers = 2 # 测试双层LSTM神经网络

        print("---------创建参数类完成---------")
        print("# 词表大小：", len(self.vocab))
        print("# 训练数据大小：", len(self.train_data))
        print("# 标签大小：", len(self.target))
        print("# 输出大小：", self.output_size)
        print("# 嵌入层大小：", self.embed_dim)
        print("# 隐藏层大小：", self.hidden_dim)
        print("# 第一层输出大小：", self.hout1)
        print("# 第二层输出大小：", self.hout2)
        print("# 学习率：", self.learning_rate)
        print("# epoch次数：", self.epoch)
        print("# dropout层：", self.dropout)
        print("# 每个LSTM中循环次数：", self.num_layers)
        print("-------------------------------")

最终输出结果如下：

训练结果

现在开始系统的测定我们的准确率，这里的召回率，我选择使用字母的对应进行测试，比如这个字符在原标签中是一个E，如果变成了其他字符，则表示错误，反之正确。当然我们的测试数据需要通过原来保存的字典转换成数据。

下面进行测试：

# 将测试数据和测试结果转换为tensor
test_data_tensor = torch.tensor(test_data_id)
test_result_tensor = torch.tensor(test_result_id)

print(test_data_tensor.shape)
print(test_result_tensor.shape)

# 输入模型进行预测
output = model(test_data_tensor)
print(output.shape)

# 将预测结果转换为numpy数组
output = output.detach().numpy()

# 将预测结果转换为标签
output = np.argmax(output, axis=2)
# print(output[0])

# 将预测结果转换为字母
output = [[list(target_dict.keys())[list(target_dict.values()).index(word)] for word in line] for line in output]
# print(output[0])

# 处理output
output = [''.join(line) for line in output]

# 去除output和test_result中的"N"
output = [line.replace("N", "") for line in output]
test_result = [line.replace("N", "") for line in test_result]

# 遍历二者中的每一行的每一个字母，计算预测结果和真实结果的准确率
acc = 0
for i in range(len(output)):
    for j in range(len(output[i])):
        if output[i][j] == test_result[i][j]:
            acc += 1
print("准确率：", acc/sum([len(line) for line in test_result]))
print("召回率：", acc/sum([len(line) for line in output]))
print("F1值：", 2*acc/(sum([len(line) for line in test_result])+sum([len(line) for line in output])))