Analiza nastrojów za pomocą transformatorów: techniki i wdrożenie
La Analiza sentymentów — lub analiza nastrojów — i najbardziej pożądane zadanie NLP w środowisko korporacyjne. Każdego dnia miliony firm analizują recenzje produktów, posty na temat media społecznościowe, zgłoszenia do pomocy technicznej i opinie klientów, aby zrozumieć, co ludzie naprawdę myślą. Wraz z pojawieniem się modeli BERT i Transformer poprawiła się jakość tych systemów radykalnie w porównaniu z klasycznym podejściem opartym na słownikach lub TF-IDF.
W tym artykule zbudujemy kompletny system analizy nastrojów: od zrozumienia zbiorów danych do produkcji, przechodząc przez dostrajanie za pomocą HuggingFace, zarządzanie klasami niezrównoważonymi, ocena metryk i strategii radzenie sobie z przypadkami granicznymi, takimi jak ironia, zaprzeczenie i dwuznaczny język.
To już trzeci artykuł z tej serii Nowoczesne NLP: od BERT do LLM. Zakłada znajomość podstaw BERT (art. 2). Specjalnie dla Włochów zobacz artykuł 4 poświęcony modelom Feel-it i AlBERTo.
Czego się nauczysz
- Klasyczne podejście vs BERT do analizy nastrojów: VADER, oparte na leksykonie, dopracowane Transformers
- Zbiory danych nastrojów społecznych: SST-2, IMDb, Amazon Reviews, SemEval
- Kompletna implementacja z HuggingFace Transformers i Trainer API
- Zarządzanie brakiem równowagi klasowej w uczuciach
- Metryki: dokładność, F1, precyzja, przypominanie, AUC-ROC
- Drobnoziarniste odczucia: aspekty (ABSA) i intensywność
- Trudne przypadki: ironia, zaprzeczenie, dwuznaczny język
- Rurociąg produkcyjny z FastAPI i wnioskowaniem wsadowym
- Optymalizacja opóźnienia: kwantyzacja i eksport ONNX
1. Ewolucja podejść: od VADER do BERT
Zanim zagłębimy się w implementację Transformers, warto zrozumieć tę podróż historia podejść do analizy nastrojów, ponieważ jest ona często stosowana w produkcji najprostsza metoda spełniająca wymagania.
1.1 Podejścia oparte na słownikach: VADER
VADER (słownik świadomy wartościowości i rozumujący sentymenty) i oparte na leksykonie analizator zoptymalizowany pod kątem mediów społecznościowych. Nie wymaga żadnego szkolenia i jest bardzo szybki i zaskakująco dobrze sprawdza się w tekstach nieformalnych.
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 Klasyczne podejścia do uczenia maszynowego
Przed Transformersami najpopularniejszymi podejściami były TF-IDF + regresja logistyczna lub SVM. Nawet dzisiaj są one przydatne jako szybkie punkty odniesienia lub gdy jest bardzo mało danych.
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, ponieważ BERT jest lepszy
Porównanie podejść do analizy nastrojów
| Zbliżać się | Dokładność (SST-2) | Utajenie | Dane szkoleniowe | Trudne przypadki |
|---|---|---|---|---|
| VADER | ~71% | <1 ms | Nikt | Rzadki |
| TF-IDF + LR | ~85% | ~5ms | Niezbędny | Średni |
| DestylBERT | ~91% | ~50ms | Niezbędny | Dobry |
| BERT-podstawowy | ~93% | ~100 ms | Niezbędny | Optymalny |
| Roberta | ~96% | ~100 ms | Niezbędny | Doskonały |
2. Zbiór danych do analizy nastrojów
Jakość dostrajania silnie zależy od jakości i rozmiaru zbioru danych. Oto najważniejsze z nich dla języka angielskiego, ze wskazaniem dla języka włoskiego w następnym artykule.
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. Dokończ dostrajanie za pomocą HuggingFace
Budujemy kompletny klasyfikator nastrojów już od przygotowania danych podczas zapisywania wyszkolonego modelu.
3.1 Przygotowanie danych
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 Definicja modelu i szkolenie
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 Zarządzanie klasami niezrównoważonymi
W wielu rzeczywistych zbiorach danych (np. recenzjach obsługi klienta) klasy są silnie zaznaczone niezrównoważone: 90% negatywne, 10% pozytywne. Bez środków ostrożności model się nauczy aby zawsze przewidzieć klasę większości.
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. Drobnoziarnisty sentyment: oparty na aspektach (ABSA)
Binarna (pozytywna/negatywna) analiza nastrojów nie uwzględnia złożoności prawdziwe opinie. Klient może być zadowolony z produktu, ale niezadowolony z przesyłki. THE'Analiza nastrojów oparta na aspektach (ABSA) identyfikuje nastroje w odniesieniu do każdego wymienionego aspektu.
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. Trudne przypadki: ironia, zaprzeczenie, dwuznaczność
Modele BERT radzą sobie z wieloma trudnymi przypadkami lepiej niż metody klasyczne, ale nie są nieomylni. Oto, jak analizować i łagodzić najczęstsze przypadki.
5.1 Zarządzanie odmową
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 błędów
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. Uruchomienie z FastAPI
Model analizy nastrojów jest wartościowy tylko wtedy, gdy jest dostępny w środowisku produkcyjnym. Oto jak zbudować szybki i skalowalny punkt końcowy REST za pomocą 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. Optymalizacja pod kątem opóźnień
W środowisku produkcyjnym opóźnienie jest często krytyczne. Oto główne techniki aby skrócić czas wnioskowania bez utraty zbyt dużej jakości.
7.1 Kwantyzacja dynamiczna
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 Eksportuj ONNX do wdrożenia
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. Pełna ocena i raportowanie
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. Optymalizacja produkcji: ONNX i kwantyzacja
Modele BERT wymagają znacznych zasobów obliczeniowych. Do zastosowań przy małych opóźnieniach lub na ograniczonym sprzęcie istnieją różne strategie optymalizacji które drastycznie skracają czas wnioskowania bez utraty jakości.
Porównanie strategii optymalizacyjnych
| Strategia | Redukcja opóźnień | Redukcja modelu | Utrata jakości | Złożoność |
|---|---|---|---|---|
| Eksport ONNX | 2-4x | ~10% | <0,1% | Niski |
| Kwantyzacja dynamiczna (INT8) | 2-3x | 75% | 0,5-1% | Niski |
| Kwantyzacja statyczna (INT8) | 3-5x | 75% | 0,3-0,8% | Przeciętny |
| DestylBERT (KD) | 2x | 40% | 3% | Przeciętny |
| TorchScript | 1,5-2x | Nic | <0,1% | Niski |
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. Najlepsze praktyki produkcyjne
Anty-wzorzec: nie używaj surowego wzorca bez sprawdzenia poprawności
Model przeszkolony na SST-2 (recenzje filmów) może działać słabo w zgłoszeniach do pomocy technicznej lub w postach w mediach społecznościowych. Zawsze sprawdzaj model w Twojej konkretnej domenie przed wdrożeniem.
Lista kontrolna dotycząca wdrożenia w środowisku produkcyjnym
- Oceń model na danych domeny docelowej (nie tylko publicznych testach porównawczych)
- Ustaw progi ufności: zwróć „niepewny” poniżej progu (np. 0,6)
- Monitoruj rozkład wyniku zaufania w czasie
- Wdrożyj mechanizm informacji zwrotnej, aby zbierać nieprawidłowe etykiety
- Wersja zarówno modelu, jak i tokenizera razem
- Zachowanie testowe przy nietypowych danych wejściowych (pusty tekst, znaki specjalne, ekstremalne długości)
- Zaimplementuj ograniczenia szybkości i limity czasu dla interfejsu API
- Rejestruj wszystkie prognozy do analizy 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
]Wnioski i dalsze kroki
W tym artykule omówiliśmy cały cykl życia systemu analizy nastrojów: od podejść klasycznych (VADER, TF-IDF) po dostrajanie modeli Transformatorów, od zarządzania niezrównoważonymi danymi po wprowadzenie ich do produkcji za pomocą FastAPI i optymalizacja opóźnień.
Kluczowe punkty
- Wybierz podejście w oparciu o swoje wymagania: VADER dla szybkości, BERT dla jakości
- Zawsze oceniaj Twoja domena konkretnych, a nie tylko na podstawie benchmarków
- Obsługuj niezrównoważone klasy ze stratą ważoną lub nadpróbkowaniem
- Używaj progów ufności w środowisku produkcyjnym zamiast wymuszonych przewidywań
- DistilBERT oferuje doskonały kompromis w zakresie szybkości i jakości produkcji
- Monitoruj prognozy w czasie, aby wykryć dryf danych
Seria trwa
- Następny: NLP dla języka włoskiego — Feel-it, AlBERTo i specyficzne wyzwania języka włoskiego
- Artykuł 5: Rozpoznawanie nazwanych podmiotów — wyodrębnij elementy z tekstu
- Artykuł 6: Klasyfikacja tekstu z wieloma etykietami — gdy tekst należy do wielu kategorii
- Artykuł 7: Transformatory HuggingFace: kompletny przewodnik — Trener API, zbiory danych, Hub
- Artykuł 10: Monitorowanie NLP w produkcji — wykrywanie dryfu i automatyczne przekwalifikowanie