Sentiment Analysis con Transformers: Tecniche e Implementazione
La Sentiment Analysis — o analisi del sentiment — e il task NLP più richiesto in ambito aziendale. Ogni giorno, milioni di aziende analizzano recensioni di prodotti, post sui social media, ticket di supporto e feedback dei clienti per capire cosa pensano davvero le persone. Con l'avvento di BERT e dei modelli Transformer, la qualità di questi sistemi e migliorata radicalmente rispetto ai classici approcci basati su dizionari o TF-IDF.
In questo articolo costruiremo un sistema completo di sentiment analysis: dalla comprensione dei dataset alla messa in produzione, passando per il fine-tuning con HuggingFace, la gestione delle classi sbilanciate, la valutazione delle metriche e le strategie per gestire i casi limite come ironia, negazione e linguaggio ambiguo.
Questo e il terzo articolo della serie NLP Moderno: da BERT ai LLM. Assume familiarita con i fondamenti BERT (articolo 2). Per l'italiano specificamente, vedi l'articolo 4 dedicato ai modelli feel-it e AlBERTo.
Cosa Imparerai
- Approcci classici vs BERT per sentiment analysis: VADER, lexicon-based, fine-tuned Transformers
- Dataset pubblici per sentiment: SST-2, IMDb, Amazon Reviews, SemEval
- Implementazione completa con HuggingFace Transformers e Trainer API
- Gestione di classi sbilanciate (class imbalance) in sentiment
- Metriche: accuracy, F1, precision, recall, AUC-ROC
- Sentiment a grana fine: aspetti (ABSA) e intensità
- Casi difficili: ironia, negazione, linguaggio ambiguo
- Pipeline di produzione con FastAPI e batch inference
- Ottimizzazione per la latenza: quantizzazione e ONNX export
1. Evoluzione degli Approcci: da VADER a BERT
Prima di immergerci nell'implementazione con Transformers, e utile capire il percorso storico degli approcci alla sentiment analysis, poichè in produzione spesso si usa il metodo più semplice che soddisfa i requisiti.
1.1 Approcci Basati su Dizionari: VADER
VADER (Valence Aware Dictionary and sEntiment Reasoner) e un lexicon-based analyzer ottimizzato per i social media. Non richiede training, e velocissimo e funziona sorprendentemente bene su testi informali.
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
analyzer = SentimentIntensityAnalyzer()
# Esempi base
texts = [
"This product is absolutely AMAZING!!!", # positivo forte
"The service was okay I guess", # neutro ambiguo
"Worst purchase I've ever made. Complete waste.", # negativo
"The food wasn't bad at all", # negazione tricky
"Yeah right, as if this would work :)", # sarcasmo
]
for text in texts:
scores = analyzer.polarity_scores(text)
print(f"Text: {text[:50]}")
print(f" neg={scores['neg']:.3f}, neu={scores['neu']:.3f}, "
f"pos={scores['pos']:.3f}, compound={scores['compound']:.3f}")
label = 'POSITIVE' if scores['compound'] >= 0.05 else \
'NEGATIVE' if scores['compound'] <= -0.05 else 'NEUTRAL'
print(f" Label: {label}\n")
# VADER gestisce bene: maiuscole, punteggiatura, emoji
# Non gestisce bene: sarcasmo, contesto complesso1.2 Approcci Machine Learning Classici
Prima dei Transformer, gli approcci più diffusi erano TF-IDF + Logistic Regression o SVM. Ancora oggi sono utili come baseline veloci o quando i dati sono pochissimi.
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report
# Dataset di esempio
train_texts = [
"Ottimo prodotto, lo consiglio a tutti",
"Pessima esperienza, non lo ricomprero",
"qualità eccellente, spedizione veloce",
"Totale spreco di soldi",
"Servizio clienti impeccabile",
"Prodotto difettoso, deluso"
]
train_labels = [1, 0, 1, 0, 1, 0]
# Pipeline TF-IDF + Logistic Regression
pipe = Pipeline([
('tfidf', TfidfVectorizer(
ngram_range=(1, 2), # unigrammi e bigrammi
max_features=50000,
sublinear_tf=True # log(1+tf) per attenuare frequenze alte
)),
('clf', LogisticRegression(C=1.0, max_iter=1000))
])
pipe.fit(train_texts, train_labels)
# Valutazione
test_texts = ["Prodotto fantastico!", "Pessimo, non funziona"]
preds = pipe.predict(test_texts)
probs = pipe.predict_proba(test_texts)
for text, pred, prob in zip(test_texts, preds, probs):
label = 'POSITIVO' if pred == 1 else 'NEGATIVO'
confidence = max(prob)
print(f"{text}: {label} ({confidence:.2f})")1.3 perchè BERT e Superiore
Confronto Approcci Sentiment Analysis
| Approccio | Accuracy (SST-2) | Latenza | Training Data | Casi Difficili |
|---|---|---|---|---|
| VADER | ~71% | <1ms | Nessuno | Scarso |
| TF-IDF + LR | ~85% | ~5ms | Necessario | Medio |
| DistilBERT | ~91% | ~50ms | Necessario | Buono |
| BERT-base | ~93% | ~100ms | Necessario | Ottimo |
| RoBERTa | ~96% | ~100ms | Necessario | Eccellente |
2. Dataset per Sentiment Analysis
La qualità del fine-tuning dipende fortemente dalla qualità e dalla dimensione del dataset. Ecco i più importanti per l'inglese, con indicazioni per l'italiano nel prossimo articolo.
from datasets import load_dataset
# SST-2: Stanford Sentiment Treebank (binario: positivo/negativo)
sst2 = load_dataset("glue", "sst2")
print(sst2)
# train: 67,349 esempi, validation: 872, test: 1,821
# IMDb Reviews (binario: positivo/negativo)
imdb = load_dataset("imdb")
print(imdb)
# train: 25,000, test: 25,000
# Amazon Reviews (1-5 stelle)
amazon = load_dataset("amazon_polarity")
print(amazon)
# train: 3,600,000, test: 400,000
# Esempio di esplorazione del dataset
print("\nSST-2 esempi:")
for i, example in enumerate(sst2['train'].select(range(3))):
label = 'POSITIVO' if example['label'] == 1 else 'NEGATIVO'
print(f" [{label}] {example['sentence']}")
# Analisi distribuzione classi
from collections import Counter
labels = sst2['train']['label']
print("\nDistribuzione SST-2 train:", Counter(labels))
# Counter({1: 37569, 0: 29780}) - leggero sbilanciamento3. Fine-tuning Completo con HuggingFace
Costruiamo un classificatore di sentiment completo, dalla preparazione dei dati al salvataggio del modello addestrato.
3.1 Preparazione dei Dati
from transformers import AutoTokenizer
from datasets import load_dataset, DatasetDict
import numpy as np
# Utilizziamo DistilBERT per velocità (97% di BERT, 60% più veloce)
MODEL_NAME = "distilbert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
# Carica SST-2 da GLUE
dataset = load_dataset("glue", "sst2")
def tokenize_function(examples):
return tokenizer(
examples["sentence"],
padding="max_length",
truncation=True,
max_length=128,
return_tensors=None # restituisce liste, non tensori
)
# Tokenizzazione del dataset completo (con cache)
tokenized = dataset.map(
tokenize_function,
batched=True,
batch_size=1000,
remove_columns=["sentence", "idx"] # rimuovi colonne non necessarie
)
# Formato PyTorch
tokenized.set_format("torch")
print(tokenized)
print("Colonne train:", tokenized['train'].column_names)
# ['input_ids', 'attention_mask', 'label']3.2 Definizione del Modello e Training
from transformers import (
AutoModelForSequenceClassification,
TrainingArguments,
Trainer
)
import evaluate
import numpy as np
# Modello con testa di classificazione
model = AutoModelForSequenceClassification.from_pretrained(
MODEL_NAME,
num_labels=2,
id2label={0: "NEGATIVE", 1: "POSITIVE"},
label2id={"NEGATIVE": 0, "POSITIVE": 1}
)
# Metriche di valutazione
accuracy = evaluate.load("accuracy")
f1 = evaluate.load("f1")
def compute_metrics(eval_pred):
logits, labels = eval_pred
predictions = np.argmax(logits, axis=-1)
return {
"accuracy": accuracy.compute(
predictions=predictions, references=labels)["accuracy"],
"f1": f1.compute(
predictions=predictions, references=labels,
average="binary")["f1"]
}
# Configurazione training
training_args = TrainingArguments(
output_dir="./results/distilbert-sst2",
num_train_epochs=3,
per_device_train_batch_size=32,
per_device_eval_batch_size=64,
warmup_ratio=0.1,
weight_decay=0.01,
learning_rate=2e-5,
lr_scheduler_type="linear",
evaluation_strategy="epoch",
save_strategy="epoch",
load_best_model_at_end=True,
metric_for_best_model="f1",
greater_is_better=True,
logging_dir="./logs",
logging_steps=100,
fp16=True, # Mixed precision (GPU con Tensor Cores)
dataloader_num_workers=4,
report_to="none", # Disabilita wandb/tensorboard per semplicità
)
trainer = Trainer(
model=model,
args=training_args,
train_dataset=tokenized["train"],
eval_dataset=tokenized["validation"],
compute_metrics=compute_metrics,
)
# Avvia training
train_result = trainer.train()
print(f"Training loss: {train_result.training_loss:.4f}")
# Valutazione finale
metrics = trainer.evaluate()
print(f"Validation accuracy: {metrics['eval_accuracy']:.4f}")
print(f"Validation F1: {metrics['eval_f1']:.4f}")
# Salva modello e tokenizer
trainer.save_model("./models/distilbert-sst2")
tokenizer.save_pretrained("./models/distilbert-sst2")3.3 Gestione delle Classi Sbilanciate
In molti dataset reali (es. recensioni di supporto clienti), le classi sono fortemente sbilanciate: 90% negativo, 10% positivo. Senza accorgimenti, il modello imparera a predire sempre la classe maggioritaria.
import torch
from torch import nn
from transformers import Trainer
# Soluzione 1: Weighted loss function
class WeightedTrainer(Trainer):
def __init__(self, class_weights, *args, **kwargs):
super().__init__(*args, **kwargs)
self.class_weights = torch.tensor(class_weights, dtype=torch.float)
def compute_loss(self, model, inputs, return_outputs=False):
labels = inputs.get("labels")
outputs = model(**inputs)
logits = outputs.get("logits")
# CrossEntropy con pesi inversamente proporzionali alla frequenza
loss_fct = nn.CrossEntropyLoss(
weight=self.class_weights.to(logits.device)
)
loss = loss_fct(logits.view(-1, self.model.config.num_labels),
labels.view(-1))
return (loss, outputs) if return_outputs else loss
# Calcola pesi dalle frequenze del dataset
from sklearn.utils.class_weight import compute_class_weight
import numpy as np
labels = tokenized['train']['label'].numpy()
weights = compute_class_weight(
class_weight='balanced',
classes=np.unique(labels),
y=labels
)
print("Class weights:", weights) # es. [2.3, 0.7] se negativo e raro
# Soluzione 2: Oversampling con imbalanced-learn
# pip install imbalanced-learn
from imblearn.over_sampling import RandomOverSampler
# (applicabile alle feature matrix, non direttamente ai tensor)
# Soluzione 3: Metriche appropriate per sbilanciamento
from sklearn.metrics import classification_report
# Usa F1 macro o F1 per la classe minoritaria, non solo accuracy4. Sentiment a Grana Fine: Aspect-Based (ABSA)
La sentiment analysis binaria (positivo/negativo) non cattura la complessità delle opinioni reali. Un cliente può essere soddisfatto del prodotto ma insoddisfatto della spedizione. L'Aspect-Based Sentiment Analysis (ABSA) identifica il sentiment per ogni aspetto menzionato.
from transformers import pipeline
# Zero-shot classification per ABSA
classifier = pipeline(
"zero-shot-classification",
model="facebook/bart-large-mnli"
)
review = "Il prodotto e eccellente ma la spedizione ha impiegato tre settimane. Il servizio clienti non ha risposto."
# Classificazione per ogni aspetto
aspects = ["prodotto", "spedizione", "servizio clienti"]
sentiments_per_aspect = {}
for aspect in aspects:
# Prompt per ogni aspetto
hypothesis = f"Il cliente e soddisfatto del {aspect}."
result = classifier(
review,
candidate_labels=["positivo", "negativo", "neutro"],
hypothesis_template=f"In questa recensione, il {{}} riguardo {aspect} e {{}}."
)
sentiments_per_aspect[aspect] = result['labels'][0]
print(f"{aspect}: {result['labels'][0]} ({result['scores'][0]:.2f})")
# Output atteso:
# prodotto: positivo (0.89)
# spedizione: negativo (0.92)
# servizio clienti: negativo (0.87)5. Casi Difficili: Ironia, Negazione, Ambiguità
I modelli BERT gestiscono molti casi difficili meglio dei metodi classici, ma non sono infallibili. Ecco come analizzare e mitigare i casi più comuni.
5.1 Gestione della Negazione
from transformers import pipeline
classifier = pipeline(
"text-classification",
model="distilbert-base-uncased-finetuned-sst-2-english"
)
# Test su casi di negazione
negation_examples = [
"This is not bad at all", # doppia negazione = positivo
"I wouldn't say it's terrible", # negazione attenuante
"Not the worst, but not great", # ambiguo
"Far from perfect", # negazione implicita
"Could have been worse", # comparativo negativo-positivo
]
for text in negation_examples:
result = classifier(text)[0]
print(f"'{text}'")
print(f" -> {result['label']} ({result['score']:.3f})\n")
# BERT gestisce bene "not bad" -> POSITIVE
# Ma può sbagliare con negazioni complesse e indirette5.2 Analisi degli Errori
import pandas as pd
from sklearn.metrics import confusion_matrix, classification_report
import matplotlib.pyplot as plt
import seaborn as sns
def analyze_errors(texts, true_labels, predicted_labels, probs):
"""Analisi dettagliata degli errori del modello."""
results = pd.DataFrame({
'text': texts,
'true_label': true_labels,
'pred_label': predicted_labels,
'confidence': [max(p) for p in probs],
'correct': [t == p for t, p in zip(true_labels, predicted_labels)]
})
# Falsi positivi: modello dice POSITIVO ma e NEGATIVO
fp = results[(results['true_label'] == 0) & (results['pred_label'] == 1)]
print(f"Falsi Positivi ({len(fp)}):")
for _, row in fp.head(5).iterrows():
print(f" Conf={row['confidence']:.2f}: {row['text'][:80]}")
# Falsi negativi: modello dice NEGATIVO ma e POSITIVO
fn = results[(results['true_label'] == 1) & (results['pred_label'] == 0)]
print(f"\nFalsi Negativi ({len(fn)}):")
for _, row in fn.head(5).iterrows():
print(f" Conf={row['confidence']:.2f}: {row['text'][:80]}")
# Confusion matrix
cm = confusion_matrix(true_labels, predicted_labels)
print(f"\nClassification Report:\n")
print(classification_report(true_labels, predicted_labels,
target_names=['NEGATIVO', 'POSITIVO']))
return results6. Messa in Produzione con FastAPI
Un modello di sentiment analysis ha valore solo se accessibile in produzione. Ecco come costruire un endpoint REST veloce e scalabile con FastAPI.
# sentiment_api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from transformers import pipeline
from typing import List
import time
app = FastAPI(title="Sentiment Analysis API", version="1.0")
# Carica il modello una sola volta all'avvio
MODEL_PATH = "./models/distilbert-sst2"
sentiment_pipeline = pipeline(
"text-classification",
model=MODEL_PATH,
device=-1, # -1 = CPU, 0 = prima GPU
batch_size=32, # batch inference per efficienza
truncation=True,
max_length=128
)
class SentimentRequest(BaseModel):
texts: List[str]
@validator('texts')
def validate_texts(cls, texts):
if not texts:
raise ValueError("Lista testi non può essere vuota")
if len(texts) > 100:
raise ValueError("Massimo 100 testi per richiesta")
for text in texts:
if len(text) > 5000:
raise ValueError("Testo troppo lungo (max 5000 caratteri)")
return texts
class SentimentResult(BaseModel):
text: str
label: str
score: float
processing_time_ms: float
@app.post("/predict", response_model=List[SentimentResult])
async def predict_sentiment(request: SentimentRequest):
start = time.time()
try:
results = sentiment_pipeline(request.texts)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
elapsed = (time.time() - start) * 1000
per_text = elapsed / len(request.texts)
return [
SentimentResult(
text=text,
label=r['label'],
score=r['score'],
processing_time_ms=per_text
)
for text, r in zip(request.texts, results)
]
@app.get("/health")
def health_check():
return {"status": "ok", "model": MODEL_PATH}
# Avvio: uvicorn sentiment_api:app --host 0.0.0.0 --port 80007. Ottimizzazione per la Latenza
In produzione, la latenza e spesso critica. Ecco le principali tecniche per ridurre il tempo di inferenza senza perdere troppa qualità.
7.1 Quantizzazione Dinamica
import torch
from transformers import AutoModelForSequenceClassification, AutoTokenizer
model = AutoModelForSequenceClassification.from_pretrained("./models/distilbert-sst2")
tokenizer = AutoTokenizer.from_pretrained("./models/distilbert-sst2")
# Quantizzazione dinamica (INT8): riduce dimensione e aumenta velocità su CPU
quantized_model = torch.quantization.quantize_dynamic(
model,
{torch.nn.Linear}, # quantizza solo i layer Linear
dtype=torch.qint8
)
# Confronto dimensioni
import os
def model_size(m):
torch.save(m.state_dict(), "tmp.pt")
size = os.path.getsize("tmp.pt") / (1024 * 1024)
os.remove("tmp.pt")
return size
print(f"Modello originale: {model_size(model):.1f} MB")
print(f"Modello quantizzato: {model_size(quantized_model):.1f} MB")
# Originale: ~250 MB, Quantizzato: ~65 MB
# Benchmark velocità
import time
def benchmark(m, tokenizer, texts, n_runs=50):
inputs = tokenizer(texts, return_tensors='pt',
padding=True, truncation=True, max_length=128)
with torch.no_grad():
# Warm-up
for _ in range(5):
_ = m(**inputs)
# Benchmark
start = time.time()
for _ in range(n_runs):
_ = m(**inputs)
elapsed = (time.time() - start) / n_runs * 1000
return elapsed
texts = ["This product is amazing!"] * 8 # batch di 8
t_orig = benchmark(model, tokenizer, texts)
t_quant = benchmark(quantized_model, tokenizer, texts)
print(f"Originale: {t_orig:.1f}ms, Quantizzato: {t_quant:.1f}ms")
print(f"Speedup: {t_orig/t_quant:.2f}x")7.2 Export ONNX per Deployment
from optimum.onnxruntime import ORTModelForSequenceClassification
from transformers import AutoTokenizer
import numpy as np
import time
# Converti in ONNX con HuggingFace Optimum
# pip install optimum[onnxruntime]
model_onnx = ORTModelForSequenceClassification.from_pretrained(
"./models/distilbert-sst2",
export=True, # esporta in ONNX al primo caricamento
provider="CPUExecutionProvider"
)
tokenizer = AutoTokenizer.from_pretrained("./models/distilbert-sst2")
# Inferenza con ONNX Runtime
text = "This product exceeded all my expectations!"
inputs = tokenizer(text, return_tensors='pt', truncation=True, max_length=128)
start = time.time()
outputs = model_onnx(**inputs)
latency = (time.time() - start) * 1000
import torch
probs = torch.softmax(outputs.logits, dim=-1)
label = model_onnx.config.id2label[probs.argmax().item()]
confidence = probs.max().item()
print(f"Label: {label}")
print(f"Confidence: {confidence:.3f}")
print(f"Latency: {latency:.1f}ms")
# ONNX e tipicamente 2-4x più veloce della versione PyTorch su CPU8. Valutazione Completa e Reporting
from sklearn.metrics import (
classification_report,
roc_auc_score,
average_precision_score,
confusion_matrix
)
import numpy as np
def evaluate_sentiment_model(model, tokenizer, test_texts, test_labels,
batch_size=64):
"""Valutazione completa del modello di sentiment."""
all_probs = []
all_preds = []
for i in range(0, len(test_texts), batch_size):
batch = test_texts[i:i+batch_size]
inputs = tokenizer(
batch, return_tensors='pt', padding=True,
truncation=True, max_length=128
)
with torch.no_grad():
outputs = model(**inputs)
probs = torch.softmax(outputs.logits, dim=-1).numpy()
preds = np.argmax(probs, axis=1)
all_probs.extend(probs[:, 1]) # probabilità classe positiva
all_preds.extend(preds)
# Report principale
print("=== Classification Report ===")
print(classification_report(
test_labels, all_preds,
target_names=['NEGATIVE', 'POSITIVE'],
digits=4
))
# Metriche aggiuntive
auc = roc_auc_score(test_labels, all_probs)
ap = average_precision_score(test_labels, all_probs)
print(f"AUC-ROC: {auc:.4f}")
print(f"Average Precision: {ap:.4f}")
# Analisi errori per fascia di confidenza
all_probs = np.array(all_probs)
all_preds = np.array(all_preds)
test_labels = np.array(test_labels)
for threshold in [0.5, 0.7, 0.9]:
high_conf = all_probs >= threshold
if high_conf.sum() > 0:
acc_high = (all_preds[high_conf] == test_labels[high_conf]).mean()
print(f"Accuracy (conf >= {threshold}): {acc_high:.4f} "
f"({high_conf.sum()} esempi)")
return np.array(all_probs), np.array(all_preds)9. Ottimizzazione per la Produzione: ONNX e Quantizzazione
I modelli BERT richiedono risorse computazionali significative. Per applicazioni a bassa latenza o su hardware limitato, existono diverse strategie di ottimizzazione che riducono drasticamente i tempi di inferenza senza perdere qualità.
Confronto Strategie di Ottimizzazione
| Strategia | Riduzione Latenza | Riduzione Modello | Perdita qualità | Complessità |
|---|---|---|---|---|
| ONNX Export | 2-4x | ~10% | <0.1% | Bassa |
| Dynamic Quantization (INT8) | 2-3x | 75% | 0.5-1% | Bassa |
| Static Quantization (INT8) | 3-5x | 75% | 0.3-0.8% | Media |
| DistilBERT (KD) | 2x | 40% | 3% | Media |
| TorchScript | 1.5-2x | Nessuna | <0.1% | Bassa |
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from optimum.onnxruntime import ORTModelForSequenceClassification
import torch
import time
# ---- Esportazione ONNX con Optimum ----
model_path = "./models/distilbert-sentiment"
# Esportazione e ottimizzazione ONNX in un comando
ort_model = ORTModelForSequenceClassification.from_pretrained(
model_path,
export=True, # Esporta automaticamente in ONNX
provider="CPUExecutionProvider"
)
tokenizer = AutoTokenizer.from_pretrained(model_path)
# Salva modello ONNX
ort_model.save_pretrained("./models/distilbert-sentiment-onnx")
# ---- Benchmark: PyTorch vs ONNX ----
def benchmark_model(predict_fn, texts, n_runs=100):
"""Misura latenza media su n_runs inferenze."""
# Warmup
for _ in range(10):
predict_fn(texts[0])
times = []
for text in texts[:n_runs]:
start = time.perf_counter()
predict_fn(text)
times.append((time.perf_counter() - start) * 1000)
import numpy as np
return {
"mean_ms": round(np.mean(times), 2),
"p50_ms": round(np.percentile(times, 50), 2),
"p95_ms": round(np.percentile(times, 95), 2),
"p99_ms": round(np.percentile(times, 99), 2),
}
# Carica modello PyTorch originale per confronto
pt_model = AutoModelForSequenceClassification.from_pretrained(model_path)
pt_model.eval()
def pt_predict(text):
inputs = tokenizer(text, return_tensors='pt', truncation=True, max_length=128)
with torch.no_grad():
return pt_model(**inputs).logits
def onnx_predict(text):
inputs = tokenizer(text, return_tensors='pt', truncation=True, max_length=128)
return ort_model(**inputs).logits
test_texts = ["Prodotto ottimo, lo consiglio!"] * 100
pt_stats = benchmark_model(pt_predict, test_texts)
onnx_stats = benchmark_model(onnx_predict, test_texts)
print("PyTorch: ", pt_stats)
print("ONNX: ", onnx_stats)
print(f"Speedup: {pt_stats['p95_ms'] / onnx_stats['p95_ms']:.1f}x")# Quantizzazione dinamica con PyTorch (nessun dato di calibrazione)
import torch
def quantize_bert_dynamic(model_path: str, output_path: str):
"""Quantizzazione INT8 dinamica per CPU inference."""
from transformers import AutoModelForSequenceClassification
model = AutoModelForSequenceClassification.from_pretrained(model_path)
model.eval()
# Quantizza solo i layer Linear (nn.Linear) dinamicamente
quantized = torch.quantization.quantize_dynamic(
model,
{torch.nn.Linear},
dtype=torch.qint8
)
# Salva il modello quantizzato
torch.save(quantized.state_dict(), f"{output_path}/quantized_model.pt")
# Confronto dimensioni
import os
original_size = sum(
os.path.getsize(f"{model_path}/{f}")
for f in os.listdir(model_path) if f.endswith('.bin')
) / 1024 / 1024
print(f"Modello originale: ~{original_size:.0f} MB")
print(f"Riduzione stimata: ~75% → ~{original_size * 0.25:.0f} MB")
return quantized
# Esempio di utilizzo
# quantized_model = quantize_bert_dynamic(
# "./models/distilbert-sentiment",
# "./models/quantized"
# )10. Best Practices per la Produzione
Anti-Pattern: Non Usare il Modello Grezzo senza Validazione
Un modello addestrato su SST-2 (recensioni cinematografiche) può performare male su ticket di supporto tecnico o post di social media. Valida sempre il modello sul tuo specifico dominio prima del deploy.
Checklist per il Deploy in Produzione
- Valutare il modello su dati del dominio target (non solo benchmark pubblici)
- Impostare soglie di confidenza: restituire "incerto" sotto soglia (es. 0.6)
- Monitorare la distribuzione del confidence score nel tempo
- Implementare un meccanismo di feedback per raccogliere label errate
- Versionare sia il modello che il tokenizer insieme
- Testare il comportamento su input anomali (testo vuoto, caratteri speciali, lunghezze estreme)
- Implementare rate limiting e timeout per l'API
- Loggare tutte le predizioni per analisi post-hoc
class ProductionSentimentClassifier:
"""Classificatore di sentiment pronto per la produzione."""
def __init__(self, model_path: str, confidence_threshold: float = 0.7):
self.pipeline = pipeline(
"text-classification",
model=model_path,
truncation=True,
max_length=128
)
self.threshold = confidence_threshold
def predict(self, text: str) -> dict:
# Validazione input
if not text or not text.strip():
return {"label": "UNKNOWN", "score": 0.0, "reason": "empty_input"}
text = text.strip()[:5000] # Trunca testi troppo lunghi
result = self.pipeline(text)[0]
# Gestione incertezza
if result['score'] < self.threshold:
return {
"label": "UNCERTAIN",
"score": result['score'],
"raw_label": result['label'],
"reason": "below_confidence_threshold"
}
return {
"label": result['label'],
"score": result['score'],
"reason": "ok"
}
def predict_batch(self, texts: list) -> list:
# Filtra testi vuoti mantenendo la posizione
valid_texts = [t.strip()[:5000] if t and t.strip() else "" for t in texts]
results = self.pipeline(valid_texts)
return [
self.predict(t) if t else {"label": "UNKNOWN", "score": 0.0}
for t in valid_texts
]Conclusioni e Prossimi Passi
In questo articolo abbiamo coperto l'intero lifecycle di un sistema di sentiment analysis: dagli approcci classici (VADER, TF-IDF) al fine-tuning di modelli Transformer, dalla gestione dei dati sbilanciati alla messa in produzione con FastAPI e ottimizzazione della latenza.
Punti Chiave
- Scegli l'approccio in base ai requisiti: VADER per velocità, BERT per qualità
- Valuta sempre sul tuo dominio specifico, non solo su benchmark
- Gestisci le classi sbilanciate con weighted loss o oversampling
- Usa soglie di confidenza in produzione invece di predizioni forzate
- DistilBERT offre un ottimo compromesso velocità/qualità per la produzione
- Monitora le predizioni nel tempo per rilevare il data drift
Continua la Serie
- Prossimo: NLP per l'Italiano — feel-it, AlBERTo e le sfide specifiche dell'italiano
- Articolo 5: Named Entity Recognition — estrarre entità dal testo
- Articolo 6: Text Classification Multi-label — quando un testo appartiene a più categorie
- Articolo 7: HuggingFace Transformers: Guida Completa — Trainer API, Datasets, Hub
- Articolo 10: Monitoring NLP in Produzione — drift detection e retraining automatico