AI 迁移学习踩坑记录

# 原始数据统计
raw_samples = 180000
raw_unique = 125000
after_dedup = 120000
after_quality_filter = 150000

# 最终用于预训练的数据
train_size = 120000
vocab_coverage = 0.85  # 医疗术语覆盖率

# 关键配置片段
from transformers import BertConfig, BertForMaskedLM, BertTokenizer

config = BertConfig.from_pretrained('hfl/chinese-roberta-wwm-ext')
config.vocab_size = tokenizer.vocab_size

model = BertForMaskedLM.from_pretrained(
    'hfl/chinese-roberta-wwm-ext',
    config=config
)

# 自定义 masking 策略
def mask_tokens(inputs, tokenizer, mlm_probability=0.15, term_mask_prob=0.20):
    """
    对医疗术语提高 masking 概率
    """
    labels = inputs.clone()
    probability_matrix = torch.full(labels.shape, mlm_probability)
    special_tokens_mask = [
        tokenizer.get_special_tokens_mask(val, already_has_special_tokens=True)
        for val in labels.tolist()
    ]
    probability_matrix.masked_fill_(torch.tensor(special_tokens_mask, dtype=torch.bool), value=0.0)

    # 识别医疗术语（这里简化处理）
    medical_terms = identify_medical_terms(inputs, tokenizer)
    for term_positions in medical_terms:
        probability_matrix[term_positions] = term_mask_prob

    masked_indices = torch.bernoulli(probability_matrix).bool()
    labels[~masked_indices] = -100
    indices_replaced = torch.bernoulli(torch.full(labels.shape, 0.8)).bool() & masked_indices
    inputs[indices_replaced] = tokenizer.convert_tokens_to_ids(tokenizer.mask_token)

    return inputs, labels

# 数据增强示例
def augment_text(text, augment_ratio=1.5):
    """对小类数据进行增强"""
    augment_methods = [
        synonym_replace,
        back_translate,
        random_insert,
    ]
    aug_texts = [text]
    for _ in range(int(augment_ratio - 1)):
        method = random.choice(augment_methods)
        aug_texts.append(method(text))
    return aug_texts

# 类别统计和平衡
class_counts = Counter(train_labels)
min_class_size = 200
balanced_data = []

for text, label in zip(train_texts, train_labels):
    if class_counts[label] < min_class_size:
        aug_texts = augment_text(text, min_class_size / class_counts[label])
        balanced_data.extend([(t, label) for t in aug_texts])
    else:
        balanced_data.append((text, label))

from transformers import BertForSequenceClassification

model = BertForSequenceClassification.from_pretrained(
    './domain_pretrained_medical_bert',
    num_labels=10
)

training_args = TrainingArguments(
    output_dir='./results',
    num_train_epochs=5,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=32,
    warmup_steps=100,
    weight_decay=0.01,
    logging_dir='./logs',
    logging_steps=50,
    evaluation_strategy='epoch',
    save_strategy='epoch',
    load_best_model_at_end=True,
    metric_for_best_model='f1',
    learning_rate=2e-5,
)

def is_valid_sample(text):
    """数据质量过滤"""
    # 过滤广告关键词
    ad_keywords = ['广告', '推广', '优惠', '购买链接']
    if any(kw in text for kw in ad_keywords):
        return False

    # 过滤过短文本
    if len(text.strip()) < 10:
        return False

    # 过滤非中文为主的内容
    chinese_ratio = len(re.findall(r'[一-鿿]', text)) / len(text)
    if chinese_ratio < 0.7:
        return False

    return True

# 混合训练数据
domain_train = load_medical_qa_data()
general_train = load_general_qa_data()[:200]  # 少量通用数据

mixed_train = domain_train + general_train
shuffle(mixed_train)

from sklearn.metrics import precision_recall_fscore_support

def compute_metrics(pred):
    labels = pred.label_ids
    preds = pred.predictions.argmax(-1)

    precision, recall, f1, _ = precision_recall_fscore_support(
        labels, preds, average='weighted'
    )
    macro_f1 = precision_recall_fscore_support(
        labels, preds, average='macro'
    )[2]

    return {
        'accuracy': (preds == labels).mean(),
        'f1': f1,
        'precision': precision,
        'recall': recall,
        'macro_f1': macro_f1  # 关注小类表现
    }