其他
带有详细注释的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.pyorpython3 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_maskdef 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会报错，所以还是转静态图吧。
总结
往期推荐