Transformers 模型库（结合源码）

class PreTrainedModel(nn.Module, EmbeddingAccessMixin, ModuleUtilsMixin, PushToHubMixin, PeftAdapterMixin):
    config_class: type[PreTrainedConfig] | None = None   # 关联的配置类
    base_model_prefix: str = ""                           # 基础模型前缀，如 "model"
    main_input_name: str = "input_ids"                   # 主输入名
    _no_split_modules: set[str] | list[str] | None = None # 设备映射时不可分割的模块
    _supports_sdpa: bool = False                          # 是否支持 SDPA 注意力
    _supports_flash_attn: bool = False                    # 是否支持 FlashAttention
    supports_gradient_checkpointing: bool = False         # 是否支持梯度检查点

def from_pretrained(
    cls,
    pretrained_model_name_or_path,  # 模型ID或本地路径
    config=None,                    # 预加载的配置对象
    cache_dir=None,                 # 缓存目录
    force_download=False,           # 强制重新下载
    local_files_only=False,         # 仅使用本地文件
    token=None,                     # HuggingFace 访问令牌
    revision="main",               # 模型版本（分支/标签/commit）
    dtype="auto",                  # 数据类型，如 torch.bfloat16
    device_map=None,               # 设备映射，如 "auto"
    quantization_config=None,      # 量化配置，如 BitsAndBytesConfig
    attn_implementation=None,      # 注意力实现："eager"/"sdpa"/"flash_attention_2"
    trust_remote_code=False,       # 是否信任远程代码
    ...
)

@dataclass_transform(kw_only_default=True)
@strict(accept_kwargs=True)
@dataclass(repr=False)
class PreTrainedConfig(PushToHubMixin, RotaryEmbeddingConfigMixin):
    model_type: str                    # 模型类型标识，如 "llama"
    vocab_size: int                    # 词表大小
    hidden_size: int                   # 隐藏层维度
    num_attention_heads: int           # 注意力头数
    num_hidden_layers: int             # Transformer 层数
    ...

# modeling_auto.py 中的映射表（部分）
MODEL_MAPPING_NAMES = OrderedDict([
    ("llama", "LlamaModel"),
    ("bert", "BertModel"),
    ("gpt2", "GPT2Model"),
    ...
])

MODEL_FOR_CAUSAL_LM_MAPPING_NAMES = OrderedDict([
    ("llama", "LlamaForCausalLM"),
    ("bert", "BertForMaskedLM"),
    ("gpt2", "GPT2LMHeadModel"),
    ...
])

class _LazyAutoMapping(OrderedDict):
    def __getitem__(self, key):
        value = super().__getitem__(key)
        # 延迟导入：当访问映射值时，才动态导入对应的模型类
        module_name = model_type_to_module_name(key)
        return dynamic_import(module_name, value)

# configuration_llama.py
@strict
class LlamaConfig(PreTrainedConfig):
    model_type = "llama"

    vocab_size: int = 32000           # 词表大小
    hidden_size: int = 4096           # 隐藏层维度
    intermediate_size: int = 11008    # FFN 中间层维度
    num_hidden_layers: int = 32       # Transformer 层数
    num_attention_heads: int = 32     # 注意力头数
    num_key_value_heads: int | None = None  # GQA 的 KV 头数
    hidden_act: str = "silu"          # 激活函数
    max_position_embeddings: int = 2048  # 最大位置编码长度
    rms_norm_eps: float = 1e-6        # RMSNorm epsilon
    rope_parameters: RopeParameters | dict | None = None  # RoPE 参数
    attention_bias: bool = False      # 注意力是否使用偏置
    mlp_bias: bool = False            # MLP 是否使用偏置

    def __post_init__(self, **kwargs):
        if self.head_dim is None:
            self.head_dim = self.hidden_size // self.num_attention_heads
        if self.num_key_value_heads is None:
            self.num_key_value_heads = self.num_attention_heads  # 默认 MHA

# modeling_llama.py
class LlamaModel(LlamaPreTrainedModel):
    def __init__(self, config: LlamaConfig):
        super().__init__(config)
        self.embed_tokens = nn.Embedding(config.vocab_size, config.hidden_size)
        self.layers = nn.ModuleList(
            [LlamaDecoderLayer(config, layer_idx) for layer_idx in range(config.num_hidden_layers)]
        )
        self.norm = LlamaRMSNorm(config.hidden_size, eps=config.rms_norm_eps)
        self.rotary_emb = LlamaRotaryEmbedding(config=config)

class LlamaForCausalLM(LlamaPreTrainedModel, GenerationMixin):
    _tied_weights_keys = {"lm_head.weight": "model.embed_tokens.weight"}  # 权重绑定

    def __init__(self, config):
        super().__init__(config)
        self.model = LlamaModel(config)        # 基座模型
        self.vocab_size = config.vocab_size
        self.lm_head = nn.Linear(config.hidden_size, config.vocab_size, bias=False)  # 语言模型头

class LlamaAttention(nn.Module):
    def __init__(self, config: LlamaConfig, layer_idx: int):
        self.head_dim = config.hidden_size // config.num_attention_heads
        self.num_key_value_groups = config.num_attention_heads // config.num_key_value_heads  # GQA 分组
        self.scaling = self.head_dim ** -0.5  # 缩放因子

        self.q_proj = nn.Linear(config.hidden_size, config.num_attention_heads * self.head_dim, bias=False)
        self.k_proj = nn.Linear(config.hidden_size, config.num_key_value_heads * self.head_dim, bias=False)
        self.v_proj = nn.Linear(config.hidden_size, config.num_key_value_heads * self.head_dim, bias=False)
        self.o_proj = nn.Linear(config.num_attention_heads * self.head_dim, config.hidden_size, bias=False)

class LlamaDecoderLayer(GradientCheckpointingLayer):
    def __init__(self, config: LlamaConfig, layer_idx: int):
        self.self_attn = LlamaAttention(config=config, layer_idx=layer_idx)
        self.mlp = LlamaMLP(config)
        self.input_layernorm = LlamaRMSNorm(config.hidden_size, eps=config.rms_norm_eps)
        self.post_attention_layernorm = LlamaRMSNorm(config.hidden_size, eps=config.rms_norm_eps)

    def forward(self, hidden_states, attention_mask, ...):
        residual = hidden_states
        hidden_states = self.input_layernorm(hidden_states)      # Pre-Norm
        hidden_states, _ = self.self_attn(hidden_states, ...)    # Self-Attention
        hidden_states = residual + hidden_states                  # 残差连接

        residual = hidden_states
        hidden_states = self.post_attention_layernorm(hidden_states)  # Pre-Norm
        hidden_states = self.mlp(hidden_states)                       # FFN
        hidden_states = residual + hidden_states                      # 残差连接

class Pipeline(_ScikitCompat, PushToHubMixin):
    """Pipeline 基类，封装了完整的推理流程"""

from transformers import AutoModelForCausalLM

model = AutoModelForCausalLM.from_pretrained(
    pretrained_model_name_or_path,  # 模型ID（如 "meta-llama/Llama-2-7b-hf"）或本地路径
    config=None,                    # PreTrainedConfig 对象，覆盖自动加载的配置
    cache_dir=None,                 # 模型缓存目录
    force_download=False,           # 强制重新下载权重
    local_files_only=False,         # 仅使用本地缓存，不联网
    token=None,                     # HuggingFace 访问令牌（用于私有模型）
    revision="main",               # Git 版本（分支名、标签或 commit ID）
    torch_dtype="auto",            # 权重数据类型
    device_map=None,               # 设备映射策略
    trust_remote_code=False,       # 是否执行远程自定义代码
    quantization_config=None,      # 量化配置（BitsAndBytesConfig等）
    attn_implementation=None,      # 注意力实现方式
    low_cpu_mem_usage=False,       # 低CPU内存模式
)

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained(
    pretrained_model_name_or_path,  # 模型ID或本地路径
    use_fast=True,                  # 是否使用 Rust 实现的快速分词器
    padding_side=None,              # 填充方向："right" 或 "left"
    truncation_side=None,           # 截断方向："right" 或 "left"
    trust_remote_code=False,        # 是否信任远程代码
    token=None,                     # 访问令牌
    revision="main",               # 版本
    add_prefix_space=None,          # 某些分词器需要在文本前加空格
)

outputs = model.generate(
    inputs=None,                    # 输入张量（input_ids）
    generation_config=None,         # GenerationConfig 对象
    max_new_tokens=128,            # 生成的新 token 最大数量
    max_length=None,               # 输入+输出的总最大长度
    temperature=1.0,               # 采样温度
    top_p=1.0,                     # 核采样概率阈值
    top_k=50,                      # Top-K 采样的 K 值
    do_sample=False,               # 是否启用随机采样
    repetition_penalty=1.0,        # 重复惩罚系数
    num_beams=1,                   # 束搜索的束数
    num_return_sequences=1,        # 返回序列数
    no_repeat_ngram_size=0,        # 禁止重复的 n-gram 大小
    early_stopping=False,          # 束搜索是否提前停止
    streamer=None,                 # 流式输出对象（TextStreamer等）
    assistant_model=None,          # 辅助模型（用于推测解码）
    pad_token_id=None,             # 填充 token ID
    eos_token_id=None,             # 结束 token ID
)

from transformers import Trainer

trainer = Trainer(
    model=None,                     # PreTrainedModel 模型
    args=None,                      # TrainingArguments 训练参数
    data_collator=None,             # DataCollator 数据整理器
    train_dataset=None,             # 训练数据集
    eval_dataset=None,              # 评估数据集
    processing_class=None,          # 分词器或处理器
    model_init=None,                # 模型初始化函数
    compute_metrics=None,           # 评估指标计算函数
    callbacks=None,                 # 训练回调列表
    optimizers=(None, None),        # (优化器, 学习率调度器)
    preprocess_logits_for_metrics=None,  # 指标计算的 logits 预处理
)

from transformers import TrainingArguments

training_args = TrainingArguments(
    output_dir="./results",              # 输出目录
    num_train_epochs=3,                  # 训练轮数
    per_device_train_batch_size=8,       # 每设备训练批次大小
    per_device_eval_batch_size=8,        # 每设备评估批次大小
    gradient_accumulation_steps=1,       # 梯度累积步数
    learning_rate=5e-5,                  # 学习率
    weight_decay=0.01,                   # 权重衰减
    warmup_steps=0,                      # 学习率预热步数
    warmup_ratio=0.0,                    # 学习率预热比例
    lr_scheduler_type="linear",          # 学习率调度类型
    fp16=False,                          # 是否使用 FP16 混合精度
    bf16=False,                          # 是否使用 BF16 混合精度
    logging_steps=500,                   # 日志记录步数间隔
    eval_strategy="no",                  # 评估策略："no"/"steps"/"epoch"
    eval_steps=None,                     # 评估步数间隔
    save_strategy="steps",               # 保存策略："no"/"steps"/"epoch"
    save_steps=500,                      # 保存步数间隔
    save_total_limit=None,               # 最多保存的检查点数
    load_best_model_at_end=False,        # 训练结束时加载最佳模型
    metric_for_best_model=None,          # 最佳模型的评估指标
    greater_is_better=None,              # 指标是否越大越好
    gradient_checkpointing=False,        # 是否启用梯度检查点
    dataloader_num_workers=0,            # DataLoader 工作进程数
    remove_unused_columns=True,          # 是否移除模型不需要的列
    report_to=None,                      # 实验跟踪后端："wandb"/"tensorboard"等
    seed=42,                             # 随机种子
    data_seed=None,                      # 数据加载随机种子
    push_to_hub=False,                   # 训练结束后推送模型
)

from transformers import DataCollatorForLanguageModeling

collator = DataCollatorForLanguageModeling(
    tokenizer=tokenizer,
    mlm=False,               # False=因果语言模型, True=掩码语言模型
    mlm_probability=0.15,    # MLM 掩码概率（仅 mlm=True 时有效）
    pad_to_multiple_of=None, # 填充到某数的倍数（如 8，用于 Tensor Core 加速）
)

from transformers import DataCollatorForSeq2Seq

collator = DataCollatorForSeq2Seq(
    tokenizer=tokenizer,
    model=None,                      # 模型（用于准备 decoder_input_ids）
    padding=True,                    # 填充策略
    max_length=None,                 # 最大长度
    pad_to_multiple_of=None,         # 填充到某数的倍数
    label_pad_token_id=-100,         # 标签填充值（-100 在损失计算中被忽略）
    return_tensors="pt",             # 返回张量类型
)

from transformers import BitsAndBytesConfig

# 4-bit 量化配置（最常用）
bnb_config_4bit = BitsAndBytesConfig(
    load_in_4bit=True,                      # 启用 4-bit 量化
    bnb_4bit_quant_type="nf4",              # 量化类型："fp4" 或 "nf4"（推荐）
    bnb_4bit_compute_dtype=torch.bfloat16,  # 计算时的数据类型
    bnb_4bit_use_double_quant=True,         # 是否使用双重量化（量化常量再次量化）
    bnb_4bit_quant_storage=torch.uint8,     # 量化参数的存储类型
)

# 8-bit 量化配置
bnb_config_8bit = BitsAndBytesConfig(
    load_in_8bit=True,                      # 启用 8-bit 量化
    llm_int8_threshold=6.0,                 # 离群值检测阈值
    llm_int8_skip_modules=None,             # 跳过量化的模块列表
    llm_int8_enable_fp32_cpu_offload=False, # 是否启用 FP32 CPU 卸载
    llm_int8_has_fp16_weight=False,         # 是否使用 FP16 主权重（微调时有用）
)

if past_key_values is not None:
    key_states, value_states = past_key_values.update(key_states, value_states, self.layer_idx)

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

# 加载模型和分词器
model_name = "meta-llama/Llama-2-7b-hf"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,    # 使用 BF16 精度
    device_map="auto",              # 自动分配到可用 GPU
)

# 准备输入
prompt = "The future of artificial intelligence is"
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)

# 生成
with torch.no_grad():
    outputs = model.generate(
        **inputs,
        max_new_tokens=128,         # 最多生成 128 个新 token
        temperature=0.7,            # 较低温度，更确定的输出
        top_p=0.9,                  # 核采样
        do_sample=True,             # 启用采样
        repetition_penalty=1.1,     # 轻微重复惩罚
    )

# 解码输出
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
print(response)

import torch
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
    DataCollatorForLanguageModeling,
)
from datasets import load_dataset

# 加载模型和分词器
model_name = "meta-llama/Llama-2-7b-hf"
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token  # 设置 pad_token

model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
)

# 准备数据集
dataset = load_dataset("tatsu-lab/alpaca", split="train")

def tokenize_function(examples):
    # 将指令和输出拼接为训练文本
    texts = [f"### Instruction:\n{inst}\n### Response:\n{out}"
             for inst, out in zip(examples["instruction"], examples["output"])]
    return tokenizer(texts, truncation=True, max_length=512)

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

# 数据整理器（因果语言模型模式）
data_collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False)

# 训练参数
training_args = TrainingArguments(
    output_dir="./llama-finetuned",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,       # 等效 batch_size=16
    learning_rate=2e-5,
    lr_scheduler_type="cosine",          # 余弦退火
    warmup_ratio=0.03,                   # 预热 3% 的步数
    bf16=True,                           # BF16 混合精度训练
    logging_steps=10,
    save_strategy="epoch",
    gradient_checkpointing=True,         # 节省显存
    report_to="tensorboard",
)

# 创建 Trainer
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_dataset,
    data_collator=data_collator,
)

# 开始训练
trainer.train()

# 保存模型
trainer.save_model("./llama-finetuned/final")

from transformers import pipeline

# 创建文本生成管道
generator = pipeline(
    "text-generation",
    model="meta-llama/Llama-2-7b-hf",
    torch_dtype=torch.bfloat16,
    device_map="auto",
)

# 单条推理
result = generator(
    "Explain quantum computing in simple terms:",
    max_new_tokens=200,
    temperature=0.7,
    do_sample=True,
)
print(result[0]["generated_text"])

# 批量推理
prompts = [
    "What is machine learning?",
    "Explain the concept of neural networks.",
    "What are transformers in AI?",
]
results = generator(prompts, max_new_tokens=100, batch_size=3)

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig

# 4-bit 量化配置（NF4 + 双重量化）
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",                  # NormalFloat4 量化
    bnb_4bit_compute_dtype=torch.bfloat16,      # 计算时使用 BF16
    bnb_4bit_use_double_quant=True,             # 双重量化节省显存
)

# 加载模型
model_name = "meta-llama/Llama-2-7b-hf"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config,
    device_map="auto",
)

# 显存占用：7B 模型从 ~14GB (FP16) 降至 ~4GB (4-bit)
# 推理
inputs = tokenizer("Hello, how are you?", return_tensors="pt").to(model.device)
outputs = model.generate(**inputs, max_new_tokens=50)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))

# 8-bit 量化（更高精度，更多显存）
bnb_config_8bit = BitsAndBytesConfig(load_in_8bit=True)
model_8bit = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config_8bit,
    device_map="auto",
)
# 显存占用：~7GB

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

# 自动设备映射：accelerate 根据可用 GPU 显存自动分配模型层
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-13b-hf",
    torch_dtype=torch.bfloat16,
    device_map="auto",              # 自动跨 GPU 分配
    max_memory={                    # 可选：指定每设备最大显存
        0: "24GiB",
        1: "24GiB",
        "cpu": "30GiB",
    },
)

# 查看设备映射
print(model.hf_device_map)
# 输出示例: {'model.embed_tokens': 0, 'model.layers.0-15': 0, 'model.layers.16-31': 1, ...}

# 手动指定设备映射
device_map = {
    "model.embed_tokens": 0,
    "model.layers": 1,
    "model.norm": 1,
    "lm_head": 1,
}
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    torch_dtype=torch.bfloat16,
    device_map=device_map,
)

# 张量并行（需要 torchrun 启动）
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    torch_dtype=torch.bfloat16,
    tp_plan="auto",     # 使用预定义的张量并行计划
    tp_size=2,          # 2 路张量并行
)
# 启动: torchrun --nproc_per_node=2 script.py

from peft import LoraConfig, get_peft_model, TaskType

# LoRA 配置
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=16,                               # LoRA 秩
    lora_alpha=32,                       # LoRA 缩放因子
    lora_dropout=0.05,                   # LoRA Dropout
    target_modules=["q_proj", "v_proj"],  # 应用 LoRA 的目标模块
)

# 加载模型并应用 LoRA
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    quantization_config=bnb_config,  # 可配合 4-bit 量化
)
model = get_peft_model(model, lora_config)
print(f"可训练参数: {model.print_trainable_parameters()}")
# 可训练参数通常仅占全部参数的 0.1%-1%

from transformers import TextStreamer

streamer = TextStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)

outputs = model.generate(
    **inputs,
    max_new_tokens=256,
    streamer=streamer,              # 流式输出
    temperature=0.7,
    do_sample=True,
)
# 输出会实时逐 token 打印