自定义数据集OCEMOTION–中文情感分类

<br/>

label_map={"sadness":'0', "happiness":'1',"disgust":'2',"like":'3',"anger":'4',"surprise":'5',"fear":'6'}

# 下载paddlenlp!pip install --upgrade paddlenlp==2.0.0b4

# 解压数据%cd ~
!unzip data/data66630/NLP中文预训练模型泛化能力挑战赛.zip -d dataset

import paddleimport paddlenlpprint(paddle.__version__, paddlenlp.__version__)

import osfrom sklearn.model_selection import train_test_splitimport pandas as pdimport numpy as np

def break_data(target, rate=0.2):
    origin_dataset = pd.read_csv("dataset/OCEMOTION_train1128.csv", delimiter="\t", header=None)  # 加入参数
    train_data, test_data = train_test_split(origin_dataset, test_size=rate)
    train_data,eval_data=train_test_split(train_data, test_size=rate)
    train_filename = os.path.join(target, 'train.txt')
    test_filename = os.path.join(target, 'test.txt')
    eval_filename = os.path.join(target, 'eval.txt')

    train_data.to_csv(train_filename, index=False, sep="\t", header=None)
    test_data.to_csv(test_filename, index=False, sep="\t", header=None)
    eval_data.to_csv(eval_filename, index=False, sep="\t", header=None)if __name__ == '__main__':
    break_data(target='dataset', rate=0.2)

%cd ~/dataset/

label_map={"sadness":'0', "happiness":'1',"disgust":'2',"like":'3',"anger":'4',"surprise":'5',"fear":'6'}

import pandas as pddef modify_data(target='.'):
    for name in ['train','test','eval']:
        source_file=os.path.join(target, name + '.txt')
        target_file=os.path.join(target, name + '.csv')
        data=pd.read_csv(source_file, delimiter="\t", header=None) 
        new_data=data[[1,2]]
        new_data.replace(label_map, inplace=True)
        new_data.to_csv(target_file, index=False, sep="\t", header=None)
        new_data=None
        data=Noneif __name__ == '__main__':
    modify_data()

!head train.csv

!head test.csv

!head eval.csv

train_data=pd.read_csv("train.csv", delimiter="\t", header=None) 
print("train_data length: ",train_data.size)
test_data=pd.read_csv("test.csv", delimiter="\t", header=None)  
print("test_data length: ",test_data.size)
eval_data=pd.read_csv("eval.csv", delimiter="\t", header=None)  
print("eval_data length: ",eval_data.size)
df1=train_data[1].value_counts()print(df1)

%cd ~/dataset/

print(label_map)

class SelfDefinedDataset(paddle.io.Dataset):
    def __init__(self, data):
        super(SelfDefinedDataset, self).__init__()
        self.data = data    def __getitem__(self, idx):
        return self.data[idx]    def __len__(self):
        return len(self.data)        
    def get_labels(self):
        return ['0','1','2','3','4','5','6']def txt_to_list(file_name):
    res_list = []    for line in open(file_name):
        res_list.append(line.strip().split('\t'))    return res_list

trainlst = txt_to_list('train.csv')
devlst = txt_to_list('eval.csv')
testlst = txt_to_list('test.csv')# 通过get_datasets()函数，将list数据转换为dataset。# get_datasets()可接收[list]参数，或[str]参数，根据自定义数据集的写法自由选择。# train_ds, dev_ds, test_ds = ppnlp.datasets.ChnSentiCorp.get_datasets(['train', 'dev', 'test'])train_ds, dev_ds, test_ds = SelfDefinedDataset.get_datasets([trainlst, devlst, testlst])

label_list = train_ds.get_labels()print(label_list)for i in range(10):    print (train_ds[i])

# 下载词汇表文件word_dict.txt，用于构造词-id映射关系。from utils import load_vocab, convert_example# !wget https://paddlenlp.bj.bcebos.com/data/senta_word_dict.txt# 加载词表vocab = load_vocab('./senta_word_dict.txt')for k, v in vocab.items():    print(k, v)    break

import numpy as npfrom functools import partialimport paddle.nn as nnimport paddle.nn.functional as Fimport paddlenlp as ppnlpfrom paddlenlp.data import Pad, Stack, Tuplefrom paddlenlp.datasets import MapDatasetWrapperfrom utils import load_vocab, convert_example

# Reads data and generates mini-batches.def create_dataloader(dataset,
                      trans_function=None,
                      mode='train',
                      batch_size=1,
                      pad_token_id=0,
                      batchify_fn=None):
    if trans_function:
        dataset = dataset.apply(trans_function, lazy=True)    # return_list 数据是否以list形式返回
    # collate_fn  指定如何将样本列表组合为mini-batch数据。传给它参数需要是一个callable对象，需要实现对组建的batch的处理逻辑，并返回每个batch的数据。在这里传入的是`prepare_input`函数，对产生的数据进行pad操作，并返回实际长度等。
    dataloader = paddle.io.DataLoader(
        dataset,
        return_list=True,
        batch_size=batch_size,
        collate_fn=batchify_fn)        
    return dataloader# python中的偏函数partial，把一个函数的某些参数固定住（也就是设置默认值），返回一个新的函数，调用这个新函数会更简单。trans_function = partial(
    convert_example,
    vocab=vocab,
    unk_token_id=vocab.get('[UNK]', 1),
    is_test=False)# 将读入的数据batch化处理，便于模型batch化运算。# batch中的每个句子将会padding到这个batch中的文本最大长度batch_max_seq_len。# 当文本长度大于batch_max_seq时，将会截断到batch_max_seq_len；当文本长度小于batch_max_seq时，将会padding补齐到batch_max_seq_len.batchify_fn = lambda samples, fn=Tuple(
    Pad(axis=0, pad_val=vocab['[PAD]']),  # input_ids
    Stack(dtype="int64"),  # seq len
    Stack(dtype="int64")  # label): [data for data in fn(samples)]


train_loader = create_dataloader(
    train_ds,
    trans_function=trans_function,
    batch_size=128,
    mode='train',
    batchify_fn=batchify_fn)
dev_loader = create_dataloader(
    dev_ds,
    trans_function=trans_function,
    batch_size=128,
    mode='validation',
    batchify_fn=batchify_fn)
test_loader = create_dataloader(
    test_ds,
    trans_function=trans_function,
    batch_size=128,
    mode='test',
    batchify_fn=batchify_fn)

class LSTMModel(nn.Layer):
    def __init__(self,
                 vocab_size,
                 num_classes,
                 emb_dim=128,
                 padding_idx=0,
                 lstm_hidden_size=198,
                 direction='forward',
                 lstm_layers=1,
                 dropout_rate=0,
                 pooling_type=None,
                 fc_hidden_size=96):
        super().__init__()        # 首先将输入word id 查表后映射成 word embedding
        self.embedder = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=emb_dim,
            padding_idx=padding_idx)        # 将word embedding经过LSTMEncoder变换到文本语义表征空间中
        self.lstm_encoder = ppnlp.seq2vec.LSTMEncoder(
            emb_dim,
            lstm_hidden_size,
            num_layers=lstm_layers,
            direction=direction,
            dropout=dropout_rate,
            pooling_type=pooling_type)        # LSTMEncoder.get_output_dim()方法可以获取经过encoder之后的文本表示hidden_size
        self.fc = nn.Linear(self.lstm_encoder.get_output_dim(), fc_hidden_size)        # 最后的分类器
        self.output_layer = nn.Linear(fc_hidden_size, num_classes)    def forward(self, text, seq_len):
        # text shape: (batch_size, num_tokens)
        # print('input :', text.shape)
        
        # Shape: (batch_size, num_tokens, embedding_dim)
        embedded_text = self.embedder(text)        # print('after word-embeding:', embedded_text.shape)

        # Shape: (batch_size, num_tokens, num_directions*lstm_hidden_size)
        # num_directions = 2 if direction is 'bidirectional' else 1
        text_repr = self.lstm_encoder(embedded_text, sequence_length=seq_len)        # print('after lstm:', text_repr.shape)


        # Shape: (batch_size, fc_hidden_size)
        fc_out = paddle.tanh(self.fc(text_repr))        # print('after Linear classifier:', fc_out.shape)

        # Shape: (batch_size, num_classes)
        logits = self.output_layer(fc_out)        # print('output:', logits.shape)
        
        # probs 分类概率值
        probs = F.softmax(logits, axis=-1)        # print('output probability:', probs.shape)
        return probs

model= LSTMModel(        len(vocab),        len(label_list),
        direction='bidirectional',
        padding_idx=vocab['[PAD]'])
model = paddle.Model(model)

optimizer = paddle.optimizer.Adam(
        parameters=model.parameters(), learning_rate=5e-5)

loss = paddle.nn.CrossEntropyLoss()
metric = paddle.metric.Accuracy()

model.prepare(optimizer, loss, metric)

# 设置visualdl路径log_dir = './visualdl'callback = paddle.callbacks.VisualDL(log_dir=log_dir)

model.fit(train_loader, dev_loader, epochs=20, save_dir='./checkpoints', save_freq=5, callbacks=callback)

results = model.evaluate(dev_loader)print("Finally test acc: %.5f" % results['acc'])

label_map={0:"sadness", 1:"happiness",2:"disgust",3:"like",4:"anger" ,5:"surprise",6:"fear"}
results = model.predict(test_loader, batch_size=128)[0]
predictions = []for batch_probs in results:    # 映射分类label
    idx = np.argmax(batch_probs, axis=-1)
    idx = idx.tolist()
    labels = [label_map[i] for i in idx]
    predictions.extend(labels)# 看看预测数据前5个样例分类结果for idx, data in enumerate(test_ds.data[:10]):    print('Data: {} \t Label: {}'.format(data[0], predictions[idx]))

<br/>