HuggingFace Transformers: 생태계에 대한 실용 가이드
포옹얼굴 머신러닝의 참조 플랫폼이 되었습니다.
현대. 500,000개 이상의 사전 훈련된 모델, 100,000개 이상의 데이터세트 및 라이브러리
transformers, datasets, peft, accelerate
e optimum는 대부분의 기반이 되는 인프라를 나타냅니다.
현재 NLP 및 컴퓨터 비전 연구 및 개발에 대해 설명합니다.
이 글에서는 HuggingFace 생태계를 실용적이고 체계적인 방식으로 살펴보겠습니다. 허브에서 올바른 모델을 선택하는 것부터 미세 조정을 위한 Trainer API까지, 대규모 데이터세트 관리, 모델 최적화 및 배포까지. 또한 사용자 정의 훈련 루프, 사용자 정의 콜백, MLOps 시스템과의 통합 및 생산에 최적화된 추론입니다.
이 시리즈의 일곱 번째 기사입니다 최신 NLP: BERT에서 LLM까지. BERT(2조) 및 감정 분석(3조)에 대한 익숙함을 전제로 합니다.
무엇을 배울 것인가
- HuggingFace 생태계: 주요 라이브러리 및 사용 시기
- 모델 허브: 모델 검색, 필터링 및 업로드
- AutoClass API: AutoModel, AutoTokenizer, AutoConfig
- API 파이프라인: 일반 작업에 대한 구성이 없는 추론
- 데이터세트 라이브러리: 대규모 데이터세트 로드, 조작 및 스트리밍
- API Trainer: 로깅, 콜백 및 체크포인트를 통한 완벽한 미세 조정
- 기본 PyTorch를 사용한 맞춤형 훈련 루프
- PEFT 및 LoRA: 몇 가지 매개변수를 사용한 효율적인 미세 조정
- 가속화: 분산 훈련 및 혼합 정밀도
- 추론 최적화: ONNX, BitsAndBytes, 양자화
- 허브로 푸시: 모델 및 데이터 세트를 공개적으로 공유
- WandB, MLflow 및 MLOps 시스템과 통합
1. HuggingFace 생태계
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 측정항목 | BLEU, ROUGE, 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 ~와 함께 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(낮은 순위 적응) 가장 많이 사용되는 방법: 업데이트를 분해합니다. 가중치를 두 개의 하위 행렬의 곱으로 계산하여 훈련 가능한 매개변수를 줄입니다. 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는 불필요하게 그래디언트를 축적하여 메모리를 낭비합니다. - 평가 모드로 변환하지 마세요. 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 임베딩
- 관련 시리즈: 고급 딥러닝 — 양자화 및 모델 최적화