Press "Enter" to skip to content

基于手动离散Prompt的MLM文本分类

本站内容均来自兴趣收集,如不慎侵害的您的相关权益,请留言告知,我们将尽快删除.谢谢.

本文旨在记录手动离散Prompt做MLM文本分类的代码,实验环境Colab

 

环境及数据

 

!pip install transformers
!pip install datasets
import os
import logging
import datasets
import transformers
import numpy as np
from sklearn import metrics
from datasets import Dataset
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, recall_score, precision_score, f1_score, precision_recall_fscore_support
from transformers import Trainer, TrainingArguments, BertTokenizer, BertForMaskedLM
transformers.set_seed(1)
logging.basicConfig(level=logging.INFO)

 

数据集是我自己随便编的48条天气状况三分类数据集,天气描述和标签之间用\t
分隔

 

昨天天气还行吧,不算好也不算坏    良
今天是艳阳天    好
明天多云转晴    好
后天有阵雨,真烦    差
明天天气不好不坏吧    良
明天有大暴雨    差
明天是晴天哦    好
今天阳光有点刺眼    差
明天是雨夹雪    差
昨天下冰雹了    差
现在也就只有点毛毛雨而已,其实还好    良
终于下雨了,干旱好久了    好
好像起雾了,开车很危险    差
雨过天晴,天上出现彩虹了    好
今天外面好大的风,把电线杆都吹倒了    差
草原上吹来一阵又一阵微风,很舒服    好
天气预报显示近期将有强风过境,大家注意安全    差
好可怕啊,外面打雷了    差
快看,出现闪电了,注意用电设备    差
台风过境,寸草不生    差
雷暴雨马上就要来了,大家快回家,别在外面玩了    差
美国加州被飓风侵袭,损失惨重    差
发洪水了,庄稼惨了    差
早上起床我看到地面接了一层霜    差
天上下冰雹了,砸的人好疼    差
面前吹来一阵轻柔的风    好
今天不冷也不热,温度非常适宜    好
天气晴朗,万里无云    好
明天是阴天,今天是晴天    好
明天是阴天,今天是晴天    差
空气有点湿润    好
地上泛起滴滴雨点,小草获得了滋润    好
现在气温25摄氏度    好
春暖花开,万物欣欣向荣    好
外面冰雪严寒,别出去    差
外面碧空万里,出去玩吧    好
乍暖乍寒,真搞不懂这天气    差
今天也是风和日丽的一天捏    好
出去一会儿感觉寒风侵肌    差
实在是太热了,出去一会儿就汗如雨下    差
刚打开水龙头就滴水成冰,太冷了    差
皓月千里    好
日和风暖的一天    好
爱就像蓝天白云,晴空万里    好
细雨斜风敲打着我的心房    良
好热好热好热好热    差
今天日丽风清    好
天寒地冻的世界,动物必须冬眠    差

 

可以看到自己编的数据还是有点乱的,而且样本及其不均衡,天气好和天气差的样本分别有20和24条,天气良的样本只有4条,因为我实在不知道有什幺样的天气描述算是「良」,而且我在中间添加了两条非常特殊的样本

 

明天是阴天,今天是晴天    好
明天是阴天,今天是晴天    差

 

同样的描述,但是标签却完全相反,这主要是为了测试在数据有噪声的情况下模型的效果,算是变相给模型负重训练了

 

Prompt设计

 

我设计的Prompt格式为:天气[Z],[X]
。例如给定上面数据的最后一行

 

天寒地冻的世界,动物必须冬眠    差

 

经过Prompt转换后,Input就是天气[MASK],天寒地冻的世界,动物必须冬眠
,而labels是天气差,天寒地冻的世界,动物必须冬眠

 

代码详解

 

代码主要分为数据处理和定义Trainer
两部分,Trainer
是Huggingface提供的一种封装性比较强的函数,具体可以看这个API文档
。从这开始,代码讲解都是代码在上,讲解在下,特此声明

 

data_show()

 

# 原始样本统计
def data_show(self, data_file='./data.txt'):
    with open(data_file, 'r', encoding='utf-8') as f:
        data = f.readlines()
    logging.info("获取数据:%s" % len(data)) # 获取数据:48
    tags_data_dict = {}
    for line in data:
        text_label = line.strip().split('\t')
        if text_label[1] in tags_data_dict:
            tags_data_dict[text_label[1]].append(text_label[0])
        else:
            tags_data_dict[text_label[1]] = [text_label[0]]
    logging.info("其中,各分类数量:")
    for k, v in tags_data_dict.items():
        logging.info("%s: %s" % (k, len(v)))
    return tags_data_dict

 

logging.info()
函数并不是将结果写入到某个文件中,与print()
类似,都是将结果输出到控制台。其中tags_data_dict
输出的结果大概如下所示:

 

{'良': ['昨天天气xxx', '明天天气xxx',...], '好': ['明天是xxx',...], '差':['天寒地冻xxx',...]}

 

这里我要声明一点,这个函数将字典return出来仅仅只是为了帮助大家调试打印查看,后面的代码完全不需要使用tags_data_dict

 

data_process()

 

def data_process(self, data_file='./data.txt'):
    with open(data_file, 'r', encoding='utf-8') as f:
        data = [line.strip().split('\t') for line in f.readlines()]  # 今天阳光明媚 好
    text = ['天气[MASK],'+_[0] for _ in data]
    label = ['天气' + _[1]+','+_[0] for _ in data]
    return text, label

 

如果打印text
label
,它们的结果大概如下所示(第一行是text
,第二行是label
):

 

['天气[MASK],今天下雨了', '天气[MASK],今天是艳阳天', '天气[MASK],后天有阵雨,真烦',...]
['天气差,今天下雨了', '天气好,今天是艳阳天', '天气差,后天有阵雨,真烦']

 

create_model_tokenizer()

 

def create_model_tokenizer(self, model_name='bert-base-chinese'):
    tokenizer = BertTokenizer.from_pretrained(model_name)
    model = BertForMaskedLM.from_pretrained(model_name)
    return tokenizer, model

 

这个函数就没什幺好说的了,如果这里您看不懂,请先去学习一下如何使用Huggingface

 

create_dataset()

 

# 构建dataset
def create_dataset(self, text, label, tokenizer, max_len):
    X_train, X_test, Y_train, Y_test = train_test_split(text, label, test_size=0.4, random_state=1)
    logging.info('训练集:%s条,
测试集:%s条' %(len(X_train), len(X_test)))
    train_dict = {'text': X_train, 'label_text': Y_train}
    test_dict = {'text': X_test, 'label_text': Y_test}
    train_dataset = Dataset.from_dict(train_dict)
    # print(train_dataset[:2]) # {'text': ['天气[MASK],实在是太热了,出去一会儿就汗如雨下', '天气[MASK],好可怕啊,外面打雷了'], 'label_text': ['天气差,实在是太热了,出去一会儿就汗如雨下', '天气差,好可怕啊,外面打雷了']}
    test_dataset = Dataset.from_dict(test_dict)
    def preprocess_function(examples):
        text_token = tokenizer(examples['text'], padding=True, truncation=True, max_length=max_len) # add three columns: attetnion_mask, input_ids, token_type_ids
        # print(tokenizer(examples['label_text'], padding=True,truncation=True, max_length=max_len)) # {'input_ids': [[101, ...], [101 ...], ...], 'attention_mask': [[], []], 'token_type_ids': [[], []]}
        text_token['labels'] = tokenizer(examples['label_text'], padding=True,truncation=True, max_length=max_len)["input_ids"]
        # print(text_token['labels']) # [[101, 1921, 3698, ...], [101, 1921, 3698, ...], ...]
        return text_token
    train_dataset = train_dataset.map(preprocess_function, batched=True, remove_columns=['text', 'label_text'])
    # print(train_dataset[:1]) # {'attention_mask': [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0]], 'input_ids': [[101, 1921, 3698, 103, 8024, 2141, 1762, 3221, 1922, 4178, 749, 8024, 1139, 1343, 671, 833, 1036, 2218, 3731, 1963, 7433, 678, 102, 0, 0, 0, 0, 0]], 'labels': [[101, 1921, 3698, 2345, 117, 2141, 1762, 3221, 1922, 4178, 749, 8024, 1139, 1343, 671, 833, 1036, 2218, 3731, 1963, 7433, 678, 102, 0, 0, 0, 0, 0]], 'token_type_ids': [[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]]}
    test_dataset = test_dataset.map(preprocess_function, batched=True, remove_columns=['text', 'label_text'])
    return train_dataset, test_dataset

 

实际上我在代码中添加了很多注释帮助大家理解每个变量保存了什幺东西,但是有些东西我也是第一次使用,因此这里就仔细讲解一下吧

 

首先是很多人会有的疑问,为什幺构建Dataset没有使用传统PyTorch的方法(重写三个函数__init__()
__getitem__()
__len__()
),实际上使用传统方法也是可以的,效果没有差别,但是秉承着学习的态度,我们应该接受新的东西

 

一开始先拆分一下训练集和验证集,然后我们需要定义一个字典类型的元素,key取什幺名字并不重要。接着使用Dataset.from_dict()
函数将其转为Dataset类型,于是我们就有了train_dataset
test_dataset
。但是现在你随便打印其中一个dataset会发现它仍然还是文本类型,没有转换为索引,所以我们需要通过一些方法将它转换为索引。实际上train_dataset.map()
这个函数本身没有做什幺很重要的事,你可以在map()
前后分别打印train_dataset
的类型看一下,变量类型没有任何变化。重要的是map()
函数中需要你提供一个参数,这个参数是一个回调函数,也就是上面代码中定义的preprocess_function()
,这个函数所做的事就是通过tokenizer
将文本转换为索引,得到attetnion_mask
input_ids
token_type_ids
(有些模型不需要这个),而label的构建其实就是将label_text
通过tokenizer
之后,取出input_ids
即可。需要注意的是map()
函数会将得到的结果「附在」原本的dataset中,例如原来train_dataset
里只有text
label_text
这两项,经过map()
之后就有text
label_text
attention_mask
input_ids
token_type_ids
这五项。当然了,text
label_text
现在没有什幺用了,因此可以通过指定参数remove_columns=['text', 'label_text']
,把它们删掉,但是实际上不删掉也没有关系。最后给出map()
这个函数的API
,大家可以看一下里面参数的具体描述

 

上面这段代码总而言之非常重要,但其中我认为最重要的是text_token['labels']
,它给label的键值起名为labels
,这个名字非常重要,Huggingface寻找label的时候就是按照labels
这个关键字来寻找的,如果不是labels
,模型就找不到标签,最后就会报错!

 

create_trainer()

 

def create_trainer(self, model, train_dataset, test_dataset, checkpoint_dir, batch_size):
    args = TrainingArguments(
        output_dir=checkpoint_dir,
        overwrite_output_dir=True,
        evaluation_strategy = "epoch",
        save_strategy = 'epoch',
        learning_rate=2e-5,
        per_device_train_batch_size=batch_size,
        per_device_eval_batch_size=batch_size,
        gradient_accumulation_steps=1,
        num_train_epochs=20,
        weight_decay=0.01,
        save_total_limit=1, # save the best model
        load_best_model_at_end=True,
        metric_for_best_model='f1',
    )
    def compute_metrics(pred):
        labels = pred.label_ids[:, 3]
        # print(labels) # [1962, 2354, 2345, ...]
        # print(tokenizer.decode(labels))
        preds = pred.predictions[:, 3].argmax(-1)
        # print(preds) # [1962, 1107, 1107, 1107, 3252, ...]
        # print(tokenizer.decode(preds))
        precision, recall, f1, _ = precision_recall_fscore_support(labels, preds, average='weighted')
        acc = accuracy_score(labels, preds)
        return {'accuracy': acc, 'f1': f1, 'precision': precision, 'recall': recall}
    trainer = Trainer(
        model,
        args,
        train_dataset=train_dataset,
        eval_dataset=test_dataset,
        compute_metrics=compute_metrics
    )
    return trainer

 

这部分代码非常长,但实际上都是为Trainer
服务的。我们看到Tranier
中的参数args
,其实就是由TrainingArguments
提供的,而TrainingArguments
里面又有许多参数,详细的参数信息可以看这里的API文档
,这里我简要说明一下,output_dir
指的是模型保存的路径;overwrite_output_dir
意思是如果保存的路径已经有模型文件了,是否覆盖;evaluation_strategy
顾名思义就是问你什幺时候进行验证,可选参数有no
steps
epoch
,如果选择steps
,那就必须和参数eval_steps
一块儿使用;save_strategy
就是问你什幺时候保存,save_strategy
可选的参数和evaluation_strategy
一样,并且他俩在设置的时候也必须一摸一样,否则就会报错(save_strategy
默认值是steps
);gradient_accumulation_steps
是为了帮助你实现梯度累积的参数,至于为什幺要梯度累积,可以看我的这篇文章;save_total_limit
是保留几个模型文件的参数,在训练过程中,每个epoch我们都会保存一个模型文件,但这实际上是没有必要的(主要是硬盘容量是有限的),设置这个参数之后,路径中就只会有一个模型文件……实际上是两个,还有一个是在验证集上表现最好的模型文件;load_best_model_at_end
就是在模型训练完之后,自动帮你加载最优的模型,那幺我们应该有一个评测指标,来定义「最优」;metric_for_best_model
就是定义「最优」的参数,可以看到,我们认为f1值越高的模型越好

 

Trainer()
中最后一个参数compute_metrics
接收一个回调函数,这个函数用于计算验证集上的各项指标,返回一个字典类型的元素

 

main()

 

def main():
    lct = LecCallTag()
    data_file = './data.txt'
    checkpoint_dir = "./checkpoint/"
    batch_size = 16
    max_len = 64
    n_label = 3
    tags_data_dict = lct.data_show(data_file)
    text, label = lct.data_process(data_file)
    tokenizer, model = lct.create_model_tokenizer("hfl/chinese-roberta-wwm-ext")
    train_dataset, test_dataset = lct.create_dataset(text, label, tokenizer, max_len)
    trainer = lct.create_trainer(model, train_dataset, test_dataset, checkpoint_dir, batch_size)
    trainer.train()
if __name__ == '__main__':
    main()

 

除了main()
函数以外,上面的所有函数都定义在类LecCallTag
中,最后给出完整代码链接
,这里面也有Inference部分的代码

Be First to Comment

发表评论

您的电子邮箱地址不会被公开。 必填项已用*标注