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")

打卡圖片：