带有详细注释的PaddlePaddle的情绪识别项目
哈喽,大家好!我是刘聪NLP。
最近,本人一直在研究PaddlePaddle框架,主要是为了支持国产(白嫖GPU),也是为了知识储备。不过,看了一些官方或非官方的项目之后,个人体验不是很好。因此抽了一个上午的时间,整理了一份情绪识别项目的代码,内部带有大量注释,与之前开源的GPT2项目相似,希望可以帮助到初学PaddlePaddle的朋友。
之前开源的GPT2项目,主要基于PyTorch,地址:超详细中文注释的GPT2新闻标题生成项目
情绪识别项目地址:https://aistudio.baidu.com/aistudio/projectdetail/2342217?contributionType=1
本文主要是对项目中的代码进行讲解,主要从数据预处理、数据类实现、模型代码实现、模型训练和模型测试,五个部分进行介绍,如下。
数据预处理
def sentiment_analysis_trans_data(path, save_path):
"""
数据预处理代码,将原始数据格式转换成模型所需格式数据,并统计各标签数据的数量
Args:
path: 原始数据路径
save_path: 保存数据路径
Returns:
"""
fin = open(save_path, "w", encoding="utf-8")
data_number = {}
with open(path, "r", encoding="utf-8") as fh:
# 加载原始数据
data = json.load(fh)
# 对原始数据进行遍历
for i, line in enumerate(data):
sample = {"text": line["content"], "label": line["label"]}
# 如果标签在data_number中,直接对其value进行加1操作;如果不在,则将标签加入的data_number中,value设为1。
if line["label"] not in data_number:
data_number[line["label"]] = 1
else:
data_number[line["label"]] += 1
# 将每一个文本和对应的标签,写入到保存文件中
fin.write(json.dumps(sample, ensure_ascii=False) + "\n")
print("data_number: ", data_number)
数据类实现
if os.path.exists(cached_feature_file) and not is_overwrite:
logger.info("已经存在缓存文件{},直接加载".format(cached_feature_file))
self.data_set = paddle.load(cached_feature_file)["data_set"]
else:
# 如果不存在缓存文件,则调用load_data函数,进行数据预处理,再将其保存成缓存文件。
logger.info("不存在缓存文件{},进行数据预处理操作".format(cached_feature_file))
self.data_set = self.load_data(path_file)
logger.info("数据预处理操作完成,将处理后的数据存到{}中,作为缓存文件".format(cached_feature_file))
paddle.save({"data_set": self.data_set}, cached_feature_file)
(2)将文本数据转换为索引数据的函数
def convert_featrue(self, sample):
"""
将单个样本转换成模型可用的id索引形式
Args:
sample: 单条样本
Returns:
"""
# 获取标签索引
label = self.label2id[sample["label"]]
# 将本文进行tokenize
tokens = self.tokenizer.tokenize(sample["text"])
# 进行长度判断,若长于最大长度,则进行截断
if len(tokens) > self.max_len - 2:
tokens = tokens[:self.max_len - 2]
# 将其头尾加上[CLS]和[SEP]
tokens = ["[CLS]"] + tokens + ["[SEP]"]
# 将token转化成id
input_ids = self.tokenizer.convert_tokens_to_ids(tokens)
# 获取模型所需的attention_mask,大小与input_ids一致
attention_mask = [1] * len(input_ids)
assert len(input_ids) == len(attention_mask)
return input_ids, attention_mask, label
(3)在模型训练时,对batch数据进行tensor转换的函数
def collate_func_sentiment_analysis(batch_data):
"""
DataLoader所需的collate_fun函数,将数据处理成tensor形式
Args:
batch_data: batch数据
Returns:
"""
# 获取batch数据的大小
batch_size = len(batch_data)
# 如果batch_size为0,则返回一个空字典
if batch_size == 0:
return {}
input_ids_list, attention_mask_list, labels_list = [], [], []
# 遍历batch数据,将每一个数据,转换成tensor的形式
for instance in batch_data:
input_ids_temp = instance["input_ids"]
attention_mask_temp = instance["attention_mask"]
labels_temp = instance["label"]
input_ids_list.append(paddle.to_tensor(input_ids_temp, dtype="int64"))
attention_mask_list.append(paddle.to_tensor(attention_mask_temp, dtype="int64"))
labels_list.append(labels_temp)
# 对一个batch内的数据,进行padding
return {"input_ids": Pad(pad_val=0, axis=0)(input_ids_list),
"attention_mask": Pad(pad_val=0, axis=0)(attention_mask_list),
"label": Stack(dtype="int64")(labels_list)}
这里的写法与Pytorch一致,感觉可扩展性更强。
模型代码实现
class SentimentAnalysisModel(BertPretrainedModel):
base_model_prefix = "bert"
def __init__(self, bert, number_label=3):
"""
情绪识别模型继承paddlenlp.transformers.BertPretrainedModel类
Args:
bert: bert模型
number_label: 标签个数
"""
super(SentimentAnalysisModel, self).__init__()
self.bert = bert
self.classifier = nn.layer.Linear(self.bert.config["hidden_size"], number_label)
self.loss_fct = nn.CrossEntropyLoss(soft_label=False, axis=-1)
def forward(self, input_ids, attention_mask, label=None):
# 将attention_mask进行维度变换,从2维变成4维。paddlenlp.transformers的实现与torch或tf不一样,不会自动进行维度扩充。
attention_mask = paddle.unsqueeze(attention_mask, axis=[1, 2])
# 获取[CLS]向量pooled_output
pooled_output = self.bert(input_ids=input_ids, attention_mask=attention_mask)[1]
# 对pooled_output进行全连接,映射到number_label上
logits = self.classifier(pooled_output)
# 使用softmax,获取每个标签类别的概率
probs = F.softmax(logits, axis=1)
# 获取标签类别概率最大的标签
pred_label = paddle.argmax(logits, axis=-1)
outputs = (pred_label, probs)
# 如果label不是None,则使用CrossEntropyLoss求解loss
if label is not None:
loss = self.loss_fct(logits, label)
outputs = (loss,) + outputs
return outputs
注意:代码中将attention_mask进行维度变换,从2维变成4维。paddlenlp.transformers的实现与torch或tf不一样,不会自动进行维度扩充。
模型训练
python3 train.py
or
python3 train.py --num_train_epochs 5--train_batch_size 64 --test_batch_size 32 --max_len 256 --output_dir./output_dir
模型训练文件主要由以下几个函数组成:(1)设置训练模型所需参数函数set_args;(2)训练模型函数train;(3)对测试数据集进行模型测试evaluate;(4)主函数main。
详细代码见AIStudio项目的train.py文件。
模型测试
def convert_featrue(sample, max_len, tokenizer):
"""
将单个文本,进行数据转换,得到模型所使用的id索引数据
Args:
sample: 单个文本,str类型
max_len: 最大长度
tokenizer: 分词器
Returns:
"""
# 对文本进行tokenize操作
tokens = tokenizer.tokenize(sample)
# 进行长度判断,若长于最大长度,则进行截断
if len(tokens) > max_len - 2:
tokens = tokens[:max_len - 2]
# 将其头尾加上[CLS]和[SEP]
tokens = ["[CLS]"] + tokens + ["[SEP]"]
# 将token转化成id,并获取模型所需的attention_mask
input_ids = tokenizer.convert_tokens_to_ids(tokens)
attention_mask = [1] * len(input_ids)
assert len(input_ids) == len(attention_mask)
# 对input_ids和attention_mask进行补全操作,补到最大长度
# 补全到最大长度,是由于后面会对动态图转onnx和静态图,输入需要定长
if len(input_ids) < max_len:
input_ids = input_ids + [0] * (max_len - len(input_ids))
attention_mask = attention_mask + [0] * (max_len - len(attention_mask))
return input_ids, attention_mask
def predict_one_sample(sample_list, model, tokenizer, max_len, id2label):
"""
对数据进行批量预测,获取每个样本对应的预测标签
Args:
sample_list: 样本序列,为一个list
model: 模型
tokenizer: 分词器
max_len: 最大长度
id2label: 标签字典
Returns:
"""
# 将数据转换成模型可使用的tensor形式
batch = batch_data(sample_list, max_len, tokenizer)
# 关掉模型的dropout
model.eval()
# 关掉模型的梯度计算
with paddle.no_grad():
input_ids = batch["input_ids"]
attention_mask = batch["attention_mask"]
# 获取模型预测结果
[pred_label, _] = model.forward(input_ids, attention_mask)
pred_label = pred_label.numpy()
# 将模型预测结果转换成标签
label_name = [id2label[pred] for pred in pred_label]
return zip(sample_list, label_name)
def test(args):
"""对模型(动态图)进行测试"""
# 设置显卡信息
os.environ["CUDA_DEVICE_ORDER"] = "PCI_BUS_ID"
os.environ["CUDA_VISIBLE_DEVICES"] = args.device
# 获取device信息,用于模型训练
device = "gpu:{}".format(args.device) if paddle.fluid.is_compiled_with_cuda() and int(args.device) >= 0 else "cpu"
paddle.device.set_device(device)
# 加载已保存模型,进行模型初始化
model = SentimentAnalysisModel.from_pretrained(args.model_path, number_label=args.num_labels)
# 实例化tokenizer
tokenizer = BertTokenizer(args.vocab_path, do_lower_case=True)
model.to(device)
id2label = {0: "angry", 1: "happy", 2: "neutral", 3: "surprise", 4: "sad", 5: "fear"}
# 计时,记录开始时间
T1 = time.time()
# 对测试集文件进行遍历,单条测试
with open(args.test_path, "r", encoding="utf-8") as fh:
for i, line in enumerate(fh):
if i >= 1000:
continue
sample_list = [json.loads(line)["text"]]
# 单条测试
# sample_list = ["妈妈说想和我聊天,她一定是有难过的事了。。。我要上课,所以我好难过。。"]
result = predict_one_sample(sample_list, model, tokenizer, args.max_len, id2label)
# 打印每个样本的结果
# for sample, label in result:
# print("label: {}, text: {}".format(label, sample))
# 计时,记录开始时间
T2 = time.time()
print("paddle模型,1000次的运行时间为{}秒".format(T2 - T1))
(3)对onnx模型进行测试
def save_onnx_model(args):
"""将paddle模型转成onnx模型"""
# 加载已保存模型,并进行参数初始化
model = SentimentAnalysisModel.from_pretrained(args.model_path, number_label=args.num_labels)
model.eval()
# 定义输入节点input_ids和attention_mask
input_ids = paddle.static.InputSpec([None, args.max_len], "int64", "input_ids")
attention_mask = paddle.static.InputSpec([None, args.max_len], "int64", "attention_mask")
# 使用paddle.onnx.export函数将模型转换成onnx模型,并保持
paddle.onnx.export(model, args.onnx_model_path, input_spec=[input_ids, attention_mask], opset_version=12)
# 检测onnx模型是否可用加载
onnx_model = onnx.load(args.onnx_model_path + ".onnx")
onnx.checker.check_model(onnx_model)
def test_onnx(args):
"""对onnx模型进行测试"""
# 设置显卡信息
os.environ["CUDA_DEVICE_ORDER"] = "PCI_BUS_ID"
os.environ["CUDA_VISIBLE_DEVICES"] = args.device
# 实例化tokenizer
tokenizer = BertTokenizer(args.vocab_path, do_lower_case=True)
id2label = {0: "angry", 1: "happy", 2: "neutral", 3: "surprise", 4: "sad", 5: "fear"}
# 加载onnx模型
ort_sess = onnxruntime.InferenceSession(args.onnx_model_path + ".onnx")
# 计时,记录开始时间
T1 = time.time()
# 对测试集文件进行遍历,单条测试
with open(args.test_path, "r", encoding="utf-8") as fh:
for i, line in enumerate(fh):
if i >= 1000:
continue
sample_list = [json.loads(line)["text"]]
# sample_list = ["妈妈说想和我聊天,她一定是有难过的事了。。。我要上课,所以我好难过。。"]
batch = batch_data(sample_list, args.max_len, tokenizer)
input_ids = batch["input_ids"]
input_ids = input_ids.numpy()
attention_mask = batch["attention_mask"]
attention_mask = attention_mask.numpy()
# 构建onnx所需的feed_dict
ort_inputs = {ort_sess.get_inputs()[0].name: input_ids, ort_sess.get_inputs()[1].name: attention_mask}
# 模型预测
pred_label = ort_sess.run(None, ort_inputs)[0]
# 标签转换
label_name = [id2label[pred] for pred in pred_label]
# 打印每个样本的结果
# for sample, label in zip(sample_list, label_name):
# print("label: {}, text: {}".format(label, sample))
T2 = time.time()
print("onnx模型,1000次的运行时间为{}秒".format(T2 - T1))
需要先将动态图,转成ONNX模型,然后再使用ONNX模型进行预测。
(4)对静态图模型进行测试
def save_static_model(args):
"""将paddle动态图转成静态图"""
# 加载已保存模型,并进行参数初始化
model = SentimentAnalysisModel.from_pretrained(args.model_path, number_label=args.num_labels)
model.eval()
# 定义输入节点input_ids和attention_mask
input_ids = paddle.static.InputSpec(shape=[None, args.max_len], dtype='int64', name='input_ids')
attention_mask = paddle.static.InputSpec(shape=[None, args.max_len], dtype='int64', name='attention_mask')
# 使用paddle.jit.to_static函数,将动态图转成静态图
model = paddle.jit.to_static(model, input_spec=[input_ids, attention_mask])
# 使用静态图进行模型预测
sample_list = ["妈妈说想和我聊天,她一定是有难过的事了。。。我要上课,所以我好难过。。"]
tokenizer = BertTokenizer(args.vocab_path, do_lower_case=True)
batch = batch_data(sample_list, args.max_len, tokenizer)
input_ids = batch["input_ids"]
attention_mask = batch["attention_mask"]
outputs = model(input_ids, attention_mask)
# 对静态进行保存
paddle.jit.save(layer=model, path=args.static_model_path, input_spec=[input_ids, attention_mask])
def test_static(args):
"""对静态图模型进行测试"""
# 设置显卡信息
os.environ["CUDA_DEVICE_ORDER"] = "PCI_BUS_ID"
os.environ["CUDA_VISIBLE_DEVICES"] = args.device
device = "gpu:{}".format(args.device) if paddle.fluid.is_compiled_with_cuda() and int(args.device) >= 0 else "cpu"
paddle.device.set_device(device)
if "gpu" in device:
use_gpu = True
else:
use_gpu = False
# 使用InferenceModel进行模型封装
model = InferenceModel(modelpath=args.static_model_path, use_gpu=use_gpu, use_mkldnn=args.use_mkldnn)
model.eval()
# 实例化tokenizer
tokenizer = BertTokenizer(args.vocab_path, do_lower_case=True)
id2label = {0: "angry", 1: "happy", 2: "neutral", 3: "surprise", 4: "sad", 5: "fear"}
# 计时,记录开始时间
T1 = time.time()
# 对测试集文件进行遍历,单条测试
with open(args.test_path, "r", encoding="utf-8") as fh:
for i, line in enumerate(fh):
if i >=1000:
continue
sample_list = [json.loads(line)["text"]]
# sample_list = ["妈妈说想和我聊天,她一定是有难过的事了。。。我要上课,所以我好难过。。"]
batch = batch_data(sample_list, args.max_len, tokenizer)
input_ids = batch["input_ids"]
attention_mask = batch["attention_mask"]
pred_label = model(input_ids, attention_mask)[0]
label_name = [id2label[pred] for pred in pred_label]
# label_name = [id2label[pred] for pred in pred_label.numpy()]
# 打印每个样本的结果
# for sample, label in zip(sample_list, label_name):
# print("label: {}, text: {}".format(label, sample))
T2 = time.time()
print("paddle静态图,1000次的运行时间为{}秒".format(T2 - T1))
需要先将动态图,转成静态图模型,然后再使用静态图模型进行预测。
测试结果如下:
动态图运行1000次耗时27.93秒,onnx运行1000次耗时10.89秒,静态图运行1000次耗时7.66秒。
可以看出,动态图最慢、静态图最快。其实这里有些超出我的认知,我一直觉得onnx的最快的。不知道是不是跟onnx的版本有关。不过动态图转onnx还是有很多坑的,目前paddlepaddle有很多操作转onnx会报错,所以还是转静态图吧。
总结
往期推荐