Hugging Face Transformers 深度指南：从 Pipeline 到生产部署

pip install transformers[torch]

python -c "import transformers; print(transformers.__version__)"

python -c "import torch; print(f'CUDA: {torch.cuda.is_available()}, GPU: {torch.cuda.get_device_name(0)}')"

from transformers import pipeline

classifier = pipeline("sentiment-analysis")
result = classifier("I love using Hugging Face Transformers!")
# [{'label': 'POSITIVE', 'score': 0.9998}]

# 命名实体识别
ner = pipeline("ner", grouped_entities=True)
ner("Hugging Face is based in New York City.")
# → [{'entity_group': 'ORG', 'word': 'Hugging Face'},
#    {'entity_group': 'LOC', 'word': 'New York City'}]

# 问答
qa = pipeline("question-answering")
qa(question="Who developed Transformers?",
   context="Transformers is a library developed by Hugging Face.")
# → {'answer': 'Hugging Face', 'score': 0.9995}

# 摘要
summarizer = pipeline("summarization")
summarizer(article, max_length=50, min_length=20)

# 翻译（需要明确指定模型）
translator = pipeline("translation_en_to_fr", model="Helsinki-NLP/opus-mt-en-fr")
translator("Hugging Face Transformers is awesome!")

# 图像分类
classifier = pipeline("image-classification", model="google/vit-base-patch16-224")
classifier("cat.jpg")

from concurrent.futures import ThreadPoolExecutor

def classify_batch(texts):
    # 每个线程创建自己的 pipeline
    classifier = pipeline("text-classification")
    results = []
    for text in texts:
        results.append(classifier(text))
    return results

from transformers import AutoModel, AutoTokenizer

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

# ❌ 错误：用 AutoModel 做分类，输出是 hidden states 不是 logits
model = AutoModel.from_pretrained("bert-base-uncased")
outputs = model(**inputs)
# outputs.last_hidden_state 是 (batch, seq_len, 768)，不是分类结果

# ✅ 正确：用 AutoModelForSequenceClassification
model = AutoModelForSequenceClassification.from_pretrained(
    "bert-base-uncased", num_labels=2
)
outputs = model(**inputs)
# outputs.logits 才是 (batch, 2)，可以对它做 softmax

tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
encoded = tokenizer("Hello, world!")
# {'input_ids': [101, 7592, 1010, 2088, 102],
#  'token_type_ids': [0, 0, 0, 0, 0],
#  'attention_mask': [1, 1, 1, 1, 1]}

# 批量编码
tokenizer(
    ["Hello world", "Transformers is great"],
    padding=True,        # 短句补零到最长
    truncation=True,     # 超长截断
    max_length=128,      # 最大长度
    return_tensors="pt"  # 返回 PyTorch tensor
)

from datasets import load_dataset

dataset = load_dataset("glue", "sst2")
# train: 67,349 条 / validation: 872 条
# 每条: {'sentence': "...", 'label': 0/1}

from transformers import AutoTokenizer

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

def tokenize_fn(examples):
    return tokenizer(
        examples["sentence"],
        truncation=True,
        padding="max_length",
        max_length=128
    )

tokenized = dataset.map(tokenize_fn, batched=True)

from transformers import (
    AutoModelForSequenceClassification,
    Trainer,
    TrainingArguments
)

model = AutoModelForSequenceClassification.from_pretrained(
    "distilbert-base-uncased", num_labels=2
)

training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=64,
    evaluation_strategy="epoch",      # 每个 epoch 评估一次
    save_strategy="epoch",            # 每个 epoch 保存一次
    load_best_model_at_end=True,      # 训练完加载最佳 checkpoint
    metric_for_best_model="accuracy", # 用 accuracy 判断最佳
    logging_dir="./logs",
    logging_steps=100,
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized["train"],
    eval_dataset=tokenized["validation"],
)

trainer.train()

import torch
from torch.utils.data import DataLoader
from transformers import get_linear_schedule_with_warmup

model = AutoModelForSequenceClassification.from_pretrained(
    "distilbert-base-uncased", num_labels=2
)
optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)

# 学习率预热：前 10% 步线性增长，之后线性衰减
total_steps = len(train_loader) * 3
scheduler = get_linear_schedule_with_warmup(
    optimizer,
    num_warmup_steps=int(0.1 * total_steps),
    num_training_steps=total_steps
)

for epoch in range(3):
    model.train()
    for batch in train_loader:
        # 只传模型需要的 key
        inputs = {k: v for k, v in batch.items()
                  if k in ["input_ids", "attention_mask", "labels"]}
        outputs = model(**inputs)
        loss = outputs.loss
        loss.backward()
        optimizer.step()
        scheduler.step()
        optimizer.zero_grad()

[原始文本] → Tokenizer → [input_ids] → Model → [logits] → softmax → [label, score]
    ↑                                                                    │
    └── 训练阶段：SST-2 数据集 → 微调 DistilBERT → 保存 checkpoint ──────┘

# 1. 微调并保存
model.save_pretrained("./my_sentiment_model")
tokenizer.save_pretrained("./my_sentiment_model")

# 2. 加载并推理
model = AutoModelForSequenceClassification.from_pretrained("./my_sentiment_model")
tokenizer = AutoTokenizer.from_pretrained("./my_sentiment_model")

# 3. 封装成服务
from fastapi import FastAPI
import torch

app = FastAPI()

@app.post("/predict")
def predict(text: str):
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=128)
    with torch.no_grad():
        logits = model(**inputs).logits
    probs = torch.softmax(logits, dim=-1)
    label_id = torch.argmax(probs, dim=-1).item()
    return {
        "text": text,
        "label": "POSITIVE" if label_id == 1 else "NEGATIVE",
        "confidence": probs[0][label_id].item()
    }

# 语音识别
asr = pipeline("automatic-speech-recognition", model="openai/whisper-base")
asr("meeting.wav")

# 图像分割
segmenter = pipeline("image-segmentation", model="facebook/detr-resnet-50-panoptic")
segmenter("photo.jpg")

# 视觉问答
vqa = pipeline("vqa", model="dandelin/vilt-b32-finetuned-vqa")
vqa(image="chart.png", question="What is the trend?")

from transformers import AutoModelForCausalLM, BitsAndBytesConfig
import torch

quant_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_compute_dtype=torch.float16,
    bnb_4bit_use_double_quant=True,  # 二次量化，进一步压缩
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    quantization_config=quant_config,
    device_map="auto"
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    attn_implementation="flash_attention_2",
    torch_dtype=torch.float16,
    device_map="auto"
)

from optimum.onnxruntime import ORTModelForSequenceClassification

model = ORTModelForSequenceClassification.from_pretrained(
    "distilbert-base-uncased-finetuned-sst-2-english", export=True
)
model.save_pretrained("./optimized_model")

# 方案 1：减小 batch size
training_args = TrainingArguments(per_device_train_batch_size=4)

# 方案 2：梯度检查点（用时间换显存）
model.gradient_checkpointing_enable()

# 方案 3：清理碎片
import torch
torch.cuda.empty_cache()

encoded = tokenizer("Hello world!", return_offsets_mapping=True)
# offset_mapping: [(0, 0), (0, 5), (5, 6), (6, 11), (11, 12), (0, 0)]
#                   [CLS]    Hello      ,      world      !      [SEP]

start, end = encoded["offset_mapping"][1]  # position 1 = "Hello"
print(f"Token 'Hello' spans characters {start} to {end}")

# 方案 1：用镜像
export HF_ENDPOINT=https://hf-mirror.com

# 方案 2：先下载再加载
from huggingface_hub import snapshot_download
snapshot_download("bert-base-uncased")

# 方案 3：用国内模型源
# ModelScope（魔搭社区）提供了大部分主流模型的镜像