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")
devicePRE_TRAINED_MODEL_NAME = '../bert-base-chinese/' # 英文bert预训练模型
tokenizer = BertTokenizer.from_pretrained(PRE_TRAINED_MODEL_NAME)
BertModel.from_pretrained(PRE_TRAINED_MODEL_NAME)
tokenizerdf = 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.shapeclass MyDataSet(Dataset):def __init__(self,texts,labels,tokenizer,max_len):self.texts = textsself.labels = labelsself.tokenizer = tokenizerself.max_len = max_lendef __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)passdef 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 = 0count = 0for 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#breakreturn 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 = 0with 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 = 0for 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_valuesy_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....