Analiza sentimentelor cu transformatoare: tehnici și implementare
La Analiza sentimentelor — sau analiza sentimentelor — și cea mai solicitată sarcină NLP în mediul corporativ. În fiecare zi, milioane de companii analizează recenzii despre produse, postări despre social media, bilete de asistență și feedback-ul clienților pentru a înțelege ce cred oamenii cu adevărat. Odată cu apariția modelelor BERT și Transformer, calitatea acestor sisteme s-a îmbunătățit radical în comparație cu abordările clasice bazate pe dicționar sau TF-IDF.
În acest articol vom construi un sistem complet de analiză a sentimentelor: de la înțelegere a seturilor de date în producție, trecând prin reglajul fin cu HuggingFace, managementul claselor dezechilibrate, evaluarea metricilor și strategiilor pentru a gestiona cazuri limită, cum ar fi ironia, negarea și limbajul ambiguu.
Acesta este al treilea articol din serie NLP modern: de la BERT la LLM. Presupune familiaritatea cu elementele fundamentale BERT (Articolul 2). În special pentru italiană, vezi articolul 4 dedicat modelelor feel-it si ALBERTo.
Ce vei învăța
- Abordări clasice vs BERT pentru analiza sentimentelor: Transformers VADER, bazate pe lexic, reglate fin
- Seturi de date privind sentimentul public: SST-2, IMDb, Amazon Reviews, SemEval
- Implementare completă cu HuggingFace Transformers și Trainer API
- Managementul dezechilibrului de clasă în sentiment
- Metrici: acuratețe, F1, precizie, rechemare, AUC-ROC
- Sentiment cu granulație fină: aspecte (ABSA) și intensitate
- Cazuri dificile: ironie, negare, limbaj ambiguu
- Conducta de producție cu FastAPI și inferență pe lot
- Optimizare pentru latență: cuantizare și export ONNX
1. Evoluția abordărilor: de la VADER la BERT
Înainte de a ne aprofunda în implementarea Transformers, este util să înțelegem călătoria istoria abordărilor analizei sentimentelor, așa cum este adesea folosită în producție cea mai simplă metodă care îndeplinește cerințele.
1.1 Abordări bazate pe dicționar: VADER
VADER (Valence Aware Dictionary and Sentiment Reasoner) și bazate pe lexic analizator optimizat pentru rețelele sociale. Nu necesită antrenament și este foarte rapid și funcționează surprinzător de bine pe textele informale.
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 Abordări clasice de învățare automată
Înainte de Transformers, cele mai populare abordări erau TF-IDF + Regresia logistică sau SVM. Chiar și astăzi sunt utile ca linii de bază rapide sau când există foarte puține date.
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 pentru că BERT este superior
Comparația abordărilor de analiză a sentimentelor
| Abordare | Precizie (SST-2) | Latența | Date de antrenament | Cazuri dificile |
|---|---|---|---|---|
| VADER | ~71% | <1 ms | Nimeni | Rar |
| TF-IDF + LR | ~85% | ~5 ms | Necesar | Mediu |
| DistilBERT | ~91% | ~50 ms | Necesar | Bun |
| BERT-de bază | ~93% | ~100 ms | Necesar | Optimal |
| Roberta | ~96% | ~100 ms | Necesar | Excelent |
2. Setul de date pentru analiza sentimentelor
Calitatea reglajului fin depinde în mare măsură de calitatea și dimensiunea setului de date. Iată cele mai importante pentru engleză, cu indicații pentru italiană în articolul următor.
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. Reglarea fină completă cu HuggingFace
Construim un clasificator complet de sentimente, de la pregătirea datelor la salvarea modelului antrenat.
3.1 Pregătirea datelor
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 Definirea modelului și instruirea
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 Managementul claselor dezechilibrate
În multe seturi de date reale (de exemplu, recenzii de asistență pentru clienți), cursurile sunt puternice dezechilibrat: 90% negativ, 10% pozitiv. Fără precauții, modelul va învăța să prezică întotdeauna clasa majoritară.
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 cu granulație fină: bazat pe aspect (ABSA)
Analiza binară (pozitivă/negativă) a sentimentelor nu surprinde complexitatea pareri reale. Un client poate fi mulțumit de produs dar nemulțumit de transport. THE'Analiza sentimentelor bazate pe aspecte (ABSA) identifică sentimentul pentru fiecare aspect menționat.
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. Cazuri dificile: ironie, negare, ambiguitate
Modelele BERT gestionează multe cazuri dificile mai bine decât metodele clasice, dar nu sunt infailibile. Iată cum să analizați și să atenuați cele mai frecvente cazuri.
5.1 Managementul negării
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 Analiza erorilor
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. Punerea în funcțiune cu FastAPI
Un model de analiză a sentimentelor este valoros doar dacă este accesibil în producție. Iată cum să construiți un punct final REST rapid și scalabil cu 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. Optimizare pentru Latență
În producție, latența este adesea critică. Iată principalele tehnici pentru a reduce timpul de inferență fără a pierde prea multă calitate.
7.1 Cuantificare dinamică
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 Exportați ONNX pentru implementare
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. Evaluare și raportare completă
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. Optimizare pentru producție: ONNX și cuantizare
Modelele BERT necesită resurse de calcul semnificative. Pentru aplicații la latență scăzută sau pe hardware limitat, există diferite strategii de optimizare care reduc drastic timpii de inferență fără a pierde calitatea.
Compararea strategiilor de optimizare
| Strategie | Reducerea latenței | Reducerea modelului | Pierderea calității | Complexitate |
|---|---|---|---|---|
| Export ONNX | 2-4x | ~10% | <0,1% | Scăzut |
| Cuantizare dinamică (INT8) | 2-3x | 75% | 0,5-1% | Scăzut |
| Cuantizare statică (INT8) | 3-5x | 75% | 0,3-0,8% | Medie |
| DistilBERT (KD) | 2x | 40% | 3% | Medie |
| TorchScript | 1,5-2x | Nici unul | <0,1% | Scăzut |
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. Cele mai bune practici pentru producție
Anti-pattern: nu utilizați modelul brut fără validare
Un model instruit pe SST-2 (recenzii de filme) poate avea rezultate slabe pe bilete de asistență tehnică sau postări pe rețelele sociale. Validați întotdeauna modelul pe domeniul dvs. specific înainte de implementare.
Lista de verificare pentru implementare în producție
- Evaluați modelul pe datele domeniului țintă (nu doar referințe publice)
- Setați praguri de încredere: returnați „incert” sub prag (de exemplu, 0,6)
- Monitorizați distribuția scorului de încredere în timp
- Implementați un mecanism de feedback pentru a colecta etichete incorecte
- Versiune atât modelul, cât și tokenizer-ul împreună
- Testați comportamentul la introducerea anormală (text gol, caractere speciale, lungimi extreme)
- Implementați limitarea ratei și timeout-urile pentru API
- Înregistrați toate predicțiile pentru analiză 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
]Concluzii și pașii următori
În acest articol am acoperit întregul ciclu de viață al unui sistem de analiză a sentimentelor: de la abordări clasice (VADER, TF-IDF) până la reglarea fină a modelelor Transformer, de la gestionarea datelor dezechilibrate până la punerea lor în producție cu FastAPI și optimizarea latenței.
Puncte cheie
- Alegeți abordarea în funcție de cerințele dvs.: VADER pentru viteză, BERT pentru calitate
- Evaluează întotdeauna domeniul dvs specifice, nu doar pe benchmark-uri
- Gestionați clasele dezechilibrate cu pierderi ponderate sau supraeșantionare
- Utilizați praguri de încredere în producție în loc de predicții forțate
- DistilBERT oferă un compromis excelent viteză/calitate pentru producție
- Monitorizați previziunile în timp pentru a detecta deriva de date
Seria continuă
- Următorul: NLP pentru italiană — feel-it, Alberto și provocările specifice italianului
- Articolul 5: Recunoașterea entității numite — extrage entități din text
- Articolul 6: Clasificarea textului cu mai multe etichete — când un text aparține mai multor categorii
- Articolul 7: HuggingFace Transformers: Ghid complet — Trainer API, seturi de date, hub
- Articolul 10: Monitorizare NLP în producție — detectarea derivei și recalificarea automată