昇思25天学习打卡营第12天|LLM-基于MindSpore实现的BERT对话情绪识别

打卡

目录

打卡

预装环境

BERT

任务说明

数据集

数据加载和数据预处理：process_dataset 函数

模型构建与训练

运行示例

模型验证

模型推理

自定义推理数据集

运行结果示例

代码

预装环境

pip install -i https://pypi.mirrors.ustc.edu.cn/simple mindspore==2.2.14
pip install mindnlp

pip show mindspore
pip show mindnlp

BERT

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

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

CASE 1 : 在用 Masked Language Model 方法训练BERT的时候，随机把语料库中15%的单词做Mask操作。对于这15%的单词做Mask操作分为三种情况：

1）80%的单词直接用[Mask]替换

2）10%的单词直接替换成另一个新的单词

3）10%的单词保持不变。

CASE 2 : 因为涉及到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），专注于识别智能对话场景中用户的情绪，针对智能对话场景中的用户文本，自动判断该文本的情绪类别并给出相应的置信度，情绪类型分为积极、消极、中性。 对话情绪识别适用于聊天、客服等多个场景，能够帮助企业更好地把握对话质量、改善产品的用户交互体验，也能分析客服服务质量、降低人工质检成本。

数据集

• 来源：百度飞桨团队
• 数据集说明：已标注的、经过分词预处理的机器人聊天数据集。数据由两列组成，以制表符（'\t'）分隔，第一列是情绪分类的类别（0表示消极；1表示中性；2表示积极），第二列是以空格分词的中文文本。文件为 utf8 编码。

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 函数

昇腾NPU环境下暂不支持动态Shape，数据预处理部分采用静态Shape处理

import numpy as np

def process_dataset(source, tokenizer, max_seq_len=64, batch_size=32, shuffle=True):
    is_ascend = mindspore.get_context('device_target') == 'Ascend'

    column_names = ["label", "text_a"]
    
    dataset = GeneratorDataset(source, column_names=column_names, shuffle=shuffle)
    # transforms
    type_cast_op = transforms.TypeCast(mindspore.int32)
    def tokenize_and_pad(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']
    # map dataset
    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')
    # batch dataset
    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


from mindnlp.transformers import BertTokenizer
tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')

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)

部分执行结果演示，看到词表大小为 21128 ，模型维度长 512 ，右侧截断，一共有5种特殊的token，其中训练、验证、测试集数据分别有302、34、33个。

模型构建与训练

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

from mindnlp.transformers import BertForSequenceClassification, BertModel
from mindnlp._legacy.amp import auto_mixed_precision


### 模型配置
# set bert config and define parameters for training
model = BertForSequenceClassification.from_pretrained('bert-base-chinese', num_labels=3)
model = auto_mixed_precision(model, 'O1')

### 优化算法选择
optimizer = nn.Adam(model.trainable_params(), learning_rate=2e-5)


### 评估指标
metric = Accuracy()
# define callbacks to save checkpoints
ckpoint_cb = CheckpointCallback(save_path='checkpoint', ckpt_name='bert_emotect', epochs=1, keep_checkpoint_max=2)
best_model_cb = BestModelCallback(save_path='checkpoint', ckpt_name='bert_emotect_best', auto_load=True)

trainer = Trainer(network=model, train_dataset=dataset_train,
                  eval_dataset=dataset_val, metrics=metric,
                  epochs=5, optimizer=optimizer, callbacks=[ckpoint_cb, best_model_cb])


### 模型训练
# start training
trainer.run(tgt_columns="labels")

运行示例

loss降低到了0.0663，精度达到了 0.9917。非常不错。

模型验证

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

evaluator = Evaluator(network=model, 
                      eval_dataset=dataset_test, 
                      metrics=metric)
evaluator.run(tgt_columns="labels")

模型推理

from mindspore import Tensor

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

def predict(text, label=None):
    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)



for label, text in dataset_infer:
    predict(text, label)

自定义推理数据集

predict("家人们咱就是说一整个无语住了 绝绝子叠buff")
predict("起开 我要开始发功了")

运行结果示例

代码

import os
import numpy as np
from mindspore import Tensor
import mindspore
from mindspore.dataset import text, GeneratorDataset, transforms
from mindspore import nn, context

from mindnlp._legacy.engine import Trainer, Evaluator
from mindnlp._legacy.engine.callbacks import CheckpointCallback, BestModelCallback
from mindnlp._legacy.metrics import Accuracy
from mindnlp.transformers import BertTokenizer
from mindnlp.transformers import BertForSequenceClassification, BertModel
from mindnlp._legacy.amp import auto_mixed_precision


# prepare dataset
class SentimentDataset:
    """Sentiment Dataset"""

    def __init__(self, path):
        self.path = path
        self._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)
    
    

def process_dataset(source, tokenizer, max_seq_len=64, batch_size=32, shuffle=True):
    """
    这个函数 process_dataset 用于处理文本数据集，包括文本的标记化、填充、类型转换和批处理，并针对不同的设备（Ascend或非Ascend）进行了不同的处理策略。
    source: 数据集的来源，可以是文件路径或数据生成器。
    tokenizer: 用于将文本转换为数字表示的标记器。
    max_seq_len: 最大序列长度，默认为64。
    batch_size: 批处理大小，默认为32。
    shuffle: 是否打乱数据集，默认为True。
    """
    ## 检查当前设备目标是否为Ascend（华为的AI处理器），并将结果存储在变量 is_ascend 中。
    is_ascend = mindspore.get_context('device_target') == 'Ascend'

    # 定义一个列表 column_names，包含数据集中列的名称。
    column_names = ["label", "text_a"]
    
    # 创建一个 GeneratorDataset 对象，它从 source 生成数据集，指定列名，并根据 shuffle 参数决定是否打乱数据。
    dataset = GeneratorDataset(source, column_names=column_names, shuffle=shuffle)
        
    # transforms
    type_cast_op = transforms.TypeCast(mindspore.int32)
    ## 定义一个内部函数 tokenize_and_pad，用于对文本进行标记化和填充。
    def tokenize_and_pad(text):
        ## 在 tokenize_and_pad 函数内部，根据 is_ascend 的值来决定是否对文本进行填充和截断。如果设备是Ascend，则使用 max_length 填充策略和截断。函数返回标记化后的 input_ids 和 attention_mask。
        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']
    # map dataset
    ## 使用 map 操作将 tokenize_and_pad 函数应用到数据集的 “text_a” 列上，并将输出列命名为 ‘input_ids’ 和 ‘attention_mask’。
    dataset = dataset.map(operations=tokenize_and_pad, input_columns="text_a", output_columns=['input_ids', 'attention_mask'])
    ## 使用 map 操作将 type_cast_op 应用于数据集的 “label” 列，并将输出列命名为 ‘labels’。
    dataset = dataset.map(operations=[type_cast_op], input_columns="label", output_columns='labels')
    # batch dataset
    ## 根据 is_ascend 的值，决定使用普通的批处理还是填充后的批处理。如果设备是Ascend，则直接进行批处理；否则，使用 padded_batch 来确保每个批次中的序列长度一致。
    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    

def predict(text, label=None):
    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)
    
    
tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')

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)
print(f"train dataset columns name: ", dataset_train.get_col_names())

print(next(dataset_train.create_tuple_iterator()))


##### 模型构建
# set bert config and define parameters for training
model = BertForSequenceClassification.from_pretrained('bert-base-chinese', num_labels=3)  ### 3个标签输出
model = auto_mixed_precision(model, 'O1')
## 优化器
optimizer = nn.Adam(model.trainable_params(), learning_rate=2e-5)
print(f"模型结构输出 model：", model)
print(f"优化器选择 optimizer：", optimizer)

####### 模型训练配置项
metric = Accuracy()
# define callbacks to save checkpoints
ckpoint_cb = CheckpointCallback(save_path='checkpoint', ckpt_name='bert_emotect', epochs=1, keep_checkpoint_max=2)
best_model_cb = BestModelCallback(save_path='checkpoint', ckpt_name='bert_emotect_best', auto_load=True)

trainer = Trainer(network=model, train_dataset=dataset_train,
                  eval_dataset=dataset_val, metrics=metric,
                  epochs=5, optimizer=optimizer, callbacks=[ckpoint_cb, best_model_cb])


######### 模型训练 
# start training
trainer.run(tgt_columns="labels")

########## 模型评估
evaluator = Evaluator(network=model, eval_dataset=dataset_test, metrics=metric)
evaluator.run(tgt_columns="labels")


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

for label, text in dataset_infer:
    predict(text, label)

print(predict("家人们咱就是说一整个无语住了 绝绝子叠buff"))
print(predict("起开 我要开始发功了"))