NLP 基础与预训练语言模型¶

import numpy as np

# 假设词表: ["猫", "狗", "鱼", "鸟"]
vocab = {"猫": 0, "狗": 1, "鱼": 2, "鸟": 3}
vocab_size = len(vocab)

def one_hot_encode(word, vocab, vocab_size):
    """One-hot编码"""
    vec = np.zeros(vocab_size)
    if word in vocab:
        vec[vocab[word]] = 1.0
    return vec

# 编码示例
cat_vec = one_hot_encode("猫", vocab, vocab_size)
dog_vec = one_hot_encode("狗", vocab, vocab_size)

print(f"猫: {cat_vec}")  # [1. 0. 0. 0.]
print(f"狗: {dog_vec}")  # [0. 1. 0. 0.]

# 问题：余弦相似度为0——"猫"和"狗"毫无关系？
cos_sim = np.dot(cat_vec, dog_vec) / (np.linalg.norm(cat_vec) * np.linalg.norm(dog_vec))
print(f"猫-狗 余弦相似度: {cos_sim}")  # 0.0

from collections import Counter
import math

class TFIDF:
    """TF-IDF文本表示"""

    def __init__(self):
        self.vocab = {}
        self.idf = {}

    def fit(self, documents):
        """构建词表和计算IDF"""
        # 构建词表
        all_words = set()
        for doc in documents:
            all_words.update(doc.split())
        self.vocab = {word: i for i, word in enumerate(sorted(all_words))}

        # 计算IDF
        N = len(documents)
        for word in self.vocab:
            # 包含该词的文档数
            df = sum(1 for doc in documents if word in doc.split())
            self.idf[word] = math.log(N / (1 + df)) + 1  # 平滑IDF

    def transform(self, document):
        """将文档转换为TF-IDF向量"""
        words = document.split()
        tf = Counter(words)  # Counter统计元素出现次数
        total = len(words)

        vec = np.zeros(len(self.vocab))
        for word, count in tf.items():
            if word in self.vocab:
                tf_val = count / total
                vec[self.vocab[word]] = tf_val * self.idf.get(word, 1.0)
        return vec

# 示例
docs = [
    "猫 喜欢 吃 鱼",
    "狗 喜欢 吃 骨头",
    "猫 和 狗 都是 宠物"
]

tfidf = TFIDF()
tfidf.fit(docs)

for doc in docs:
    vec = tfidf.transform(doc)
    print(f"'{doc}' → 非零维度: {np.count_nonzero(vec)}, 向量模长: {np.linalg.norm(vec):.4f}")

# BPE（Byte Pair Encoding）算法核心流程演示
def bpe_train(text, vocab_size=1000):
    """
    简化的 BPE 训练过程
    核心思想：反复合并最高频的相邻 token 对
    """
    # 步骤1：将文本拆分为字符序列
    words = text.split()
    # 每个词表示为字符元组，末尾加 </w> 标记词边界
    vocab = {}
    for word in words:
        symbols = tuple(list(word) + ['</w>'])
        vocab[symbols] = vocab.get(symbols, 0) + 1

    print(f"初始词表（字符级）: {len(set(s for symbols in vocab for s in symbols))} 个符号")

    # 步骤2：迭代合并最高频的符号对
    num_merges = 10  # 简化演示
    for i in range(num_merges):
        # 统计所有相邻符号对的频率
        pairs = {}
        for symbols, freq in vocab.items():
            for j in range(len(symbols) - 1):
                pair = (symbols[j], symbols[j+1])
                pairs[pair] = pairs.get(pair, 0) + freq

        if not pairs:
            break

        # 找到频率最高的符号对
        best_pair = max(pairs, key=pairs.get)

        # 合并该符号对
        new_vocab = {}
        for symbols, freq in vocab.items():
            new_symbols = []
            j = 0
            while j < len(symbols):
                if j < len(symbols) - 1 and symbols[j] == best_pair[0] and symbols[j+1] == best_pair[1]:
                    new_symbols.append(best_pair[0] + best_pair[1])
                    j += 2
                else:
                    new_symbols.append(symbols[j])
                    j += 1
            new_vocab[tuple(new_symbols)] = freq

        vocab = new_vocab
        print(f"合并 {i+1}: {best_pair} → '{best_pair[0]+best_pair[1]}' (频率: {pairs[best_pair]})")

    return vocab

# 示例
sample_text = "low lower newest wide low low lower newest widest lowest"
print("=== BPE 训练演示 ===")
result = bpe_train(sample_text)

import torch
import torch.nn as nn
import torch.optim as optim

class SkipGram(nn.Module):
    """
    Word2Vec Skip-Gram模型
    给定中心词，预测上下文词
    """
    def __init__(self, vocab_size, embedding_dim):
        super().__init__()  # super()调用父类方法
        # 中心词嵌入矩阵
        self.center_embeddings = nn.Embedding(vocab_size, embedding_dim)
        # 上下文词嵌入矩阵
        self.context_embeddings = nn.Embedding(vocab_size, embedding_dim)

        # 初始化
        nn.init.xavier_uniform_(self.center_embeddings.weight)
        nn.init.xavier_uniform_(self.context_embeddings.weight)

    def forward(self, center_ids, context_ids, negative_ids):
        """
        Negative Sampling训练
        Args:
            center_ids:   [batch_size] 中心词ID
            context_ids:  [batch_size] 正样本上下文词ID
            negative_ids: [batch_size, num_neg] 负样本ID
        """
        # 获取嵌入向量
        center = self.center_embeddings(center_ids)      # [B, D]
        context = self.context_embeddings(context_ids)    # [B, D]
        negatives = self.context_embeddings(negative_ids) # [B, num_neg, D]

        # 正样本得分: center · context
        pos_score = torch.sum(center * context, dim=-1)      # [B]
        pos_loss = -torch.log(torch.sigmoid(pos_score) + 1e-8)

        # 负样本得分: center · negative
        neg_score = torch.bmm(negatives, center.unsqueeze(-1)).squeeze(-1)  # [B, num_neg]  # unsqueeze增加一个维度
        neg_loss = -torch.log(torch.sigmoid(-neg_score) + 1e-8).sum(dim=-1)

        return (pos_loss + neg_loss).mean()

    def get_embedding(self, word_id):
        """获取词的最终向量（使用中心词嵌入）"""
        return self.center_embeddings(torch.tensor([word_id])).detach()

# --- 训练示例 ---
# 构建简单语料
corpus = "the cat sat on the mat the dog sat on the rug".split()
vocab = {w: i for i, w in enumerate(set(corpus))}
vocab_size = len(vocab)
embedding_dim = 32

model = SkipGram(vocab_size, embedding_dim)
optimizer = optim.Adam(model.parameters(), lr=0.01)

print(f"词表大小: {vocab_size}, 嵌入维度: {embedding_dim}")
print(f"词表: {vocab}")

# 生成训练样本（简化版，window_size=2）
window_size = 2
training_pairs = []
for i, word in enumerate(corpus):  # enumerate同时获取索引和元素
    center_id = vocab[word]
    for j in range(max(0, i - window_size), min(len(corpus), i + window_size + 1)):
        if i != j:
            context_id = vocab[corpus[j]]
            # 随机负采样
            neg_ids = []
            while len(neg_ids) < 5:
                neg = torch.randint(0, vocab_size, (1,)).item()
                if neg != context_id:
                    neg_ids.append(neg)
            training_pairs.append((center_id, context_id, neg_ids))

print(f"训练样本数: {len(training_pairs)}")

# 训练
for epoch in range(100):
    total_loss = 0
    for center, context, negs in training_pairs:
        loss = model(
            torch.tensor([center]),
            torch.tensor([context]),
            torch.tensor([negs])
        )
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        total_loss += loss.item()

    if (epoch + 1) % 20 == 0:
        print(f"Epoch {epoch+1}, Loss: {total_loss/len(training_pairs):.4f}")

# 词向量相似度
def cosine_similarity(v1, v2):
    return torch.cosine_similarity(v1, v2, dim=-1).item()

cat_emb = model.get_embedding(vocab["cat"])
dog_emb = model.get_embedding(vocab["dog"])
mat_emb = model.get_embedding(vocab["mat"])

print(f"\n词向量相似度:")
print(f"  cat-dog: {cosine_similarity(cat_emb, dog_emb):.4f}")
print(f"  cat-mat: {cosine_similarity(cat_emb, mat_emb):.4f}")

# ELMo的核心思想示意（简化版）
class SimpleBiLM(nn.Module):
    """简化版双向语言模型（ELMo核心思想）"""

    def __init__(self, vocab_size, embed_dim, hidden_dim, num_layers=2):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)

        # 前向LSTM
        self.forward_lstm = nn.LSTM(
            embed_dim, hidden_dim, num_layers,
            batch_first=True, dropout=0.1
        )
        # 反向LSTM
        self.backward_lstm = nn.LSTM(
            embed_dim, hidden_dim, num_layers,
            batch_first=True, dropout=0.1
        )

        # 层权重（可学习的标量混合参数）
        self.layer_weights = nn.Parameter(torch.ones(num_layers + 1) / (num_layers + 1))
        self.gamma = nn.Parameter(torch.ones(1))

    def forward(self, input_ids):
        """
        Args:
            input_ids: [batch_size, seq_len]
        Returns:
            contextualized: [batch_size, seq_len, hidden_dim * 2]
        """
        embeds = self.embedding(input_ids)  # [B, L, E]

        # 前向
        fwd_out, _ = self.forward_lstm(embeds)        # [B, L, H]
        # 反向（翻转序列）
        bwd_out, _ = self.backward_lstm(embeds.flip(1))
        bwd_out = bwd_out.flip(1)                     # 翻回来 [B, L, H]

        # 拼接双向表示
        contextualized = torch.cat([fwd_out, bwd_out], dim=-1)  # [B, L, 2H]

        return contextualized

# ELMo提供了上下文相关的词表示
# 同一个 "bank" 在 "river bank" 和 "bank account" 中有不同向量

import torch
import torch.nn as nn

class TextCNN(nn.Module):
    """
    TextCNN (Kim, 2014) — 经典文本分类模型
    用不同大小的卷积核捕捉n-gram特征
    """
    def __init__(self, vocab_size, embed_dim, num_classes,
                 filter_sizes=(3, 4, 5), num_filters=100):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)

        # 多尺度卷积
        self.convs = nn.ModuleList([
            nn.Conv1d(embed_dim, num_filters, fs)
            for fs in filter_sizes
        ])

        self.dropout = nn.Dropout(0.5)
        self.fc = nn.Linear(num_filters * len(filter_sizes), num_classes)

    def forward(self, x):
        """
        Args: x [batch_size, seq_len]
        Returns: logits [batch_size, num_classes]
        """
        # [B, L, E] → [B, E, L] (Conv1d需要 channels-first)
        embedded = self.embedding(x).transpose(1, 2)

        # 多尺度卷积 + ReLU + 最大池化
        pooled = []
        for conv in self.convs:
            h = torch.relu(conv(embedded))       # [B, num_filters, L-fs+1]
            h = torch.max(h, dim=-1).values      # [B, num_filters]
            pooled.append(h)

        # 拼接所有尺度的特征
        cat = torch.cat(pooled, dim=-1)           # [B, num_filters * len(filter_sizes)]
        return self.fc(self.dropout(cat))

# 使用示例
model = TextCNN(vocab_size=10000, embed_dim=128, num_classes=4)
dummy_input = torch.randint(0, 10000, (8, 50))  # batch=8, seq_len=50
output = model(dummy_input)
print(f"TextCNN输出形状: {output.shape}")  # [8, 4]

class BiLSTM_CRF(nn.Module):
    """
    BiLSTM-CRF — 经典序列标注模型
    BiLSTM提取特征，CRF建模标签间的转移约束
    """
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_tags):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(
            embed_dim, hidden_dim // 2,
            bidirectional=True, batch_first=True
        )
        self.hidden2tag = nn.Linear(hidden_dim, num_tags)

        # CRF转移矩阵: transitions[i][j] = 从标签j转移到标签i的分数
        self.transitions = nn.Parameter(torch.randn(num_tags, num_tags))
        self.num_tags = num_tags

    def _get_emissions(self, x):
        """获取发射分数"""
        embeds = self.embedding(x)
        lstm_out, _ = self.lstm(embeds)
        return self.hidden2tag(lstm_out)  # [B, L, num_tags]

    def forward(self, x):
        """推理时使用维特比解码"""
        emissions = self._get_emissions(x)  # [B, L, T]
        # 简化版：逐batch维特比解码
        batch_size, seq_len, num_tags = emissions.shape

        best_paths = []
        for b in range(batch_size):
            # 维特比算法
            viterbi = emissions[b, 0]  # [T]
            backpointers = []

            for t in range(1, seq_len):
                # viterbi_expand: [T_prev] + transitions: [T_cur, T_prev] → [T_cur, T_prev]
                scores = viterbi.unsqueeze(0) + self.transitions  # [T, T]
                scores = scores + emissions[b, t].unsqueeze(1)    # [T, T]

                best_scores, best_ids = scores.max(dim=1)        # [T]
                viterbi = best_scores
                backpointers.append(best_ids)

            # 回溯
            best_last = viterbi.argmax().item()
            best_path = [best_last]
            for bp in reversed(backpointers):
                best_path.append(bp[best_path[-1]].item())  # [-1]负索引取最后一个元素
            best_path.reverse()
            best_paths.append(best_path)

        return best_paths

# NER标签示例: BIO标注
tag2id = {"O": 0, "B-PER": 1, "I-PER": 2, "B-ORG": 3, "I-ORG": 4, "B-LOC": 5, "I-LOC": 6}
model = BiLSTM_CRF(vocab_size=5000, embed_dim=128, hidden_dim=256, num_tags=len(tag2id))

dummy_input = torch.randint(0, 5000, (4, 20))
predictions = model(dummy_input)
print(f"BiLSTM-CRF预测: 4个序列, 各{len(predictions[0])}个标签")

class Seq2SeqWithAttention(nn.Module):
    """
    带注意力的Seq2Seq（Bahdanau Attention）
    经典的编码器-解码器架构，为Transformer奠定基础
    """
    def __init__(self, src_vocab, tgt_vocab, embed_dim, hidden_dim):
        super().__init__()
        # 编码器
        self.src_embedding = nn.Embedding(src_vocab, embed_dim)
        self.encoder = nn.GRU(embed_dim, hidden_dim, bidirectional=True, batch_first=True)

        # 解码器
        self.tgt_embedding = nn.Embedding(tgt_vocab, embed_dim)
        self.decoder = nn.GRU(embed_dim + hidden_dim * 2, hidden_dim, batch_first=True)

        # 注意力
        self.attn = nn.Linear(hidden_dim * 3, hidden_dim)
        self.v = nn.Linear(hidden_dim, 1, bias=False)

        # 输出
        self.output = nn.Linear(hidden_dim, tgt_vocab)

    def attention(self, decoder_hidden, encoder_outputs):
        """
        Bahdanau (Additive) Attention
        Args:
            decoder_hidden: [B, 1, H]
            encoder_outputs: [B, src_len, 2H]
        Returns:
            context: [B, 1, 2H]
        """
        src_len = encoder_outputs.shape[1]
        hidden_expanded = decoder_hidden.repeat(1, src_len, 1)  # [B, src_len, H]

        # 拼接 + 打分
        energy = torch.tanh(self.attn(
            torch.cat([hidden_expanded, encoder_outputs], dim=-1)  # [B, src_len, 3H]
        ))
        scores = self.v(energy).squeeze(-1)      # [B, src_len]
        weights = torch.softmax(scores, dim=-1)   # [B, src_len]

        context = torch.bmm(weights.unsqueeze(1), encoder_outputs)  # [B, 1, 2H]
        return context, weights

print("Seq2Seq with Attention: 注意力机制是Transformer的前身")
print("关键洞察: 让解码器在每一步'看'编码器的不同位置")

# 三种预训练目标的直观对比
print("""
┌─────────────────────────────────────────────────────────┐
│ MLM (BERT): 完形填空                                     │
│   输入: "小明在[MASK]读大学, 他的专业是[MASK]科学"        │
│   目标: 预测被遮住的词 → "北京", "计算机"                 │
│   特点: 双向上下文, 15%的词被随机遮盖                     │
├─────────────────────────────────────────────────────────┤
│ CLM (GPT): 从左到右续写                                  │
│   输入: "小明在北京读大学"                                │
│   目标: P(在|小明) × P(北京|小明,在) × P(读|小明,在,北京)│
│   特点: 单向(从左到右), 自回归生成                        │
├─────────────────────────────────────────────────────────┤
│ DAE (T5): 去噪重建                                      │
│   输入: "小明在<X>读大学, 他的专业是<Y>"                  │
│   目标: "<X> 北京 <Y> 计算机科学"                        │
│   特点: Encoder看全文, Decoder生成被替换的片段            │
└─────────────────────────────────────────────────────────┘
""")

import torch
import torch.nn as nn

class BERTForClassification(nn.Module):
    """
    BERT用于文本分类（微调范式的典型用法）
    """
    def __init__(self, hidden_size=768, num_classes=2, dropout=0.1):
        super().__init__()
        # 实际使用时这里加载预训练的BERT
        # self.bert = BertModel.from_pretrained('bert-base-chinese')

        # 简化版：模拟BERT encoder
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=hidden_size, nhead=12,
            dim_feedforward=3072, dropout=dropout,
            batch_first=True
        )
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=12)
        self.embedding = nn.Embedding(21128, hidden_size)  # BERT中文词表大小
        self.pos_embedding = nn.Embedding(512, hidden_size)

        # 分类头
        self.classifier = nn.Sequential(
            nn.Dropout(dropout),
            nn.Linear(hidden_size, hidden_size),
            nn.Tanh(),
            nn.Dropout(dropout),
            nn.Linear(hidden_size, num_classes)
        )

    def forward(self, input_ids, attention_mask=None):
        """
        Args:
            input_ids: [B, L] token IDs
            attention_mask: [B, L] 1=有效token, 0=padding
        """
        B, L = input_ids.shape
        positions = torch.arange(L, device=input_ids.device).unsqueeze(0).expand(B, L)

        # Token + Position Embedding
        x = self.embedding(input_ids) + self.pos_embedding(positions)

        # Transformer Encoder
        if attention_mask is not None:
            # 将 0/1 mask 转换为 True/False（True表示被遮蔽）
            src_key_padding_mask = (attention_mask == 0)
        else:
            src_key_padding_mask = None

        encoded = self.encoder(x, src_key_padding_mask=src_key_padding_mask)

        # 取 [CLS] 位置的输出做分类
        cls_output = encoded[:, 0, :]  # [B, H]
        return self.classifier(cls_output)

# 使用示例
model = BERTForClassification(hidden_size=768, num_classes=3)
dummy_ids = torch.randint(0, 21128, (4, 128))
dummy_mask = torch.ones(4, 128)

logits = model(dummy_ids, dummy_mask)
print(f"BERT分类输出: {logits.shape}")  # [4, 3]

# RoBERTa vs BERT 预训练配置对比
config_comparison = {
    "BERT-base": {
        "参数量": "110M",
        "预训练数据": "13GB (3.3B tokens)",
        "Batch Size": 256,
        "训练步数": "1M",
        "词表大小": "30K (WordPiece)",
        "预训练任务": "MLM + NSP",
        "遮蔽策略": "静态（4次随机Mask）",
    },
    "RoBERTa-base": {
        "参数量": "125M",
        "预训练数据": "160GB (约10x BERT)",
        "Batch Size": "8K",
        "训练步数": "500K",
        "词表大小": "50K (BPE)",
        "预训练任务": "仅MLM（去掉NSP）",
        "遮蔽策略": "动态（每epoch实时生成）",
    }
}

# 核心洞察：同样的架构，更充分的训练 → 显著性能提升
# 证明了"训练方法"与"模型架构"同等重要

# ALBERT 参数效率演示
def albert_param_savings(vocab_size, hidden_size, embed_size, num_layers):
    """
    计算 ALBERT 通过因式分解和参数共享节省的参数量
    """
    # 标准 BERT 的 Embedding 参数
    bert_embed_params = vocab_size * hidden_size

    # ALBERT 因式分解后的 Embedding 参数
    albert_embed_params = vocab_size * embed_size + embed_size * hidden_size

    # 单层 Encoder 参数（简化估计）
    single_layer_params = 4 * hidden_size * hidden_size  # Attention + FFN

    # BERT: 每层独立参数
    bert_encoder_params = num_layers * single_layer_params

    # ALBERT: 所有层共享参数
    albert_encoder_params = single_layer_params  # 仅一层

    bert_total = bert_embed_params + bert_encoder_params
    albert_total = albert_embed_params + albert_encoder_params

    print(f"BERT 参数量: {bert_total / 1e6:.1f}M")
    print(f"ALBERT 参数量: {albert_total / 1e6:.1f}M")
    print(f"参数压缩比: {bert_total / albert_total:.1f}x")

# 示例：24层模型
albert_param_savings(vocab_size=30000, hidden_size=1024,
                     embed_size=128, num_layers=24)
# BERT 参数量: ~340M → ALBERT 参数量: ~59M（压缩约 5.8x ）

# GPT系列参数规模演进
gpt_evolution = {
    "GPT-1 (2018)":   {"参数": "117M",  "数据": "5GB",    "关键创新": "预训练+微调范式"},
    "GPT-2 (2019)":   {"参数": "1.5B",  "数据": "40GB",   "关键创新": "Zero-shot多任务"},
    "GPT-3 (2020)":   {"参数": "175B",  "数据": "570GB",  "关键创新": "In-context Learning"},
    "GPT-3.5 (2022)": {"参数": "~175B", "数据": "更大",   "关键创新": "RLHF/InstructGPT"},
    "GPT-4 (2023)":   {"参数": "未公开", "数据": "未公开", "关键创新": "多模态/MoE传闻"},
}

print("GPT系列关键里程碑：")
print("=" * 60)
for name, info in gpt_evolution.items():
    print(f"{name}:")
    print(f"  参数量: {info['参数']}")
    print(f"  训练数据: {info['数据']}")
    print(f"  关键创新: {info['关键创新']}")
    print()

print("核心洞察：")
print("GPT-1→2：规模增大 → 零样本能力出现")
print("GPT-2→3：10x规模 → In-context Learning涌现")
print("GPT-3→3.5：RLHF对齐 → 真正可用的AI助手")
print("GPT-3.5→4：多模态 + 推理能力飞跃")

# In-context Learning示例：不需要微调，直接在prompt中提供示例

# Zero-shot（零样本）
zero_shot_prompt = """
请将以下英文翻译成中文：
The weather is beautiful today.
"""

# One-shot（单样本）
one_shot_prompt = """
英文翻译成中文示例：
English: Hello, how are you?
中文: 你好，你怎么样？

请翻译：
English: The weather is beautiful today.
中文:
"""

# Few-shot（少样本）
few_shot_prompt = """
将英文情感分类为正面或负面：

"This movie is amazing!" → 正面
"I hated every minute of it." → 负面
"The food was delicious and service was great." → 正面

"The product broke after one day." →
"""

print("In-context Learning的三种模式：")
print("  Zero-shot: 不给示例，直接问")
print("  One-shot:  给1个示例")
print("  Few-shot:  给2-5个示例")
print()
print("关键洞察：GPT-3不需要更新参数就能适应新任务！")
print("这是'涌现能力'的典型体现。")

# T5的统一框架思想
class T5Conceptual(nn.Module):
    """
    T5概念性实现：展示Encoder-Decoder统一框架
    """
    def __init__(self, vocab_size, d_model, nhead, num_layers):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, d_model)
        self.pos_embedding = nn.Embedding(512, d_model)

        # 编码器
        encoder_layer = nn.TransformerEncoderLayer(
            d_model, nhead, d_model * 4, batch_first=True
        )
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers)

        # 解码器
        decoder_layer = nn.TransformerDecoderLayer(
            d_model, nhead, d_model * 4, batch_first=True
        )
        self.decoder = nn.TransformerDecoder(decoder_layer, num_layers)

        # 输出投影
        self.output_proj = nn.Linear(d_model, vocab_size)

    def forward(self, src_ids, tgt_ids):
        """
        Encoder-Decoder前向传播
        src_ids: 编码器输入 [B, src_len]
        tgt_ids: 解码器输入 [B, tgt_len]
        """
        B, src_len = src_ids.shape
        _, tgt_len = tgt_ids.shape

        # 嵌入
        src_pos = torch.arange(src_len, device=src_ids.device).unsqueeze(0)
        tgt_pos = torch.arange(tgt_len, device=tgt_ids.device).unsqueeze(0)

        src = self.embedding(src_ids) + self.pos_embedding(src_pos)
        tgt = self.embedding(tgt_ids) + self.pos_embedding(tgt_pos)

        # 编码
        memory = self.encoder(src)  # [B, src_len, D]

        # 因果mask（解码器只能看到之前的token）
        causal_mask = nn.Transformer.generate_square_subsequent_mask(tgt_len)
        causal_mask = causal_mask.to(src_ids.device)

        # 解码
        decoded = self.decoder(tgt, memory, tgt_mask=causal_mask)  # [B, tgt_len, D]

        return self.output_proj(decoded)  # [B, tgt_len, vocab_size]

model = T5Conceptual(vocab_size=32000, d_model=512, nhead=8, num_layers=6)
src = torch.randint(0, 32000, (2, 30))
tgt = torch.randint(0, 32000, (2, 20))
output = model(src, tgt)
print(f"T5输出形状: {output.shape}")  # [2, 20, 32000]

comparison = """
┌──────────────────┬──────────────────┬──────────────────┬──────────────────┐
│                  │ Encoder-only     │ Decoder-only     │ Encoder-Decoder  │
│                  │ (BERT系列)       │ (GPT系列)        │ (T5/BART)        │
├──────────────────┼──────────────────┼──────────────────┼──────────────────┤
│ 注意力方向       │ 双向(全可见)     │ 单向(因果mask)   │ 编码器双向       │
│                  │                  │                  │ 解码器单向+交叉  │
├──────────────────┼──────────────────┼──────────────────┼──────────────────┤
│ 预训练目标       │ MLM(完形填空)    │ CLM(下一词预测)  │ DAE(去噪重建)    │
├──────────────────┼──────────────────┼──────────────────┼──────────────────┤
│ 输入             │ 完整文本         │ 前缀/提示        │ 源文本           │
│ 输出             │ 每位置的表示     │ 自回归生成       │ 目标文本         │
├──────────────────┼──────────────────┼──────────────────┼──────────────────┤
│ 擅长任务         │ 分类/NER/匹配    │ 文本生成/对话    │ 翻译/摘要/QA     │
├──────────────────┼──────────────────┼──────────────────┼──────────────────┤
│ 微调方式         │ 加分类头微调     │ 自回归微调       │ Seq2Seq微调      │
├──────────────────┼──────────────────┼──────────────────┼──────────────────┤
│ 代表模型         │ BERT, RoBERTa    │ GPT-4, LLaMA     │ T5, BART, mT5    │
│                  │ DeBERTa, ELECTRA │ Claude, Gemini   │ FLAN-T5          │
├──────────────────┼──────────────────┼──────────────────┼──────────────────┤
│ 当前地位         │ NLU任务首选      │ LLM主流架构      │ 翻译/摘要经典    │
│                  │ (但被LLM逐渐     │ (统治性地位)     │ (但被LLM逐渐     │
│                  │  取代)           │                  │  取代)           │
└──────────────────┴──────────────────┴──────────────────┴──────────────────┘
"""
print(comparison)

reasons = """
Decoder-only架构(GPT/LLaMA/Claude)为何成为主流？

1. 统一输入输出格式
   - Encoder-Decoder需要区分"源"和"目标"
   - Decoder-only只需要一个连续序列，更简洁
   - 所有任务统一为: prompt → completion

2. 规模扩展效率更高
   - 只有一个Transformer栈，参数利用更高效
   - 相同参数量下，Decoder-only通常效果更好
   - KV Cache加速推理，Encoder-Decoder需要额外存编码器cache

3. In-context Learning天然适配
   - 自回归生成 = 自然的多轮对话
   - Few-shot提示 = 把示例放在前面然后续写
   - BERT的[CLS] + Fine-tune范式在LLM时代不够灵活

4. 训练数据利用率
   - CLM利用每一个token位置做预测（N-1个预测目标）
   - MLM只预测15%被mask的token
   - 在海量数据上，CLM的数据利用率更高

5. 生态与工程优势
   - 推理只需要一个统一的Decoder，工程简单
   - KV Cache可以高效缓存，支持流式输出
   - 但BERT在检索、Embedding等任务上仍有独特优势！
"""
print(reasons)

import torch
import torch.nn as nn

class MiniEncoder(nn.Module):
    """最小Encoder-only（BERT风格）"""
    def __init__(self, vocab_size=1000, d_model=128, nhead=4, num_layers=2):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, d_model)
        encoder_layer = nn.TransformerEncoderLayer(d_model, nhead, d_model*4, batch_first=True)
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers)
        self.mlm_head = nn.Linear(d_model, vocab_size)

    def forward(self, x, mask_positions=None):
        h = self.encoder(self.embedding(x))
        if mask_positions is not None:
            # 只在mask位置预测
            return self.mlm_head(h[:, mask_positions, :])
        return h  # 返回所有位置的表示

class MiniDecoder(nn.Module):
    """最小Decoder-only（GPT风格）"""
    def __init__(self, vocab_size=1000, d_model=128, nhead=4, num_layers=2):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, d_model)
        decoder_layer = nn.TransformerEncoderLayer(d_model, nhead, d_model*4, batch_first=True)
        self.decoder = nn.TransformerEncoder(decoder_layer, num_layers)
        self.lm_head = nn.Linear(d_model, vocab_size)

    def forward(self, x):
        L = x.shape[1]
        # 因果mask: 每个位置只能看到自己和之前
        causal_mask = torch.triu(torch.ones(L, L), diagonal=1).bool().to(x.device)
        h = self.decoder(self.embedding(x), src_key_padding_mask=None,
                         mask=causal_mask)
        return self.lm_head(h)  # 每个位置预测下一个token

class MiniEncDec(nn.Module):
    """最小Encoder-Decoder（T5风格）"""
    def __init__(self, vocab_size=1000, d_model=128, nhead=4, num_layers=2):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, d_model)
        self.transformer = nn.Transformer(
            d_model, nhead, num_layers, num_layers, d_model*4, batch_first=True
        )
        self.output = nn.Linear(d_model, vocab_size)

    def forward(self, src, tgt):
        L = tgt.shape[1]
        causal_mask = torch.triu(torch.ones(L, L), diagonal=1).bool().to(src.device)

        src_emb = self.embedding(src)
        tgt_emb = self.embedding(tgt)
        h = self.transformer(src_emb, tgt_emb, tgt_mask=causal_mask)
        return self.output(h)

# 对比三种架构
print("三种架构参数量对比（相同配置）:")
print("=" * 50)

for name, model in [
    ("Encoder-only (BERT)", MiniEncoder()),
    ("Decoder-only (GPT)", MiniDecoder()),
    ("Encoder-Decoder (T5)", MiniEncDec())
]:
    params = sum(p.numel() for p in model.parameters())
    print(f"  {name}: {params:,} 参数")

# 推理对比
print("\n推理特性对比:")
print("  Encoder-only: 一次性编码整个序列 → 产出固定表示")
print("  Decoder-only: 自回归逐token生成 → 可配合KV Cache")
print("  Encoder-Decoder: 先编码源序列, 再自回归解码目标序列")

emergent_abilities = """
涌现能力: 在小模型中不存在, 在大模型中突然出现的能力

┌────────────────────────────────────────────────┐
│           准确率                                │
│    100% │                    ╭──── GPT-4       │
│         │                  ╭─╯                  │
│     50% │              ╭──╯   ← 涌现点         │
│         │          ╭──╯       (能力突然跳跃)    │
│      0% │──────────╯                            │
│         └────────────────────────── 模型规模     │
│         1B    10B    100B   1T                   │
└────────────────────────────────────────────────┘

已观测到的涌现能力:
├── 算术推理 (多位数加法)
├── 思维链 (Chain-of-Thought)
├── 指令遵循 (Instruction Following)
├── 代码生成
├── 多步推理
└── 世界知识问答

注: 涌现能力是否真的存在引发争论（Wei et al. vs Schaeffer et al.）
    有研究认为用不同的评测指标可能不存在"突变"
"""
print(emergent_abilities)

"""
实验: 在文本分类任务上对比三种架构
数据集: 简单的情感分类 (正面/负面)
目标: 理解不同架构在NLU任务上的差异
"""

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import random
import numpy as np

# 设置随机种子
def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
set_seed()

# 模拟数据集
class SentimentDataset(Dataset):
    """简单情感分类数据集"""
    # 正面关键词和负面关键词
    POS_WORDS = ["good", "great", "excellent", "wonderful", "amazing", "love", "best", "happy"]
    NEG_WORDS = ["bad", "terrible", "awful", "horrible", "hate", "worst", "sad", "poor"]
    NEUTRAL_WORDS = ["the", "is", "a", "this", "that", "it", "very", "really", "so", "and"]

    def __init__(self, num_samples=500, seq_len=20, vocab_size=100):
        self.data = []
        all_words = self.POS_WORDS + self.NEG_WORDS + self.NEUTRAL_WORDS
        self.word2id = {w: i+1 for i, w in enumerate(all_words)}  # 0 for PAD
        self.vocab_size = max(self.word2id.values()) + 1

        for _ in range(num_samples):
            label = random.choice([0, 1])  # 0=负面, 1=正面
            words = []
            for _ in range(seq_len):
                if random.random() < 0.3:  # 30%概率选情感词
                    if label == 1:
                        words.append(random.choice(self.POS_WORDS))
                    else:
                        words.append(random.choice(self.NEG_WORDS))
                else:
                    words.append(random.choice(self.NEUTRAL_WORDS))
            ids = [self.word2id.get(w, 0) for w in words]
            self.data.append((torch.tensor(ids), label))

    def __len__(self):
        return len(self.data)

    def __getitem__(self, idx):
        return self.data[idx]

dataset = SentimentDataset(num_samples=1000, seq_len=20)
train_set, val_set = torch.utils.data.random_split(dataset, [800, 200])
train_loader = DataLoader(train_set, batch_size=32, shuffle=True)
val_loader = DataLoader(val_set, batch_size=32)

print(f"词表大小: {dataset.vocab_size}")
print(f"训练集: {len(train_set)}, 验证集: {len(val_set)}")

class EncoderClassifier(nn.Module):
    """Encoder-only分类器（BERT风格）"""
    def __init__(self, vocab_size, d_model=64, nhead=4, num_layers=2):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, d_model, padding_idx=0)
        encoder_layer = nn.TransformerEncoderLayer(d_model, nhead, d_model*4,
                                                    batch_first=True, dropout=0.1)
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers)
        self.classifier = nn.Linear(d_model, 2)

    def forward(self, x):
        h = self.encoder(self.embedding(x))
        # 平均池化作为句子表示
        h_mean = h.mean(dim=1)
        return self.classifier(h_mean)

class DecoderClassifier(nn.Module):
    """Decoder-only分类器（GPT风格）"""
    def __init__(self, vocab_size, d_model=64, nhead=4, num_layers=2):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, d_model, padding_idx=0)
        encoder_layer = nn.TransformerEncoderLayer(d_model, nhead, d_model*4,
                                                    batch_first=True, dropout=0.1)
        self.decoder = nn.TransformerEncoder(encoder_layer, num_layers)
        self.classifier = nn.Linear(d_model, 2)

    def forward(self, x):
        L = x.shape[1]
        causal_mask = torch.triu(torch.ones(L, L, device=x.device), diagonal=1).bool()
        h = self.decoder(self.embedding(x), mask=causal_mask)
        # 取最后一个位置（因为因果注意力，最后位置看到全部信息）
        h_last = h[:, -1, :]
        return self.classifier(h_last)

class EncDecClassifier(nn.Module):
    """Encoder-Decoder分类器（T5风格）"""
    def __init__(self, vocab_size, d_model=64, nhead=4, num_layers=2):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, d_model, padding_idx=0)
        self.transformer = nn.Transformer(d_model, nhead, num_layers, num_layers,
                                           d_model*4, batch_first=True, dropout=0.1)
        self.classifier = nn.Linear(d_model, 2)
        # 用一个可学习的query token
        self.query = nn.Parameter(torch.randn(1, 1, d_model))

    def forward(self, x):
        B = x.shape[0]
        src = self.embedding(x)
        # 用query token做解码
        tgt = self.query.expand(B, -1, -1)
        h = self.transformer(src, tgt)
        return self.classifier(h.squeeze(1))