240705_昇思学习打卡-Day17-基于 MindSpore 实现 BERT对话情绪识别

近期确实太忙，此处仅作简单记录：

模型简介

BERT全称是来自变换器的双向编码器表征量（Bidirectional Encoder Representations from Transformers），它是Google于2018年末开发并发布的一种新型语言模型。与BERT模型相似的预训练语言模型例如问答、命名实体识别、自然语言推理、文本分类等在许多自然语言处理任务中发挥着重要作用。模型是基于Transformer中的Encoder并加上双向的结构，因此一定要熟练掌握Transformer的Encoder的结构。

关于Transformer的Encoder的结构在这篇中有提及，可以去参考看看240701_昇思学习打卡-Day13-Vision Transformer图像分类-CSDN博客

BERT模型的主要创新点都在pre-train方法上，即用了Masked Language Model和Next Sentence Prediction两种方法分别捕捉词语和句子级别的representation。

在用Masked Language Model方法训练BERT的时候，随机把语料库中15%的单词做Mask操作。对于这15%的单词做Mask操作分为三种情况：80%的单词直接用[Mask]替换、10%的单词直接替换成另一个新的单词、10%的单词保持不变。

因为涉及到Question Answering (QA) 和 Natural Language Inference (NLI)之类的任务，增加了Next Sentence Prediction预训练任务，目的是让模型理解两个句子之间的联系。与Masked Language Model任务相比，Next Sentence Prediction更简单些，训练的输入是句子A和B，B有一半的几率是A的下一句，输入这两个句子，BERT模型预测B是不是A的下一句。

BERT预训练之后，会保存它的Embedding table和12层Transformer权重（BERT-BASE）或24层Transformer权重（BERT-LARGE）。使用预训练好的BERT模型可以对下游任务进行Fine-tuning，比如：文本分类、相似度判断、阅读理解等。

对话情绪识别（Emotion Detection，简称EmoTect），专注于识别智能对话场景中用户的情绪，针对智能对话场景中的用户文本，自动判断该文本的情绪类别并给出相应的置信度，情绪类型分为积极、消极、中性。 对话情绪识别适用于聊天、客服等多个场景，能够帮助企业更好地把握对话质量、改善产品的用户交互体验，也能分析客服服务质量、降低人工质检成本。

下面以一个文本情感分类任务为例子来说明BERT模型的整个应用过程。

我们假设已经装好了MindSpore环境

# 该案例在 mindnlp 0.3.1 版本完成适配，如果发现案例跑不通，可以指定mindnlp版本，执行`!pip install mindnlp==0.3.1`
!pip install mindnlp

import osimport mindspore
from mindspore.dataset import text, GeneratorDataset, transforms
from mindspore import nn, contextfrom mindnlp._legacy.engine import Trainer, Evaluator
from mindnlp._legacy.engine.callbacks import CheckpointCallback, BestModelCallback
from mindnlp._legacy.metrics import Accuracy

# prepare dataset
class SentimentDataset:"""Sentiment Dataset"""def __init__(self, path):self.path = pathself._labels, self._text_a = [], []self._load()def _load(self):with open(self.path, "r", encoding="utf-8") as f:dataset = f.read()lines = dataset.split("\n")for line in lines[1:-1]:label, text_a = line.split("\t")self._labels.append(int(label))self._text_a.append(text_a)def __getitem__(self, index):return self._labels[index], self._text_a[index]def __len__(self):return len(self._labels)

# 准备数据集
class 情感分析数据集(SentimentDataset):"""情感分析数据集类，用于加载和管理数据集。参数:path (str): 数据集文件的路径。属性:_labels (list): 存储情感标签的列表。_text_a (list): 存储文本内容的列表。方法:_load(): 从指定路径加载数据集文件，解析内容并存储到_labels和_text_a中。__getitem__(index): 根据索引返回特定样本的标签和文本。__len__(): 返回数据集的样本数量。"""def __init__(self, path):"""初始化情感分析数据集对象，设置数据路径并加载数据。参数:path (str): 数据集文件的路径。"""self.path = pathself._labels, self._text_a = [], []self._load()def _load(self):"""私有方法：读取数据集文件，按行处理数据，分割标签和文本，并存储到实例变量中。"""with open(self.path, "r", encoding="utf-8") as f:dataset = f.read()lines = dataset.split("\n")for line in lines[1:-1]:  # 跳过首行（假设为列名）和末尾的空行label, text_a = line.split("\t")self._labels.append(int(label))  # 添加标签到_labels列表self._text_a.append(text_a)  # 添加文本到_text_a列表def __getitem__(self, index):"""通过索引获取数据集中对应样本的标签和文本。参数:index (int): 数据样本的索引位置。返回:tuple: 包含样本标签和文本的元组 (label, text)。"""return self._labels[index], self._text_a[index]def __len__(self):"""返回数据集中的样本数量。返回:int: 数据集样本数量。"""return len(self._labels)

数据集

这里提供一份已标注的、经过分词预处理的机器人聊天数据集，来自于百度飞桨团队。数据由两列组成，以制表符（‘\t’）分隔，第一列是情绪分类的类别（0表示消极；1表示中性；2表示积极），第二列是以空格分词的中文文本，如下示例，文件为 utf8 编码。

label–text_a

0–谁骂人了？我从来不骂人，我骂的都不是人，你是人吗 ？

1–我有事等会儿就回来和你聊

2–我见到你很高兴谢谢你帮我

这部分主要包括数据集读取，数据格式转换，数据 Tokenize 处理和 pad 操作。

# download dataset
!wget https://baidu-nlp.bj.bcebos.com/emotion_detection-dataset-1.0.0.tar.gz -O emotion_detection.tar.gz
!tar xvf emotion_detection.tar.gz

数据加载和数据预处理

新建 process_dataset 函数用于数据加载和数据预处理，具体内容可见下面代码注释。

import numpy as npdef process_dataset(source, tokenizer, max_seq_len=64, batch_size=32, shuffle=True):"""处理数据集，将其转换为适合模型训练的格式。参数:source: 数据集的来源，可以是文件路径或数据生成器。tokenizer: 用于将文本序列化为模型输入的标记化器。max_seq_len: 最大序列长度，超过这个长度的序列将被截断。batch_size: 每个批次的样本数量。shuffle: 是否在处理数据集前打乱数据顺序。返回:经过处理后的数据集，包括输入序列和标签。"""# 判断是否在昇腾设备上运行is_ascend = mindspore.get_context('device_target') == 'Ascend'# 定义数据集的列名column_names = ["label", "text_a"]# 创建数据集对象dataset = GeneratorDataset(source, column_names=column_names, shuffle=shuffle)# 将字符串类型转换为整型type_cast_op = transforms.TypeCast(mindspore.int32)# 定义文本标记化和填充函数def tokenize_and_pad(text):"""对文本进行标记化和填充，以适应模型的要求。参数:text: 需要处理的文本。返回:标记化和填充后的输入序列和注意力掩码。"""if is_ascend:# 在昇腾设备上，使用特定的处理方式tokenized = tokenizer(text, padding='max_length', truncation=True, max_length=max_seq_len)else:# 在其他设备上，直接进行标记化tokenized = tokenizer(text)return tokenized['input_ids'], tokenized['attention_mask']# 对文本列进行标记化和填充处理dataset = dataset.map(operations=tokenize_and_pad, input_columns="text_a", output_columns=['input_ids', 'attention_mask'])# 对标签列进行类型转换dataset = dataset.map(operations=[type_cast_op], input_columns="label", output_columns='labels')# 根据设备类型选择合适的批次处理方式if is_ascend:# 在昇腾设备上，使用简单的批次处理dataset = dataset.batch(batch_size)else:# 在其他设备上，使用带填充的批次处理dataset = dataset.padded_batch(batch_size, pad_info={'input_ids': (None, tokenizer.pad_token_id),'attention_mask': (None, 0)})return dataset

数据预处理部分采用静态Shape处理：

# 导入BertTokenizer类，用于BERT模型的预训练 tokenizer
from mindnlp.transformers import BertTokenizer# 初始化一个BertTokenizer实例，用于处理中文文本
# 这里使用了预训练的'bert-base-chinese'模型，该模型已经在中文文本上进行了预训练
# 选择这个预训练模型是因为我们的任务是处理中文文本，需要一个针对中文优化的tokenizer
tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')

tokenizer.pad_token_id

dataset_train = process_dataset(SentimentDataset("data/train.tsv"), tokenizer)
dataset_val = process_dataset(SentimentDataset("data/dev.tsv"), tokenizer)
dataset_test = process_dataset(SentimentDataset("data/test.tsv"), tokenizer, shuffle=False)

dataset_train.get_col_names()

print(next(dataset_train.create_tuple_iterator()))

模型构建

通过 BertForSequenceClassification 构建用于情感分类的 BERT 模型，加载预训练权重，设置情感三分类的超参数自动构建模型。后面对模型采用自动混合精度操作，提高训练的速度，然后实例化优化器，紧接着实例化评价指标，设置模型训练的权重保存策略，最后就是构建训练器，模型开始训练。

# 导入MindNLP库中用于序列分类任务的BertForSequenceClassification模型与用于获取文本编码表示的BertModel
from mindnlp.transformers import BertForSequenceClassification, BertModel
# 导入auto_mixed_precision函数以启用混合精度训练，能够加速训练过程并减少内存占用
from mindnlp._legacy.amp import auto_mixed_precision# 根据预训练的'bert-base-chinese'模型初始化BertForSequenceClassification模型，设置类别数为3
# 此模型适用于如文本分类任务，将输入文本归类到三个预定义类别中的一个
# 设置BERT模型配置及训练所需参数
model = BertForSequenceClassification.from_pretrained('bert-base-chinese', num_labels=3)
# 使用auto_mixed_precision函数对模型应用混合精度训练策略，采用'O1'优化级别
# 混合精度训练通过结合使用float16和float32数据类型来提升训练速度并节省内存资源
model = auto_mixed_precision(model, 'O1')# 定义模型训练使用的优化器为Adam算法，设置学习率为2e-5，并仅针对模型中可训练参数进行优化
optimizer = nn.Adam(model.trainable_params(), learning_rate=2e-5)

# 初始化Accuracy类，用于计算模型预测的准确率
metric = Accuracy()# 定义回调函数以保存训练过程中的检查点
# CheckpointCallback用于在指定的epoch后保存模型，保存路径为'checkpoint'，检查点文件名为'bert_emotect'
# 参数epochs设为1表示每个epoch后保存一次，keep_checkpoint_max=2表示最多保留2个检查点文件
ckpoint_cb = CheckpointCallback(save_path='checkpoint', ckpt_name='bert_emotect', epochs=1, keep_checkpoint_max=2)# BestModelCallback用于自动保存验证性能最优的模型，同样保存在'checkpoint'路径下，文件名为'bert_emotect_best'
# 设置auto_load=True可在训练结束后自动加载该最优模型
best_model_cb = BestModelCallback(save_path='checkpoint', ckpt_name='bert_emotect_best', auto_load=True)# 创建Trainer实例以组织训练流程
# network参数指定训练的模型，train_dataset和eval_dataset分别指定了训练集和验证集
# metrics参数指定了评估模型性能的指标，此处为刚刚定义的准确率Accuracy
# epochs设置训练轮次为5，optimizer为训练使用的优化器，callbacks列表包含了之前定义的保存检查点和最佳模型的回调函数
trainer = Trainer(network=model, train_dataset=dataset_train,eval_dataset=dataset_val, metrics=metric,epochs=5, optimizer=optimizer, callbacks=[ckpoint_cb, best_model_cb])

%%time
# start training
trainer.run(tgt_columns="labels")

模型验证

将验证数据集加再进训练好的模型，对数据集进行验证，查看模型在验证数据上面的效果，此处的评价指标为准确率。

# 初始化Evaluator对象，用于评估模型性能
# 参数说明：
# network: 待评估的模型
# eval_dataset: 用于评估的测试数据集
# metrics: 评估指标
evaluator = Evaluator(network=model, eval_dataset=dataset_test, metrics=metric)# 执行模型评估，指定目标列作为评估标签
# 该步骤将计算模型在测试数据集上的指定评估指标
evaluator.run(tgt_columns="labels")

dataset_infer = SentimentDataset("data/infer.tsv")

def predict(text, label=None):"""根据给定的文本进行情感分析预测。参数:text (str): 需要进行情感分析的文本。label (int, optional): 用于比较的预定义标签。如果提供，将打印预测标签和给定标签的比较。返回:无返回值，但打印了模型预测的情感标签以及输入文本。"""# 映射预测结果的标签到人类可读的情感描述label_map = {0: "消极", 1: "中性", 2: "积极"}# 将文本转换为模型输入所需的格式text_tokenized = Tensor([tokenizer(text).input_ids])# 使用模型预测文本的情感logits = model(text_tokenized)predict_label = logits[0].asnumpy().argmax()# 构建包含预测信息的字符串info = f"inputs: '{text}', predict: '{label_map[predict_label]}'"if label is not None:# 如果提供了标签，则添加实际标签的信息info += f" , label: '{label_map[label]}'"# 打印预测结果print(info)

from mindspore import Tensorfor label, text in dataset_infer:predict(text, label)

自定义推理数据集

自己输入一句话，进行测试

predict("家人们咱就是说一整个无语住了 绝绝子叠buff")

打卡图片：