from transformers import pipeline

generator = pipeline("text-generation", model="gpt2")
result = generator("This movie was really", max_length=20, num_return_sequences=1)
print(result[0]['generated_text'])
# 输出可能是: "This movie was really good. I was so excited to see it..."
# 但它不会告诉你"正面"还是"负面"！

预训练模型（通用语言知识） + 特定任务数据（如情感分析） = 专业化模型

from transformers import DistilBertForSequenceClassification, DistilBertTokenizer
import torch

# Step 1: 加载 tokenizer 和模型
tokenizer = DistilBertTokenizer.from_pretrained(
    "distilbert-base-uncased-finetuned-sst-2-english"
)
model = DistilBertForSequenceClassification.from_pretrained(
    "distilbert-base-uncased-finetuned-sst-2-english"
)

# Step 2: 文本 → Token IDs
text = "This movie is absolutely amazing!"
inputs = tokenizer(text, return_tensors="pt")
# inputs 包含: input_ids（token索引）和 attention_mask（注意力掩码）

# Step 3: 前向传播
with torch.no_grad():  # 推理时不需要计算梯度
    outputs = model(**inputs)

# Step 4: Logits → 预测结果
logits = outputs.logits  # 原始输出，shape: [1, 2]
probs = torch.softmax(logits, dim=-1)  # 转换为概率
predicted_class = torch.argmax(probs, dim=-1)

labels = ["NEGATIVE", "POSITIVE"]
print(f"预测结果: {labels[predicted_class]}")
# 输出: 预测结果: POSITIVE

from transformers import pipeline

# 情感分析
classifier = pipeline(
    "text-classification",
    model="distilbert-base-uncased-finetuned-sst-2-english"
)
result = classifier("This movie is absolutely amazing!")
print(result)
# [{'label': 'POSITIVE', 'score': 0.9998}]

from transformers import GPT2LMHeadModel, GPT2Tokenizer

# 手动加载
tokenizer = GPT2Tokenizer.from_pretrained("gpt2")
model = GPT2LMHeadModel.from_pretrained("gpt2")

prompt = "Once upon a time"
inputs = tokenizer(prompt, return_tensors="pt")

output_ids = model.generate(
    inputs.input_ids,
    attention_mask=inputs.attention_mask,
    pad_token_id=tokenizer.eos_token_id,
    max_length=50,
    num_return_sequences=1
)

generated_text = tokenizer.decode(output_ids[0], skip_special_tokens=True)
print(generated_text)

generator = pipeline("text-generation", model="gpt2")
result = generator("Once upon a time", max_length=50, truncation=True)
print(result[0]['generated_text'])

generator = pipeline("text2text-generation", model="t5-small")
result = generator("translate English to French: How are you?", max_length=50)
print(result[0]['generated_text'])
# 输出: Comment allez-vous?

输入: "The capital of France is [MASK]."
预测: "The capital of France is Paris."

from transformers import pipeline

fill_mask = pipeline("fill-mask", model="bert-base-uncased")
result = fill_mask("The capital of France is [MASK].")
for r in result:
    print(f"预测: {r['token_str']}, 置信度: {r['score']:.2f}")
# 预测: paris, 置信度: 0.88
# 预测: lyon, 置信度: 0.02
# ...

句子A: "The dog is running."
句子B: "It is chasing a ball."  → IsNext（是连续的）
句子B: "The stock market rose."  → NotNext（不是连续的）

输入: "The quick brown"
预测: "fox"

from transformers import (
    BertConfig, BertForMaskedLM, BertTokenizerFast,
    DataCollatorForLanguageModeling, TrainingArguments, Trainer
)
from datasets import load_dataset

# 1. 加载数据集
dataset = load_dataset("wikitext", "wikitext-2-raw-v1")

# 2. 创建 Tokenizer
bert_tokenizer = BertTokenizerFast.from_pretrained("bert-base-uncased")

# 3. 定义模型配置
config = BertConfig(
    vocab_size=30522,       # 词汇表大小
    hidden_size=768,        # 隐藏层维度
    num_hidden_layers=12,   # Transformer 层数
    num_attention_heads=12, # 注意力头数
    intermediate_size=3072, # FFN 中间层维度
)

# 4. 创建全新的 BERT 模型（随机初始化权重）
model = BertForMaskedLM(config)

# 5. Tokenize 数据
def tokenize_function(examples):
    return bert_tokenizer(
        examples["text"],
        truncation=True,
        padding="max_length",
        max_length=512
    )

tokenized_datasets = dataset.map(tokenize_function, batched=True, remove_columns=["text"])

# 6. 设置 MLM 数据收集器（自动遮盖 15% 的 token）
data_collator = DataCollatorForLanguageModeling(
    tokenizer=bert_tokenizer,
    mlm=True,
    mlm_probability=0.15  # 15% 的 token 被遮盖
)

# 7. 训练配置
training_args = TrainingArguments(
    output_dir="./trained_model",
    num_train_epochs=10,
    per_device_train_batch_size=2,
    learning_rate=5e-5,
    save_total_limit=2,
)

# 8. 开始训练
trainer = Trainer(
    model=model,
    args=training_args,
    data_collator=data_collator,
    train_dataset=tokenized_datasets["train"],
    eval_dataset=tokenized_datasets["test"],
)
trainer.train()

import math

eval_results = trainer.evaluate()
perplexity = math.exp(eval_results['eval_loss'])
print(f"困惑度: {perplexity:.2f}")

import torch
from torch.optim import AdamW
from torch.utils.data import DataLoader
from transformers import AutoModelForSequenceClassification, AutoTokenizer
from datasets import load_dataset
from torchmetrics import Accuracy

# 1. 加载数据
dataset = load_dataset("yelp_review_full")
# Yelp 有 5 个类别（1-5 星）

# 2. 加载预训练模型和 tokenizer
tokenizer = AutoTokenizer.from_pretrained("bert-base-cased")
model = AutoModelForSequenceClassification.from_pretrained(
    "bert-base-cased",
    num_labels=5  # 5 分类
)

# 3. Tokenize 数据
def tokenize_function(examples):
    return tokenizer(examples["text"], padding="max_length", truncation=True)

tokenized_datasets = dataset.map(tokenize_function, batched=True)

# 4. 数据预处理
tokenized_datasets = tokenized_datasets.remove_columns(["text"])
tokenized_datasets = tokenized_datasets.rename_column("label", "labels")
tokenized_datasets.set_format("torch")

# 5. DataLoader
train_dataloader = DataLoader(tokenized_datasets["train"], shuffle=True, batch_size=2)
eval_dataloader = DataLoader(tokenized_datasets["test"], batch_size=2)

# 6. 优化器和学习率调度
optimizer = AdamW(model.parameters(), lr=5e-4)
num_epochs = 10
num_training_steps = num_epochs * len(train_dataloader)

# 7. 训练循环
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)
model.train()

for epoch in range(num_epochs):
    total_loss = 0
    for batch in train_dataloader:
        batch = {k: v.to(device) for k, v in batch.items()}

        outputs = model(**batch)        # 前向传播
        loss = outputs.loss             # 计算 loss
        loss.backward()                 # 反向传播

        optimizer.step()                # 更新参数
        optimizer.zero_grad()           # 清零梯度

        total_loss += loss.item()

    avg_loss = total_loss / len(train_dataloader)
    print(f"Epoch {epoch+1}/{num_epochs}, Loss: {avg_loss:.4f}")

def evaluate_model(model, eval_dataloader):
    metric = Accuracy(task="multiclass", num_classes=5).to(device)
    model.eval()

    with torch.no_grad():
        for batch in eval_dataloader:
            batch = {k: v.to(device) for k, v in batch.items()}
            outputs = model(**batch)
            predictions = torch.argmax(outputs.logits, dim=-1)
            metric(predictions, batch["labels"])

    accuracy = metric.compute()
    print(f"准确率: {accuracy.item():.4f}")

from trl import SFTConfig, SFTTrainer, DataCollatorForCompletionOnlyLM
from transformers import AutoModelForCausalLM, AutoTokenizer
from datasets import load_dataset

# 1. 加载对话数据集
dataset = load_dataset("timdettmers/openassistant-guanaco", split="train")
# 数据格式: "### Human: ...\n### Assistant: ..."

# 2. 加载生成式模型
model = AutoModelForCausalLM.from_pretrained("facebook/opt-350m")
tokenizer = AutoTokenizer.from_pretrained("facebook/opt-350m")

# 3. 定义指令和回复模板
instruction_template = "### Human:"
response_template = "### Assistant:"

# 4. 创建数据收集器（只在 Assistant 回复部分计算 loss）
collator = DataCollatorForCompletionOnlyLM(
    instruction_template=instruction_template,
    response_template=response_template,
    tokenizer=tokenizer,
    mlm=False
)

# 5. 训练配置
training_args = SFTConfig(
    output_dir="/tmp",
    num_train_epochs=10,
    per_device_train_batch_size=2,
    max_seq_length=1024,
    fp16=True,  # 混合精度训练
)

# 6. 创建 Trainer 并训练
trainer = SFTTrainer(
    model,
    args=training_args,
    train_dataset=dataset,
    dataset_text_field="text",
    data_collator=collator,
)
trainer.train()

预训练数据: AG News（新闻分类，4类）
微调数据: IMDB（电影评论情感，2类）
模型架构: 基于 PyTorch TransformerEncoder 的分类器
词向量: GloVe-6B-100d（预训练词向量）

class Net(nn.Module):
    """基于 TransformerEncoder 的文本分类器"""

    def __init__(self, num_class, vocab_size, freeze=True,
                 nhead=2, dim_feedforward=128, num_layers=2,
                 dropout=0.1):
        super().__init__()

        # 使用 GloVe 预训练词向量
        self.emb = nn.Embedding.from_pretrained(
            glove_embedding.vectors, freeze=freeze
        )
        embedding_dim = self.emb.embedding_dim  # 100

        # 位置编码
        self.pos_encoder = PositionalEncoding(
            d_model=embedding_dim, dropout=dropout
        )

        # Transformer 编码器
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=embedding_dim,
            nhead=nhead,
            dim_feedforward=dim_feedforward,
            dropout=dropout,
        )
        self.transformer_encoder = nn.TransformerEncoder(
            encoder_layer, num_layers=num_layers
        )

        # 分类头
        self.classifier = nn.Linear(embedding_dim, num_class)

    def forward(self, x):
        x = self.emb(x) * math.sqrt(self.d_model)
        x = self.pos_encoder(x)
        x = self.transformer_encoder(x)
        x = x.mean(dim=1)  # 平均池化
        x = self.classifier(x)
        return x

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, vocab_size=5000, dropout=0.1):
        super().__init__()
        self.dropout = nn.Dropout(p=dropout)

        pe = torch.zeros(vocab_size, d_model)
        position = torch.arange(0, vocab_size, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(
            torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)
        )
        pe[:, 0::2] = torch.sin(position * div_term)  # 偶数位用 sin
        pe[:, 1::2] = torch.cos(position * div_term)  # 奇数位用 cos
        pe = pe.unsqueeze(0)
        self.register_buffer("pe", pe)

    def forward(self, x):
        x = x + self.pe[:, :x.size(1), :]
        return self.dropout(x)

model = Net(num_class=2, vocab_size=vocab_size).to(device)
# 训练 100 epochs...
# 结果: ~83% 准确率

# 1. 加载 AG News 上预训练好的模型
model_fine1 = Net(vocab_size=vocab_size, num_class=4).to(device)
model_fine1.load_state_dict(pretrained_ag_news_weights)

# 2. 替换最后一层（4类 → 2类），但不冻结其他层
model_fine1.classifier = nn.Linear(in_features, 2)

# 3. 所有参数都参与训练
for name, param in model_fine1.named_parameters():
    print(f"{name} requires_grad: {param.requires_grad}")
    # 全部是 True！

# 训练 100 epochs...
# 结果: ~86% 准确率 ✨

# 1. 加载预训练模型
model_fine2 = Net(vocab_size=vocab_size, num_class=4).to(device)
model_fine2.load_state_dict(pretrained_ag_news_weights)

# 2. 冻结所有层
for param in model_fine2.parameters():
    param.requires_grad = False

# 3. 替换最后一层（新层默认 requires_grad=True）
model_fine2.classifier = nn.Linear(dim, 2)

# 训练 100 epochs...
# 结果: ~64% 准确率 ❌

# 只解冻 Transformer 中的 linear2 层和分类头
for param in model.parameters():
    param.requires_grad = False

# 解冻特定层
for i in range(2):  # 假设有 2 层 Transformer
    for param in model.transformer_encoder.layers[i].linear2.parameters():
        param.requires_grad = True

# 替换分类头
model.classifier = nn.Linear(100, 2)

原始层输出 (d维)
     ↓
  ┌──────────┐
  │ 降维 (d→m) │  ← 全连接层，m << d
  │  ReLU      │  ← 非线性激活
  │ 升维 (m→d) │  ← 全连接层
  └──────────┘
     ↓
  + 残差连接（原始输出 + Adapter输出）
     ↓
  最终输出 (d维)

class FeatureAdapter(nn.Module):
    """适配器模块 - 瓶颈结构 + 残差连接"""

    def __init__(self, input_dim, bottleneck_dim):
        super().__init__()
        self.adapter = nn.Sequential(
            nn.Linear(input_dim, bottleneck_dim),   # 降维
            nn.ReLU(),                               # 激活
            nn.Linear(bottleneck_dim, input_dim)     # 升维
        )

    def forward(self, x):
        return x + self.adapter(x)  # 残差连接！

class Adapted(nn.Module):
    """将 Adapter 注入到现有模型中"""

    def __init__(self, base_model, adapter_dim=64):
        super().__init__()
        self.base_model = base_model

        # 冻结所有原始参数
        for param in self.base_model.parameters():
            param.requires_grad = False

        # 为每一层 Transformer 添加 Adapter
        d_model = base_model.emb.embedding_dim
        self.adapters = nn.ModuleList([
            FeatureAdapter(d_model, adapter_dim)
            for _ in range(len(base_model.transformer_encoder.layers))
        ])

        # 新的分类头
        self.classifier = nn.Linear(d_model, num_classes)

    def forward(self, x):
        x = self.base_model.emb(x)
        x = self.base_model.pos_encoder(x)

        for i, layer in enumerate(self.base_model.transformer_encoder.layers):
            x = layer(x)
            x = self.adapters[i](x)  # 在每层后面插入 Adapter

        x = x.mean(dim=1)
        x = self.classifier(x)
        return x

class LoRALayer(nn.Module):
    """LoRA 低秩适配层"""

    def __init__(self, in_features, out_features, rank=4, alpha=1.0):
        super().__init__()
        self.rank = rank
        self.alpha = alpha

        # A 矩阵：用高斯分布初始化
        self.A = nn.Parameter(torch.randn(rank, in_features))
        # B 矩阵：初始化为零！
        self.B = nn.Parameter(torch.zeros(out_features, rank))

    def forward(self, x):
        # ΔW·x = B·A·x
        # 先算 A·x（降维），再算 B·(A·x)（升维）
        return self.alpha * (x @ self.A.T @ self.B.T)

class LinearWithLoRA(nn.Module):
    """带有 LoRA 的线性层"""

    def __init__(self, original_linear, rank=4, alpha=1.0):
        super().__init__()
        self.original_linear = original_linear  # 冻结的原始层
        self.lora = LoRALayer(
            original_linear.in_features,
            original_linear.out_features,
            rank=rank,
            alpha=alpha
        )

    def forward(self, x):
        # 原始输出 + LoRA 修正
        return self.original_linear(x) + self.lora(x)

# 1. 加载在 AG News 上预训练的模型
pretrained_model = TextClassifier(...)
pretrained_model.load_state_dict(ag_news_weights)

# 2. 冻结所有参数
for param in pretrained_model.parameters():
    param.requires_grad = False

# 3. 替换分类头
pretrained_model.output_layer = nn.Linear(embed_dim, 2)

# 4. 给线性层注入 LoRA
for layer in pretrained_model.transformer_encoder.layers:
    # 替换 self-attention 中的线性层
    layer.self_attn.out_proj = LinearWithLoRA(
        layer.self_attn.out_proj, rank=8
    )
    # 替换 FFN 中的线性层
    layer.linear1 = LinearWithLoRA(layer.linear1, rank=8)
    layer.linear2 = LinearWithLoRA(layer.linear2, rank=8)

# 5. 检查可训练参数
trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
total_params = sum(p.numel() for p in model.parameters())
print(f"可训练参数: {trainable_params:,} / {total_params:,}")
print(f"占比: {100 * trainable_params / total_params:.2f}%")
# 输出类似: 可训练参数: 12,288 / 5,000,000 (约 0.25%)

基线（无 LoRA 的预训练模型）: 73.62% 准确率
加入 LoRA 微调后:              76.40% 准确率（+2.78%）

FP32 (32-bit):  3.141592653589793  → 32 bit/参数
FP16 (16-bit):  3.14159             → 16 bit/参数
INT8 (8-bit):   3                   → 8 bit/参数
NF4 (4-bit):    ≈3                  → 4 bit/参数 ← QLoRA 使用这个

7B 参数模型的显存占用:
FP32: ~28 GB
FP16: ~14 GB
INT8: ~7 GB
NF4:  ~3.5 GB ← QLoRA!

from transformers import (
    AutoModelForSequenceClassification,
    AutoTokenizer,
    BitsAndBytesConfig,
    TrainingArguments,
    Trainer
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training

# 1. 配置 4-bit 量化
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,                # 加载为 4-bit
    bnb_4bit_quant_type="nf4",        # 使用 NF4 量化
    bnb_4bit_use_double_quant=True,   # 双重量化（进一步压缩）
    bnb_4bit_compute_dtype=torch.float16  # 计算时用 FP16
)

# 2. 加载量化模型
model = AutoModelForSequenceClassification.from_pretrained(
    "distilbert-base-uncased",
    num_labels=2,
    quantization_config=bnb_config
)

# 3. 准备模型用于 k-bit 训练
model = prepare_model_for_kbit_training(model)

# 4. 配置 LoRA
lora_config = LoraConfig(
    r=8,                              # LoRA 秩
    lora_alpha=32,                    # LoRA 缩放因子
    target_modules=["q_lin", "v_lin"], # 在 Q 和 V 矩阵上应用 LoRA
    lora_dropout=0.05,
    bias="none",
    task_type="SEQ_CLS"
)

# 5. 应用 LoRA
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出: trainable params: 888,578 || all params: 67,544,834 || trainable%: 1.3154%

tokenizer = AutoTokenizer.from_pretrained("distilbert-base-uncased")
dataset = load_dataset("imdb")

def tokenize_function(examples):
    return tokenizer(examples["text"], padding="max_length", truncation=True, max_length=512)

tokenized_dataset = dataset.map(tokenize_function, batched=True)

training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=8,
    learning_rate=2e-4,
    fp16=True,
    logging_steps=100,
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_dataset["train"],
    eval_dataset=tokenized_dataset["test"],
)

trainer.train()

输入 → [Embedding] → [Transformer层1] → [Transformer层2] → ... → [分类头] → 输出
         可训练 ↑        可训练 ↑           可训练 ↑                可训练 ↑

输入 → [Embedding❄️] → [Transformer层1❄️] → [Adapter🔥] → [Transformer层2❄️] → [Adapter🔥] → [分类头🔥]
                                                 ↑ 新增小模块                        ↑ 新增小模块

                    ┌───────────┐
输入 → [原始权重W₀❄️]──→ W₀·x ──┐
  │                              ├──→ 输出 = W₀·x + α·B·A·x
  └→ [A矩阵🔥] → [B矩阵🔥] ──→ α·B·A·x ─┘
      (d→r)       (r→d)

你有足够的 GPU 吗？
├── 是 → 数据集大吗？
│   ├── 是 → 全量微调（最佳性能）
│   └── 否 → LoRA（避免过拟合）
│
└── 否 → 模型能放进显存吗？
    ├── 能 → LoRA
    └── 不能 → QLoRA（4-bit 量化后再 LoRA）

model = AutoModelForSequenceClassification.from_pretrained("bert-base-cased", num_labels=2)
# 所有参数默认可训练

for param in model.parameters():
    param.requires_grad = False
model.classifier = nn.Linear(768, 2)  # 新层可训练

from peft import LoraConfig, get_peft_model
lora_config = LoraConfig(r=8, lora_alpha=32, target_modules=["query", "value"])
model = get_peft_model(model, lora_config)

bnb_config = BitsAndBytesConfig(load_in_4bit=True, bnb_4bit_quant_type="nf4")
model = AutoModel.from_pretrained("model_name", quantization_config=bnb_config)
model = prepare_model_for_kbit_training(model)
model = get_peft_model(model, lora_config)

Step 1: 学会使用 HuggingFace pipeline 做推理
        ↓
Step 2: 理解预训练的概念（MLM、NSP）
        ↓
Step 3: 手写 PyTorch 训练循环，理解全量微调
        ↓
Step 4: 学习 LoRA 的数学原理和代码实现
        ↓
Step 5: 使用 PEFT 库实践 LoRA/QLoRA
        ↓
Step 6: 在自己的项目中应用！