BERT预训练模型实战：从零构建影评情感分析系统

1次阅读

共计 2552 个字符，预计需要花费 7 分钟才能阅读完成。

情感分析能帮助企业理解用户反馈、优化产品体验，而 BERT 模型凭借其强大的上下文理解能力，在各类 NLP 任务中表现出色。相比传统方法，BERT 能自动捕获文本中的复杂语义关系，特别适合处理影评这类包含讽刺、反语等复杂情感的文本。本文将带新手快速实现一个基于 BERT 的端到端情感分类器。

在 IMDB 影评数据集上的测试结果对比（测试集准确率）：

模型类型	准确率	训练时间	显存占用
TF-IDF+SVM	82.3%	2 分钟	<1GB
LSTM	85.7%	30 分钟	3GB
BERT-base	92.1%	1 小时	6GB
BERT-large	93.4%	2 小时	12GB

安装必要的库：

pip install transformers torch pandas

加载中文 BERT 模型和 tokenizer：

from transformers import BertTokenizer, BertForSequenceClassification

tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
model = BertForSequenceClassification.from_pretrained('bert-base-chinese', 
                                                      num_labels=2)  # 二分类

处理特殊文本的示例代码：

def clean_text(text):
    # 处理颜文字
    text = re.sub(r'\[.+?\]', '[EMOTION]', text)  
    # 去除 HTML 标签
    text = re.sub(r'<[^>]+>', '', text)
    return text

from torch.utils.data import Dataset

class ReviewDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_len=128):
        self.tokenizer = tokenizer
        self.texts = texts
        self.labels = labels
        self.max_len = max_len

    def __getitem__(self, idx):
        text = str(self.texts[idx])
        inputs = self.tokenizer.encode_plus(
            text,
            None,
            add_special_tokens=True,
            max_length=self.max_len,
            padding='max_length',
            truncation=True,
            return_tensors='pt'
        )
        return {'input_ids': inputs['input_ids'].flatten(),
            'attention_mask': inputs['attention_mask'].flatten(),
            'labels': torch.tensor(self.labels[idx], dtype=torch.long)
        }

包含显存优化的关键代码：

# 梯度累积（显存不足时使用）accumulation_steps = 4  
optimizer = AdamW(model.parameters(), lr=2e-5)  # BERT 推荐初始学习率

for epoch in range(3):  # 通常 3 - 4 个 epoch 足够
    model.train()
    total_loss = 0

    for step, batch in enumerate(train_loader):
        inputs = {'input_ids': batch['input_ids'].to(device),
            'attention_mask': batch['attention_mask'].to(device),
            'labels': batch['labels'].to(device)
        }

        outputs = model(**inputs)
        loss = outputs.loss
        loss = loss / accumulation_steps  # 梯度累积
        loss.backward()

        if (step+1) % accumulation_steps == 0:
            optimizer.step()
            optimizer.zero_grad()

重采样：对少数类过采样或多数类欠采样
类别权重：在损失函数中设置 class_weight
Focal Loss：降低易分类样本的权重

from transformers import TrainerCallback

class EarlyStoppingCallback(TrainerCallback):
    def __init__(self, patience=3):
        self.patience = patience
        self.best_metric = None
        self.wait = 0

    def on_evaluate(self, args, state, control, metrics, **kwargs):
        current = metrics['eval_accuracy']
        if self.best_metric is None or current > self.best_metric:
            self.best_metric = current
            self.wait = 0
        else:
            self.wait += 1
            if self.wait >= self.patience:
                control.should_training_stop = True