BERT+PET方式数据处理介绍

发布于 2025-12-31

4.3 BERT+PET方式数据预处理

基于BERT+PET方式数据预处理介绍

学习目标

了解本项目数据类型和表现格式
掌握数据处理的工具函数代码实现

BERT+PET方式数据预处理

本项目中对数据部分的预处理步骤如下:
查看项目数据集
编写Config类项目文件配置代码
编写数据处理相关代码

1 查看项目数据集

数据存放位置：/Users/***/PycharmProjects/llm/prompt_tasks/PET/data
data文件夹里面包含4个txt文档，分别为：train.txt、dev.txt、prompt.txt、verbalizer.txt

1.1 train.txt

train.txt为训练数据集，其部分数据展示如下：

水果  脆脆的，甜味可以，可能时间有点长了，水分不是很足。
平板  华为机器肯定不错，但第一次碰上京东最糟糕的服务，以后不想到京东购物了。
书籍  为什么不认真的检查一下， 发这么一本脏脏的书给顾客呢！
衣服  手感不错，用料也很好，不知道水洗后怎样，相信大品牌，质量过关，五星好评！！！
水果  苹果有点小，不过好吃，还有几个烂的。估计是故意的放的。差评。
衣服  掉色掉的厉害，洗一次就花了

train.txt一共包含63条样本数据，每一行用\t分开，前半部分为标签(label)，后半部分为原始输入 (用户评论)。

如果想使用自定义数据训练，只需要仿照上述示例数据构建数据集即可。

1.2 dev.txt

dev.txt为验证数据集，其部分数据展示如下：

书籍  "一点都不好笑,很失望,内容也不是很实用"
衣服  完全是一条旧裤子。
手机  相机质量不错，如果阳光充足，可以和数码相机媲美．界面比较人性化，容易使用．软件安装简便
书籍  明明说有货，结果送货又没有了。并且也不告诉我，怎么评啊
洗浴  非常不满意，晚上洗的头发，第二天头痒痒的不行了，还都是头皮屑。
水果  这个苹果感觉是长熟的苹果，没有打蜡，不错，又甜又脆

dev.txt一共包含590条样本数据，每一行用\t分开，前半部分为标签(label)，后半部分为原始输入 (用户评论)。

如果想使用自定义数据训练，只需要仿照上述示例数据构建数据集即可。

1.3 prompt.txt

prompt.txt为人工设定提示模版，其数据展示如下：

这是一条{MASK}评论：{textA}。

其中，用大括号括起来的部分为「自定义参数」，可以自定义设置大括号内的值。

示例中 {MASK} 代表 [MASK] token 的位置，{textA} 代表评论数据的位置。

你可以改为自己想要的模板，例如想新增一个 {textB} 参数：
{textA}和{textB}是{MASK}同的意思。

1.4 verbalizer.txt

verbalizer.txt 主要用于定义「真实标签」到「标签预测词」之间的映射。在有些情况下，将「真实标签」作为 [MASK] 去预测可能不具备很好的语义通顺性，因此，我们会对「真实标签」做一定的映射。
例如：

"中国爆冷2-1战胜韩国"是一则[MASK][MASK]新闻。 体育

这句话中的标签为「体育」，但如果我们将标签设置为「足球」会更容易预测。
因此，我们可以对「体育」这个 label 构建许多个子标签，在推理时，只要预测到子标签最终推理出真实标签即可，如下：

体育 -> 足球,篮球,网球,棒球,乒乓,体育

项目中标签词映射数据展示如下：

电脑  电脑
水果  水果
平板  平板
衣服  衣服
酒店  酒店
洗浴  洗浴
书籍  书籍
蒙牛  蒙牛
手机  手机
电器  电器

verbalizer.txt 一共包含10个类别，上述数据中，我们使用了1对1的verbalizer, 如果想定义一对多的映射，只需要在后面用”,“分割即可， eg:
水果    苹果,香蕉,橘子
若想使用自定义数据训练，只需要仿照示例数据构建数据集

2 编写Config类项目文件配置代码

代码路径：/Users/***/PycharmProjects/llm/prompt_tasks/PET/pet_config.py
config文件目的：配置项目常用变量，一般这些变量属于不经常改变的，比如：训练文件路径、模型训练次数、模型超参数等等

具体代码实现：

# coding:utf-8
import torch
import sys
print(sys.path)

class ProjectConfig(object):
    def __init__(self):
        # 是否使用GPU
        self.device = 'cuda:0' if torch.cuda.is_available() else 'cpu'
        # 预训练模型bert路径
        self.pre_model = '/home/prompt_project/bert-base-chinese'
        self.train_path = '/home/prompt_project/PET/data/train.txt'
        self.dev_path = '/home/prompt_project/PET/data/dev.txt'
        self.prompt_file = '/home/prompt_project/PET/data/prompt.txt'
        self.verbalizer = '/home/prompt_project/PET/data/verbalizer.txt'
        self.max_seq_len = 512
        self.batch_size = 8
        self.learning_rate = 5e-5
        # 权重衰减参数（正则化，抑制模型过拟合）
        self.weight_decay = 0
        # 预热学习率(用来定义预热的步数)
        self.warmup_ratio = 0.06
        self.max_label_len = 2
        self.epochs = 50
        self.logging_steps = 10
        self.valid_steps = 20
        self.save_dir = '/home/prompt_project/PET/checkpoints'


if __name__ == '__main__':
    pc = ProjectConfig()
    print(pc.prompt_file)
    print(pc.pre_model)

3 编写数据处理相关代码

代码路径：/Users/***/PycharmProjects/llm/prompt_tasks/PET/data_handle.
data_handle文件夹中一共包含三个py脚本：template.py、data_preprocess.py、data_loader.py

3.1 template.py

目的：构建固定模版类，text2id的转换
导入必备工具包

# -*- coding:utf-8 -*-
from rich import print # 终端层次显示
from transformers import AutoTokenizer
import numpy as np
import sys
sys.path.append('..')
from pet_config import *

定义HardTemplate类

class HardTemplate(object):
    """
    硬模板，人工定义句子和[MASK]之间的位置关系。
    """

    def __init__(self, prompt: str):
        """
        Args:
            prompt (str): prompt格式定义字符串, e.g. -> "这是一条{MASK}评论：{textA}。"
        """
        self.prompt = prompt
        self.inputs_list = []                       # 根据文字prompt拆分为各part的列表
        self.custom_tokens = set(['MASK'])          # 从prompt中解析出的自定义token集合
        self.prompt_analysis()                         # 解析prompt模板

    def prompt_analysis(self):
        """
        将prompt文字模板拆解为可映射的数据结构。

        Examples:
         prompt -> "这是一条{MASK}评论：{textA}。"
         inputs_list -> ['这', '是', '一', '条', 'MASK', '评', '论', '：', 'textA', '。']
         custom_tokens -> {'textA', 'MASK'}
        """
        idx = 0
        while idx < len(self.prompt):
            str_part = ''
            if self.prompt[idx] not in ['{', '}']:
                self.inputs_list.append(self.prompt[idx])
            if self.prompt[idx] == '{':                  # 进入自定义字段
                idx += 1
                while self.prompt[idx] != '}':
                    str_part += self.prompt[idx]             # 拼接该自定义字段的值
                    idx += 1
            elif self.prompt[idx] == '}':
                raise ValueError("Unmatched bracket '}', check your prompt.")
            if str_part:
                self.inputs_list.append(str_part)
                # 将所有自定义字段存储，后续会检测输入信息是否完整
                self.custom_tokens.add(str_part)
            idx += 1

    def __call__(self,
                 inputs_dict: dict,
                 tokenizer,
                 mask_length,
                 max_seq_len=512):
        """
        输入一个样本，转换为符合模板的格式。

        Args:
            inputs_dict (dict): prompt中的参数字典, e.g. -> {
                                                            "textA": "这个手机也太卡了",
                                                            "MASK": "[MASK]"
                                                        }
            tokenizer: 用于encoding文本
            mask_length (int): MASK token 的长度

        Returns:
            dict -> {
                'text': '[CLS]这是一条[MASK]评论：这个手机也太卡了。[SEP]',
                'input_ids': [1, 47, 10, 7, 304, 3, 480, 279, 74, 47,
                                            27, 247, 98, 105, 512, 777, 15, 12043, 2],
                'token_type_ids': [0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
                                                        0, 0, 0, 0, 0, 0, 0, 0, 0],
                'attention_mask': [1, 1, 1, 1, 1, 1, 1, 1, 1,
                                                    1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
                'mask_position': [0, 0, 0, 0, 0, 1, 0, 0, 0, 0,
                                                    0, 0, 0, 0, 0, 0, 0, 0, 0]
            }
        """
        # 定义输出格式
        outputs = {
            'text': '',
            'input_ids': [],
            'token_type_ids': [],
            'attention_mask': [],
            'mask_position': []
        }

        str_formated = ''
        for value in self.inputs_list:
            if value in self.custom_tokens:
                if value == 'MASK':
                    str_formated += inputs_dict[value] * mask_length
                else:
                    str_formated += inputs_dict[value]
            else:
                str_formated += value
        # print(f'str_formated-->{str_formated}')
        encoded = tokenizer(text=str_formated,
                            truncation=True,
                            max_length=max_seq_len,
                            padding='max_length')
        # print(f'encoded--->{encoded}')
        outputs['input_ids'] = encoded['input_ids']
        outputs['token_type_ids'] = encoded['token_type_ids']
        outputs['attention_mask'] = encoded['attention_mask']
        token_list = tokenizer.convert_ids_to_tokens(encoded['input_ids'])
        outputs['text'] = ''.join(token_list)
        mask_token_id = tokenizer.convert_tokens_to_ids(['[MASK]'])[0]
        condition = np.array(outputs['input_ids']) == mask_token_id
        mask_position = np.where(condition)[0].tolist()
        outputs['mask_position'] = mask_position
        return outputs


if __name__ == '__main__':
    pc = ProjectConfig()
    tokenizer = AutoTokenizer.from_pretrained(pc.pre_model)
    hard_template = HardTemplate(prompt='这是一条{MASK}评论：{textA}')
    print(hard_template.inputs_list)
    print(hard_template.custom_tokens)
    tep = hard_template(
                inputs_dict={'textA': '包装不错，苹果挺甜的，个头也大。', 'MASK': '[MASK]'},
                tokenizer=tokenizer,
                max_seq_len=30,
                mask_length=2)
    print(tep)

    print(tokenizer.convert_ids_to_tokens([3819, 3352]))
    print(tokenizer.convert_tokens_to_ids(['水', '果']))

3.2 data_preprocess.py

目的: 将样本数据转换为模型接受的输入数据
导入必备的工具包

from template import *
from rich import print
from datasets import load_dataset
# partial：把一个函数的某些参数给固定住（也就是设置默认值），返回一个新的函数，调用这个新函数会更简单
from functools import partial
from pet_config import *

定义数据转换方法convert_example()

def convert_example(
        examples: dict,
        tokenizer,
        max_seq_len: int,
        max_label_len: int,
        hard_template: HardTemplate,
        train_mode=True,
        return_tensor=False) -> dict:
    """
    将样本数据转换为模型接收的输入数据。

    Args:
        examples (dict): 训练数据样本, e.g. -> {
                                                "text": [
                                                            '手机 这个手机也太卡了。',
                                                            '体育 世界杯为何迟迟不见宣传',
                                                            ...
                                                ]
                                            }
        max_seq_len (int): 句子的最大长度，若没有达到最大长度，则padding为最大长度
        max_label_len (int): 最大label长度，若没有达到最大长度，则padding为最大长度
        hard_template (HardTemplate): 模板类。
        train_mode (bool): 训练阶段 or 推理阶段。
        return_tensor (bool): 是否返回tensor类型，如不是，则返回numpy类型。

    Returns:
        dict (str: np.array) -> tokenized_output = {
                            'input_ids': [[1, 47, 10, 7, 304, 3, 3, 3, 3, 47, 27,
                                                        247, 98, 105, 512, 777, 15, 12043, 2], ...],
                            'token_type_ids': [[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
                                                                    0, 0, 0, 0, 0, 0, 0, 0], ...],
                            'attention_mask': [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
                                                                    1, 1, 1, 1, 1, 1, 1, 1], ...],
                            'mask_positions': [[5, 6, 7, 8], ...],
                            'mask_labels': [[2372, 3442, 0, 0],
                                                            [2643, 4434, 2334, 0], ...]
                        }
    """
    tokenized_output = {
        'input_ids': [],
        'token_type_ids': [],
        'attention_mask': [],
        'mask_positions': [],
        'mask_labels': []
    }

    for i, example in enumerate(examples['text']):
        if train_mode:
            label, content = example.strip().split('\t')
        else:
            content = example.strip()

        inputs_dict = {
            'textA': content,
            'MASK': '[MASK]'
        }
        encoded_inputs = hard_template(
            inputs_dict=inputs_dict,
            tokenizer=tokenizer,
            max_seq_len=max_seq_len,
            mask_length=max_label_len)
        tokenized_output['input_ids'].append(encoded_inputs["input_ids"])
        tokenized_output['token_type_ids'].append(encoded_inputs["token_type_ids"])
        tokenized_output['attention_mask'].append(encoded_inputs["attention_mask"])
        tokenized_output['mask_positions'].append(encoded_inputs["mask_position"])

        if train_mode:
            label_encoded = tokenizer(text=[label])  # 将label补到最大长度
            # print(f'label_encoded-->{label_encoded}')
            label_encoded = label_encoded['input_ids'][0][1:-1]
            label_encoded = label_encoded[:max_label_len]
            add_pad = [tokenizer.pad_token_id] * (max_label_len - len(label_encoded))
            label_encoded = label_encoded + add_pad
            tokenized_output['mask_labels'].append(label_encoded)

    for k, v in tokenized_output.items():
        if return_tensor:
            tokenized_output[k] = torch.LongTensor(v)
        else:
            tokenized_output[k] = np.array(v)

    return tokenized_output


if __name__ == '__main__':
    pc = ProjectConfig()
    train_dataset = load_dataset('text', data_files=pc.train_path)
    print(type(train_dataset))
    print(train_dataset)
    # print('*'*80)
    # print(train_dataset['train']['text'])
    tokenizer = AutoTokenizer.from_pretrained(pc.pre_model)
    hard_template = HardTemplate(prompt='这是一条{MASK}评论：{textA}')

    convert_func = partial(convert_example,
                           tokenizer=tokenizer,
                           hard_template=hard_template,
                           max_seq_len=30,
                           max_label_len=2)
    dataset = train_dataset.map(convert_func, batched=True)
    for value in dataset['train']:
        print(value)
        print(len(value['input_ids']))
        break

3.3 data_loader.py

目的：定义数据加载器
导入必备的工具包

# coding:utf-8
from torch.utils.data import DataLoader
from transformers import default_data_collator
from data_preprocess import *
from pet_config import *

pc = ProjectConfig() # 实例化项目配置文件
tokenizer = AutoTokenizer.from_pretrained(pc.pre_model)

定义获取数据加载器的方法get_data()

def get_data():
    # prompt定义
    prompt = open(pc.prompt_file, 'r', encoding='utf8').readlines()[0].strip()
    hard_template = HardTemplate(prompt=prompt)  # 模板转换器定义
    dataset = load_dataset('text', data_files={'train': pc.train_path,
                                               'dev': pc.dev_path})
    # print(dataset)
    # print(f'Prompt is -> {prompt}')
    new_func = partial(convert_example,
                       tokenizer=tokenizer,
                       hard_template=hard_template,
                       max_seq_len=pc.max_seq_len,
                       max_label_len=pc.max_label_len)

    dataset = dataset.map(new_func, batched=True)

    train_dataset = dataset["train"]
    dev_dataset = dataset["dev"]
    # print('train_dataset', train_dataset[:2])
    # print('*'*80)
    train_dataloader = DataLoader(train_dataset,
                                  shuffle=True,
                                  collate_fn=default_data_collator,
                                  batch_size=pc.batch_size)
    dev_dataloader = DataLoader(dev_dataset,
                                collate_fn=default_data_collator,
                                batch_size=pc.batch_size)
    return train_dataloader, dev_dataloader


if __name__ == '__main__':
    train_dataloader, dev_dataloader = get_data()
    print(len(train_dataloader))
    print(len(dev_dataloader))
    for i, value in enumerate(train_dataloader):
        print(i)
        print(value)
        print(value['input_ids'].dtype)
        break

发布于 2025-12-31