ハギングフェイストランスフォーマー: エコシステムへの実践ガイド
ハグ顔 機械学習のリファレンスプラットフォームとなっています
現代的な。 500,000 を超える事前トレーニングされたモデル、100,000 以上のデータセットとライブラリを備えています。
transformers, datasets, peft, accelerate
e optimum、ほとんどの基盤となるインフラストラクチャを表します
現在の NLP とコンピューター ビジョンの研究開発について説明します。
この記事では、HuggingFace エコシステムを実践的かつ体系的に探っていきます。 ハブでの適切なモデルの選択から、微調整のためのトレーナー API まで、 大規模なデータセットの管理、モデルの最適化とデプロイまで。 カスタム トレーニング ループ、カスタム コールバック、 本番環境および MLOps システムとの統合向けに最適化された推論。
これはシリーズの 7 番目の記事です 最新の NLP: BERT から LLM へ。 BERT (記事 2) と感情分析 (記事 3) に精通していることを前提としています。
何を学ぶか
- HuggingFace エコシステム: 主要なライブラリとそれらをいつ使用するか
- モデルハブ: モデルの検索、フィルター、アップロード
- AutoClass API: AutoModel、AutoTokenizer、AutoConfig
- API パイプライン: 一般的なタスクのゼロ構成推論
- データセット ライブラリ: 大規模なデータセットの読み込み、操作、ストリーミング
- API トレーナー: ロギング、コールバック、チェックポイントによる完全な微調整
- ネイティブ PyTorch を使用したカスタム トレーニング ループ
- PEFT と LoRA: 少ないパラメータで効率的な微調整
- 加速: 分散トレーニングと混合精度
- 推論の最適化: ONNX、BitsAndBytes、量子化
- ハブへのプッシュ: モデルとデータセットをパブリックに共有します
- WandB、MLflow、MLOps システムとの統合
1. ハギングフェイスエコシステム
HuggingFace エコシステムは、多くの個別だが統合されたライブラリで構成されています。 車輪の再発明を避けるためには、シナリオごとにどれを使用するかを理解することが重要です そしてコミュニティの取り組みを最大限に活用してください。
主なライブラリと利用シーン
| 本棚 | 範囲 | 典型的なシナリオ | インストール |
|---|---|---|---|
transformers |
モデル、トークナイザー、トレーニング | BERT、推論パイプラインの微調整 | pip install transformers |
datasets |
データセット管理 | ロード、前処理、ストリーミング | pip install datasets |
peft |
効率的な微調整 | LoRA、プレフィックス チューニング、P チューニング | pip install peft |
accelerate |
分散型トレーニング | マルチ GPU、TPU、混合精度 | pip install accelerate |
optimum |
推論の最適化 | ONNX エクスポート、量子化、TensorRT | pip install optimum |
evaluate |
標準の NLP メトリクス | ブルー、ルージュ、F1、精度、シーケンス | pip install evaluate |
trl |
LLM の RLHF および SFT | 教育後の報酬モデリング | pip install trl |
safetensors |
重みの安全な形式 | 高速かつ安全な保存/読み込み | pip install safetensors |
sentence-transformers |
文の埋め込み | 意味的類似性、クラスタリング、RAG | pip install sentence-transformers |
tokenizers |
高速トークン化 | BPE、WordPiece、Unigram カスタム | pip install tokenizers |
適切なライブラリの選択はコンテキストによって異なります。ラピッドプロトタイピング用途向け
pipeline() da transformers。実稼働グレードの微調整用
使う Trainer con datasets。限られた GPU 上の大規模モデルの場合
使う peft。最適化された推論の使用のために optimum.
2. モデルハブ: 適切なモデルを見つける
Il ハグフェイスハブ 500,000 を超えるモデルをホストします。適切なものを見つけてください
利用可能なフィルタと命名規則を理解する必要があります。モデルが特定される
から username/model-name、言語、タスク、フレームワーク、およびデータセットのタグを使用します。
from huggingface_hub import HfApi, list_models, ModelFilter
import pandas as pd
api = HfApi()
# Cerca modelli per task e lingua
models = list(list_models(
filter=ModelFilter(
task="text-classification",
language="it", # italiano
),
sort="downloads",
direction=-1, # decrescente
limit=10
))
print("Top 10 modelli italiani per text-classification:")
for i, model in enumerate(models, 1):
print(f" {i}. {model.modelId} "
f"(downloads: {model.downloads:,}, likes: {model.likes})")
# Cerca modelli BERT italiani specificamente
bert_it_models = list(list_models(
search="bert italian",
sort="downloads",
direction=-1,
limit=5
))
# Carica informazioni dettagliate su un modello
model_info = api.model_info("dbmdz/bert-base-italian-cased")
print(f"\nModello: {model_info.modelId}")
print(f"Task: {model_info.pipeline_tag}")
print(f"Tag: {model_info.tags}")
print(f"Downloads/mese: {model_info.downloads:,}")
# Confronta modelli per benchmark
print("\n=== Modelli Italiani Consigliati per Task ===")
italian_models = {
"sentiment": [
"neuraly/bert-base-italian-cased-sentiment",
"MilaNLProc/feel-it-italian-sentiment",
"morenolq/bert-base-italian-cased-sentiment"
],
"ner": [
"osiria/bert-base-italian-uncased-ner",
"Babelscape/wikineural-multilingual-ner"
],
"embeddings": [
"nickprock/sentence-bert-base-italian-uncased",
"paraphrase-multilingual-mpnet-base-v2"
],
"base_models": [
"dbmdz/bert-base-italian-cased",
"dbmdz/bert-base-italian-uncased",
"idb-ita/gilberto-uncased-from-camembert"
]
}
for task, models_list in italian_models.items():
print(f"\n{task.upper()}:")
for m in models_list:
print(f" - {m}")ハブからテンプレートを選択する基準
- 毎月のダウンロード: コミュニティでの採用と信頼性の指標
- タスクタグ: モデルにタスク固有のヘッドがあることを確認します (例:
text-classification) - モデルカード: トレーニング ドキュメント、使用されるデータセット、ベンチマーク、制限事項
- 舌： ターゲット言語をサポートしていることを確認してください（タグ
it,multilingual) - サイズ： パフォーマンスと速度のバランスを保ちます。モデル
base(110M) は 3 ～ 4 倍高速ですlarge(340M) - データトレーニング: 同じアーキテクチャでも、最近のモデルは古いモデルよりも優れたパフォーマンスを発揮することがよくあります
3. AutoClass API: 柔軟なロード
Le オートクラス 任意の HuggingFace モデルをロードできるようにします
基盤となるアーキテクチャに関係なく、同じ API を使用できます。
ファイルのおかげでこれが可能になります config.json 各モデルに付属するもの
そして、インスタンス化する正確なクラスを指定します。
from transformers import (
AutoModel,
AutoModelForSequenceClassification,
AutoModelForTokenClassification,
AutoModelForCausalLM,
AutoModelForSeq2SeqLM,
AutoModelForQuestionAnswering,
AutoModelForMaskedLM,
AutoTokenizer,
AutoConfig,
AutoFeatureExtractor
)
import torch
# Carica configurazione senza scaricare i pesi (molto veloce)
config = AutoConfig.from_pretrained("bert-base-uncased")
print(f"Architettura: {config.architectures}")
print(f"Hidden size: {config.hidden_size}")
print(f"Num layers: {config.num_hidden_layers}")
print(f"Num attention heads: {config.num_attention_heads}")
print(f"Vocab size: {config.vocab_size}")
# Tokenizer (funziona per qualsiasi modello)
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
# Codifica testo
inputs = tokenizer(
"HuggingFace e fantastico!",
return_tensors="pt", # "pt" per PyTorch, "tf" per TensorFlow, "np" per NumPy
truncation=True,
max_length=128,
padding="max_length"
)
print(f"\nToken IDs shape: {inputs['input_ids'].shape}") # [1, 128]
# Modello base (senza testa task-specifica) - per feature extraction
model_base = AutoModel.from_pretrained("bert-base-uncased")
with torch.no_grad():
outputs = model_base(**inputs)
hidden_states = outputs.last_hidden_state # [1, 128, 768]
cls_embedding = hidden_states[:, 0, :] # CLS token [1, 768]
print(f"CLS embedding shape: {cls_embedding.shape}")
# Modello con testa per classificazione
model_clf = AutoModelForSequenceClassification.from_pretrained(
"distilbert-base-uncased-finetuned-sst-2-english"
)
print(f"\nModello classificazione labels: {model_clf.config.id2label}")
# Tabella delle AutoClass per task
autoclass_map = {
"AutoModelForSequenceClassification": "Classificazione testo, sentiment",
"AutoModelForTokenClassification": "NER, POS tagging",
"AutoModelForQuestionAnswering": "Extractive QA (SQuAD-style)",
"AutoModelForCausalLM": "Generazione testo (GPT-style)",
"AutoModelForSeq2SeqLM": "Traduzione, summarization (T5/mBART)",
"AutoModelForMaskedLM": "Masked language modeling (BERT)",
"AutoModelForMultipleChoice": "Multiple choice (SWAG, HellaSwag)",
}
print("\nMappa AutoClass -> Task:")
for cls, task in autoclass_map.items():
print(f" {cls}: {task}")
# Opzioni avanzate di caricamento
model_optimized = AutoModelForSequenceClassification.from_pretrained(
"bert-base-uncased",
num_labels=3,
torch_dtype=torch.float16, # fp16 per risparmiare memoria GPU (~50%)
device_map="auto", # distribuisce automaticamente su GPU disponibili
low_cpu_mem_usage=True, # carica parametri progressivamente
attn_implementation="flash_attention_2" # Flash Attention 2 se disponibile
)4. API パイプライン: 高速推論
La APIパイプライン そして、HuggingFace テンプレートを使用する最も簡単な方法です。 トークン化、推論、後処理を自動的に処理します。 プロトタイピングに最適ですが、バッチ処理を使用して実稼働環境でも使用できます。
from transformers import pipeline
import torch
# =========================================
# Task 1: Text Classification / Sentiment
# =========================================
sentiment = pipeline(
"sentiment-analysis",
model="distilbert-base-uncased-finetuned-sst-2-english",
device=0 if torch.cuda.is_available() else -1 # GPU se disponibile
)
results = sentiment(["I love this product!", "This is terrible.", "It's okay I guess."])
for r in results:
print(f" Label: {r['label']}, Score: {r['score']:.3f}")
# =========================================
# Task 2: Named Entity Recognition
# =========================================
ner = pipeline(
"ner",
model="dslim/bert-base-NER",
aggregation_strategy="simple" # aggrega token dello stesso entity
)
entities = ner("Apple CEO Tim Cook announced a new iPhone in Cupertino.")
for ent in entities:
print(f" '{ent['word']}' -> {ent['entity_group']} ({ent['score']:.3f})")
# =========================================
# Task 3: Question Answering
# =========================================
qa = pipeline("question-answering", model="deepset/roberta-base-squad2")
result = qa(
question="Chi ha fondato Tesla?",
context="Elon Musk ha co-fondato Tesla Motors nel 2003 insieme a Martin Eberhard."
)
print(f"\nQA Answer: '{result['answer']}' (score={result['score']:.3f})")
# =========================================
# Task 4: Text Generation
# =========================================
generator = pipeline(
"text-generation",
model="gpt2",
max_new_tokens=80,
num_return_sequences=2,
do_sample=True,
temperature=0.8,
top_p=0.95,
repetition_penalty=1.2
)
outputs = generator("Il futuro dell'intelligenza artificiale e")
for i, out in enumerate(outputs):
print(f"\nGeneration {i+1}: {out['generated_text']}")
# =========================================
# Task 5: Zero-Shot Classification
# =========================================
zero_shot = pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
result = zero_shot(
"L'economia italiana ha subito una contrazione dello 0.5% nel Q3 2024.",
candidate_labels=["economia", "politica", "sport", "tecnologia", "salute"]
)
print("\nZero-shot classification:")
for label, score in zip(result['labels'][:3], result['scores'][:3]):
print(f" {label}: {score:.3f}")
# =========================================
# Task 6: Summarization
# =========================================
summarizer = pipeline("summarization", model="facebook/bart-large-cnn",
min_length=30, max_length=130)
text = """
L'intelligenza artificiale generativa ha rivoluzionato il settore tecnologico nel 2024.
I modelli di linguaggio di grandi dimensioni come GPT-4, Claude e Gemini hanno dimostrato
capacità sorprendenti nel ragionamento, nella scrittura creativa e nella risoluzione di problemi
complessi. Le aziende di tutto il mondo stanno integrando queste tecnologie nei loro processi
produttivi, dalla customer service all'analisi dei dati, dalla generazione di codice alla
creazione di contenuti multimediali.
"""
summary = summarizer(text)[0]['summary_text']
print(f"\nSummary: {summary}")
# =========================================
# Task 7: Translation
# =========================================
translator = pipeline("translation_it_to_en", model="Helsinki-NLP/opus-mt-it-en")
italian_text = "Il machine learning sta trasformando il mondo moderno."
translated = translator(italian_text)[0]['translation_text']
print(f"\nTranslation: {translated}")
# =========================================
# Batch Processing per Performance
# =========================================
print("\n=== Batch Processing ===")
texts = [
"Ottimo prodotto, lo consiglio!",
"Pessima qualità, non lo ricomprero.",
"Nella media, niente di speciale."
] * 100 # 300 testi
# Batch inference e molto più efficiente di loop singoli
sentiment_batch = pipeline(
"sentiment-analysis",
model="distilbert-base-uncased-finetuned-sst-2-english",
batch_size=32 # processa 32 testi alla volta
)
results_batch = sentiment_batch(texts)
print(f"Processati {len(results_batch)} testi")5. データセット ライブラリ: 効率的なデータ管理
図書館 datasets Apache Arrow をバックエンドとして使用するため、
大規模なデータに対して非常に効率的です。すべての操作は遅延し、メモリマップされます。
RAM に収まらないデータセットを操作できるようになります。
from datasets import (
load_dataset,
Dataset,
DatasetDict,
concatenate_datasets,
interleave_datasets,
Features,
Value,
ClassLabel
)
import pandas as pd
from typing import Dict, List
# =========================================
# Caricamento da HuggingFace Hub
# =========================================
# Dataset pubblico con split
sst2 = load_dataset("glue", "sst2")
print("SST-2 dataset:", sst2)
print("Training size:", len(sst2["train"]))
print("Features:", sst2["train"].features)
# Con streaming (per dataset enormi - non scarica tutto)
wiki_stream = load_dataset(
"wikipedia",
"20220301.it",
split="train",
streaming=True,
trust_remote_code=True
)
# Prendi solo 5 esempi senza scaricare tutto il dataset
for i, example in enumerate(wiki_stream.take(5)):
print(f"Titolo: {example['title']} - Lunghezza: {len(example['text'])} chars")
# =========================================
# Creazione da sorgenti locali
# =========================================
# Da dizionario Python
data = Dataset.from_dict({
"text": [
"Ottimo prodotto, lo consiglio!",
"Pessima qualità, non vale i soldi.",
"Prodotto nella media, niente di eccezionale.",
"Fantastimo! Superato le aspettative."
],
"label": [1, 0, 0, 1]
})
# Da pandas DataFrame con tipi espliciti
df = pd.DataFrame({
"text": ["Sample text 1", "Sample text 2"],
"label": [0, 1],
"split": ["train", "train"]
})
dataset_from_df = Dataset.from_pandas(df)
# Da file con schema esplicito
features = Features({
"text": Value("string"),
"label": ClassLabel(names=["negativo", "positivo"]),
"confidence": Value("float32")
})
dataset_json = load_dataset("json", data_files="data.jsonl", features=features)
# =========================================
# Manipolazione avanzata
# =========================================
# map: tokenizzazione parallela
def tokenize_function(examples, tokenizer, max_length=128):
return tokenizer(
examples["text"],
truncation=True,
padding="max_length",
max_length=max_length,
return_token_type_ids=False
)
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
tokenized = data.map(
lambda x: tokenize_function(x, tokenizer),
batched=True, # processa batch per efficienza
batch_size=1000, # 1000 esempi per batch
num_proc=4, # usa 4 processi paralleli
remove_columns=["text"] # rimuovi colonne non necessarie
)
tokenized.set_format("torch") # converte in PyTorch tensors
# filter: rimuove esempi corti
long_texts = data.filter(
lambda x: len(x["text"].split()) > 5,
num_proc=4
)
# Balancing: sovracampionamento classe minoritaria
class_0 = data.filter(lambda x: x["label"] == 0)
class_1 = data.filter(lambda x: x["label"] == 1)
# Ripeti classe_0 per bilanciare
if len(class_0) < len(class_1):
factor = len(class_1) // len(class_0)
class_0_repeated = concatenate_datasets([class_0] * factor)
balanced = concatenate_datasets([class_0_repeated, class_1]).shuffle(seed=42)
# train_test_split
splits = data.train_test_split(test_size=0.2, seed=42, stratify_by_column="label")
train_ds = splits["train"]
test_ds = splits["test"]
print(f"\nTrain: {len(train_ds)}, Test: {len(test_ds)}")
# =========================================
# DatasetDict: gestione multi-split
# =========================================
dataset_dict = DatasetDict({
"train": train_ds,
"test": test_ds,
"validation": data.select(range(2))
})
# Salva e ricarica (formato Arrow efficiente)
dataset_dict.save_to_disk("./data/my_dataset")
loaded = DatasetDict.load_from_disk("./data/my_dataset")
# =========================================
# Statistiche e ispezione
# =========================================
print("\n=== Statistiche Dataset ===")
print(f"Numero esempi: {len(data)}")
print(f"Distribuzione label: {data.to_pandas()['label'].value_counts().to_dict()}")
print(f"Lunghezza media testi: {data.to_pandas()['text'].str.len().mean():.0f} chars")6. API トレーナー: 完全な微調整
La APIトレーナー そして、HuggingFace でのトレーニングのための高レベルの抽象化。 トレーニング ループ、評価、チェックポイント、ロギングなどを処理します。 混合精度、勾配累積、分散トレーニング、すぐに使用できる機能をサポート WandB、TensorBoard、MLflow との統合。
from transformers import (
AutoModelForSequenceClassification,
AutoTokenizer,
TrainingArguments,
Trainer,
EarlyStoppingCallback,
TrainerCallback,
TrainerControl,
TrainerState
)
from datasets import load_dataset
import evaluate
import numpy as np
import torch
MODEL = "distilbert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(MODEL)
model = AutoModelForSequenceClassification.from_pretrained(MODEL, num_labels=2)
# Preparazione dataset
dataset = load_dataset("glue", "sst2")
def tokenize(examples):
return tokenizer(
examples["sentence"],
truncation=True,
padding="max_length",
max_length=128
)
tokenized = dataset.map(
tokenize,
batched=True,
remove_columns=["sentence", "idx"]
)
tokenized.set_format("torch")
# Metriche composite
accuracy = evaluate.load("accuracy")
f1 = evaluate.load("f1")
def compute_metrics(eval_pred):
logits, labels = eval_pred
preds = np.argmax(logits, axis=-1)
probs = torch.softmax(torch.tensor(logits), dim=-1).numpy()
acc = accuracy.compute(predictions=preds, references=labels)["accuracy"]
f1_score = f1.compute(predictions=preds, references=labels, average="binary")["f1"]
# Aggiungi calibration metric (ECE - Expected Calibration Error)
from sklearn.calibration import calibration_curve
# Semplificato: usa solo le metriche base
return {"accuracy": acc, "f1": f1_score}
# TrainingArguments: configurazione completa
args = TrainingArguments(
# I/O e checkpoint
output_dir="./results/distilbert-sst2",
logging_dir="./logs",
logging_steps=50,
logging_strategy="steps",
# Epochs e batch
num_train_epochs=5,
per_device_train_batch_size=32,
per_device_eval_batch_size=64,
# Learning rate schedule
learning_rate=2e-5,
lr_scheduler_type="cosine", # cosine, linear, cosine_with_restarts, polynomial
warmup_ratio=0.1, # 10% di warm-up steps
weight_decay=0.01, # L2 regularization
# Valutazione e salvataggio
eval_strategy="epoch", # "no", "steps", "epoch"
save_strategy="epoch",
load_best_model_at_end=True,
metric_for_best_model="f1",
greater_is_better=True,
save_total_limit=3, # mantieni solo i 3 migliori checkpoint
# Ottimizzazione computazionale
fp16=True, # mixed precision FP16
# bf16=True, # alternativa su hardware recente (A100, RTX3090+)
dataloader_num_workers=4,
gradient_accumulation_steps=2, # batch effettivo = 32*2 = 64
# gradient_checkpointing=True, # riduce memoria a costo di +30% calcolo
max_grad_norm=1.0, # gradient clipping
# Reporting
report_to="none", # "wandb", "tensorboard", "mlflow", "comet_ml"
# Seed e reproducibilita
seed=42,
data_seed=42,
)
# =========================================
# Custom Callback: monitoring avanzato
# =========================================
class TrainingMonitorCallback(TrainerCallback):
def __init__(self, patience: int = 3):
self.patience = patience
self.best_metric = None
self.steps_without_improvement = 0
def on_evaluate(self, args, state: TrainerState, control: TrainerControl, metrics, **kwargs):
current_metric = metrics.get("eval_f1", 0)
if self.best_metric is None or current_metric > self.best_metric:
self.best_metric = current_metric
self.steps_without_improvement = 0
print(f"\n[Callback] Nuovo best F1: {current_metric:.4f}")
else:
self.steps_without_improvement += 1
print(f"\n[Callback] No improvement ({self.steps_without_improvement}/{self.patience})")
def on_log(self, args, state: TrainerState, control: TrainerControl, logs=None, **kwargs):
if logs and "loss" in logs:
if state.global_step % 200 == 0:
print(f" Step {state.global_step}: loss={logs['loss']:.4f}")
# Trainer con callbacks multipli
trainer = Trainer(
model=model,
args=args,
train_dataset=tokenized["train"],
eval_dataset=tokenized["validation"],
compute_metrics=compute_metrics,
callbacks=[
EarlyStoppingCallback(
early_stopping_patience=2,
early_stopping_threshold=0.001
),
TrainingMonitorCallback(patience=3)
]
)
# Training e valutazione
train_result = trainer.train()
print(f"\nTraining completato!")
print(f"Train loss: {train_result.training_loss:.4f}")
print(f"Train time: {train_result.metrics['train_runtime']:.1f}s")
print(f"Samples/sec: {train_result.metrics['train_samples_per_second']:.1f}")
# Valutazione finale
metrics = trainer.evaluate(eval_dataset=tokenized["validation"])
print(f"Validation F1: {metrics['eval_f1']:.4f}")
print(f"Validation Acc: {metrics['eval_accuracy']:.4f}")
# Salva tutto
trainer.save_model("./models/distilbert-sst2-final")
tokenizer.save_pretrained("./models/distilbert-sst2-final")
# Log history
import pandas as pd
log_history = pd.DataFrame(trainer.state.log_history)
print(f"\nLog history columns: {list(log_history.columns)}")7. PyTorch を使用したカスタム トレーニング ループ
Trainer API では不十分な高度なケースでは、トレーニング ループを作成できます。 すべての最適化を維持しながらカスタマイズします。これにより、最大限の制御が可能になります 損失関数のカスタム、サンプリング戦略、カリキュラム学習など。
from transformers import AutoModelForSequenceClassification, AutoTokenizer, get_linear_schedule_with_warmup
from torch.optim import AdamW
from torch.utils.data import DataLoader
from datasets import load_dataset
import torch
import numpy as np
from tqdm import tqdm
# =========================================
# Setup
# =========================================
MODEL = "bert-base-uncased"
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
EPOCHS = 3
BATCH_SIZE = 32
LR = 2e-5
tokenizer = AutoTokenizer.from_pretrained(MODEL)
model = AutoModelForSequenceClassification.from_pretrained(MODEL, num_labels=2).to(DEVICE)
# Dataset
dataset = load_dataset("glue", "sst2")
def tokenize(examples):
return tokenizer(examples["sentence"], truncation=True, padding="max_length", max_length=128)
tokenized = dataset.map(tokenize, batched=True, remove_columns=["sentence", "idx"])
tokenized.set_format("torch")
train_loader = DataLoader(
tokenized["train"],
batch_size=BATCH_SIZE,
shuffle=True,
num_workers=4,
pin_memory=True # velocizza trasferimento CPU->GPU
)
val_loader = DataLoader(tokenized["validation"], batch_size=64, num_workers=4)
# Optimizer con weight decay selettivo (no bias e LayerNorm)
no_decay = ["bias", "LayerNorm.weight", "LayerNorm.bias"]
optimizer_grouped = [
{"params": [p for n, p in model.named_parameters() if not any(nd in n for nd in no_decay)],
"weight_decay": 0.01},
{"params": [p for n, p in model.named_parameters() if any(nd in n for nd in no_decay)],
"weight_decay": 0.0}
]
optimizer = AdamW(optimizer_grouped, lr=LR, eps=1e-8)
# Learning rate scheduler con warmup
total_steps = len(train_loader) * EPOCHS
warmup_steps = int(total_steps * 0.1)
scheduler = get_linear_schedule_with_warmup(
optimizer,
num_warmup_steps=warmup_steps,
num_training_steps=total_steps
)
# Mixed precision scaler
scaler = torch.cuda.amp.GradScaler(enabled=torch.cuda.is_available())
# =========================================
# Training Loop
# =========================================
best_f1 = 0.0
for epoch in range(EPOCHS):
model.train()
total_loss = 0
progress_bar = tqdm(train_loader, desc=f"Epoch {epoch+1}/{EPOCHS}")
for step, batch in enumerate(progress_bar):
# Sposta batch su device
batch = {k: v.to(DEVICE) for k, v in batch.items()}
# Mixed precision forward pass
with torch.cuda.amp.autocast(enabled=torch.cuda.is_available()):
outputs = model(**batch)
loss = outputs.loss
# Backward pass con scaler
scaler.scale(loss).backward()
# Gradient clipping
scaler.unscale_(optimizer)
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
scaler.step(optimizer)
scaler.update()
optimizer.zero_grad()
scheduler.step()
total_loss += loss.item()
progress_bar.set_postfix({"loss": f"{loss.item():.4f}", "lr": f"{scheduler.get_last_lr()[0]:.2e}"})
avg_loss = total_loss / len(train_loader)
# Validation
model.eval()
all_preds, all_labels = [], []
with torch.no_grad():
for batch in val_loader:
batch = {k: v.to(DEVICE) for k, v in batch.items()}
outputs = model(**batch)
preds = torch.argmax(outputs.logits, dim=-1)
all_preds.extend(preds.cpu().numpy())
all_labels.extend(batch["labels"].cpu().numpy())
from sklearn.metrics import f1_score, accuracy_score
f1 = f1_score(all_labels, all_preds)
acc = accuracy_score(all_labels, all_preds)
print(f"\nEpoch {epoch+1}: loss={avg_loss:.4f}, val_f1={f1:.4f}, val_acc={acc:.4f}")
# Salva best model
if f1 > best_f1:
best_f1 = f1
model.save_pretrained("./models/best_model")
tokenizer.save_pretrained("./models/best_model")
print(f" Salvato nuovo best model (F1={f1:.4f})")8. PEFT: LoRA を使用した効率的な微調整
図書館 PEFT (パラメータ効率の良い微調整) できるようになります パラメータのごく一部のみを更新することで、大規模なモデルを微調整します。 LoRA (低ランク適応) 最もよく使用される方法: 更新を分解する 重みを 2 つの低ランク行列の積として計算し、トレーニング可能なパラメーターを減らす 99%程度。
PEFT 法の比較
| 方法 | トレーニング可能なパラメータ | メモリ | パフォーマンス | 使用事例 |
|---|---|---|---|---|
| 完全な微調整 | 100% | 高 (7B の場合は 40GB 以上) | 最大 | 大規模なデータセット、エンタープライズ GPU |
| LoRA (r=16) | ~0.5% | 低 (-70%) | ほぼ満員に等しい | コンシューマー向けGPU (8～24GB) |
| QLoRA | ~0.5% (4ビットモデル) | 非常に低い (-85%) | やや低め | GPU 8～16GB、大型モデル |
| プレフィックスチューニング | ~0.1% | 非常に低い | 劣る | ジェネレーション、LLM |
| 即時チューニング | ～0.01% | 最小限 | 変数 | 大規模な LLM (>10B) |
| アダプター層 | ~1-3% | 低い | 良い | マルチタスク、モジュール式 |
from peft import (
LoraConfig,
get_peft_model,
TaskType,
PeftModel,
prepare_model_for_kbit_training,
get_peft_config
)
from transformers import AutoModelForSequenceClassification, BitsAndBytesConfig
import torch
# =========================================
# LoRA Standard
# =========================================
model = AutoModelForSequenceClassification.from_pretrained(
"roberta-base",
num_labels=3
)
# Configurazione LoRA
lora_config = LoraConfig(
task_type=TaskType.SEQ_CLS,
r=16, # rank della decomposizione
lora_alpha=32, # scaling factor (alpha/r = scaling ratio)
lora_dropout=0.1, # dropout sui layer LoRA
target_modules=["query", "value", "key"], # layer da modificare
# Alternativa: target_modules="all-linear" per tutti i layer lineari
bias="none", # "none", "all", "lora_only"
inference_mode=False
)
peft_model = get_peft_model(model, lora_config)
peft_model.print_trainable_parameters()
# trainable params: 888,578 || all params: 125,535,234 || trainable%: 0.71%
# =========================================
# QLoRA: LoRA con quantizzazione 4-bit
# =========================================
bnb_config = BitsAndBytesConfig(
load_in_4bit=True, # quantizza il modello a 4bit
bnb_4bit_use_double_quant=True, # double quantization per efficienza
bnb_4bit_quant_type="nf4", # NormalFloat4 (migliore per LM)
bnb_4bit_compute_dtype=torch.bfloat16
)
# Solo per modelli grandi (>1B parametri)
model_4bit = AutoModelForSequenceClassification.from_pretrained(
"bert-large-uncased",
quantization_config=bnb_config,
device_map="auto",
num_labels=3
)
# Prepara per QLoRA training
model_4bit = prepare_model_for_kbit_training(model_4bit)
lora_config_qlora = LoraConfig(
task_type=TaskType.SEQ_CLS,
r=16,
lora_alpha=32,
lora_dropout=0.05,
target_modules=["query", "value"],
bias="none"
)
peft_4bit_model = get_peft_model(model_4bit, lora_config_qlora)
peft_4bit_model.print_trainable_parameters()
# =========================================
# Salvataggio e caricamento LoRA
# =========================================
# Salva solo i pesi LoRA (molto leggeri ~1-5MB)
peft_model.save_pretrained("./models/roberta-lora-classification")
# Caricamento: base model + adapter LoRA
base = AutoModelForSequenceClassification.from_pretrained("roberta-base", num_labels=3)
model_with_lora = PeftModel.from_pretrained(base, "./models/roberta-lora-classification")
model_with_lora.eval()
# Merge per inference più veloce (elimina overhead LoRA)
merged = model_with_lora.merge_and_unload()
merged.save_pretrained("./models/roberta-merged") # salva modello completo fuso9. 加速: 分散トレーニング
スピードアップ トレーニングの複雑さを自動的に管理します。 さまざまなハードウェア構成: シングル CPU、シングル GPU、マルチ GPU、マルチノード、 TPU、混合精度。コードの変更は最小限です。
# accelerate_training.py
# Avvio: accelerate launch accelerate_training.py
# Multi-GPU: accelerate launch --num_processes 4 accelerate_training.py
# Config: accelerate config (wizard interattivo)
from accelerate import Accelerator
from accelerate.utils import set_seed, ProjectConfiguration
from transformers import AutoModelForSequenceClassification, AutoTokenizer, get_cosine_schedule_with_warmup
from torch.optim import AdamW
from torch.utils.data import DataLoader
from datasets import load_dataset
import torch
from tqdm import tqdm
# =========================================
# Inizializzazione Accelerate
# =========================================
project_config = ProjectConfiguration(
project_dir="./accelerate_project",
logging_dir="./logs"
)
accelerator = Accelerator(
mixed_precision="fp16", # "no", "fp16", "bf16"
gradient_accumulation_steps=4, # accumula gradienti per batch più grandi
log_with="tensorboard", # logging integrato
project_config=project_config
)
# Importante: usa accelerator.print per evitare output duplicati su multi-GPU
accelerator.print(f"Training su: {accelerator.device}")
accelerator.print(f"Num processi: {accelerator.num_processes}")
accelerator.print(f"Mixed precision: {accelerator.mixed_precision}")
set_seed(42)
MODEL = "bert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(MODEL)
model = AutoModelForSequenceClassification.from_pretrained(MODEL, num_labels=2)
# Dataset
dataset = load_dataset("glue", "sst2")
def tokenize(examples):
return tokenizer(examples["sentence"], truncation=True, padding="max_length", max_length=128)
tokenized = dataset.map(tokenize, batched=True, remove_columns=["sentence", "idx"])
tokenized.set_format("torch")
train_loader = DataLoader(tokenized["train"], batch_size=32, shuffle=True, num_workers=4)
val_loader = DataLoader(tokenized["validation"], batch_size=64, num_workers=4)
optimizer = AdamW(model.parameters(), lr=2e-5, weight_decay=0.01)
total_steps = (len(train_loader) // accelerator.gradient_accumulation_steps) * 3
scheduler = get_cosine_schedule_with_warmup(optimizer, num_warmup_steps=total_steps//10, num_training_steps=total_steps)
# Prepara TUTTI gli oggetti con Accelerate
model, optimizer, train_loader, val_loader, scheduler = accelerator.prepare(
model, optimizer, train_loader, val_loader, scheduler
)
# =========================================
# Training loop con gradient accumulation
# =========================================
for epoch in range(3):
model.train()
total_loss = 0
for step, batch in enumerate(tqdm(train_loader, desc=f"Epoch {epoch+1}")):
# Accumulation context manager
with accelerator.accumulate(model):
outputs = model(**batch)
loss = outputs.loss
accelerator.backward(loss) # invece di loss.backward()
if accelerator.sync_gradients:
# Gradient clipping (solo quando si aggiornano i pesi)
accelerator.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
scheduler.step()
optimizer.zero_grad()
total_loss += loss.item()
avg_loss = total_loss / len(train_loader)
accelerator.print(f"\nEpoch {epoch+1}: avg_loss={avg_loss:.4f}")
# Salvataggio checkpoint (gestisce correttamente multi-GPU)
accelerator.save_state(f"./checkpoints/epoch_{epoch+1}")
# Salvataggio modello finale
unwrapped_model = accelerator.unwrap_model(model) # rimuove wrapper Accelerate
unwrapped_model.save_pretrained(
"./models/final_model",
save_function=accelerator.save
)10. ハブへのプッシュ: テンプレートの共有
from huggingface_hub import HfApi, login, create_repo
from transformers import AutoModelForSequenceClassification, AutoTokenizer
# Autenticazione (usa token da https://huggingface.co/settings/tokens)
login(token="hf_YOUR_TOKEN_HERE")
# =========================================
# Metodo 1: Trainer API (push automatico)
# =========================================
from transformers import TrainingArguments
training_args = TrainingArguments(
output_dir="tuo-username/bert-italian-sentiment", # formato: username/repo-name
push_to_hub=True,
hub_strategy="every_save", # "end", "every_save", "checkpoint", "all_checkpoints"
hub_token="hf_YOUR_TOKEN",
)
# =========================================
# Metodo 2: Push manuale dopo training
# =========================================
model = AutoModelForSequenceClassification.from_pretrained("./models/distilbert-sst2-final")
tokenizer = AutoTokenizer.from_pretrained("./models/distilbert-sst2-final")
# Push modello e tokenizer
model.push_to_hub("tuo-username/bert-italian-sentiment")
tokenizer.push_to_hub("tuo-username/bert-italian-sentiment")
# =========================================
# Metodo 3: HfApi completo con Model Card
# =========================================
api = HfApi()
# Crea repository se non esiste
create_repo(
repo_id="tuo-username/bert-italian-sentiment",
private=False, # True per repository privato
exist_ok=True
)
# Crea Model Card completa
model_card = """---
language:
- it
tags:
- sentiment-analysis
- bert
- italian
- transformers
license: apache-2.0
metrics:
- f1
- accuracy
model-index:
- name: bert-italian-sentiment
results:
- task:
type: text-classification
name: Sentiment Analysis
dataset:
name: SENTIPOLC 2016
type: custom
metrics:
- type: f1
value: 0.924
name: F1 Score
- type: accuracy
value: 0.931
name: Accuracy
---
# BERT Italian Sentiment Analysis
Modello BERT fine-tuned per sentiment analysis in italiano, basato su `dbmdz/bert-base-italian-cased`.
## Utilizzo
```python
from transformers import pipeline
sentiment = pipeline("sentiment-analysis", model="tuo-username/bert-italian-sentiment")
result = sentiment("Ottimo prodotto, lo consiglio!")
print(result) # [{'label': 'POSITIVE', 'score': 0.998}]
```
## Metriche
| Metrica | Valore |
|---------|--------|
| F1 | 0.924 |
| Accuracy | 0.931 |
## Training
- **Base model**: dbmdz/bert-base-italian-cased
- **Dataset**: 50.000 recensioni in italiano
- **Epochs**: 5
- **Batch size**: 32
- **Learning rate**: 2e-5
## Limitazioni
- Addestrato su recensioni di prodotti. Performance può variare su altri domini.
- Non adatto per ironia/sarcasmo sottile.
"""
# Carica README
api.upload_file(
path_or_fileobj=model_card.encode(),
path_in_repo="README.md",
repo_id="tuo-username/bert-italian-sentiment"
)
print("Model card caricata con successo!")11. 本番環境での推論の最適化
運用環境では、推論は高速、効率的、スケーラブルである必要があります。ハグ顔 さまざまな最適化戦略を提供します: ONNX ランタイム、静的/動的量子化、 TorchScript、およびテキスト推論サーバー (TGI/TEI)。
from optimum.onnxruntime import ORTModelForSequenceClassification
from optimum.exporters.onnx import main_export
from transformers import AutoTokenizer
import torch
import time
import numpy as np
# =========================================
# Export ONNX con ottimizzazione
# =========================================
main_export(
model_name_or_path="./models/distilbert-sst2-final",
output="./models/onnx-optimized",
task="text-classification",
optimize="O2" # O1: base, O2: extended, O3: layout opt, O4: full + float16
)
# Carica e usa il modello ONNX
ort_model = ORTModelForSequenceClassification.from_pretrained(
"./models/onnx-optimized",
provider="CPUExecutionProvider" # "CUDAExecutionProvider" per GPU
)
tokenizer = AutoTokenizer.from_pretrained("./models/distilbert-sst2-final")
# =========================================
# Benchmark: PyTorch vs ONNX vs Quantizzato
# =========================================
texts = ["This product is absolutely amazing!"] * 200
def benchmark_model(model, tokenizer, texts, batch_size=32, num_runs=5):
"""Misura latenza media su batch di testi."""
times = []
for _ in range(num_runs):
start = time.perf_counter()
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
inputs = tokenizer(batch, return_tensors='pt', padding=True,
truncation=True, max_length=128)
with torch.no_grad():
if hasattr(model, '__call__'):
_ = model(**inputs)
times.append(time.perf_counter() - start)
return np.mean(times), np.std(times)
# Test ONNX
avg_time, std_time = benchmark_model(ort_model, tokenizer, texts)
print(f"ONNX (200 testi): {avg_time*1000:.1f}ms ± {std_time*1000:.1f}ms")
# =========================================
# Quantizzazione dinamica PyTorch
# =========================================
from transformers import AutoModelForSequenceClassification
pt_model = AutoModelForSequenceClassification.from_pretrained(
"./models/distilbert-sst2-final"
)
pt_model.eval()
# Quantizzazione dinamica INT8 (nessun dataset di calibrazione necessario)
quantized_model = torch.quantization.quantize_dynamic(
pt_model,
{torch.nn.Linear}, # quantizza solo layer lineari
dtype=torch.qint8 # INT8
)
# Confronto dimensioni
import os
pt_size = sum(p.numel() * p.element_size() for p in pt_model.parameters()) / 1e6
print(f"\nDimensione PyTorch FP32: {pt_size:.1f}MB")
# INT8 e circa 4x più piccolo
# =========================================
# Serving con Text Embeddings Inference (TEI)
# =========================================
# Docker: docker run -p 8080:80 ghcr.io/huggingface/text-embeddings-inference:latest
# --model-id BAAI/bge-base-en-v1.5 --pooling mean
# Esempio client per TEI
import requests
def get_embeddings_tei(texts: list, url: str = "http://localhost:8080") -> np.ndarray:
"""Chiama Text Embeddings Inference server."""
response = requests.post(
f"{url}/embed",
json={"inputs": texts, "normalize": True}
)
response.raise_for_status()
return np.array(response.json())
# embeddings = get_embeddings_tei(["Test sentence"]) # Richiede TEI server attivo
print("\nTEI server endpoint: POST http://localhost:8080/embed")12. MLOps との統合: WandB および MLflow
本番グレードのコンテキストでは、トレーニングを監視し、バージョン管理する必要があります そして再現可能です。 HuggingFace Trainer は主要なトレーナーとネイティブに統合されます MLOps ツール。
import os
import wandb
import mlflow
import mlflow.pytorch
from transformers import (
AutoModelForSequenceClassification,
AutoTokenizer,
TrainingArguments,
Trainer
)
# =========================================
# Integrazione WandB
# =========================================
wandb.init(
project="bert-italian-sentiment",
name="distilbert-sst2-run1",
config={
"model": "distilbert-base-uncased",
"learning_rate": 2e-5,
"epochs": 3,
"batch_size": 32,
"dataset": "SST-2"
},
tags=["bert", "sentiment", "italian", "fine-tuning"]
)
# Il Trainer usa automaticamente WandB se disponibile
args_wandb = TrainingArguments(
output_dir="./results",
report_to="wandb", # abilita WandB logging
run_name="distilbert-run1", # nome nella dashboard WandB
num_train_epochs=3,
# ... altri parametri
)
# =========================================
# Integrazione MLflow
# =========================================
mlflow.set_tracking_uri("./mlflow_runs")
mlflow.set_experiment("bert-nlp-experiments")
with mlflow.start_run(run_name="distilbert-sst2"):
# Log parametri
mlflow.log_params({
"model": "distilbert-base-uncased",
"lr": 2e-5,
"epochs": 3,
"batch_size": 32
})
# Trainer con MLflow
args_mlflow = TrainingArguments(
output_dir="./results",
report_to="mlflow",
num_train_epochs=3,
)
model = AutoModelForSequenceClassification.from_pretrained("distilbert-base-uncased", num_labels=2)
tokenizer = AutoTokenizer.from_pretrained("distilbert-base-uncased")
trainer = Trainer(model=model, args=args_mlflow)
# trainer.train() # avvia training con MLflow tracking
# Log metriche e artefatti
mlflow.log_metrics({"eval_f1": 0.924, "eval_accuracy": 0.931})
# Salva modello come artefatto MLflow
mlflow.pytorch.log_model(
model,
"model",
registered_model_name="bert-italian-sentiment"
)
print("MLflow run completato!")
print(f"Run ID: {mlflow.active_run().info.run_id}")HuggingFace の一般的なアンチパターン
- バッチ処理は使用しないでください。 個々のテキストに対してループでパイプラインを呼び出すため、バッチ全体を渡すよりも 10 ～ 50 倍遅くなります
- 呼び出しごとにモデルをロードします。 モデルをメモリにロードしたままにして再利用します。ディスクからのロードには数秒かかります
- max_length を無視します: 切り詰めないと、長いシーケンスはメモリを指数関数的に消費します。常に設定します
truncation=True - 使用しないでください
torch.no_grad()推論では： PyTorch は不必要に勾配を蓄積し、メモリを浪費します - eval モードに変換しないでください。 BatchNorm と Dropout は、トレーニングと推論で動作が異なります。いつも電話する
model.eval() - 常にpadding="max_length"を使用してください: 固定長パディングは計算を無駄にします。推論では動的パディングを使用します
- 語彙サイズを無視する: 大きな語彙を持つ埋め込みレイヤーはモデルのメモリを支配する可能性があります
結論と次のステップ
HuggingFace エコシステムは、現代の NLP の事実上の標準になっています。
その主要なライブラリを知る — transformers, datasets,
peft, accelerate —そしてあらゆるものにとって基本的なもの
2025 年には NLP エンジニアまたは ML エンジニア。
エコシステムの強みはコンポーネント間の統合にあります。データセットが提供するものは次のとおりです。 データ、モデルの変換、パラメータの最適化、加速 ハードウェアのスケーラビリティと最適な生産推論。どの図書館でもできること 単独で使用することも、他のものと組み合わせて使用することもできます。
重要なポイント
- アメリカ合衆国 オートクラス 同じコードを使用して任意のアーキテクチャをロードする
- La APIパイプライン ラピッドプロトタイピングに最適です。アメリカ合衆国
batch_size生産用 - La APIトレーナー カスタマイズ可能なコールバックを使用して、標準的な微調整ケースの 90% を処理します
- Il カスタムトレーニングループ カスタム損失関数、カリキュラム学習、および高度なトレーニングに必要です
- PEFT/LoRA メモリを大幅に削減: ほぼ同等のパフォーマンスでパラメータの最大 0.5%
- スピードアップ コードを変更せずに分散トレーニングが可能
- ONNX ネイティブ PyTorch と比較して CPU 推論の速度が 2 ～ 5 倍向上します
- 統合する WandB または MLflow 実験のトレーサビリティのために最初から
モダンNLPシリーズは続く
- 前の： テキスト分類: マルチラベルおよびマルチクラス — BERT によるテキスト分類
- 次： LLM をローカルで微調整する: コンシューマー GPU 上の LoRA — LLM 7B+ 向けの高度な QLoRA
- 第9条： 意味的な類似性とテキストの一致 — SBERT、FAISS、密検索
- 第10条： 本番環境での NLP モデルのモニタリング — ドリフト検出、自動再トレーニング
- 関連シリーズ： AIエンジニアリング/RAG — RAG パイプラインの HuggingFace 埋め込み
- 関連シリーズ： 高度なディープラーニング — 量子化とモデルの最適化