【Bug已解决】How to fine tune BERT on unlabeled data? 解决方案问题标签BERT、无标签数据、微调、半监督学习、自监督学习、领域适应适用版本Transformers 4.x、PyTorch 1.8难度等级⭐⭐⭐⭐较高一、问题描述在实际业务场景中标注数据往往稀缺且昂贵而无标签数据则大量存在。很多开发者面临的问题是如何利用大量无标签数据来微调 BERT 模型提升模型在下游任务上的表现这是一个经典的半监督学习/自监督学习问题。BERT 本身就是通过自监督方式MLM NSP在无标签数据上预训练的我们可以在下游任务中继续利用无标签数据来进一步提升模型性能。核心挑战挑战说明无标签数据无法直接用于监督学习需要设计自监督或半监督策略领域差异通用 BERT 可能不了解领域术语计算资源限制大规模无标签数据处理耗时何时停止自监督训练容易过拟合二、错误复现场景直接用无标签数据做监督训练必然失败import torch from transformers import BertTokenizer, BertForSequenceClassification # 加载预训练 BERT tokenizer BertTokenizer.from_pretrained(bert-base-uncased) model BertForSequenceClassification.from_pretrained(bert-base-uncased, num_labels2) # 模拟无标签数据 unlabeled_texts [ The new machine learning model achieves state-of-the-art results., Quantum computing may revolutionize cryptography., The stock market experienced significant volatility today., # ... 大量无标签文本 ] # ❌ 错误无标签数据没有标签无法计算 CrossEntropyLoss texts unlabeled_texts # labels ??? # 没有标签 # 以下代码会失败因为没有标签 # loss model(input_ids, attention_mask, labelslabels)问题无标签数据没有标签无法直接使用监督学习的损失函数。三、根因分析3.1 为什么需要利用无标签数据领域适应通用 BERT 在法律、医学、金融等垂直领域表现不佳需要领域数据继续预训练。数据效率少量标注 大量无标签 仅用少量标注。语言/风格适应社交媒体文本、口语化表达与标准文本差异大。3.2 可行的策略概览策略原理适用场景继续预训练Domain-Adaptive Pretraining, DAPT在无标签数据上继续 MLM 训练领域适应任务自适应预训练TAPT在任务相关无标签数据上 MLM提升特定任务性能伪标签Pseudo-Labeling用模型预测作为伪标签半监督学习一致性正则化Consistency Regularization对输入做扰动要求输出一致半监督学习对比学习Contrastive Learning拉近相似样本推远不同样本表示学习四、解决方案方案一领域自适应预训练DAPT—— 最推荐在无标签数据上继续 BERT 的 Masked Language ModelMLM训练让模型适应领域语言。import torch import torch.nn as nn from transformers import BertTokenizer, BertForMaskedLM, BertConfig from torch.utils.data import Dataset, DataLoader import random import numpy as np # MLM 数据集 class MLMDataset(Dataset): Masked Language Model 数据集 def __init__(self, texts, tokenizer, max_length128, mlm_probability0.15): self.texts texts self.tokenizer tokenizer self.max_length max_length self.mlm_probability mlm_probability def __len__(self): return len(self.texts) def __getitem__(self, idx): text self.texts[idx] encoding self.tokenizer( text, truncationTrue, max_lengthself.max_length, paddingmax_length, return_tensorspt, ) input_ids encoding[input_ids].squeeze(0) attention_mask encoding[attention_mask].squeeze(0) # 创建 MLM 标签 labels input_ids.clone() # 随机 mask 15% 的 token probability_matrix torch.full(input_ids.shape, self.mlm_probability) special_tokens_mask [ self.tokenizer.get_special_tokens_mask(val.tolist(), already_has_special_tokensTrue) for val in input_ids ] special_tokens_mask torch.tensor(special_tokens_mask, dtypetorch.bool) probability_matrix.masked_fill_(special_tokens_mask, value0.0) masked_indices torch.bernoulli(probability_matrix).bool() labels[~masked_indices] -100 # 只计算 masked 位置的 loss # 80% 替换为 [MASK] indices_replaced torch.bernoulli(torch.full(input_ids.shape, 0.8)).bool() masked_indices input_ids[indices_replaced] self.tokenizer.convert_tokens_to_ids(self.tokenizer.mask_token) # 10% 替换为随机 token indices_random torch.bernoulli(torch.full(input_ids.shape, 0.5)).bool() masked_indices ~indices_replaced random_words torch.randint(len(self.tokenizer), input_ids.shape, dtypetorch.long) input_ids[indices_random] random_words[indices_random] # 10% 保持不变 return { input_ids: input_ids, attention_mask: attention_mask, labels: labels, } # DAPT 训练 def domain_adaptive_pretraining(texts, model_namebert-base-uncased, epochs3, batch_size16, lr5e-5): 领域自适应预训练 device torch.device(cuda if torch.cuda.is_available() else cpu) print(f使用设备: {device}) # 加载 tokenizer 和模型 tokenizer BertTokenizer.from_pretrained(model_name) model BertForMaskedLM.from_pretrained(model_name) model.to(device) # 创建数据集 dataset MLMDataset(texts, tokenizer, max_length128) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) # 优化器 optimizer torch.optim.AdamW(model.parameters(), lrlr, weight_decay0.01) # 学习率调度器 num_training_steps epochs * len(dataloader) scheduler torch.optim.lr_scheduler.LinearLR( optimizer, start_factor0.1, total_itersnum_training_steps // 10 ) print(f无标签数据量: {len(texts)}) print(f训练步数: {num_training_steps}) # 训练循环 for epoch in range(epochs): model.train() total_loss 0 for batch_idx, batch in enumerate(dataloader): input_ids batch[input_ids].to(device) attention_mask batch[attention_mask].to(device) labels batch[labels].to(device) optimizer.zero_grad() outputs model(input_idsinput_ids, attention_maskattention_mask, labelslabels) loss outputs.loss loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() scheduler.step() total_loss loss.item() if batch_idx % 50 0: print(fEpoch [{epoch1}/{epochs}] Batch [{batch_idx}/{len(dataloader)}] fLoss: {loss.item():.4f}) avg_loss total_loss / len(dataloader) print(fEpoch [{epoch1}/{epochs}] Average Loss: {avg_loss:.4f}) return model, tokenizer # 运行示例 # 模拟领域无标签数据 unlabeled_texts [ The patient presented with acute myocardial infarction and was admitted to the ICU., Clinical trials show that the new drug reduces mortality by 30% in high-risk patients., The radiologist identified a mass in the left lung during routine screening., Laboratory results indicate elevated levels of troponin and creatine kinase., The surgical team performed a coronary artery bypass graft successfully., Post-operative care includes monitoring for signs of infection and blood clots., The epidemiological study tracked 5000 participants over a ten-year period., Genetic testing revealed a mutation in the BRCA1 gene associated with breast cancer., The emergency department handled a surge of trauma cases during the holiday weekend., Telemedicine platforms have expanded access to specialist consultations in rural areas., ] * 50 # 复制以增加数据量 # 运行 DAPT # model, tokenizer domain_adaptive_pretraining(unlabeled_texts, epochs3)方案二伪标签Pseudo-Labeling先用少量标注数据训练初始模型然后用该模型对无标签数据预测将高置信度预测作为伪标签。import torch import torch.nn as nn from transformers import BertTokenizer, BertForSequenceClassification from torch.utils.data import Dataset, DataLoader, TensorDataset import numpy as np class PseudoLabelTrainer: 伪标签训练器 def __init__(self, model_namebert-base-uncased, num_labels2, confidence_threshold0.9, deviceauto): self.model_name model_name self.num_labels num_labels self.confidence_threshold confidence_threshold if device auto: self.device torch.device(cuda if torch.cuda.is_available() else cpu) else: self.device torch.device(device) self.tokenizer BertTokenizer.from_pretrained(model_name) self.model BertForSequenceClassification.from_pretrained( model_name, num_labelsnum_labels ).to(self.device) def train_on_labeled(self, labeled_texts, labeled_labels, epochs5, batch_size16, lr2e-5): 在有标签数据上训练 # 编码 encodings self.tokenizer( labeled_texts, truncationTrue, max_length128, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(labeled_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) optimizer torch.optim.AdamW(self.model.parameters(), lrlr) criterion nn.CrossEntropyLoss() self.model.train() for epoch in range(epochs): total_loss 0 correct 0 total 0 for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] optimizer.zero_grad() outputs self.model(input_idsinput_ids, attention_maskattention_mask) loss criterion(outputs.logits, labels) loss.backward() optimizer.step() total_loss loss.item() _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() print(f[有标签训练] Epoch [{epoch1}/{epochs}] fLoss: {total_loss/len(dataloader):.4f}, fAcc: {100.*correct/total:.2f}%) def generate_pseudo_labels(self, unlabeled_texts, batch_size32): 为无标签数据生成伪标签 encodings self.tokenizer( unlabeled_texts, truncationTrue, max_length128, paddingTrue, return_tensorspt ) dataset TensorDataset(encodings[input_ids], encodings[attention_mask]) dataloader DataLoader(dataset, batch_sizebatch_size) self.model.eval() all_probs [] all_preds [] with torch.no_grad(): for batch in dataloader: input_ids, attention_mask [b.to(self.device) for b in batch] outputs self.model(input_idsinput_ids, attention_maskattention_mask) probs torch.softmax(outputs.logits, dim-1) confidence, preds probs.max(dim-1) all_probs.extend(confidence.cpu().tolist()) all_preds.extend(preds.cpu().tolist()) # 筛选高置信度样本 pseudo_labeled [] pseudo_labels [] for i, (text, pred, conf) in enumerate(zip(unlabeled_texts, all_preds, all_probs)): if conf self.confidence_threshold: pseudo_labeled.append(text) pseudo_labels.append(pred) print(f伪标签生成: {len(pseudo_labeled)}/{len(unlabeled_texts)} f({100.*len(pseudo_labeled)/len(unlabeled_texts):.1f}%) 样本置信度 {self.confidence_threshold}) return pseudo_labeled, pseudo_labels def train_with_pseudo_labels(self, labeled_texts, labeled_labels, unlabeled_texts, epochs5, batch_size16, lr2e-5): 结合有标签和伪标签数据训练 # 生成伪标签 pseudo_texts, pseudo_labels self.generate_pseudo_labels(unlabeled_texts) # 合并数据 all_texts labeled_texts pseudo_texts all_labels labeled_labels pseudo_labels # 编码 encodings self.tokenizer( all_texts, truncationTrue, max_length128, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(all_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) optimizer torch.optim.AdamW(self.model.parameters(), lrlr) criterion nn.CrossEntropyLoss() self.model.train() for epoch in range(epochs): total_loss 0 correct 0 total 0 for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] optimizer.zero_grad() outputs self.model(input_idsinput_ids, attention_maskattention_mask) loss criterion(outputs.logits, labels) loss.backward() optimizer.step() total_loss loss.item() _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() print(f[伪标签训练] Epoch [{epoch1}/{epochs}] fLoss: {total_loss/len(dataloader):.4f}, fAcc: {100.*correct/total:.2f}%) # 使用示例 # trainer PseudoLabelTrainer(confidence_threshold0.9) # trainer.train_on_labeled(labeled_texts, labeled_labels, epochs5) # trainer.train_with_pseudo_labels(labeled_texts, labeled_labels, unlabeled_texts, epochs5)方案三一致性正则化对同一输入施加不同扰动要求模型输出保持一致。import torch import torch.nn as nn from transformers import BertTokenizer, BertForSequenceClassification import random class ConsistencyRegularizationTrainer: 一致性正则化训练器 def __init__(self, model_namebert-base-uncased, num_labels2): self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.tokenizer BertTokenizer.from_pretrained(model_name) self.model BertForSequenceClassification.from_pretrained( model_name, num_labelsnum_labels ).to(self.device) def augment_text(self, text, aug_prob0.1): 文本数据增强 words text.split() augmented [] for word in words: if random.random() aug_prob: # 随机删除 continue elif random.random() aug_prob: # 随机替换为 [MASK] augmented.append([MASK]) else: augmented.append(word) # 随机打乱 if random.random() 0.1 and len(augmented) 2: idx random.randint(0, len(augmented) - 2) augmented[idx], augmented[idx 1] augmented[idx 1], augmented[idx] return .join(augmented) if augmented else text def train_step(self, labeled_batch, unlabeled_batch, optimizer, lambda_consistency1.0, lambda_supervised1.0): 一步训练 self.model.train() optimizer.zero_grad() # 有标签数据的监督损失 labeled_input_ids labeled_batch[input_ids].to(self.device) labeled_attention_mask labeled_batch[attention_mask].to(self.device) labeled_labels labeled_batch[labels].to(self.device) labeled_outputs self.model( input_idslabeled_input_ids, attention_masklabeled_attention_mask ) supervised_loss nn.CrossEntropyLoss()( labeled_outputs.logits, labeled_labels ) # 无标签数据的一致性损失 unlabeled_input_ids unlabeled_batch[input_ids].to(self.device) unlabeled_attention_mask unlabeled_batch[attention_mask].to(self.device) # 原始预测 with torch.no_grad(): original_outputs self.model( input_idsunlabeled_input_ids, attention_maskunlabeled_attention_mask ) original_probs torch.softmax(original_outputs.logits, dim-1) # 增强后预测 augmented_input_ids unlabeled_batch[augmented_input_ids].to(self.device) augmented_attention_mask unlabeled_batch[augmented_attention_mask].to(self.device) augmented_outputs self.model( input_idsaugmented_input_ids, attention_maskaugmented_attention_mask ) augmented_probs torch.softmax(augmented_outputs.logits, dim-1) # KL 散度作为一致性损失 consistency_loss nn.KLDivLoss(reductionbatchmean)( torch.log(augmented_probs 1e-8), original_probs.detach() ) # 总损失 total_loss (lambda_supervised * supervised_loss lambda_consistency * consistency_loss) total_loss.backward() optimizer.step() return { supervised_loss: supervised_loss.item(), consistency_loss: consistency_loss.item(), total_loss: total_loss.item(), }五、完整修复代码import torch import torch.nn as nn from transformers import ( BertTokenizer, BertForMaskedLM, BertForSequenceClassification, get_linear_schedule_with_warmup ) from torch.utils.data import Dataset, DataLoader, TensorDataset import numpy as np import random import os # 完整 DAPT 下游微调流程 class BertUnlabeledFineTuner: BERT 无标签数据微调完整流程 def __init__(self, model_namebert-base-uncased, num_labels2, max_length128, deviceauto): self.model_name model_name self.num_labels num_labels self.max_length max_length if device auto: self.device torch.device(cuda if torch.cuda.is_available() else cpu) else: self.device torch.device(device) self.tokenizer BertTokenizer.from_pretrained(model_name) print(f设备: {self.device}) def step1_domain_adaptive_pretraining(self, unlabeled_texts, epochs3, batch_size16, lr5e-5): 步骤 1: 领域自适应预训练 (DAPT) print(\n * 60) print(步骤 1: 领域自适应预训练 (DAPT)) print( * 60) # 加载 MLM 模型 model BertForMaskedLM.from_pretrained(self.model_name).to(self.device) # 创建 MLM 数据集 dataset self._create_mlm_dataset(unlabeled_texts) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) # 优化器 optimizer torch.optim.AdamW(model.parameters(), lrlr, weight_decay0.01) total_steps epochs * len(dataloader) scheduler get_linear_schedule_with_warmup( optimizer, num_warmup_stepstotal_steps // 10, num_training_stepstotal_steps ) # 训练 for epoch in range(epochs): model.train() total_loss 0 for batch in dataloader: input_ids batch[input_ids].to(self.device) attention_mask batch[attention_mask].to(self.device) labels batch[labels].to(self.device) optimizer.zero_grad() outputs model(input_idsinput_ids, attention_maskattention_mask, labelslabels) loss outputs.loss loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() scheduler.step() total_loss loss.item() print(f Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(dataloader):.4f}) # 保存领域适应后的模型 self.dapt_model_path ./dapt_bert model.save_pretrained(self.dapt_model_path) self.tokenizer.save_pretrained(self.dapt_model_path) print(f 领域适应模型已保存到: {self.dapt_model_path}) return model def step2_supervised_finetuning(self, labeled_texts, labeled_labels, epochs5, batch_size16, lr2e-5): 步骤 2: 在有标签数据上微调 print(\n * 60) print(步骤 2: 有标签数据监督微调) print( * 60) # 加载 DAPT 后的模型如果有否则用原始 BERT model_path getattr(self, dapt_model_path, self.model_name) model BertForSequenceClassification.from_pretrained( model_path, num_labelsself.num_labels ).to(self.device) # 编码有标签数据 encodings self.tokenizer( labeled_texts, truncationTrue, max_lengthself.max_length, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(labeled_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) optimizer torch.optim.AdamW(model.parameters(), lrlr, weight_decay0.01) total_steps epochs * len(dataloader) scheduler get_linear_schedule_with_warmup( optimizer, num_warmup_stepstotal_steps // 10, num_training_stepstotal_steps ) criterion nn.CrossEntropyLoss() for epoch in range(epochs): model.train() total_loss 0 correct 0 total 0 for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] optimizer.zero_grad() outputs model(input_idsinput_ids, attention_maskattention_mask) loss criterion(outputs.logits, labels) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() scheduler.step() total_loss loss.item() _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() print(f Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(dataloader):.4f}, fAcc: {100.*correct/total:.2f}%) self.classification_model model return model def step3_pseudo_label_training(self, labeled_texts, labeled_labels, unlabeled_texts, confidence_threshold0.9, epochs3, batch_size16, lr1e-5): 步骤 3: 伪标签训练 print(\n * 60) print(步骤 3: 伪标签训练) print( * 60) model self.classification_model # 生成伪标签 model.eval() pseudo_texts [] pseudo_labels [] with torch.no_grad(): for text in unlabeled_texts: encoding self.tokenizer( text, truncationTrue, max_lengthself.max_length, paddingTrue, return_tensorspt ) input_ids encoding[input_ids].to(self.device) attention_mask encoding[attention_mask].to(self.device) output model(input_idsinput_ids, attention_maskattention_mask) probs torch.softmax(output.logits, dim-1) confidence, pred probs.max(dim-1) if confidence.item() confidence_threshold: pseudo_texts.append(text) pseudo_labels.append(pred.item()) print(f 伪标签样本数: {len(pseudo_texts)}/{len(unlabeled_texts)}) if len(pseudo_texts) 0: print( 无高置信度伪标签跳过此步骤) return model # 合并数据 all_texts labeled_texts pseudo_texts all_labels labeled_labels pseudo_labels # 训练 encodings self.tokenizer( all_texts, truncationTrue, max_lengthself.max_length, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(all_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) optimizer torch.optim.AdamW(model.parameters(), lrlr, weight_decay0.01) criterion nn.CrossEntropyLoss() for epoch in range(epochs): model.train() total_loss 0 correct 0 total 0 for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] optimizer.zero_grad() outputs model(input_idsinput_ids, attention_maskattention_mask) loss criterion(outputs.logits, labels) loss.backward() optimizer.step() total_loss loss.item() _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() print(f Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(dataloader):.4f}, fAcc: {100.*correct/total:.2f}%) return model def _create_mlm_dataset(self, texts, mlm_probability0.15): 创建 MLM 数据集 encoded self.tokenizer( texts, truncationTrue, max_lengthself.max_length, paddingmax_length, return_tensorspt ) input_ids encoded[input_ids] attention_mask encoded[attention_mask] labels input_ids.clone() # Mask probability_matrix torch.full(input_ids.shape, mlm_probability) special_tokens_mask [ self.tokenizer.get_special_tokens_mask(ids.tolist(), already_has_special_tokensTrue) for ids in input_ids ] special_tokens_mask torch.tensor(special_tokens_mask, dtypetorch.bool) probability_matrix.masked_fill_(special_tokens_mask, value0.0) masked_indices torch.bernoulli(probability_matrix).bool() labels[~masked_indices] -100 # 80% [MASK] indices_replaced torch.bernoulli(torch.full(input_ids.shape, 0.8)).bool() masked_indices input_ids[indices_replaced] self.tokenizer.mask_token_id # 10% random indices_random (torch.bernoulli(torch.full(input_ids.shape, 0.5)).bool() masked_indices ~indices_replaced) random_tokens torch.randint(len(self.tokenizer), input_ids.shape, dtypetorch.long) input_ids[indices_random] random_tokens[indices_random] return TensorDataset(input_ids, attention_mask, labels) def evaluate(self, test_texts, test_labels, batch_size32): 评估模型 model self.classification_model model.eval() encodings self.tokenizer( test_texts, truncationTrue, max_lengthself.max_length, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(test_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size) correct 0 total 0 with torch.no_grad(): for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] outputs model(input_idsinput_ids, attention_maskattention_mask) _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() accuracy 100. * correct / total print(f\n测试准确率: {accuracy:.2f}%) return accuracy # 运行完整流程 def run_complete_pipeline(): 运行完整的无标签数据微调流程 # 模拟数据 np.random.seed(42) # 无标签数据大量 unlabeled_texts [ fSample domain text number {i} with various technical terms. for i in range(200) ] # 有标签数据少量 labeled_texts [fLabeled sample {i} for classification. for i in range(50)] labeled_labels [i % 2 for i in range(50)] # 测试数据 test_texts [fTest sample {i}. for i in range(20)] test_labels [i % 2 for i in range(20)] # 初始化 finetuner BertUnlabeledFineTuner( model_namebert-base-uncased, num_labels2, max_length64, # 演示用小长度 ) # 步骤 1: DAPT在无标签数据上继续预训练 # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs2) # 步骤 2: 有标签微调 # finetuner.step2_supervised_finetuning(labeled_texts, labeled_labels, epochs3) # 步骤 3: 伪标签训练 # finetuner.step3_pseudo_label_training( # labeled_texts, labeled_labels, unlabeled_texts, # confidence_threshold0.9, epochs2 # ) # 评估 # finetuner.evaluate(test_texts, test_labels) print(\n完整流程:) print(1. DAPT: 在无标签数据上继续 MLM 预训练) print(2. 监督微调: 在有标签数据上训练分类器) print(3. 伪标签: 用模型预测高置信度样本作为伪标签再次训练) print(4. 评估: 在测试集上评估最终模型) if __name__ __main__: run_complete_pipeline()六、常见陷阱与注意事项陷阱 1DAPT 训练过拟合# ❌ 在小数据集上训练太多 epoch # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs50) # 过多 # ✓ 通常 2-5 个 epoch 足够 # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs3)陷阱 2伪标签置信度阈值过低# ❌ 阈值过低引入大量错误伪标签 # trainer PseudoLabelTrainer(confidence_threshold0.5) # ✓ 使用高阈值0.9只保留高置信度预测 # trainer PseudoLabelTrainer(confidence_threshold0.95)陷阱 3忘记保存 DAPT 模型# ❌ DAPT 训练后直接加载原始 BERT 做分类 # model BertForSequenceClassification.from_pretrained(bert-base-uncased) # ✓ 使用 DAPT 后的模型 # model BertForSequenceClassification.from_pretrained(./dapt_bert)陷阱 4MLM 的 mask 策略不正确# ❌ mask 了特殊 token[CLS], [SEP], [PAD] # masked_indices torch.bernoulli(probability_matrix).bool() # 没有排除特殊 token # ✓ 排除特殊 token # special_tokens_mask get_special_tokens_mask(...) # probability_matrix.masked_fill_(special_tokens_mask, value0.0)陷阱 5学习率设置不当# DAPT 和监督微调需要不同的学习率 # DAPT: 5e-5较大因为只是继续预训练 # 监督微调: 2e-5较小避免破坏预训练知识 # 伪标签训练: 1e-5更小因为已有基础七、总结策略适用场景优点缺点DAPT领域差异大效果最好计算成本高TAPT任务特定提升针对性强需要任务相关数据伪标签有少量标注简单易行可能引入噪声一致性正则化半监督学习鲁棒性好实现复杂推荐流程1. 收集领域无标签数据 2. DAPT: 在无标签数据上继续 MLM 预训练2-5 epochs 3. 监督微调: 在有标签数据上训练分类器3-5 epochs 4. 伪标签: 生成高置信度伪标签阈值 ≥ 0.9 5. 混合训练: 有标签 伪标签数据联合训练 6. 评估: 在独立测试集上评估核心原则无标签数据的价值在于让模型理解领域语言而非直接提供监督信号。DAPT 是利用无标签数据最有效的方式伪标签是锦上添花的补充手段。通过合理利用无标签数据可以在标注数据有限的情况下显著提升 BERT 模型的性能这在实际业务场景中具有极高的实用价值。
【Bug已解决】How to fine tune BERT on unlabeled data? 解决方案问题标签BERT、无标签数据、微调、半监督学习、自监督学习、领域适应适用版本Transformers 4.x、PyTorch 1.8难度等级⭐⭐⭐⭐较高一、问题描述在实际业务场景中标注数据往往稀缺且昂贵而无标签数据则大量存在。很多开发者面临的问题是如何利用大量无标签数据来微调 BERT 模型提升模型在下游任务上的表现这是一个经典的半监督学习/自监督学习问题。BERT 本身就是通过自监督方式MLM NSP在无标签数据上预训练的我们可以在下游任务中继续利用无标签数据来进一步提升模型性能。核心挑战挑战说明无标签数据无法直接用于监督学习需要设计自监督或半监督策略领域差异通用 BERT 可能不了解领域术语计算资源限制大规模无标签数据处理耗时何时停止自监督训练容易过拟合二、错误复现场景直接用无标签数据做监督训练必然失败import torch from transformers import BertTokenizer, BertForSequenceClassification # 加载预训练 BERT tokenizer BertTokenizer.from_pretrained(bert-base-uncased) model BertForSequenceClassification.from_pretrained(bert-base-uncased, num_labels2) # 模拟无标签数据 unlabeled_texts [ The new machine learning model achieves state-of-the-art results., Quantum computing may revolutionize cryptography., The stock market experienced significant volatility today., # ... 大量无标签文本 ] # ❌ 错误无标签数据没有标签无法计算 CrossEntropyLoss texts unlabeled_texts # labels ??? # 没有标签 # 以下代码会失败因为没有标签 # loss model(input_ids, attention_mask, labelslabels)问题无标签数据没有标签无法直接使用监督学习的损失函数。三、根因分析3.1 为什么需要利用无标签数据领域适应通用 BERT 在法律、医学、金融等垂直领域表现不佳需要领域数据继续预训练。数据效率少量标注 大量无标签 仅用少量标注。语言/风格适应社交媒体文本、口语化表达与标准文本差异大。3.2 可行的策略概览策略原理适用场景继续预训练Domain-Adaptive Pretraining, DAPT在无标签数据上继续 MLM 训练领域适应任务自适应预训练TAPT在任务相关无标签数据上 MLM提升特定任务性能伪标签Pseudo-Labeling用模型预测作为伪标签半监督学习一致性正则化Consistency Regularization对输入做扰动要求输出一致半监督学习对比学习Contrastive Learning拉近相似样本推远不同样本表示学习四、解决方案方案一领域自适应预训练DAPT—— 最推荐在无标签数据上继续 BERT 的 Masked Language ModelMLM训练让模型适应领域语言。import torch import torch.nn as nn from transformers import BertTokenizer, BertForMaskedLM, BertConfig from torch.utils.data import Dataset, DataLoader import random import numpy as np # MLM 数据集 class MLMDataset(Dataset): Masked Language Model 数据集 def __init__(self, texts, tokenizer, max_length128, mlm_probability0.15): self.texts texts self.tokenizer tokenizer self.max_length max_length self.mlm_probability mlm_probability def __len__(self): return len(self.texts) def __getitem__(self, idx): text self.texts[idx] encoding self.tokenizer( text, truncationTrue, max_lengthself.max_length, paddingmax_length, return_tensorspt, ) input_ids encoding[input_ids].squeeze(0) attention_mask encoding[attention_mask].squeeze(0) # 创建 MLM 标签 labels input_ids.clone() # 随机 mask 15% 的 token probability_matrix torch.full(input_ids.shape, self.mlm_probability) special_tokens_mask [ self.tokenizer.get_special_tokens_mask(val.tolist(), already_has_special_tokensTrue) for val in input_ids ] special_tokens_mask torch.tensor(special_tokens_mask, dtypetorch.bool) probability_matrix.masked_fill_(special_tokens_mask, value0.0) masked_indices torch.bernoulli(probability_matrix).bool() labels[~masked_indices] -100 # 只计算 masked 位置的 loss # 80% 替换为 [MASK] indices_replaced torch.bernoulli(torch.full(input_ids.shape, 0.8)).bool() masked_indices input_ids[indices_replaced] self.tokenizer.convert_tokens_to_ids(self.tokenizer.mask_token) # 10% 替换为随机 token indices_random torch.bernoulli(torch.full(input_ids.shape, 0.5)).bool() masked_indices ~indices_replaced random_words torch.randint(len(self.tokenizer), input_ids.shape, dtypetorch.long) input_ids[indices_random] random_words[indices_random] # 10% 保持不变 return { input_ids: input_ids, attention_mask: attention_mask, labels: labels, } # DAPT 训练 def domain_adaptive_pretraining(texts, model_namebert-base-uncased, epochs3, batch_size16, lr5e-5): 领域自适应预训练 device torch.device(cuda if torch.cuda.is_available() else cpu) print(f使用设备: {device}) # 加载 tokenizer 和模型 tokenizer BertTokenizer.from_pretrained(model_name) model BertForMaskedLM.from_pretrained(model_name) model.to(device) # 创建数据集 dataset MLMDataset(texts, tokenizer, max_length128) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) # 优化器 optimizer torch.optim.AdamW(model.parameters(), lrlr, weight_decay0.01) # 学习率调度器 num_training_steps epochs * len(dataloader) scheduler torch.optim.lr_scheduler.LinearLR( optimizer, start_factor0.1, total_itersnum_training_steps // 10 ) print(f无标签数据量: {len(texts)}) print(f训练步数: {num_training_steps}) # 训练循环 for epoch in range(epochs): model.train() total_loss 0 for batch_idx, batch in enumerate(dataloader): input_ids batch[input_ids].to(device) attention_mask batch[attention_mask].to(device) labels batch[labels].to(device) optimizer.zero_grad() outputs model(input_idsinput_ids, attention_maskattention_mask, labelslabels) loss outputs.loss loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() scheduler.step() total_loss loss.item() if batch_idx % 50 0: print(fEpoch [{epoch1}/{epochs}] Batch [{batch_idx}/{len(dataloader)}] fLoss: {loss.item():.4f}) avg_loss total_loss / len(dataloader) print(fEpoch [{epoch1}/{epochs}] Average Loss: {avg_loss:.4f}) return model, tokenizer # 运行示例 # 模拟领域无标签数据 unlabeled_texts [ The patient presented with acute myocardial infarction and was admitted to the ICU., Clinical trials show that the new drug reduces mortality by 30% in high-risk patients., The radiologist identified a mass in the left lung during routine screening., Laboratory results indicate elevated levels of troponin and creatine kinase., The surgical team performed a coronary artery bypass graft successfully., Post-operative care includes monitoring for signs of infection and blood clots., The epidemiological study tracked 5000 participants over a ten-year period., Genetic testing revealed a mutation in the BRCA1 gene associated with breast cancer., The emergency department handled a surge of trauma cases during the holiday weekend., Telemedicine platforms have expanded access to specialist consultations in rural areas., ] * 50 # 复制以增加数据量 # 运行 DAPT # model, tokenizer domain_adaptive_pretraining(unlabeled_texts, epochs3)方案二伪标签Pseudo-Labeling先用少量标注数据训练初始模型然后用该模型对无标签数据预测将高置信度预测作为伪标签。import torch import torch.nn as nn from transformers import BertTokenizer, BertForSequenceClassification from torch.utils.data import Dataset, DataLoader, TensorDataset import numpy as np class PseudoLabelTrainer: 伪标签训练器 def __init__(self, model_namebert-base-uncased, num_labels2, confidence_threshold0.9, deviceauto): self.model_name model_name self.num_labels num_labels self.confidence_threshold confidence_threshold if device auto: self.device torch.device(cuda if torch.cuda.is_available() else cpu) else: self.device torch.device(device) self.tokenizer BertTokenizer.from_pretrained(model_name) self.model BertForSequenceClassification.from_pretrained( model_name, num_labelsnum_labels ).to(self.device) def train_on_labeled(self, labeled_texts, labeled_labels, epochs5, batch_size16, lr2e-5): 在有标签数据上训练 # 编码 encodings self.tokenizer( labeled_texts, truncationTrue, max_length128, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(labeled_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) optimizer torch.optim.AdamW(self.model.parameters(), lrlr) criterion nn.CrossEntropyLoss() self.model.train() for epoch in range(epochs): total_loss 0 correct 0 total 0 for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] optimizer.zero_grad() outputs self.model(input_idsinput_ids, attention_maskattention_mask) loss criterion(outputs.logits, labels) loss.backward() optimizer.step() total_loss loss.item() _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() print(f[有标签训练] Epoch [{epoch1}/{epochs}] fLoss: {total_loss/len(dataloader):.4f}, fAcc: {100.*correct/total:.2f}%) def generate_pseudo_labels(self, unlabeled_texts, batch_size32): 为无标签数据生成伪标签 encodings self.tokenizer( unlabeled_texts, truncationTrue, max_length128, paddingTrue, return_tensorspt ) dataset TensorDataset(encodings[input_ids], encodings[attention_mask]) dataloader DataLoader(dataset, batch_sizebatch_size) self.model.eval() all_probs [] all_preds [] with torch.no_grad(): for batch in dataloader: input_ids, attention_mask [b.to(self.device) for b in batch] outputs self.model(input_idsinput_ids, attention_maskattention_mask) probs torch.softmax(outputs.logits, dim-1) confidence, preds probs.max(dim-1) all_probs.extend(confidence.cpu().tolist()) all_preds.extend(preds.cpu().tolist()) # 筛选高置信度样本 pseudo_labeled [] pseudo_labels [] for i, (text, pred, conf) in enumerate(zip(unlabeled_texts, all_preds, all_probs)): if conf self.confidence_threshold: pseudo_labeled.append(text) pseudo_labels.append(pred) print(f伪标签生成: {len(pseudo_labeled)}/{len(unlabeled_texts)} f({100.*len(pseudo_labeled)/len(unlabeled_texts):.1f}%) 样本置信度 {self.confidence_threshold}) return pseudo_labeled, pseudo_labels def train_with_pseudo_labels(self, labeled_texts, labeled_labels, unlabeled_texts, epochs5, batch_size16, lr2e-5): 结合有标签和伪标签数据训练 # 生成伪标签 pseudo_texts, pseudo_labels self.generate_pseudo_labels(unlabeled_texts) # 合并数据 all_texts labeled_texts pseudo_texts all_labels labeled_labels pseudo_labels # 编码 encodings self.tokenizer( all_texts, truncationTrue, max_length128, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(all_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) optimizer torch.optim.AdamW(self.model.parameters(), lrlr) criterion nn.CrossEntropyLoss() self.model.train() for epoch in range(epochs): total_loss 0 correct 0 total 0 for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] optimizer.zero_grad() outputs self.model(input_idsinput_ids, attention_maskattention_mask) loss criterion(outputs.logits, labels) loss.backward() optimizer.step() total_loss loss.item() _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() print(f[伪标签训练] Epoch [{epoch1}/{epochs}] fLoss: {total_loss/len(dataloader):.4f}, fAcc: {100.*correct/total:.2f}%) # 使用示例 # trainer PseudoLabelTrainer(confidence_threshold0.9) # trainer.train_on_labeled(labeled_texts, labeled_labels, epochs5) # trainer.train_with_pseudo_labels(labeled_texts, labeled_labels, unlabeled_texts, epochs5)方案三一致性正则化对同一输入施加不同扰动要求模型输出保持一致。import torch import torch.nn as nn from transformers import BertTokenizer, BertForSequenceClassification import random class ConsistencyRegularizationTrainer: 一致性正则化训练器 def __init__(self, model_namebert-base-uncased, num_labels2): self.device torch.device(cuda if torch.cuda.is_available() else cpu) self.tokenizer BertTokenizer.from_pretrained(model_name) self.model BertForSequenceClassification.from_pretrained( model_name, num_labelsnum_labels ).to(self.device) def augment_text(self, text, aug_prob0.1): 文本数据增强 words text.split() augmented [] for word in words: if random.random() aug_prob: # 随机删除 continue elif random.random() aug_prob: # 随机替换为 [MASK] augmented.append([MASK]) else: augmented.append(word) # 随机打乱 if random.random() 0.1 and len(augmented) 2: idx random.randint(0, len(augmented) - 2) augmented[idx], augmented[idx 1] augmented[idx 1], augmented[idx] return .join(augmented) if augmented else text def train_step(self, labeled_batch, unlabeled_batch, optimizer, lambda_consistency1.0, lambda_supervised1.0): 一步训练 self.model.train() optimizer.zero_grad() # 有标签数据的监督损失 labeled_input_ids labeled_batch[input_ids].to(self.device) labeled_attention_mask labeled_batch[attention_mask].to(self.device) labeled_labels labeled_batch[labels].to(self.device) labeled_outputs self.model( input_idslabeled_input_ids, attention_masklabeled_attention_mask ) supervised_loss nn.CrossEntropyLoss()( labeled_outputs.logits, labeled_labels ) # 无标签数据的一致性损失 unlabeled_input_ids unlabeled_batch[input_ids].to(self.device) unlabeled_attention_mask unlabeled_batch[attention_mask].to(self.device) # 原始预测 with torch.no_grad(): original_outputs self.model( input_idsunlabeled_input_ids, attention_maskunlabeled_attention_mask ) original_probs torch.softmax(original_outputs.logits, dim-1) # 增强后预测 augmented_input_ids unlabeled_batch[augmented_input_ids].to(self.device) augmented_attention_mask unlabeled_batch[augmented_attention_mask].to(self.device) augmented_outputs self.model( input_idsaugmented_input_ids, attention_maskaugmented_attention_mask ) augmented_probs torch.softmax(augmented_outputs.logits, dim-1) # KL 散度作为一致性损失 consistency_loss nn.KLDivLoss(reductionbatchmean)( torch.log(augmented_probs 1e-8), original_probs.detach() ) # 总损失 total_loss (lambda_supervised * supervised_loss lambda_consistency * consistency_loss) total_loss.backward() optimizer.step() return { supervised_loss: supervised_loss.item(), consistency_loss: consistency_loss.item(), total_loss: total_loss.item(), }五、完整修复代码import torch import torch.nn as nn from transformers import ( BertTokenizer, BertForMaskedLM, BertForSequenceClassification, get_linear_schedule_with_warmup ) from torch.utils.data import Dataset, DataLoader, TensorDataset import numpy as np import random import os # 完整 DAPT 下游微调流程 class BertUnlabeledFineTuner: BERT 无标签数据微调完整流程 def __init__(self, model_namebert-base-uncased, num_labels2, max_length128, deviceauto): self.model_name model_name self.num_labels num_labels self.max_length max_length if device auto: self.device torch.device(cuda if torch.cuda.is_available() else cpu) else: self.device torch.device(device) self.tokenizer BertTokenizer.from_pretrained(model_name) print(f设备: {self.device}) def step1_domain_adaptive_pretraining(self, unlabeled_texts, epochs3, batch_size16, lr5e-5): 步骤 1: 领域自适应预训练 (DAPT) print(\n * 60) print(步骤 1: 领域自适应预训练 (DAPT)) print( * 60) # 加载 MLM 模型 model BertForMaskedLM.from_pretrained(self.model_name).to(self.device) # 创建 MLM 数据集 dataset self._create_mlm_dataset(unlabeled_texts) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) # 优化器 optimizer torch.optim.AdamW(model.parameters(), lrlr, weight_decay0.01) total_steps epochs * len(dataloader) scheduler get_linear_schedule_with_warmup( optimizer, num_warmup_stepstotal_steps // 10, num_training_stepstotal_steps ) # 训练 for epoch in range(epochs): model.train() total_loss 0 for batch in dataloader: input_ids batch[input_ids].to(self.device) attention_mask batch[attention_mask].to(self.device) labels batch[labels].to(self.device) optimizer.zero_grad() outputs model(input_idsinput_ids, attention_maskattention_mask, labelslabels) loss outputs.loss loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() scheduler.step() total_loss loss.item() print(f Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(dataloader):.4f}) # 保存领域适应后的模型 self.dapt_model_path ./dapt_bert model.save_pretrained(self.dapt_model_path) self.tokenizer.save_pretrained(self.dapt_model_path) print(f 领域适应模型已保存到: {self.dapt_model_path}) return model def step2_supervised_finetuning(self, labeled_texts, labeled_labels, epochs5, batch_size16, lr2e-5): 步骤 2: 在有标签数据上微调 print(\n * 60) print(步骤 2: 有标签数据监督微调) print( * 60) # 加载 DAPT 后的模型如果有否则用原始 BERT model_path getattr(self, dapt_model_path, self.model_name) model BertForSequenceClassification.from_pretrained( model_path, num_labelsself.num_labels ).to(self.device) # 编码有标签数据 encodings self.tokenizer( labeled_texts, truncationTrue, max_lengthself.max_length, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(labeled_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) optimizer torch.optim.AdamW(model.parameters(), lrlr, weight_decay0.01) total_steps epochs * len(dataloader) scheduler get_linear_schedule_with_warmup( optimizer, num_warmup_stepstotal_steps // 10, num_training_stepstotal_steps ) criterion nn.CrossEntropyLoss() for epoch in range(epochs): model.train() total_loss 0 correct 0 total 0 for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] optimizer.zero_grad() outputs model(input_idsinput_ids, attention_maskattention_mask) loss criterion(outputs.logits, labels) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) optimizer.step() scheduler.step() total_loss loss.item() _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() print(f Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(dataloader):.4f}, fAcc: {100.*correct/total:.2f}%) self.classification_model model return model def step3_pseudo_label_training(self, labeled_texts, labeled_labels, unlabeled_texts, confidence_threshold0.9, epochs3, batch_size16, lr1e-5): 步骤 3: 伪标签训练 print(\n * 60) print(步骤 3: 伪标签训练) print( * 60) model self.classification_model # 生成伪标签 model.eval() pseudo_texts [] pseudo_labels [] with torch.no_grad(): for text in unlabeled_texts: encoding self.tokenizer( text, truncationTrue, max_lengthself.max_length, paddingTrue, return_tensorspt ) input_ids encoding[input_ids].to(self.device) attention_mask encoding[attention_mask].to(self.device) output model(input_idsinput_ids, attention_maskattention_mask) probs torch.softmax(output.logits, dim-1) confidence, pred probs.max(dim-1) if confidence.item() confidence_threshold: pseudo_texts.append(text) pseudo_labels.append(pred.item()) print(f 伪标签样本数: {len(pseudo_texts)}/{len(unlabeled_texts)}) if len(pseudo_texts) 0: print( 无高置信度伪标签跳过此步骤) return model # 合并数据 all_texts labeled_texts pseudo_texts all_labels labeled_labels pseudo_labels # 训练 encodings self.tokenizer( all_texts, truncationTrue, max_lengthself.max_length, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(all_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) optimizer torch.optim.AdamW(model.parameters(), lrlr, weight_decay0.01) criterion nn.CrossEntropyLoss() for epoch in range(epochs): model.train() total_loss 0 correct 0 total 0 for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] optimizer.zero_grad() outputs model(input_idsinput_ids, attention_maskattention_mask) loss criterion(outputs.logits, labels) loss.backward() optimizer.step() total_loss loss.item() _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() print(f Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(dataloader):.4f}, fAcc: {100.*correct/total:.2f}%) return model def _create_mlm_dataset(self, texts, mlm_probability0.15): 创建 MLM 数据集 encoded self.tokenizer( texts, truncationTrue, max_lengthself.max_length, paddingmax_length, return_tensorspt ) input_ids encoded[input_ids] attention_mask encoded[attention_mask] labels input_ids.clone() # Mask probability_matrix torch.full(input_ids.shape, mlm_probability) special_tokens_mask [ self.tokenizer.get_special_tokens_mask(ids.tolist(), already_has_special_tokensTrue) for ids in input_ids ] special_tokens_mask torch.tensor(special_tokens_mask, dtypetorch.bool) probability_matrix.masked_fill_(special_tokens_mask, value0.0) masked_indices torch.bernoulli(probability_matrix).bool() labels[~masked_indices] -100 # 80% [MASK] indices_replaced torch.bernoulli(torch.full(input_ids.shape, 0.8)).bool() masked_indices input_ids[indices_replaced] self.tokenizer.mask_token_id # 10% random indices_random (torch.bernoulli(torch.full(input_ids.shape, 0.5)).bool() masked_indices ~indices_replaced) random_tokens torch.randint(len(self.tokenizer), input_ids.shape, dtypetorch.long) input_ids[indices_random] random_tokens[indices_random] return TensorDataset(input_ids, attention_mask, labels) def evaluate(self, test_texts, test_labels, batch_size32): 评估模型 model self.classification_model model.eval() encodings self.tokenizer( test_texts, truncationTrue, max_lengthself.max_length, paddingTrue, return_tensorspt ) dataset TensorDataset( encodings[input_ids], encodings[attention_mask], torch.tensor(test_labels) ) dataloader DataLoader(dataset, batch_sizebatch_size) correct 0 total 0 with torch.no_grad(): for batch in dataloader: input_ids, attention_mask, labels [b.to(self.device) for b in batch] outputs model(input_idsinput_ids, attention_maskattention_mask) _, predicted outputs.logits.max(1) total labels.size(0) correct predicted.eq(labels).sum().item() accuracy 100. * correct / total print(f\n测试准确率: {accuracy:.2f}%) return accuracy # 运行完整流程 def run_complete_pipeline(): 运行完整的无标签数据微调流程 # 模拟数据 np.random.seed(42) # 无标签数据大量 unlabeled_texts [ fSample domain text number {i} with various technical terms. for i in range(200) ] # 有标签数据少量 labeled_texts [fLabeled sample {i} for classification. for i in range(50)] labeled_labels [i % 2 for i in range(50)] # 测试数据 test_texts [fTest sample {i}. for i in range(20)] test_labels [i % 2 for i in range(20)] # 初始化 finetuner BertUnlabeledFineTuner( model_namebert-base-uncased, num_labels2, max_length64, # 演示用小长度 ) # 步骤 1: DAPT在无标签数据上继续预训练 # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs2) # 步骤 2: 有标签微调 # finetuner.step2_supervised_finetuning(labeled_texts, labeled_labels, epochs3) # 步骤 3: 伪标签训练 # finetuner.step3_pseudo_label_training( # labeled_texts, labeled_labels, unlabeled_texts, # confidence_threshold0.9, epochs2 # ) # 评估 # finetuner.evaluate(test_texts, test_labels) print(\n完整流程:) print(1. DAPT: 在无标签数据上继续 MLM 预训练) print(2. 监督微调: 在有标签数据上训练分类器) print(3. 伪标签: 用模型预测高置信度样本作为伪标签再次训练) print(4. 评估: 在测试集上评估最终模型) if __name__ __main__: run_complete_pipeline()六、常见陷阱与注意事项陷阱 1DAPT 训练过拟合# ❌ 在小数据集上训练太多 epoch # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs50) # 过多 # ✓ 通常 2-5 个 epoch 足够 # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs3)陷阱 2伪标签置信度阈值过低# ❌ 阈值过低引入大量错误伪标签 # trainer PseudoLabelTrainer(confidence_threshold0.5) # ✓ 使用高阈值0.9只保留高置信度预测 # trainer PseudoLabelTrainer(confidence_threshold0.95)陷阱 3忘记保存 DAPT 模型# ❌ DAPT 训练后直接加载原始 BERT 做分类 # model BertForSequenceClassification.from_pretrained(bert-base-uncased) # ✓ 使用 DAPT 后的模型 # model BertForSequenceClassification.from_pretrained(./dapt_bert)陷阱 4MLM 的 mask 策略不正确# ❌ mask 了特殊 token[CLS], [SEP], [PAD] # masked_indices torch.bernoulli(probability_matrix).bool() # 没有排除特殊 token # ✓ 排除特殊 token # special_tokens_mask get_special_tokens_mask(...) # probability_matrix.masked_fill_(special_tokens_mask, value0.0)陷阱 5学习率设置不当# DAPT 和监督微调需要不同的学习率 # DAPT: 5e-5较大因为只是继续预训练 # 监督微调: 2e-5较小避免破坏预训练知识 # 伪标签训练: 1e-5更小因为已有基础七、总结策略适用场景优点缺点DAPT领域差异大效果最好计算成本高TAPT任务特定提升针对性强需要任务相关数据伪标签有少量标注简单易行可能引入噪声一致性正则化半监督学习鲁棒性好实现复杂推荐流程1. 收集领域无标签数据 2. DAPT: 在无标签数据上继续 MLM 预训练2-5 epochs 3. 监督微调: 在有标签数据上训练分类器3-5 epochs 4. 伪标签: 生成高置信度伪标签阈值 ≥ 0.9 5. 混合训练: 有标签 伪标签数据联合训练 6. 评估: 在独立测试集上评估核心原则无标签数据的价值在于让模型理解领域语言而非直接提供监督信号。DAPT 是利用无标签数据最有效的方式伪标签是锦上添花的补充手段。通过合理利用无标签数据可以在标注数据有限的情况下显著提升 BERT 模型的性能这在实际业务场景中具有极高的实用价值。
