📖 第10章：预训练语言模型¶

pretraining_paradigms = {
    "自编码器(AE)": {
        "代表": "BERT",
        "方式": "掩码语言模型(MLM) - 双向",
        "训练": "随机遮盖15%的token，预测被遮盖的token",
        "优点": "可以看到上下文，适合理解任务",
        "缺点": "不能直接生成文本",
    },
    "自回归(AR)": {
        "代表": "GPT",
        "方式": "因果语言模型(CLM) - 单向（左到右）",
        "训练": "根据前文预测下一个token",
        "优点": "天然适合生成文本",
        "缺点": "只能看到左边，可能损失双向信息",
    },
    "编码器-解码器(Enc-Dec)": {
        "代表": "T5, BART",
        "方式": "文本到文本的统一框架",
        "训练": "将所有任务转化为文本生成",
        "优点": "统一框架，灵活性高",
    },
}

for paradigm, info in pretraining_paradigms.items():
    print(f"\n📌 {paradigm}: {info['代表']}")
    print(f"   训练方式: {info['方式']}")
    print(f"   优点: {info['优点']}")

import torch
import torch.nn as nn

class SimpleELMo(nn.Module):  # 继承nn.Module定义网络层
    """简化版ELMo"""

    def __init__(self, vocab_size, embed_dim=128, hidden_dim=256, num_layers=2):
        super().__init__()  # super()调用父类方法

        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.embed_proj = nn.Linear(embed_dim, hidden_dim)  # 将embedding投影到hidden_dim

        self.forward_lstms = nn.ModuleList([
            nn.LSTM(embed_dim if i == 0 else hidden_dim, hidden_dim, batch_first=True)
            for i in range(num_layers)
        ])

        self.backward_lstms = nn.ModuleList([
            nn.LSTM(embed_dim if i == 0 else hidden_dim, hidden_dim, batch_first=True)
            for i in range(num_layers)
        ])

        # 各层权重（可学习）
        self.layer_weights = nn.Parameter(torch.ones(num_layers + 1) / (num_layers + 1))
        self.gamma = nn.Parameter(torch.tensor(1.0))

    def forward(self, x):
        embedded = self.embedding(x)

        all_layers = [self.embed_proj(embedded)]  # 投影到hidden_dim以匹配后续层维度

        fwd_input = embedded
        bwd_input = torch.flip(embedded, [1])

        for fwd_lstm, bwd_lstm in zip(self.forward_lstms, self.backward_lstms):  # zip按位置配对
            fwd_out, _ = fwd_lstm(fwd_input)
            bwd_out, _ = bwd_lstm(bwd_input)
            bwd_out = torch.flip(bwd_out, [1])

            combined = fwd_out + bwd_out
            all_layers.append(combined)
            fwd_input = fwd_out
            bwd_input = torch.flip(bwd_out, [1])

        # 加权求和
        weights = torch.softmax(self.layer_weights, dim=0)
        elmo_repr = self.gamma * sum(w * layer for w, layer in zip(weights, all_layers))

        return elmo_repr

model = SimpleELMo(vocab_size=5000)
x = torch.randint(0, 5000, (2, 10))
output = model(x)
print(f"ELMo输出: {output.shape}")  # (2, 10, 256)

gpt_evolution = {
    "GPT-1 (2018)": {
        "参数量": "1.17亿",
        "训练数据": "BooksCorpus (8亿词)",
        "特点": "首次证明Transformer预训练+微调的有效性",
    },
    "GPT-2 (2019)": {
        "参数量": "15亿",
        "训练数据": "WebText (40GB)",
        "特点": "Zero-shot能力，文本生成质量飞跃",
    },
    "GPT-3 (2020)": {
        "参数量": "1750亿",
        "训练数据": "CommonCrawl等 (570GB)",
        "特点": "In-Context Learning，Few-shot能力",
    },
    "GPT-4 (2023)": {
        "参数量": "未公开",
        "训练数据": "未公开",
        "特点": "多模态，推理能力大幅提升",
    },
}

for name, info in gpt_evolution.items():
    print(f"\n🔹 {name}")
    for k, v in info.items():
        print(f"   {k}: {v}")

"""
GPT训练目标：因果语言模型 (CLM)

给定前面的token，预测下一个token：
L = -∑ log P(xᵢ | x₁, x₂, ..., xᵢ₋₁)

示例：
输入:  [BOS] 我 爱 自然 语言
标签:       我 爱 自然 语言 处理

每个位置只能看到自己和左边的token（因果掩码）
"""

import torch
import torch.nn as nn
import math

class CausalLanguageModel(nn.Module):
    """因果语言模型"""

    def __init__(self, vocab_size, d_model=256, num_heads=4, num_layers=4):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, d_model)
        self.pos_embedding = nn.Embedding(512, d_model)

        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model, nhead=num_heads, dim_feedforward=d_model*4,
            batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        self.lm_head = nn.Linear(d_model, vocab_size)

    def forward(self, x):
        seq_len = x.size(1)
        positions = torch.arange(seq_len, device=x.device).unsqueeze(0)  # unsqueeze增加一个维度

        # 因果掩码
        causal_mask = nn.Transformer.generate_square_subsequent_mask(seq_len).to(x.device)

        h = self.embedding(x) + self.pos_embedding(positions)
        h = self.transformer(h, mask=causal_mask)
        logits = self.lm_head(h)

        return logits

clm = CausalLanguageModel(vocab_size=5000)
x = torch.randint(0, 5000, (2, 20))
logits = clm(x)
print(f"CLM输出: {logits.shape}")  # (2, 20, 5000)

"""
BERT两个预训练任务：

1. MLM (Masked Language Model):
   - 随机遮盖15%的token
   - 其中80%替换为[MASK]，10%替换为随机词，10%保持不变
   - 预测被遮盖的token

   **80%-10%-10%策略的设计原因：**
   - 80%替换为[MASK]：主要训练方式，让模型学习上下文理解
   - 10%替换为随机词：迫使模型学习鲁棒性，不能完全依赖[MASK]标记
   - 10%保持不变：**关键设计**！避免预训练与微调之间的mismatch
     * 预训练时模型会看到[MASK]标记，但微调/实际使用时没有
     * 如果100%都替换为[MASK]，模型可能过度依赖这个特殊标记
     * 保留10%原始token让模型学会对任意输入都能产生有意义的表示
     * 这是一种"去噪自编码器"的思想，提升模型的泛化能力

2. NSP (Next Sentence Prediction):
   - 输入两个句子A和B
   - 50%的概率B是A的真实下一句(IsNext)
   - 50%的概率B是随机句子(NotNext)
   - 预测B是否是A的下一句
"""

class BERTPretraining:
    """BERT预训练数据准备"""

    def __init__(self, tokenizer_vocab_size=21128, mask_prob=0.15):
        self.vocab_size = tokenizer_vocab_size
        self.mask_prob = mask_prob
        self.mask_token_id = 103    # [MASK]
        self.cls_token_id = 101     # [CLS]
        self.sep_token_id = 102     # [SEP]

    def create_mlm_data(self, token_ids):
        """创建MLM训练数据"""
        import random

        masked_ids = list(token_ids)
        labels = [-100] * len(token_ids)  # -100表示不计算loss

        for i in range(len(token_ids)):
            if token_ids[i] in [self.cls_token_id, self.sep_token_id, 0]:
                continue

            if random.random() < self.mask_prob:
                labels[i] = token_ids[i]

                rand = random.random()
                if rand < 0.8:
                    masked_ids[i] = self.mask_token_id  # 80% → [MASK]
                elif rand < 0.9:
                    masked_ids[i] = random.randint(1, self.vocab_size - 1)  # 10% → random
                # else: 10% → keep original

        return masked_ids, labels

    def create_nsp_data(self, sent_a_ids, sent_b_ids, is_next):
        """创建NSP训练数据"""
        input_ids = [self.cls_token_id] + sent_a_ids + [self.sep_token_id] + sent_b_ids + [self.sep_token_id]
        segment_ids = [0] * (len(sent_a_ids) + 2) + [1] * (len(sent_b_ids) + 1)

        return input_ids, segment_ids, 1 if is_next else 0

# 演示
bert_pre = BERTPretraining()

# MLM示例
original = [101, 2769, 4263, 5765, 4197, 6427, 6404, 1164, 4415, 102]
masked, labels = bert_pre.create_mlm_data(original)
print("MLM示例:")
print(f"  原始: {original}")
print(f"  掩码: {masked}")
print(f"  标签: {labels}")

bert_configs = {
    "BERT-Base": {
        "层数(L)": 12,
        "隐藏维度(H)": 768,
        "注意力头(A)": 12,
        "参数量": "110M",
        "说明": "最常用的标准版本",
    },
    "BERT-Large": {
        "层数(L)": 24,
        "隐藏维度(H)": 1024,
        "注意力头(A)": 16,
        "参数量": "340M",
        "说明": "更大更强，但需要更多算力",
    },
}

for name, config in bert_configs.items():
    print(f"\n📌 {name}:")
    for k, v in config.items():
        print(f"   {k}: {v}")

# 参数量计算
def count_bert_params(L=12, H=768, A=12, V=30522):
    """计算BERT参数量"""
    # Embedding层
    token_emb = V * H
    pos_emb = 512 * H
    seg_emb = 2 * H
    emb_layer_norm = 2 * H
    emb_total = token_emb + pos_emb + seg_emb + emb_layer_norm

    # 每层Transformer
    # Self-Attention: Q, K, V各有W(H×H)+b(H)，输出投影W(H×H)+b(H)
    attn = 4 * (H * H + H)
    # LayerNorm: 2 * H
    attn_ln = 2 * H
    # FFN: W1(H×4H)+b1(4H) + W2(4H×H)+b2(H)
    ffn = H * 4 * H + 4 * H + 4 * H * H + H
    # LayerNorm: 2 * H
    ffn_ln = 2 * H

    per_layer = attn + attn_ln + ffn + ffn_ln
    all_layers = L * per_layer

    # 输出层
    pooler = H * H + H

    total = emb_total + all_layers + pooler
    return total

params = count_bert_params()
print(f"\nBERT-Base参数量计算: {params:,} ≈ {params/1e6:.0f}M")

bert_variants = {
    "RoBERTa (2019)": {
        "改进": "去掉NSP、动态Masking、更大batch和数据、更长训练",
        "效果": "比BERT平均提升2-3个点",
    },
    "ALBERT (2019)": {
        "改进": "嵌入矩阵分解、跨层参数共享",
        "效果": "参数量减少80%+，但训练速度不一定快",
    },
    "XLNet (2019)": {
        "改进": "排列语言模型，融合AR和AE的优点",
        "效果": "在多个任务上超过BERT",
    },
    "ELECTRA (2020)": {
        "改进": "替换词检测(RTD)代替MLM，训练效率更高",
        "效果": "小模型效果显著提升",
    },
    "DeBERTa (2020)": {
        "改进": "解耦注意力(内容+位置分别计算)",
        "效果": "SuperGLUE首次超过人类",
    },
    "T5 (2019)": {
        "改进": "所有任务统一为文本到文本格式",
        "效果": "极其灵活的统一框架",
    },
}

print("BERT变体对比:")
for name, info in bert_variants.items():
    print(f"\n  🔹 {name}")
    print(f"     改进: {info['改进']}")
    print(f"     效果: {info['效果']}")

"""
RoBERTa vs BERT 的关键差异:

1. 去掉NSP任务
   - BERT: MLM + NSP
   - RoBERTa: 只用MLM（实验证明NSP伤效果）

2. 动态Masking
   - BERT: 数据预处理时做静态masking（每个epoch相同）
   - RoBERTa: 每次取数据时动态masking（增加数据多样性）

3. 更大的Batch Size
   - BERT: 256
   - RoBERTa: 8192

4. 更多训练数据
   - BERT: 16GB
   - RoBERTa: 160GB

5. 更长的训练时间
   - BERT: 1M steps
   - RoBERTa: 500K steps（但batch更大，等效更多）
"""
print("RoBERTa要点已展示在注释中")

chinese_models = {
    "BERT-Base-Chinese": {
        "来源": "Google",
        "词汇量": 21128,
        "特点": "字级别分词，中文NLP基础模型",
        "HF名称": "bert-base-chinese",
    },
    "ERNIE (百度)": {
        "来源": "百度",
        "特点": "知识增强、实体级别Masking",
        "HF名称": "nghuyong/ernie-3.0-base-zh",
    },
    "MacBERT (哈工大)": {
        "来源": "哈工大讯飞",
        "特点": "用近义词替换代替[MASK]，缓解预训练-微调差距",
        "HF名称": "hfl/chinese-macbert-base",
    },
    "RoBERTa-wwm-ext (哈工大)": {
        "来源": "哈工大讯飞",
        "特点": "全词Masking + 更多数据",
        "HF名称": "hfl/chinese-roberta-wwm-ext",
    },
    "ChatGLM": {
        "来源": "清华智谱",
        "特点": "中文对话大模型",
        "HF名称": "THUDM/chatglm3-6b",
    },
}

print("常用中文预训练模型:")
for name, info in chinese_models.items():
    print(f"\n  📌 {name} ({info['来源']})")
    print(f"     特点: {info['特点']}")
    print(f"     Hugging Face: {info['HF名称']}")

from transformers import AutoTokenizer, AutoModel, AutoModelForSequenceClassification
import torch

# ==================
# 1. 加载模型和分词器
# ==================

model_name = "bert-base-chinese"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModel.from_pretrained(model_name)

# 分词
text = "自然语言处理是人工智能的重要分支"
tokens = tokenizer(text, return_tensors="pt")

print(f"Token IDs: {tokens['input_ids']}")
print(f"Tokens: {tokenizer.convert_ids_to_tokens(tokens['input_ids'][0])}")
print(f"Attention Mask: {tokens['attention_mask']}")

# 获取encoder输出
with torch.no_grad():  # 禁用梯度计算，节省内存
    outputs = model(**tokens)

print(f"\nLast hidden state: {outputs.last_hidden_state.shape}")
print(f"Pooler output: {outputs.pooler_output.shape}")

# ==================
# 2. 文本分类
# ==================

clf_model = AutoModelForSequenceClassification.from_pretrained(
    model_name, num_labels=3
)

with torch.no_grad():
    outputs = clf_model(**tokens)
    logits = outputs.logits
    probs = torch.softmax(logits, dim=-1)

print(f"\n分类logits: {logits}")
print(f"分类概率: {probs}")

from transformers import pipeline

# 情感分析
# classifier = pipeline("sentiment-analysis", model="uer/roberta-base-finetuned-chinanews-chinese")

# 命名实体识别
# ner = pipeline("ner", model="ckiplab/bert-base-chinese-ner")

# 问答
# qa = pipeline("question-answering", model="uer/roberta-base-chinese-extractive-qa")

# 文本生成
# generator = pipeline("text-generation", model="uer/gpt2-chinese-cluecorpussmall")

# 零样本分类
# zero_shot = pipeline("zero-shot-classification", model="MoritzLaurer/mDeBERTa-v3-base-mnli-xnli")

print("Hugging Face Pipeline示例:")
print("  classifier = pipeline('sentiment-analysis')")
print("  result = classifier('这个电影很好看')")
print("  # 输出: [{'label': 'POSITIVE', 'score': 0.99}]")

from transformers import Trainer, TrainingArguments
import torch

class FineTuneExample:
    """微调示例框架"""

    @staticmethod  # @staticmethod不需要实例即可调用
    def prepare_dataset(texts, labels, tokenizer, max_len=128):
        """准备数据集"""
        class TextDataset(torch.utils.data.Dataset):
            def __init__(self, texts, labels, tokenizer, max_len):
                self.encodings = tokenizer(texts, truncation=True,
                                           padding=True, max_length=max_len,
                                           return_tensors='pt')
                self.labels = torch.tensor(labels)

            def __getitem__(self, idx):  # __getitem__定义索引访问行为
                item = {k: v[idx] for k, v in self.encodings.items()}
                item['labels'] = self.labels[idx]
                return item

            def __len__(self):  # __len__定义len()行为
                return len(self.labels)

        return TextDataset(texts, labels, tokenizer, max_len)

    @staticmethod
    def get_training_args(output_dir="./results", epochs=3, batch_size=16, lr=2e-5):
        """获取训练参数"""
        return TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=epochs,
            per_device_train_batch_size=batch_size,
            per_device_eval_batch_size=batch_size,
            learning_rate=lr,
            weight_decay=0.01,
            warmup_ratio=0.1,
            logging_steps=10,
            eval_strategy="epoch",
            save_strategy="epoch",
            load_best_model_at_end=True,
        )

print("微调框架代码已定义")
print("使用: Trainer(model=model, args=args, train_dataset=train_ds, eval_dataset=eval_ds).train()")

fine_tuning_tips = {
    "学习率": {
        "推荐值": "1e-5 ~ 5e-5",
        "说明": "太大会破坏预训练参数，太小收敛慢",
        "技巧": "底层学习率更小(Layer-wise LR Decay)",
    },
    "Batch Size": {
        "推荐值": "16 或 32",
        "说明": "显存允许下尽量大",
        "技巧": "梯度累积模拟大batch",
    },
    "Epochs": {
        "推荐值": "2-5",
        "说明": "过多会过拟合（数据少时尤其注意）",
        "技巧": "使用Early Stopping",
    },
    "Warmup": {
        "推荐值": "总步数的6%-10%",
        "说明": "防止训练初期梯度过大",
    },
    "对抗训练": {
        "推荐值": "FGM或PGD",
        "说明": "对输入嵌入加扰动，提升鲁棒性",
        "效果": "通常提升0.5-1.0 F1",
    },
    "R-Drop": {
        "说明": "对同一输入做两次前向传播，最小化输出分布的KL散度",
        "效果": "有效的正则化方法",
    },
}

print("BERT微调最佳实践:")
for key, info in fine_tuning_tips.items():
    print(f"\n  📌 {key}:")
    for k, v in info.items():
        print(f"     {k}: {v}")

class FGM:
    """Fast Gradient Method 对抗训练"""

    def __init__(self, model, epsilon=1.0, emb_name='word_embeddings'):
        self.model = model
        self.epsilon = epsilon
        self.emb_name = emb_name
        self.backup = {}

    def attack(self):
        """添加扰动"""
        for name, param in self.model.named_parameters():
            if param.requires_grad and self.emb_name in name:
                self.backup[name] = param.data.clone()
                norm = torch.norm(param.grad)
                if norm != 0 and not torch.isnan(norm):
                    r_at = self.epsilon * param.grad / norm
                    param.data.add_(r_at)

    def restore(self):
        """恢复原始参数"""
        for name, param in self.model.named_parameters():
            if param.requires_grad and self.emb_name in name:
                assert name in self.backup  # assert断言
                param.data = self.backup[name]
        self.backup = {}

"""
对抗训练流程:
1. 正常前向+反向
2. fgm.attack()  # 加扰动
3. 扰动后再次前向+反向
4. fgm.restore()  # 恢复参数
5. optimizer.step()  # 更新参数
"""
print("FGM对抗训练类已定义")