企业官网建设流程全解析

【Bug已解决】How to fine tune BERT on unlabeled data? 解决方案

问题标签：BERT、无标签数据、微调、半监督学习、自监督学习、领域适应
适用版本：Transformers 4.x、PyTorch 1.8+
难度等级：⭐⭐⭐⭐（较高）

一、问题描述

在实际业务场景中，标注数据往往稀缺且昂贵，而无标签数据则大量存在。很多开发者面临的问题是：

如何利用大量无标签数据来微调 BERT 模型，提升模型在下游任务上的表现？

这是一个经典的半监督学习/自监督学习问题。BERT 本身就是通过自监督方式（MLM + NSP）在无标签数据上预训练的，我们可以在下游任务中继续利用无标签数据来进一步提升模型性能。

核心挑战

二、错误复现

场景：直接用无标签数据做监督训练（必然失败）

import torch from transformers import BertTokenizer, BertForSequenceClassification # 加载预训练 BERT tokenizer = BertTokenizer.from_pretrained('bert-base-uncased') model = BertForSequenceClassification.from_pretrained('bert-base-uncased', num_labels=2) # 模拟无标签数据 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, labels=labels)

问题：无标签数据没有标签，无法直接使用监督学习的损失函数。

三、根因分析

3.1 为什么需要利用无标签数据

1. 领域适应：通用 BERT 在法律、医学、金融等垂直领域表现不佳，需要领域数据继续预训练。
2. 数据效率：少量标注 + 大量无标签 > 仅用少量标注。
3. 语言/风格适应：社交媒体文本、口语化表达与标准文本差异大。

3.2 可行的策略概览

四、解决方案

方案一：领域自适应预训练（DAPT）—— 最推荐

在无标签数据上继续 BERT 的 Masked Language Model（MLM）训练，让模型适应领域语言。

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_length=128, mlm_probability=0.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, truncation=True, max_length=self.max_length, padding='max_length', return_tensors='pt', ) 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_tokens=True) for val in input_ids ] special_tokens_mask = torch.tensor(special_tokens_mask, dtype=torch.bool) probability_matrix.masked_fill_(special_tokens_mask, value=0.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, dtype=torch.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_name='bert-base-uncased', epochs=3, batch_size=16, lr=5e-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_length=128) dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True) # 优化器 optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.01) # 学习率调度器 num_training_steps = epochs * len(dataloader) scheduler = torch.optim.lr_scheduler.LinearLR( optimizer, start_factor=0.1, total_iters=num_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_ids=input_ids, attention_mask=attention_mask, labels=labels) loss = outputs.loss loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() scheduler.step() total_loss += loss.item() if batch_idx % 50 == 0: print(f"Epoch [{epoch+1}/{epochs}] Batch [{batch_idx}/{len(dataloader)}] " f"Loss: {loss.item():.4f}") avg_loss = total_loss / len(dataloader) print(f"Epoch [{epoch+1}/{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, epochs=3)

方案二：伪标签（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_name='bert-base-uncased', num_labels=2, confidence_threshold=0.9, device='auto'): 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_labels=num_labels ).to(self.device) def train_on_labeled(self, labeled_texts, labeled_labels, epochs=5, batch_size=16, lr=2e-5): """在有标签数据上训练""" # 编码 encodings = self.tokenizer( labeled_texts, truncation=True, max_length=128, padding=True, return_tensors='pt' ) dataset = TensorDataset( encodings['input_ids'], encodings['attention_mask'], torch.tensor(labeled_labels) ) dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True) optimizer = torch.optim.AdamW(self.model.parameters(), lr=lr) 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_ids=input_ids, attention_mask=attention_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 [{epoch+1}/{epochs}] " f"Loss: {total_loss/len(dataloader):.4f}, " f"Acc: {100.*correct/total:.2f}%") def generate_pseudo_labels(self, unlabeled_texts, batch_size=32): """为无标签数据生成伪标签""" encodings = self.tokenizer( unlabeled_texts, truncation=True, max_length=128, padding=True, return_tensors='pt' ) dataset = TensorDataset(encodings['input_ids'], encodings['attention_mask']) dataloader = DataLoader(dataset, batch_size=batch_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_ids=input_ids, attention_mask=attention_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, epochs=5, batch_size=16, lr=2e-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, truncation=True, max_length=128, padding=True, return_tensors='pt' ) dataset = TensorDataset( encodings['input_ids'], encodings['attention_mask'], torch.tensor(all_labels) ) dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True) optimizer = torch.optim.AdamW(self.model.parameters(), lr=lr) 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_ids=input_ids, attention_mask=attention_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 [{epoch+1}/{epochs}] " f"Loss: {total_loss/len(dataloader):.4f}, " f"Acc: {100.*correct/total:.2f}%") # 使用示例 # trainer = PseudoLabelTrainer(confidence_threshold=0.9) # trainer.train_on_labeled(labeled_texts, labeled_labels, epochs=5) # trainer.train_with_pseudo_labels(labeled_texts, labeled_labels, unlabeled_texts, epochs=5)

方案三：一致性正则化

对同一输入施加不同扰动，要求模型输出保持一致。

import torch import torch.nn as nn from transformers import BertTokenizer, BertForSequenceClassification import random class ConsistencyRegularizationTrainer: """一致性正则化训练器""" def __init__(self, model_name='bert-base-uncased', num_labels=2): 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_labels=num_labels ).to(self.device) def augment_text(self, text, aug_prob=0.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_consistency=1.0, lambda_supervised=1.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_ids=labeled_input_ids, attention_mask=labeled_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_ids=unlabeled_input_ids, attention_mask=unlabeled_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_ids=augmented_input_ids, attention_mask=augmented_attention_mask ) augmented_probs = torch.softmax(augmented_outputs.logits, dim=-1) # KL 散度作为一致性损失 consistency_loss = nn.KLDivLoss(reduction='batchmean')( 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_name='bert-base-uncased', num_labels=2, max_length=128, device='auto'): 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, epochs=3, batch_size=16, lr=5e-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_size=batch_size, shuffle=True) # 优化器 optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.01) total_steps = epochs * len(dataloader) scheduler = get_linear_schedule_with_warmup( optimizer, num_warmup_steps=total_steps // 10, num_training_steps=total_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_ids=input_ids, attention_mask=attention_mask, labels=labels) 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 [{epoch+1}/{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, epochs=5, batch_size=16, lr=2e-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_labels=self.num_labels ).to(self.device) # 编码有标签数据 encodings = self.tokenizer( labeled_texts, truncation=True, max_length=self.max_length, padding=True, return_tensors='pt' ) dataset = TensorDataset( encodings['input_ids'], encodings['attention_mask'], torch.tensor(labeled_labels) ) dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True) optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.01) total_steps = epochs * len(dataloader) scheduler = get_linear_schedule_with_warmup( optimizer, num_warmup_steps=total_steps // 10, num_training_steps=total_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_ids=input_ids, attention_mask=attention_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 [{epoch+1}/{epochs}] Loss: {total_loss/len(dataloader):.4f}, " f"Acc: {100.*correct/total:.2f}%") self.classification_model = model return model def step3_pseudo_label_training(self, labeled_texts, labeled_labels, unlabeled_texts, confidence_threshold=0.9, epochs=3, batch_size=16, lr=1e-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, truncation=True, max_length=self.max_length, padding=True, return_tensors='pt' ) input_ids = encoding['input_ids'].to(self.device) attention_mask = encoding['attention_mask'].to(self.device) output = model(input_ids=input_ids, attention_mask=attention_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, truncation=True, max_length=self.max_length, padding=True, return_tensors='pt' ) dataset = TensorDataset( encodings['input_ids'], encodings['attention_mask'], torch.tensor(all_labels) ) dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True) optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.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_ids=input_ids, attention_mask=attention_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 [{epoch+1}/{epochs}] Loss: {total_loss/len(dataloader):.4f}, " f"Acc: {100.*correct/total:.2f}%") return model def _create_mlm_dataset(self, texts, mlm_probability=0.15): """创建 MLM 数据集""" encoded = self.tokenizer( texts, truncation=True, max_length=self.max_length, padding='max_length', return_tensors='pt' ) 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_tokens=True) for ids in input_ids ] special_tokens_mask = torch.tensor(special_tokens_mask, dtype=torch.bool) probability_matrix.masked_fill_(special_tokens_mask, value=0.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, dtype=torch.long) input_ids[indices_random] = random_tokens[indices_random] return TensorDataset(input_ids, attention_mask, labels) def evaluate(self, test_texts, test_labels, batch_size=32): """评估模型""" model = self.classification_model model.eval() encodings = self.tokenizer( test_texts, truncation=True, max_length=self.max_length, padding=True, return_tensors='pt' ) dataset = TensorDataset( encodings['input_ids'], encodings['attention_mask'], torch.tensor(test_labels) ) dataloader = DataLoader(dataset, batch_size=batch_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_ids=input_ids, attention_mask=attention_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 = [ f"Sample domain text number {i} with various technical terms." for i in range(200) ] # 有标签数据（少量） labeled_texts = [f"Labeled sample {i} for classification." for i in range(50)] labeled_labels = [i % 2 for i in range(50)] # 测试数据 test_texts = [f"Test sample {i}." for i in range(20)] test_labels = [i % 2 for i in range(20)] # 初始化 finetuner = BertUnlabeledFineTuner( model_name='bert-base-uncased', num_labels=2, max_length=64, # 演示用小长度 ) # 步骤 1: DAPT（在无标签数据上继续预训练） # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs=2) # 步骤 2: 有标签微调 # finetuner.step2_supervised_finetuning(labeled_texts, labeled_labels, epochs=3) # 步骤 3: 伪标签训练 # finetuner.step3_pseudo_label_training( # labeled_texts, labeled_labels, unlabeled_texts, # confidence_threshold=0.9, epochs=2 # ) # 评估 # finetuner.evaluate(test_texts, test_labels) print("\n完整流程:") print("1. DAPT: 在无标签数据上继续 MLM 预训练") print("2. 监督微调: 在有标签数据上训练分类器") print("3. 伪标签: 用模型预测高置信度样本作为伪标签，再次训练") print("4. 评估: 在测试集上评估最终模型") if __name__ == '__main__': run_complete_pipeline()

六、常见陷阱与注意事项

陷阱 1：DAPT 训练过拟合

# ❌ 在小数据集上训练太多 epoch # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs=50) # 过多！ # ✓ 通常 2-5 个 epoch 足够 # finetuner.step1_domain_adaptive_pretraining(unlabeled_texts, epochs=3)

陷阱 2：伪标签置信度阈值过低

# ❌ 阈值过低，引入大量错误伪标签 # trainer = PseudoLabelTrainer(confidence_threshold=0.5) # ✓ 使用高阈值（0.9+），只保留高置信度预测 # trainer = PseudoLabelTrainer(confidence_threshold=0.95)

陷阱 3：忘记保存 DAPT 模型

# ❌ DAPT 训练后直接加载原始 BERT 做分类 # model = BertForSequenceClassification.from_pretrained('bert-base-uncased') # ✓ 使用 DAPT 后的模型 # model = BertForSequenceClassification.from_pretrained('./dapt_bert')

陷阱 4：MLM 的 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, value=0.0)

陷阱 5：学习率设置不当

# DAPT 和监督微调需要不同的学习率 # DAPT: 5e-5（较大，因为只是继续预训练） # 监督微调: 2e-5（较小，避免破坏预训练知识） # 伪标签训练: 1e-5（更小，因为已有基础）

七、总结

推荐流程

1. 收集领域无标签数据 2. DAPT: 在无标签数据上继续 MLM 预训练（2-5 epochs） 3. 监督微调: 在有标签数据上训练分类器（3-5 epochs） 4. 伪标签: 生成高置信度伪标签（阈值 ≥ 0.9） 5. 混合训练: 有标签 + 伪标签数据联合训练 6. 评估: 在独立测试集上评估

核心原则

无标签数据的价值在于让模型"理解"领域语言，而非直接提供监督信号。DAPT 是利用无标签数据最有效的方式，伪标签是锦上添花的补充手段。

通过合理利用无标签数据，可以在标注数据有限的情况下显著提升 BERT 模型的性能，这在实际业务场景中具有极高的实用价值。