Day09 【基于LSTM实现文本加标点的任务】

目标

本文基于给定的词表，将输入的文本基于jieba分词分割为若干个词，然后基于词表将词初步序列化，之后经过embedding``LSTM等网络结构层，输出在已知类别标点符号标签上的概率分布，从而实现一个简单文本加标点任务。

数据准备

词表文件chars.txt

类别标签文件schema.json

{"": 0,"，": 1,"。": 2,"？": 3
}

本文的任务只处理三类标点符号，逗号"，"、句号"。"和问号"？"，不需要加标点词的后面相当于加上空字符""，因此文本标签总共有4类。

训练集数据train_corpus.txt训练集数据

验证集数据valid_corpus.txt验证集数据

参数配置

config.py

# -*- coding: utf-8 -*-"""
配置参数信息
"""Config = {"model_path": "model_output","schema_path": "data/schema.json","train_data_path": "data/train_corpus.txt","valid_data_path": "data/valid_corpus.txt","vocab_path":"chars.txt","max_length": 50,"hidden_size": 128,"epoch": 10,"batch_size": 128,"optimizer": "adam","learning_rate": 1e-3,"use_crf": False,"class_num": None
}

数据处理

loader.py

# -*- coding: utf-8 -*-import json
import re
import os
import torch
import random
import jieba
import numpy as np
from torch.utils.data import Dataset, DataLoader"""
数据加载
"""class DataGenerator:def __init__(self, data_path, config):self.config = configself.path = data_pathself.vocab = load_vocab(config["vocab_path"])self.config["vocab_size"] = len(self.vocab)self.sentences = []self.schema = self.load_schema(config["schema_path"])self.config["class_num"] = len(self.schema)self.max_length = config["max_length"]self.load()def load(self):self.data = []with open(self.path, encoding="utf8") as f:for line in f:if len(line) > self.max_length:for i in range(len(line) // self.max_length):input_id, label = self.process_sentence(line[i * self.max_length:(i+1) * self.max_length])self.data.append([torch.LongTensor(input_id), torch.LongTensor(label)])else:input_id, label = self.process_sentence(line)self.data.append([torch.LongTensor(input_id), torch.LongTensor(label)])returndef process_sentence(self, line):sentence_without_sign = []label = []for index, char in enumerate(line[:-1]):if char in self.schema:  #准备加的标点，在训练数据中不应该存在continuesentence_without_sign.append(char)next_char = line[index + 1]if next_char in self.schema:  #下一个字符是标点，计入对应labellabel.append(self.schema[next_char])else:label.append(0)assert len(sentence_without_sign) == len(label)encode_sentence = self.encode_sentence(sentence_without_sign)label = self.padding(label, -1)assert len(encode_sentence) == len(label)self.sentences.append("".join(sentence_without_sign))return encode_sentence, labeldef encode_sentence(self, text, padding=True):input_id = []if self.config["vocab_path"] == "words.txt":for word in jieba.cut(text):input_id.append(self.vocab.get(word, self.vocab["[UNK]"]))else:for char in text:input_id.append(self.vocab.get(char, self.vocab["[UNK]"]))if padding:input_id = self.padding(input_id)return input_id#补齐或截断输入的序列，使其可以在一个batch内运算def padding(self, input_id, pad_token=0):input_id = input_id[:self.config["max_length"]]input_id += [pad_token] * (self.config["max_length"] - len(input_id))return input_iddef __len__(self):return len(self.data)def __getitem__(self, index):return self.data[index]def load_schema(self, path):with open(path, encoding="utf8") as f:return json.load(f)#加载字表或词表
def load_vocab(vocab_path):token_dict = {}with open(vocab_path, encoding="utf8") as f:for index, line in enumerate(f):token = line.strip()token_dict[token] = index + 1  #0留给padding位置，所以从1开始return token_dict#用torch自带的DataLoader类封装数据
def load_data(data_path, config, shuffle=True):dg = DataGenerator(data_path, config)dl = DataLoader(dg, batch_size=config["batch_size"], shuffle=shuffle)return dl

这段代码实现了一个数据生成器（DataGenerator），主要用于加载、处理和准备文本数据，以供深度学习模型进行训练。该数据生成器类利用 PyTorch 的 DataLoader 进行数据加载，并通过以下几个步骤处理原始数据：

数据生成器的逻辑思路如下：

1. 文本加载与切割：首先，加载原始文本数据，并根据设定的最大长度将文本切割成多个片段。每个片段的长度不超过最大长度，确保在模型训练时，输入文本的长度统一。

2. 文本编码：接下来，将每个片段的文本进行编码。根据配置，可以选择字符级别或词级别的编码方式。文本中的每个字符或词会被映射为词汇表中的数字索引。这样，文本从字符串形式转化为模型可以处理的数字形式。

3. 标签处理：标签生成是该过程的重要部分。对于每个字符，判断其后是否是标点符号。如果是标点符号，则生成标签1，表示该字符后是标点；否则，标签为0，表示后续不是标点符号。这一步有助于模型学习文本中的标点符号位置。

4. 填充与截断：为了保证输入序列的长度一致，所有文本片段都会根据最大长度进行填充或截断。如果文本长度不足最大长度，会在其后填充特殊符号；如果超出最大长度，则会被截断。

5. 数据批量处理：所有处理后的文本和标签被封装为一个数据集，通过 DataLoader 进行批量加载。DataLoader 不仅可以将数据按批次划分，还能在训练过程中打乱数据，确保训练的多样性和随机性。

6. 输出数据：最终，经过上述处理的文本和标签将作为输入提供给模型进行训练。每个批次的数据包括编码后的文本和相应的标签序列，保证数据格式的正确性和一致性。

通过这个过程，文本被转化为模型可以接受的数字形式，同时为每个字符提供了对应的标记信息，确保了训练数据的高效处理和准确性。

模型构建

model.py

# -*- coding: utf-8 -*-import torch
import torch.nn as nn
from torch.optim import Adam, SGD
from torchcrf import CRF
"""
建立网络模型结构
"""class TorchModel(nn.Module):def __init__(self, config):super(TorchModel, self).__init__()hidden_size = config["hidden_size"]vocab_size = config["vocab_size"] + 1class_num = config["class_num"]self.embedding = nn.Embedding(vocab_size, hidden_size, padding_idx=0)self.layer = nn.LSTM(hidden_size, hidden_size, batch_first=True, bidirectional=True, num_layers=1)self.classify = nn.Linear(hidden_size * 2, class_num)self.crf_layer = CRF(class_num, batch_first=True)self.use_crf = config["use_crf"]self.loss = torch.nn.CrossEntropyLoss(ignore_index=-1)  #loss采用交叉熵损失#当输入真实标签，返回loss值；无真实标签，返回预测值def forward(self, x, target=None):x = self.embedding(x)  #input shape:(batch_size, sen_len)x, _ = self.layer(x)      #input shape:(batch_size, sen_len, input_dim)predict = self.classify(x)                if target is not None:if self.use_crf:mask = target.gt(-1) # 返回判断每个值是否比-1大的结果return -self.crf_layer(predict, target, mask, reduction="mean")else:return self.loss(predict.view(-1, predict.shape[-1]), target.view(-1))else:if self.use_crf:return self.crf_layer.decode(predict)else:return predictdef choose_optimizer(config, model):optimizer = config["optimizer"]learning_rate = config["learning_rate"]if optimizer == "adam":return Adam(model.parameters(), lr=learning_rate)elif optimizer == "sgd":return SGD(model.parameters(), lr=learning_rate)

这段代码展示了如何定义一个用于序列标注任务的神经网络模型，结合 CRF 层进行精确的标签预测，并根据配置选择不同的优化器进行训练。。以下是逻辑顺序的详细解析：

定义模型结构

• TorchModel 类：该类继承自 nn.Module，表示一个自定义的神经网络模型。

1.1 初始化方法（__init__）：

• 从传入的 config 字典中读取配置项，如：
  • hidden_size：隐藏层大小，决定了 LSTM 层的输出维度。
  • vocab_size：词汇表大小，模型使用的词汇表包括一个额外的索引（用于填充），因此 vocab_size + 1。
  • class_num：分类任务的类别数。
• 初始化各个层：
  • self.embedding：使用 nn.Embedding 创建词嵌入层，将词汇索引转化为固定维度的嵌入向量，输入的 vocab_size 和 hidden_size 作为参数。
  • self.layer：使用 nn.LSTM 创建一个双向 LSTM 层，hidden_size 表示 LSTM 的输入和输出维度。batch_first=True 表示输入数据的格式是 (batch_size, sequence_length)，bidirectional=True 使得 LSTM 双向处理输入序列。
  • self.classify：使用 nn.Linear 创建一个全连接层，将 LSTM 的输出映射到分类数目 class_num。
  • self.crf_layer：使用自定义的 CRF（条件随机场）层进行序列标注任务的解码，通常在命名实体识别、分词等任务中使用。
  • self.use_crf：从 config 中读取是否使用 CRF 层的标志。
  • self.loss：定义交叉熵损失函数 CrossEntropyLoss，用于计算模型的损失，忽略标签为 -1 的样本。

前向传播方法

2.1 输入处理：

• 输入 x 是一个包含词汇索引的序列，首先通过 self.embedding 进行词嵌入，将词汇索引转化为嵌入向量，输出形状为 (batch_size, sequence_length, hidden_size)。

2.2 LSTM 处理：

• 将嵌入后的数据输入到 self.layer，LSTM 层会输出一个双向的序列表示，输出形状为 (batch_size, sequence_length, 2 * hidden_size)（因为使用了双向 LSTM）。

2.3 分类层：

• 将 LSTM 输出传入 self.classify 层，输出的形状为 (batch_size, sequence_length, class_num)，表示每个时间步的类别预测。

2.4 有目标标签时计算损失：

• 如果输入包含目标标签 target，判断是否使用 CRF 层：
  • 使用 CRF：如果 use_crf=True，通过 CRF 层计算负对数似然损失。首先创建一个掩码（mask）表示有效的标签位置，然后使用 CRF 层计算损失。
  • 不使用 CRF：使用标准的交叉熵损失函数计算损失，predict.view(-1, predict.shape[-1]) 和 target.view(-1) 将预测值和目标标签拉平成一维，进行损失计算。

2.5 没有目标标签时进行预测：

• 如果没有目标标签 target，返回模型的预测结果：
  • 使用 CRF：通过 CRF 层进行解码，返回最可能的标签序列。
  • 不使用 CRF：直接返回 LSTM 输出的分类结果。

优化器选择

3.1 输入配置读取：

• 从 config 中读取优化器类型 optimizer（如 Adam 或 SGD）和学习率 learning_rate。

3.2 优化器选择：

• 根据 optimizer 字段的值选择不同的优化器：
  • Adam：使用 Adam 优化器，传入模型参数和学习率。
  • SGD：使用 SGD 优化器，传入模型参数和学习率。

总结

1. TorchModel：构建了一个包含嵌入层、LSTM 层、分类层和 CRF 层的神经网络模型，支持两种任务模式：训练时计算损失，预测时返回结果。
2. 优化器选择：根据配置选择合适的优化器（Adam 或 SGD），并初始化学习率。

主程序

main.py

# -*- coding: utf-8 -*-import torch
import os
import numpy as np
import logging
from config import Config
from model import TorchModel, choose_optimizer
from evaluate import Evaluator
from loader import load_datalogging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)"""
模型训练主程序
"""def main(config):#创建保存模型的目录if not os.path.isdir(config["model_path"]):os.mkdir(config["model_path"])#加载训练数据train_data = load_data(config["train_data_path"], config)#加载模型model = TorchModel(config)# 标识是否使用gpucuda_flag = torch.cuda.is_available()if cuda_flag:logger.info("gpu可以使用，迁移模型至gpu")model = model.cuda()#加载优化器optimizer = choose_optimizer(config, model)#加载效果测试类evaluator = Evaluator(config, model, logger)#训练for epoch in range(config["epoch"]):epoch += 1model.train()logger.info("epoch %d begin" % epoch)train_loss = []for index, batch_data in enumerate(train_data):optimizer.zero_grad()if cuda_flag:batch_data = [d.cuda() for d in batch_data]input_id, labels = batch_data   #输入变化时这里需要修改，比如多输入，多输出的情况loss = model(input_id, labels)loss.backward()optimizer.step()train_loss.append(loss.item())if index % int(len(train_data) / 2) == 0:logger.info("batch loss %f" % loss)logger.info("epoch average loss: %f" % np.mean(train_loss))evaluator.eval(epoch)model_path = os.path.join(config["model_path"], "epoch_%d.pth" % epoch)torch.save(model.state_dict(), model_path)return model, train_dataif __name__ == "__main__":model, train_data = main(Config)

该代码实现了一个深度学习模型的训练过程，包括数据加载、模型初始化、训练、评估和模型保存等步骤。首先，配置文件加载训练所需的参数，并创建模型保存目录。接着，检测是否有可用的 GPU 来加速训练，并通过配置文件加载优化器。模型进入训练模式后，在每个 epoch 中遍历数据批次，计算损失并执行反向传播与优化器更新，同时记录每个批次的损失。在每个 epoch 结束后，通过评估函数计算模型的表现，并保存当前模型权重。训练完成后，返回训练后的模型和数据，整个过程通过日志输出损失变化，以便监控模型训练效果。

测试与评估

evaluate.py

# -*- coding: utf-8 -*-
import torch
import re
import numpy as np
from collections import defaultdict
from loader import load_data"""
模型效果测试
"""class Evaluator:def __init__(self, config, model, logger):self.config = configself.model = modelself.logger = loggerself.valid_data = load_data(config["valid_data_path"], config, shuffle=False)self.schema = self.valid_data.dataset.schemaself.index_to_label = dict((y, x) for x, y in self.schema.items())def eval(self, epoch):self.logger.info("开始测试第%d轮模型效果：" % epoch)self.stats_dict = dict(zip(self.schema.keys(), [defaultdict(int) for i in range(len(self.schema))]))self.model.eval()for index, batch_data in enumerate(self.valid_data):sentences = self.valid_data.dataset.sentences[index * self.config["batch_size"]: (index+1) * self.config["batch_size"]]if torch.cuda.is_available():batch_data = [d.cuda() for d in batch_data]input_id, labels = batch_data   #输入变化时这里需要修改，比如多输入，多输出的情况with torch.no_grad():pred_results = self.model(input_id) #不输入labels，使用模型当前参数进行预测self.write_stats(labels, pred_results, sentences)self.show_stats()returndef write_stats(self, labels, pred_results, sentences):assert len(labels) == len(pred_results) == len(sentences), print(len(labels), len(pred_results), len(sentences))if not self.config["use_crf"]:pred_results = torch.argmax(pred_results, dim=-1)for true_label, pred_label, sentence in zip(labels, pred_results, sentences):if not self.config["use_crf"]:pred_label = pred_label.cpu().detach().tolist()[:len(sentence)]true_label = true_label.cpu().detach().tolist()[:len(sentence)]for pred, gold in zip(pred_label, true_label):key = self.index_to_label[gold]self.stats_dict[key]["correct"] += 1 if pred == gold else 0self.stats_dict[key]["total"] += 1returndef show_stats(self):total = []for key in self.schema:acc = self.stats_dict[key]["correct"] / (1e-5 + self.stats_dict[key]["total"])self.logger.info("符号%s预测准确率：%f"%(key, acc))total.append(acc)self.logger.info("平均acc：%f" % np.mean(total))self.logger.info("--------------------")return

这段代码是一个用于模型评估的类 Evaluator，主要用于在验证集上评估深度学习模型的性能。它通过与配置文件中的设置相配合，加载验证数据并在模型上进行推理，最后计算预测的准确性。

类初始化

class Evaluator:def __init__(self, config, model, logger):self.config = configself.model = modelself.logger = loggerself.valid_data = load_data(config["valid_data_path"], config, shuffle=False)self.schema = self.valid_data.dataset.schemaself.index_to_label = dict((y, x) for x, y in self.schema.items())

• config：包含了模型训练的配置信息，比如验证数据集的路径、批次大小等。
• model：训练好的模型，通常是一个神经网络。
• logger：日志记录器，用于记录评估过程中的信息。
• valid_data：通过 load_data 函数加载验证数据集，这里使用 config["valid_data_path"] 提供的路径。
• schema：数据集中的标签集合，用于定义每个标签的名称，可能表示分类或标签类别。
• index_to_label：将标签的索引映射到标签名称的字典。self.schema.items() 返回标签的键值对，dict((y, x) for x, y in self.schema.items()) 将键值对调换，得到标签从索引到名称的映射。

模型评估

def eval(self, epoch):self.logger.info("开始测试第%d轮模型效果：" % epoch)self.stats_dict = dict(zip(self.schema.keys(), [defaultdict(int) for i in range(len(self.schema))]))self.model.eval()for index, batch_data in enumerate(self.valid_data):sentences = self.valid_data.dataset.sentences[index * self.config["batch_size"]: (index+1) * self.config["batch_size"]]if torch.cuda.is_available():batch_data = [d.cuda() for d in batch_data]input_id, labels = batch_datawith torch.no_grad():pred_results = self.model(input_id)self.write_stats(labels, pred_results, sentences)self.show_stats()return

• epoch：当前的训练轮数，用于日志记录，标识模型评估的轮次。
• self.stats_dict：一个字典，记录每个标签类别的预测统计数据，包括正确预测和总预测数。使用 defaultdict(int) 来自动初始化为整数，方便后续累加。
• self.model.eval()：将模型设置为评估模式（eval），这会关闭像 dropout 这样的训练时特有的操作。
• 数据批次处理：通过遍历验证数据集中的每个批次来进行评估。self.valid_data 是一个数据加载器，按批次返回数据。
  • batch_data：包含了输入数据和标签的数据。
  • torch.cuda.is_available()：检查是否有可用的 GPU，如果有，将数据迁移到 GPU 上进行加速计算。
  • input_id, labels = batch_data：假设 batch_data 中包含了输入数据（input_id）和真实标签（labels）。
  • with torch.no_grad()：在此上下文中，关闭梯度计算，节省内存并加速计算，因为评估时不需要进行反向传播。
  • pred_results = self.model(input_id)：使用当前模型预测输入数据的结果。
• write_stats：将模型的预测结果与真实标签对比，并记录统计信息。
• show_stats：显示最终的评估统计结果。

统计记录

def write_stats(self, labels, pred_results, sentences):assert len(labels) == len(pred_results) == len(sentences), print(len(labels), len(pred_results), len(sentences))if not self.config["use_crf"]:pred_results = torch.argmax(pred_results, dim=-1)for true_label, pred_label, sentence in zip(labels, pred_results, sentences):if not self.config["use_crf"]:pred_label = pred_label.cpu().detach().tolist()[:len(sentence)]true_label = true_label.cpu().detach().tolist()[:len(sentence)]for pred, gold in zip(pred_label, true_label):key = self.index_to_label[gold]self.stats_dict[key]["correct"] += 1 if pred == gold else 0self.stats_dict[key]["total"] += 1return

• assert：确保标签、预测结果和句子的长度一致。如果不一致，则输出错误信息。
• torch.argmax(pred_results, dim=-1)：如果没有使用 CRF（条件随机场），将模型的输出结果转化为最大值的索引，表示每个位置的预测标签。
• for true_label, pred_label, sentence in zip(labels, pred_results, sentences)：遍历每个样本的标签、预测标签和句子。
  • pred_label = pred_label.cpu().detach().tolist()：将预测标签从 GPU 转移到 CPU，去除梯度信息，并转换为列表，截取与句子长度相同的部分。
  • true_label = true_label.cpu().detach().tolist()：同样地，处理真实标签。
  • for pred, gold in zip(pred_label, true_label)：遍历每个单词的预测标签和真实标签。
    • key = self.index_to_label[gold]：将标签的索引转换为标签名称。
    • 统计正确预测：如果预测标签与真实标签一致，则增加 correct，否则不变；无论如何，都会增加 total。

显示统计结果

def show_stats(self):total = []for key in self.schema:acc = self.stats_dict[key]["correct"] / (1e-5 + self.stats_dict[key]["total"])self.logger.info("符号%s预测准确率：%f"%(key, acc))total.append(acc)self.logger.info("平均acc：%f" % np.mean(total))self.logger.info("--------------------")return

• total = []：用于记录每个标签的准确率。
• for key in self.schema：遍历所有标签。
  • acc = self.stats_dict[key]["correct"] / (1e-5 + self.stats_dict[key]["total"])：计算每个标签的准确率。分母中加上 1e-5 来避免除以零的情况。
  • self.logger.info()：记录每个标签的准确率。
• np.mean(total)：计算所有标签的平均准确率，并打印出来。

总结

• 该类主要用于在模型训练后进行模型效果的评估。通过对验证集数据的批量处理、预测结果与真实标签的对比，计算每个标签的预测准确率，并输出平均准确率。
• eval 方法是评估的主流程，依次执行数据加载、模型推理、统计记录和结果展示。
• write_stats 方法负责记录每个标签的预测准确度，show_stats 方法负责展示整体的评估结果。

测试结果

predict.py

# -*- coding: utf-8 -*-
import torch
import json
from config import Config
from model import TorchModel
"""
模型效果测试
"""class SentenceLabel:def __init__(self, config, model_path):self.config = configself.schema = self.load_schema(config["schema_path"])self.index_to_sign = dict((y, x) for x, y in self.schema.items())self.vocab = self.load_vocab(config["vocab_path"])self.model = TorchModel(config)self.model.load_state_dict(torch.load(model_path))self.model.eval()print("模型加载完毕!")def load_schema(self, path):with open(path, encoding="utf8") as f:schema = json.load(f)self.config["class_num"] = len(schema)return schema# 加载字表或词表def load_vocab(self, vocab_path):token_dict = {}with open(vocab_path, encoding="utf8") as f:for index, line in enumerate(f):token = line.strip()token_dict[token] = index + 1  # 0留给padding位置，所以从1开始self.config["vocab_size"] = len(token_dict)return token_dictdef predict(self, sentence):input_id = []for char in sentence:input_id.append(self.vocab.get(char, self.vocab["[UNK]"]))with torch.no_grad():res = self.model(torch.LongTensor([input_id]))[0]res = torch.argmax(res, dim=-1)labeled_sentence = ""for char, label_index in zip(sentence, res):labeled_sentence += char + self.index_to_sign[int(label_index)]return labeled_sentenceif __name__ == "__main__":sl = SentenceLabel(Config, "model_output/epoch_10.pth")sentence = "客厅的颜色比较稳重但不沉重相反很好的表现了欧式的感觉给人高雅的味道"res = sl.predict(sentence)print(res)sentence = "双子座的健康运势也呈上升的趋势但下半月有所回落"res = sl.predict(sentence)print(res)

这部分实现了一个基于深度学习模型的中文文本标注功能。SentenceLabel 类负责加载模型、词表和标签映射。模型通过 TorchModel 加载，并使用 load_state_dict 加载预训练的权重。load_schema 和 load_vocab 方法分别加载标签映射和词表，将句子中的每个字符转换为词汇表中的索引。predict 方法将输入句子转换为模型输入，进行预测并输出带标签的句子。

输入文本：
"客厅的颜色比较稳重但不沉重相反很好的表现了欧式的感觉给人高雅的味道"
"双子座的健康运势也呈上升的趋势但下半月有所回落"

文本加标点效果：
客厅的颜色比较稳重，但不沉重相反很好的表现了欧式的感觉给人高雅的味道。
双子座的健康运势也呈上升的趋势，但下半月有所回落。