NLP基础

【了解】今日内容介绍

• 中文分类任务
• 中文填空任务

【区别】自动模型和指定模型

#1.自动模型：
tokenizer：AutoTokenizer
model：AutoModel/ AutoModelForXXX

#2.指定模型（具体模型）
Tokenizer：XXXTokenizer，XXX就是模型名称，比如BertTokenizer
model：XXX / XXXForYYY，XXX就是模型名称，比如BertForTokenClassification

微调怎么做？

之前的预训练模型中，使用方式：

把数据送给模型，直接出结果（除了特征提取任务外）

因此，也叫开箱即用。

微调，顾名思义，需要稍微调整一下模型。因此需要有一些模型结果的变化：

预训练模型 + 下游线性层

# 解释
预训练模型：专门用来做特征提取
下游线性层：用来实现多分类（开发者自定义），因此非常


# 结论
部分参数微调：微调下游自定义线性层参数。
全参微调：预训练模型参数和自定义下游线性层参数都会改变。

【实现】中文分类任务

需求说明

直接加载预训练模型进行输入文本的特征表示, 后接自定义网络进行微调输出结果。

label,text
1,这个宾馆比较陈旧了，特价的房间也很一般。总体来说一般
1,交通方便；环境很好；服务态度很好 房间较小

获取数据

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from transformers import BertModel, BertTokenizer, AdamW
from datasets import load_dataset
from tqdm import tqdm

# 加载预训练模型
bert_tokenizer = BertTokenizer.from_pretrained("../model/bert-base-chinese")
bert_model = BertModel.from_pretrained("../model/bert-base-chinese")


# 批量数据处理函数（二次数据处理函数）
def collate_fn(data):
    # data：[{"label":1, "text":"xxxx"}, {},....]
    # print(data)
    inputs = [i['text'] for i in data]
    labels = [i['label'] for i in data]
    # print(inputs)
    # print(labels)
    inputs = bert_tokenizer.batch_encode_plus(inputs, padding='max_length', truncation=True, max_length=32, return_tensors='pt')
    input_ids = inputs['input_ids']
    attention_mask = inputs['attention_mask']
    token_type_ids = inputs['token_type_ids']
    labels = torch.LongTensor(labels)
    return input_ids, attention_mask, token_type_ids, labels


# 准备dataloader
def get_dataloader():
    #1.加载数据
    dataset = load_dataset("csv", data_files='../data/train.csv', split="train")
    dataset = dataset.filter(lambda x:len(x['text']) > 32)
    #2.构建dataloader
    my_dataloader = DataLoader(dataset, batch_size=8, shuffle=True, collate_fn=collate_fn, drop_last=True)

    # input_ids, attention_mask, token_type_ids, labels = next(iter(my_dataloader))
    # print("input_ids-->", input_ids, input_ids.shape)
    # print("attention_mask-->", attention_mask, attention_mask.shape)
    # print("token_type_ids-->", token_type_ids, token_type_ids.shape)
    # print("labels-->", labels, labels.shape)
    #3.返回dataloader
    return my_dataloader


if __name__ == '__main__':
    get_dataloader()

定义模型

# 定义模型
class MyModel(nn.Module):
    def __init__(self):
        super(MyModel, self).__init__()
        self.linear = nn.Linear(768, 2)

    def forward(self, input_ids, attention_mask, token_type_ids):
        # bert模型要接收这3个参数
        with torch.no_grad():
            bert_out = bert_model(input_ids, attention_mask, token_type_ids)
        # print("bert_out-->", bert_out['last_hidden_state'].shape)
        # print("bert_out-->", bert_out['pooler_output'].shape)
        # output = self.linear(bert_out['pooler_output']) # 直接使用pooler_output，这就是句子的向量表示
        output = self.linear(bert_out['last_hidden_state'][:, 0]) # 取第一个位置的词向量表示，代表该条样本的结果
        # print("output-->", output, output.shape)
        return output



# 测试模型
def test_my_model():
    my_dataloader = get_dataloader()
    my_model = MyModel()
    # input_ids, attention_mask, token_type_ids, labels = next(iter(my_dataloader))
    for input_ids, attention_mask, token_type_ids, labels in my_dataloader:
        output = my_model(input_ids, attention_mask, token_type_ids)
        print("output-->", output)
        break


if __name__ == '__main__':
    # get_dataloader()
    test_my_model()

模型训练

# 模型训练
def train_model():
    # 准备数据
    my_dataloader = get_dataloader()
    # 准备模型
    my_model = MyModel()
    # 准备损失函数，优化器
    my_loss_fn = nn.CrossEntropyLoss()
    my_optimizer = AdamW(params=my_model.parameters(), lr=5e-4)

    for p in bert_model.parameters():
        p.requires_grad_(False)

    epochs = 1
    my_model.train()

    for epoch in range(epochs):
        for i,(input_ids, attention_mask, token_type_ids, labels) in enumerate(my_dataloader):
            out = my_model(input_ids=input_ids, attention_mask=attention_mask, token_type_ids=token_type_ids)
            loss = my_loss_fn(out, labels)
            my_optimizer.zero_grad()
            loss.backward()
            my_optimizer.step()

            # 打印观测值
            if i % 10 == 0:
                tmp = torch.argmax(out, dim=-1)
                acc = (tmp == labels).sum().item() / len(labels)
                print('轮次为-->',i, '\t准确率为-->',acc)

            # 保存模型
            torch.save(my_model.state_dict(), '../save/text_classify_%d.bin' % epoch)


if __name__ == '__main__':
    # get_dataloader()
    # test_my_model()
    train_model()

模型评估

# 模型评估
def evalute_model():
    # 1.加载数据
    test_dataset = load_dataset("csv", data_files='../data/test.csv', split='train')
    test_dataloader = DataLoader(test_dataset, batch_size=8, shuffle=True, drop_last=True, collate_fn=collate_fn)
    # 2.加载模型
    my_model = MyModel()
    my_model.load_state_dict(torch.load("../save/text_classify_1.bin"))

    epochs = 1
    my_model.eval()

    total = 0
    accuracy = 0

    for i, (input_ids, attention_mask, token_type_ids, labels) in enumerate(tqdm(test_dataloader)):
        with torch.no_grad():
            my_out = my_model(input_ids=input_ids, attention_mask=attention_mask, token_type_ids=token_type_ids)
        # print("my_out-->", my_out, my_out.shape)

        out = torch.argmax(my_out, dim=-1)
        acc = (out == labels).sum().item() / len(labels)
        accuracy += acc
        total += len(labels)

        if i % 10 == 0:
            print('轮次为-->', i, '\t准确率为-->', acc, end=' ')
            str1 = bert_tokenizer.decode(input_ids[0], skip_special_tokens=True)
            print(str1, end=' ')
            predict_label = out[0].item()
            true_label = labels[0].item()
            print("predict_label-->", predict_label, "\ttrue_label-->", true_label)


if __name__ == '__main__':
    # get_dataloader()
    # test_my_model()
    # train_model()
    evalute_model()

所有代码

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from transformers import BertModel, BertTokenizer, AdamW
from datasets import load_dataset
from tqdm import tqdm

# 加载预训练模型
bert_tokenizer = BertTokenizer.from_pretrained("../model/bert-base-chinese")
bert_model = BertModel.from_pretrained("../model/bert-base-chinese")


# 批量数据处理函数（二次数据处理函数）
def collate_fn(data):
    # data：[{"label":1, "text":"xxxx"}, {},....]
    # print(data)
    inputs = [i['text'] for i in data]
    labels = [i['label'] for i in data]
    # print(inputs)
    # print(labels)
    inputs = bert_tokenizer.batch_encode_plus(inputs, padding='max_length', truncation=True, max_length=32, return_tensors='pt')
    input_ids = inputs['input_ids']
    attention_mask = inputs['attention_mask']
    token_type_ids = inputs['token_type_ids']
    labels = torch.LongTensor(labels)
    return input_ids, attention_mask, token_type_ids, labels


# 准备dataloader
def get_dataloader():
    #1.加载数据
    dataset = load_dataset("csv", data_files='../data/train.csv', split="train")
    dataset = dataset.filter(lambda x:len(x['text']) > 32)
    #2.构建dataloader
    my_dataloader = DataLoader(dataset, batch_size=8, shuffle=True, collate_fn=collate_fn, drop_last=True)

    # input_ids, attention_mask, token_type_ids, labels = next(iter(my_dataloader))
    # print("input_ids-->", input_ids, input_ids.shape)
    # print("attention_mask-->", attention_mask, attention_mask.shape)
    # print("token_type_ids-->", token_type_ids, token_type_ids.shape)
    # print("labels-->", labels, labels.shape)
    #3.返回dataloader
    return my_dataloader


# 定义模型
class MyModel(nn.Module):
    def __init__(self):
        super(MyModel, self).__init__()
        self.linear = nn.Linear(768, 2)

    def forward(self, input_ids, attention_mask, token_type_ids):
        # bert模型要接收这3个参数
        with torch.no_grad():
            bert_out = bert_model(input_ids, attention_mask, token_type_ids)
        # print("bert_out-->", bert_out['last_hidden_state'].shape)
        # print("bert_out-->", bert_out['pooler_output'].shape)
        # output = self.linear(bert_out['pooler_output']) # 直接使用pooler_output，这就是句子的向量表示
        output = self.linear(bert_out['last_hidden_state'][:, 0]) # 取第一个位置的词向量表示，代表该条样本的结果
        # print("output-->", output, output.shape)
        return output



# 测试模型
def test_my_model():
    my_dataloader = get_dataloader()
    my_model = MyModel()
    # input_ids, attention_mask, token_type_ids, labels = next(iter(my_dataloader))
    for input_ids, attention_mask, token_type_ids, labels in my_dataloader:
        output = my_model(input_ids, attention_mask, token_type_ids)
        print("output-->", output)
        break

# 模型训练
def train_model():
    # 准备数据
    my_dataloader = get_dataloader()
    # 准备模型
    my_model = MyModel()
    # 准备损失函数，优化器
    my_loss_fn = nn.CrossEntropyLoss()
    my_optimizer = AdamW(params=my_model.parameters(), lr=5e-4)

    for p in bert_model.parameters():
        p.requires_grad_(False)

    epochs = 1
    my_model.train()

    for epoch in range(epochs):
        for i,(input_ids, attention_mask, token_type_ids, labels) in enumerate(my_dataloader):
            out = my_model(input_ids=input_ids, attention_mask=attention_mask, token_type_ids=token_type_ids)
            loss = my_loss_fn(out, labels)
            my_optimizer.zero_grad()
            loss.backward()
            my_optimizer.step()

            # 打印观测值
            if i % 10 == 0:
                tmp = torch.argmax(out, dim=-1)
                acc = (tmp == labels).sum().item() / len(labels)
                print('轮次为-->',i, '\t准确率为-->',acc)

            # 保存模型
            torch.save(my_model.state_dict(), '../save/text_classify_%d.bin' % epoch)


# 模型评估
def evalute_model():
    # 1.加载数据
    test_dataset = load_dataset("csv", data_files='../data/test.csv', split='train')
    test_dataloader = DataLoader(test_dataset, batch_size=8, shuffle=True, drop_last=True, collate_fn=collate_fn)
    # 2.加载模型
    my_model = MyModel()
    my_model.load_state_dict(torch.load("../save/text_classify_1.bin"))

    epochs = 1
    my_model.eval()

    total = 0
    accuracy = 0

    for i, (input_ids, attention_mask, token_type_ids, labels) in enumerate(tqdm(test_dataloader)):
        with torch.no_grad():
            my_out = my_model(input_ids=input_ids, attention_mask=attention_mask, token_type_ids=token_type_ids)
        # print("my_out-->", my_out, my_out.shape)

        out = torch.argmax(my_out, dim=-1)
        acc = (out == labels).sum().item() / len(labels)
        accuracy += acc
        total += len(labels)

        if i % 10 == 0:
            print('轮次为-->', i, '\t准确率为-->', acc, end=' ')
            str1 = bert_tokenizer.decode(input_ids[0], skip_special_tokens=True)
            print(str1, end=' ')
            predict_label = out[0].item()
            true_label = labels[0].item()
            print("predict_label-->", predict_label, "\ttrue_label-->", true_label)



if __name__ == '__main__':
    # get_dataloader()
    # test_my_model()
    # train_model()
    evalute_model()

【实现】中文填空任务

需求说明

完型填空任务又被叫做“遮蔽语言建模任务”。

数据文件有三个train.csv，test.csv，validation.csv，数据样式都是一样的。

label,text
1,这个宾馆比较陈旧了，特价的房间也很一般。总体来说一般
1,交通方便；环境很好；服务态度很好 房间较小

获取数据

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from transformers import BertTokenizer, BertModel, AdamW
from datasets import load_dataset

tokenizer = BertTokenizer.from_pretrained("../model/bert-base-chinese")
bert_model = BertModel.from_pretrained("../model/bert-base-chinese")

# 批量处理函数
def collate_fn(data):
    # data就是一批数据
    sents = [i['text'] for i in data]

    # 编码
    inputs = tokenizer.batch_encode_plus(sents, padding='max_length',
                                         truncation=True,
                                         max_length=32,
                                         return_tensors='pt')
    input_ids = inputs['input_ids']
    attention_mask = inputs['attention_mask']
    token_type_ids = inputs['token_type_ids']

    # 取出下标为16的值（标签），并且把该位置mask
    labels = input_ids[:, 16].reshape(-1).clone()
    labels = torch.LongTensor(labels)
    input_ids[:, 16] = tokenizer.get_vocab()[tokenizer.mask_token]

    # 返回结果
    return input_ids, attention_mask, token_type_ids, labels


# 获取数据集
def get_dataloader():
    my_datasets = load_dataset('csv', data_files="../data/train.csv", split='train')
    my_datasets = my_datasets.filter(lambda x: len(x['text']) > 32)
    dataloader = DataLoader(my_datasets, batch_size=8, shuffle=True, collate_fn=collate_fn, drop_last=True)

    # input_ids, attention_mask, token_type_ids,labels = next(iter(dataloader))
    # for i, (input_ids, attention_mask, token_type_ids, labels) in enumerate(dataloader):
    #     print("input_ids-->", input_ids, input_ids.shape)
    #     print('attention_mask-->', attention_mask, attention_mask.shape)
    #     print('token_type_ids-->', token_type_ids, token_type_ids.shape)
    #     print('labels-->', labels, labels.shape)
    #     break
    return dataloader


if __name__ == '__main__':
    get_dataloader()

定义模型

# 定义模型
class MyModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear = nn.Linear(in_features=768, out_features=tokenizer.vocab_size, bias=False)

    def forward(self, input_ids, attention_mask, token_type_ids):
        with torch.no_grad():
            output = bert_model(input_ids, attention_mask, token_type_ids)
            # last_hidden_state：词向量， pooler_output：句向量
            # print('output-->', output)
        # 千万注意不要写在with里面
        out = self.linear(output.last_hidden_state[:, 16])
        return out


def test_mymodel():
    dataloader = get_dataloader()
    my_model = MyModel()
    for i, (input_ids, attention_mask, token_type_ids, labels) in enumerate(dataloader):
        out = my_model(input_ids, attention_mask, token_type_ids)
        print(out, out.shape)
        break


if __name__ == '__main__':
    # get_dataloader()
    test_mymodel()

模型训练

# 训练模型
def train_model():
    # 准备物料 模型，损失函数，优化器
    my_dataloader = get_dataloader()
    my_model = MyModel()
    my_loss = nn.CrossEntropyLoss()
    my_optimizer = AdamW(params=my_model.parameters(), lr=5e-4)
    epochs = 1
    for p in bert_model.parameters():
        p.requires_grad_(False)

    # 训练模型
    my_model.train()
    for epoch in range(epochs):
        for i, (input_ids, attention_mask, token_type_ids, labels) in enumerate(my_dataloader):
            output = my_model(input_ids, attention_mask, token_type_ids)
            # print("output-->", output, output.shape)
            # print("label-->", labels, labels.shape)
            loss = my_loss(output, labels)
            my_optimizer.zero_grad()
            loss.backward()
            my_optimizer.step()

            if i % 10 == 0:
                tmp = torch.argmax(output, dim=-1)
                acc = (tmp == labels).sum().item() / len(labels)
                print("i-->", i, "\tloss-->", loss.item(), "\t准确率-->", acc)
        # 保存模型
        torch.save(my_model.state_dict(), "../save/day11/my_model_%d.bin" % epoch)


if __name__ == '__main__':
    # get_dataloader()
    # test_mymodel()
    train_model()

模型评估

加载训练好的模型，换一份数据集，看看模型的表现情况。

# 预测模型
def evalute_model():
    # 1.准备数据
    test_dataset = load_dataset("csv", data_files="../data/test.csv", split='train')
    test_dataset = test_dataset.filter(lambda x:len(x['text']) > 32)
    test_dataloader = DataLoader(test_dataset, batch_size=8, shuffle=True, drop_last=True, collate_fn=collate_fn)

    #2.准备模型
    my_model = MyModel()
    my_model.load_state_dict(torch.load("../save/fill_mask_1.bin"))

    #3.预测
    my_model.eval()
    accuracy = 0
    total = 0
    for i,(input_ids, attention_mask, input_type_ids, labels) in enumerate(test_dataloader):
        with torch.no_grad():
            output = my_model(input_ids, attention_mask, input_type_ids)

        output = torch.argmax(output, dim=-1)
        accuracy += (output == labels).sum().item()
        total += len(labels)

        # 打印观测值
        if i % 10 == 0:
            print('平均准确率为-->\t',accuracy/total)
    print("模型总准确率为-->", accuracy/total)


if __name__ == '__main__':
    # get_dataloader()
    # test_mymodel()
    # train_model()
    evalute_model()

所有代码

该案例的所有代码汇总如下：

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from transformers import BertTokenizer, BertModel, AdamW
from datasets import load_dataset

tokenizer = BertTokenizer.from_pretrained("../model/bert-base-chinese")
bert_model = BertModel.from_pretrained("../model/bert-base-chinese")

# 批量处理函数
def collate_fn(data):
    # data就是一批数据
    sents = [i['text'] for i in data]

    # 编码
    inputs = tokenizer.batch_encode_plus(sents, padding='max_length',
                                         truncation=True,
                                         max_length=32,
                                         return_tensors='pt')
    input_ids = inputs['input_ids']
    attention_mask = inputs['attention_mask']
    token_type_ids = inputs['token_type_ids']

    # 取出下标为16的值（标签），并且把该位置mask
    labels = input_ids[:, 16].reshape(-1).clone()
    labels = torch.LongTensor(labels)
    input_ids[:, 16] = tokenizer.get_vocab()[tokenizer.mask_token]

    # 返回结果
    return input_ids, attention_mask, token_type_ids, labels


# 获取数据集
def get_dataloader():
    my_datasets = load_dataset('csv', data_files="../data/train.csv", split='train')
    my_datasets = my_datasets.filter(lambda x: len(x['text']) > 32)
    dataloader = DataLoader(my_datasets, batch_size=8, shuffle=True, collate_fn=collate_fn, drop_last=True)

    # input_ids, attention_mask, token_type_ids,labels = next(iter(dataloader))
    # for i, (input_ids, attention_mask, token_type_ids, labels) in enumerate(dataloader):
    #     print("input_ids-->", input_ids, input_ids.shape)
    #     print('attention_mask-->', attention_mask, attention_mask.shape)
    #     print('token_type_ids-->', token_type_ids, token_type_ids.shape)
    #     print('labels-->', labels, labels.shape)
    #     break
    return dataloader


# 定义模型
class MyModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear = nn.Linear(in_features=768, out_features=tokenizer.vocab_size, bias=False)

    def forward(self, input_ids, attention_mask, token_type_ids):
        with torch.no_grad():
            output = bert_model(input_ids, attention_mask, token_type_ids)
            # last_hidden_state：词向量， pooler_output：句向量
            # print('output-->', output)
        # 千万注意不要写在with里面
        out = self.linear(output.last_hidden_state[:, 16])
        return out


def test_mymodel():
    dataloader = get_dataloader()
    my_model = MyModel()
    for i, (input_ids, attention_mask, token_type_ids, labels) in enumerate(dataloader):
        out = my_model(input_ids, attention_mask, token_type_ids)
        print(out, out.shape)
        break

# 训练模型
def train_model():
    # 准备物料 模型，损失函数，优化器
    my_dataloader = get_dataloader()
    my_model = MyModel()
    my_loss = nn.CrossEntropyLoss()
    my_optimizer = AdamW(params=my_model.parameters(), lr=5e-4)
    epochs = 1
    for p in bert_model.parameters():
        p.requires_grad_(False)

    # 训练模型
    my_model.train()
    for epoch in range(epochs):
        for i, (input_ids, attention_mask, token_type_ids, labels) in enumerate(my_dataloader):
            output = my_model(input_ids, attention_mask, token_type_ids)
            # print("output-->", output, output.shape)
            # print("label-->", labels, labels.shape)
            loss = my_loss(output, labels)
            my_optimizer.zero_grad()
            loss.backward()
            my_optimizer.step()

            if i % 10 == 0:
                tmp = torch.argmax(output, dim=-1)
                acc = (tmp == labels).sum().item() / len(labels)
                print("i-->", i, "\tloss-->", loss.item(), "\t准确率-->", acc)
        # 保存模型
        torch.save(my_model.state_dict(), "../save/day11/my_model_%d.bin" % epoch)



# 预测模型
def evalute_model():
    # 1.准备数据
    test_dataset = load_dataset("csv", data_files="../data/test.csv", split='train')
    test_dataset = test_dataset.filter(lambda x:len(x['text']) > 32)
    test_dataloader = DataLoader(test_dataset, batch_size=8, shuffle=True, drop_last=True, collate_fn=collate_fn)

    #2.准备模型
    my_model = MyModel()
    my_model.load_state_dict(torch.load("../save/fill_mask_1.bin"))

    #3.预测
    my_model.eval()
    accuracy = 0
    total = 0
    for i,(input_ids, attention_mask, input_type_ids, labels) in enumerate(test_dataloader):
        with torch.no_grad():
            output = my_model(input_ids, attention_mask, input_type_ids)

        output = torch.argmax(output, dim=-1)
        accuracy += (output == labels).sum().item()
        total += len(labels)

        # 打印观测值
        if i % 10 == 0:
            print('平均准确率为-->\t',accuracy/total)
    print("模型总准确率为-->", accuracy/total)


if __name__ == '__main__':
    # get_dataloader()
    # test_mymodel()
    # train_model()
    evalute_model()