3 文本分类入门finetune：bert-base-chinese

项目实战：

数据准备工作

????????`bert-base-chinese` 是一种预训练的语言模型，基于 BERT（Bidirectional Encoder Representations from Transformers）架构，专门用于中文自然语言处理任务。BERT 是由 Google 在 2018 年提出的一种革命性的预训练模型，通过大规模的无监督训练，能够学习到丰富的语言表示。

????????`bert-base-chinese` 是 BERT 在中文语料上进行预训练的版本，它包含了 12 层 Transformer 编码器和 110 万个参数。这个模型在中文文本上进行了大规模的预训练，可以用于各种中文自然语言处理任务，如文本分类、命名实体识别、情感分析等。

使用 `bert-base-chinese` 模型时，可以将其作为一个特征提取器，将输入的文本转换为固定长度的向量表示，然后将这些向量输入到其他机器学习模型中进行训练或推断。也可以对 `bert-base-chinese` 进行微调，将其用于特定任务的训练。

????????预训练的 `bert-base-chinese` 模型可以通过 Hugging Face 的 Transformers 库进行加载和使用。在加载模型后，可以使用它的 `encode` 方法将文本转换为向量表示，或者使用 `forward` 方法对文本进行特定任务的预测。

????????需要注意的是，`bert-base-chinese` 是一个通用的中文语言模型，但它可能在特定的任务上表现不佳。在某些情况下，可能需要使用更大的模型或进行微调来获得更好的性能。

进行微调时，可以按照以下步骤进行操作：

1. 准备数据集：首先，你需要准备一个与你的任务相关的标注数据集。这个数据集应该包含输入文本以及相应的标签或注释，用于训练和评估模型。

2. 加载预训练模型：使用 Hugging Face 的 Transformers 库加载预训练的 `bert-base-chinese` 模型。你可以选择加载整个模型或只加载其中的一部分，具体取决于你的任务需求。

3. 创建模型架构：根据你的任务需求，创建一个适当的模型架构。这通常包括在 `bert-base-chinese` 模型之上添加一些额外的层，用于适应特定的任务。

4. 数据预处理：将你的数据集转换为适合模型输入的格式。这可能包括将文本转换为输入的编码表示，进行分词、填充和截断等操作。

5. 定义损失函数和优化器：选择适当的损失函数来衡量模型预测与真实标签之间的差异，并选择合适的优化器来更新模型的参数。

6. 微调模型：使用训练集对模型进行训练。在每个训练步骤中，将输入文本提供给模型，计算损失并进行反向传播，然后使用优化器更新模型的参数。

7. 评估模型：使用验证集或测试集评估模型的性能。可以计算准确率、精确率、召回率等指标来评估模型在任务上的表现。

8. 调整和优化：根据评估结果，对模型进行调整和优化。你可以尝试不同的超参数设置、模型架构或训练策略，以获得更好的性能。

9. 推断和应用：在微调完成后，你可以使用微调后的模型进行推断和应用。将新的输入文本提供给模型，获取预测结果，并根据任务需求进行后续处理。

需要注意的是，微调的过程可能需要大量的计算资源和时间，并且需要对模型和数据进行仔细的调整和优化。此外，合适的数据集规模和质量对于获得良好的微调结果也非常重要。

准备模型

https://huggingface.co/bert-base-chinese/tree/main

数据集

与之前的训练数据一样使用；

代码部分

# 导入transformers
import transformers
from transformers import BertModel, BertTokenizer, AdamW, get_linear_schedule_with_warmup

# 导入torch
import torch
from torch import nn, optim
from torch.utils.data import Dataset, DataLoader
import torch.nn.functional as F
 
# 常用包
import re
import numpy as np
import pandas as pd
import seaborn as sns
from pylab import rcParams
import matplotlib.pyplot as plt
from matplotlib import rc
from sklearn.model_selection import train_test_split
from sklearn.metrics import confusion_matrix, classification_report
from collections import defaultdict
from textwrap import wrap
from tqdm import tqdm
 
%matplotlib inline
%config InlineBackend.figure_format='retina' # 主题
 
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
device


PRE_TRAINED_MODEL_NAME = '../bert-base-chinese/' # 英文bert预训练模型
tokenizer = BertTokenizer.from_pretrained(PRE_TRAINED_MODEL_NAME)
BertModel.from_pretrained(PRE_TRAINED_MODEL_NAME)
tokenizer

df = pd.read_csv("../data/online_shopping_10_cats.csv")
#myDataset[0]
RANDOM_SEED = 1012
df_train, df_test = train_test_split(df, test_size=0.1, random_state=RANDOM_SEED)
df_val, df_test = train_test_split(df_test, test_size=0.5, random_state=RANDOM_SEED)
df_train.shape, df_val.shape, df_test.shape



class MyDataSet(Dataset):
    def __init__(self,texts,labels,tokenizer,max_len):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_len = max_len
    def __len__(self):
        return len(self.texts)
    def __getitem__(self,item):
        text = str(self.texts[item])
        label = self.labels[item]
 
        encoding = self.tokenizer(text=text,
                      max_length=self.max_len,
                      pad_to_max_length=True,
                      add_special_tokens=True,
                      return_attention_mask=True,
                      return_token_type_ids=True,
                      return_tensors='pt')
        return {"text":text,
               "input_ids":encoding['token_type_ids'].flatten(),
               "attention_mask":encoding['attention_mask'].flatten(),
               "labels":torch.tensor(label,dtype=torch.long)}
def create_data_loader(df,tokenizer,max_len,batch_size=4):
    ds = MyDataSet(texts=df['review'].values,
                  labels=df['label'].values,
                   tokenizer = tokenizer,
                   max_len=max_len
                  )
    return DataLoader(ds,batch_size=batch_size)
MAX_LEN = 512
BATCH_SIZE = 32
train_data_loader = create_data_loader(df_train,tokenizer,max_len=MAX_LEN, batch_size=BATCH_SIZE)
val_data_loader = create_data_loader(df_val,tokenizer,max_len=MAX_LEN, batch_size=BATCH_SIZE)
test_data_loader = create_data_loader(df_test,tokenizer,max_len=MAX_LEN, batch_size=BATCH_SIZE)

class BaseBertModel(nn.Module):
    def __init__(self,n_class=2):
        super(BaseBertModel,self).__init__()
        self.bert = BertModel.from_pretrained(PRE_TRAINED_MODEL_NAME)
        self.drop = nn.Dropout(0.2)
        self.out = nn.Linear(self.bert.config.hidden_size, n_class)
        pass
    def forward(self,input_ids,attention_mask):
        _,pooled_output = self.bert(input_ids=input_ids,
                 attention_mask=attention_mask,
                 return_dict = False)
        out = self.drop(pooled_output)
        return self.out(out)
 
#pooled_output
model = BaseBertModel()
model = model.to(device)

EPOCHS = 10 # 训练轮数
optimizer = AdamW(model.parameters(),lr=2e-4,correct_bias=False)
total_steps = len(train_data_loader) * EPOCHS
scheduler = get_linear_schedule_with_warmup(optimizer,num_warmup_steps=0,
                               num_training_steps=total_steps)
loss_fn = nn.CrossEntropyLoss().to(device)

def train_epoch(model,data_loader,loss_fn,optimizer,device,schedule,n_examples):
    model = model.train()
    losses = []
    correct_predictions = 0
    count = 0
    for d in tqdm(data_loader):
        input_ids = d['input_ids'].to(device)
        attention_mask = d['attention_mask'].to(device)
        targets = d['labels'].to(device)
        outputs = model(
            input_ids=input_ids,
            attention_mask=attention_mask
        )
        _,preds = torch.max(outputs, dim=1)
        loss = loss_fn(outputs,targets)
        losses.append(loss.item())
        
        correct_predictions += torch.sum(preds==targets)
        loss.backward()
        nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        scheduler.step()
        optimizer.zero_grad()
        
        count += 1
        # if count > 100:
        #     break
        #break
    return correct_predictions.double() / n_examples, np.mean(losses)
 
def eval_model(model, data_loader, loss_fn, device, n_examples):
    model = model.eval() # 验证预测模式
 
    losses = []
    correct_predictions = 0
 
    with torch.no_grad():
        for d in data_loader:
            input_ids = d["input_ids"].to(device)
            attention_mask = d["attention_mask"].to(device)
            targets = d["labels"].to(device)
 
            outputs = model(
                input_ids=input_ids,
                attention_mask=attention_mask
            )
            _, preds = torch.max(outputs, dim=1)
 
            loss = loss_fn(outputs, targets)
 
            correct_predictions += torch.sum(preds == targets)
            losses.append(loss.item())
 
    return correct_predictions.double() / n_examples, np.mean(losses)
 
 
# train model
history = defaultdict(list) # 记录10轮loss和acc
best_accuracy = 0
 
for epoch in range(EPOCHS):
 
    print(f'Epoch {epoch + 1}/{EPOCHS}')
    print('-' * 10)
 #(model,data_loader,loss_fn,device,schedule,n_exmaples)
    train_acc, train_loss = train_epoch(
        model,
        train_data_loader,
        loss_fn,
        optimizer,
        device,
        scheduler,
        len(df_train)
    )
 
    print(f'Train loss {train_loss} accuracy {train_acc}')
 
    val_acc, val_loss = eval_model(
        model,
        val_data_loader,
        loss_fn,
        device,
        len(df_val)
    )
 
    print(f'Val loss {val_loss} accuracy {val_acc}')
    print()
 
    history['train_acc'].append(train_acc)
    history['train_loss'].append(train_loss)
    history['val_acc'].append(val_acc)
    history['val_loss'].append(val_loss)
 
    if val_acc > best_accuracy:
        torch.save(model.state_dict(), 'best_model_state.bin')
        best_accuracy = val_acc
 
 
# 模型评估
test_acc, _ = eval_model(
  model,
  test_data_loader,
  loss_fn,
  device,
  len(df_test)
)
test_acc.item()
 
 
def get_predictions(model, data_loader):
    model = model.eval()
 
    texts = []
    predictions = []
    prediction_probs = []
    real_values = []
 
    with torch.no_grad():
        for d in data_loader:
            texts = d["texts"]
            input_ids = d["input_ids"].to(device)
            attention_mask = d["attention_mask"].to(device)
            targets = d["labels"].to(device)
 
            outputs = model(
                input_ids=input_ids,
                attention_mask=attention_mask
            )
            _, preds = torch.max(outputs, dim=1)
 
            probs = F.softmax(outputs, dim=1)
 
            texts.extend(texts)
            predictions.extend(preds)
            prediction_probs.extend(probs)
            real_values.extend(targets)
 
    predictions = torch.stack(predictions).cpu()
    prediction_probs = torch.stack(prediction_probs).cpu()
    real_values = torch.stack(real_values).cpu()
    return texts, predictions, prediction_probs, real_values
 
y_texts, y_pred, y_pred_probs, y_test = get_predictions(
  model,
  test_data_loader
)
print(classification_report(y_test, y_pred, target_names=[str(label) for label in class_names]))
 
# 模型预测
 
sample_text='Hard but Robust, Easy but Sensitive: How Encod.'
encoded_text = tokenizer.encode_plus(
  sample_text,
  max_length=MAX_LEN,
  add_special_tokens=True,
  return_token_type_ids=False,
  pad_to_max_length=True,
  return_attention_mask=True,
  return_tensors='pt',
)
 
input_ids = encoded_text['input_ids'].to(device)
attention_mask = encoded_text['attention_mask'].to(device)
 
output = model(input_ids, attention_mask)
_, prediction = torch.max(output, dim=1)
 
print(f'Sample text: {sample_text}')
print(f'Danger label  : {label_id2cate[prediction.cpu().numpy()[0]]}')

runing....