最近在做一个中文文本分类的项目,用到了哈工大和科大讯飞联合发布的 Chinese-RoBERTa-wwm 模型。这个模型在不少中文 NLP 榜单上表现都挺亮眼,但实际微调起来,发现从数据准备到最终部署上线,中间有不少“坑”。今天就把我这次实战的经验和踩过的雷整理一下,希望能帮到有同样需求的同学。
Chinese-RoBERTa-wwm 可以看作是 BERT-wwm 的升级版。它最大的特点就是“全词掩码”(Whole Word Masking, WWM)。简单说,对于中文,BERT 原始的掩码策略是按字(character)来随机掩码的,比如“人工智能”可能被掩码成“人[MASK]智能”。而 WWM 策略会把一个完整的词(如“人工”)作为一个整体来掩码,这样模型在预训练时学习到的就是更完整的语义单元,对中文这种没有明显空格分隔的语言更友好。ROBERTa 架构本身去掉了 BERT 的下一句预测任务,采用了动态掩码和更大的批次,训练更充分。在实际对比中,Chinese-RoBERTa-wwm 在诸如 CLUE 这样的中文评测基准上,通常比原始的 BERT-wwm 有 1-2 个百分点的提升,尤其是在需要理解词语内部关系的任务上。
不过,预训练模型再强,直接拿来微调也可能“水土不服”。尤其是在我们常见的业务场景里,标注数据往往不多(小样本),这时候微调效果就很不稳定。我总结下来,主要有这么几个原因:
- 标签噪声:人工标注难免有误,几百条数据里混入几条错误标签,对模型的影响会被放大。
- 领域偏移:预训练语料(如百科、新闻)和你的业务数据(如医疗报告、客服对话)分布差异大,模型学到的通用知识不能直接套用。
- 过拟合:参数庞大的模型在少量数据上很容易记住训练样本,导致在验证集上表现好,一上真实场景就“拉胯”。
针对这些问题,我尝试了一套组合优化方案,效果还不错。
1. 数据增强:基于 TF-IDF 的关键词替换直接回译或者 EDA 有时会改变句子的专业术语。我采用了一种更保守的方法:用 TF-IDF 找出句子中最重要的词(非停用词),然后用同义词词林或哈工大同义词词库中的同义词进行替换。这样既增加了数据多样性,又最大程度保留了原句的核心语义。实现起来也不复杂,先分词,计算 TF-IDF,对每个句子选取 TF-IDF 值最低的 N 个词(因为这些词可能是相对不重要的修饰词)尝试替换。
2. 分层学习率设置这是从 HuggingFace 社区学来的一招。模型的不同层学习速度应该不同。通常,靠近输出的顶层(分类器)需要更快地学习新任务,而底层的 Embedding 和 Transformer 层已经包含了丰富的通用语义知识,应该用更小的学习率微调,避免“灾难性遗忘”。在 PyTorch 中,可以很方便地为不同的参数组设置不同的学习率。
3. 对抗训练(FGM)对抗训练能提升模型的鲁棒性。我实现了 Fast Gradient Method(FGM)。核心思想是在 embedding 层上添加一个小的扰动,这个扰动的方向是损失函数梯度上升的方向,让模型在面对这种“恶意”扰动时也能保持预测稳定。这相当于给模型做了正则化,对缓解过拟合有帮助。
下面是我的 PyTorch 微调代码核心部分,包含了上述技巧的实现:
import torch import torch.nn as nn from transformers import BertTokenizer, BertForSequenceClassification, AdamW, get_linear_schedule_with_warmup from torch.utils.data import DataLoader, Dataset import numpy as np # 1. 动态 Padding 的 DataLoader class TextDataset(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, idx): text = str(self.texts[idx]) label = self.labels[idx] encoding = self.tokenizer.encode_plus( text, add_special_tokens=True, max_length=self.max_len, return_token_type_ids=False, padding='max_length', # 这里先pad到最大长度,DataLoader中通过collate_fn动态处理 truncation=True, return_attention_mask=True, return_tensors='pt', ) # 实际collate_fn中会根据batch内最大长度重新padding return { 'input_ids': encoding['input_ids'].flatten(), 'attention_mask': encoding['attention_mask'].flatten(), 'labels': torch.tensor(label, dtype=torch.long) } def collate_fn(batch): # 动态padding:取batch中最长的序列长度 max_len = max([len(item['input_ids']) for item in batch]) input_ids = [] attention_masks = [] labels = [] for item in batch: pad_len = max_len - len(item['input_ids']) input_ids.append(torch.cat([item['input_ids'], torch.zeros(pad_len, dtype=torch.long)])) attention_masks.append(torch.cat([item['attention_mask'], torch.zeros(pad_len, dtype=torch.long)])) labels.append(item['labels']) return { 'input_ids': torch.stack(input_ids), 'attention_mask': torch.stack(attention_masks), 'labels': torch.stack(labels) } # 2. FGM 对抗训练类 class FGM: def __init__(self, model, epsilon=0.25): self.model = model self.epsilon = epsilon self.backup = {} # 用于备份原始embedding参数 def attack(self, emb_name='word_embeddings'): # 默认攻击embedding层 for name, param in self.model.named_parameters(): if param.requires_grad and emb_name in name: self.backup[name] = param.data.clone() norm = torch.norm(param.grad) if norm != 0: r_at = self.epsilon * param.grad / norm param.data.add_(r_at) def restore(self, emb_name='word_embeddings'): for name, param in self.model.named_parameters(): if param.requires_grad and emb_name in name: assert name in self.backup param.data = self.backup[name] self.backup = {} # 训练循环片段(包含梯度累积和分层学习率) def train_epoch(model, data_loader, optimizer, scheduler, device, fgm=None, gradient_accumulation_steps=4): model.train() total_loss = 0 optimizer.zero_grad() # 梯度累积,每accumulation步清空一次 for step, batch in enumerate(data_loader): input_ids = batch['input_ids'].to(device) attention_mask = batch['attention_mask'].to(device) labels = batch['labels'].to(device) outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels) loss = outputs.loss loss = loss / gradient_accumulation_steps # 损失按累积步数平均 loss.backward() # 对抗训练 if fgm: fgm.attack() outputs_adv = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels) loss_adv = outputs_adv.loss / gradient_accumulation_steps loss_adv.backward() fgm.restore() if (step + 1) % gradient_accumulation_steps == 0: torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 梯度裁剪 optimizer.step() scheduler.step() optimizer.zero_grad() total_loss += loss.item() * gradient_accumulation_steps return total_loss / len(data_loader) # 主函数 def main(): # 参数设置 MODEL_NAME = 'hfl/chinese-roberta-wwm-ext' MAX_LEN = 128 BATCH_SIZE = 16 EPOCHS = 5 LEARNING_RATE = 2e-5 GRADIENT_ACCUMULATION_STEPS = 2 # 模拟更大batch size,优化显存 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') tokenizer = BertTokenizer.from_pretrained(MODEL_NAME) model = BertForSequenceClassification.from_pretrained(MODEL_NAME, num_labels=2).to(device) # 分层学习率设置:顶层分类器用大学习率,底层bert用小学习率 no_decay = ['bias', 'LayerNorm.weight'] optimizer_grouped_parameters = [ {'params': [p for n, p in model.named_parameters() if 'classifier' in n and not any(nd in n for nd in no_decay)], 'lr': LEARNING_RATE * 10}, # 分类器学习率放大10倍 {'params': [p for n, p in model.named_parameters() if 'classifier' in n and any(nd in n for nd in no_decay)], 'lr': LEARNING_RATE * 10, 'weight_decay': 0.0}, {'params': [p for n, p in model.named_parameters() if 'classifier' not in n and not any(nd in n for nd in no_decay)], 'lr': LEARNING_RATE}, {'params': [p for n, p in model.named_parameters() if 'classifier' not in n and any(nd in n for nd in no_decay)], 'lr': LEARNING_RATE, 'weight_decay': 0.0}, ] optimizer = AdamW(optimizer_grouped_parameters, lr=LEARNING_RATE, correct_bias=False) total_steps = len(train_loader) // GRADIENT_ACCUMULATION_STEPS * EPOCHS scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps=0.1*total_steps, num_training_steps=total_steps) fgm = FGM(model, epsilon=0.25) # 初始化FGM # 训练循环... # 模型保存最佳实践:保存整个模型和tokenizer,方便后续加载 # model.save_pretrained('./saved_model') # tokenizer.save_pretrained('./saved_model') if __name__ == '__main__': main()关键参数说明:
epsilon=0.25(FGM):扰动大小,经验值,太大可能破坏语义,太小效果不明显。LEARNING_RATE=2e-5:BERT类模型微调的经典起点学习率。GRADIENT_ACCUMULATION_STEPS=2:当GPU显存不足时,通过累积梯度来等效增大批次大小。max_norm=1.0(梯度裁剪):防止梯度爆炸的阈值。
踩坑是进步的阶梯,下面分享几个我遇到的典型问题及解决办法:
1. 验证集数据泄露这是最隐蔽的坑。表现是验证集指标奇高,但测试集一塌糊涂。检查方法:确保在任何数据预处理(如分词、构建词表、TF-IDF计算)步骤之前,就将训练集、验证集、测试集严格分开。特别是使用整个数据集统计信息(如均值、方差)去做标准化,或者用整个数据集训练 tokenizer,都会导致信息泄露。我的做法是,在代码一开始就用sklearn的train_test_split固定随机种子划分好,并且将预处理对象(如scaler)只在训练集上拟合,然后分别应用到验证集和测试集。
2. 混合精度训练导致 NaN 值为了加快训练和节省显存,我启用了torch.cuda.amp自动混合精度训练。但有时会出现损失变成 NaN 的情况。解决方案:
- 梯度缩放:使用
GradScaler是必须的,它能防止梯度下溢。 - 检查输入:确保输入数据中没有异常值或无穷值。
- 降低学习率:混合精度下有时需要更保守的学习率。
- 跳过有问题的批次:在
scaler.scale(loss).backward()后,检查梯度是否有 NaN,如果有,跳过本次参数更新。scaler.step(optimizer)内部其实有类似机制。
3. 生产环境中的并发推理优化模型上线后,面对高并发请求,直接加载原生 PyTorch 模型调用model.eval()和model.to(device)可能效率不高且显存占用大。
- 模型导出:使用
torch.jit.trace或torch.jit.script将模型转换为 TorchScript,可以获得更快的加载速度和一定的优化。对于 Transformer 模型,HuggingFace 的transformers库也提供了torch.jit支持。 - 服务化:推荐使用TorchServe或Triton Inference Server来部署模型。它们支持多模型、版本管理、动态批处理(Dynamic Batching),能显著提高 GPU 利用率和吞吐量。在我的测试中,使用动态批处理,在 batch size=32 时,相比单条推理,吞吐量提升了约 15 倍,平均延迟从 50ms 降至 15ms(RTX 3090)。
- 量化:如果对延迟极度敏感,可以考虑使用 PyTorch 的动态量化或静态量化来减小模型体积、加速推理,INT8 量化通常能带来 2-4 倍的推理速度提升,但可能会有轻微精度损失,需要仔细评估。
最后,留一个开放性问题给大家思考:在我们这次微调中,我们假设训练数据和未来线上数据是独立同分布的。但现实中,业务数据的分布可能会随时间缓慢变化(概念漂移),或者我们需要将一个在通用领域微调好的模型,快速应用到另一个新领域(如从新闻分类迁移到法律文书分类)。如何设计一种领域自适应(Domain Adaptation)的微调策略,让模型能更好地泛化到与训练数据分布不同但相关的目标领域呢?是使用领域对抗训练(DANN),还是在预训练阶段就融入多领域数据,抑或是设计更精巧的迁移学习架构?这是一个值得深入探索的方向。