Üretimde NLP İzleme: Sapma Tespiti ve Yeniden Eğitim
Dağıtımı mükemmel olan bir NLP modeli hızla geçerliliğini yitirebilir. Dil gelişir, kalıplar değişir, gerçek veriler eğitim setinden sapar. Bu fenomene denir veri kayması ve eğer izlenmezse, Genellikle çok geç fark edilen sessiz performans düşüşü — müşteriler şikayet ettiğinde veya şirketin KPI'ları düştüğünde.
Bu yazıda NLP modelleri için eksiksiz bir izleme sistemi oluşturacağız. üretim: gerçek zamanlı tahmin takibinden otomatik tespite kadar MLflow ve Airflow ile uyarıdan otomatik yeniden eğitime kadar sürüklenme. Bu makale seriyi kapatıyor Modern NLP: BERT'ten Yüksek Lisans'a operasyonlara ileri derecede odaklanılmıştır.
Ne Öğreneceksiniz
- Kayma türleri: veri kayması, kavram kayması, etiket kayması, özellik kayması
- Üretimdeki NLP modelleri için izlenecek ölçümler
- Metin kayması tespiti: Nüfus İstikrar Endeksi (PSI) ile sürüklenmenin yerleştirilmesi
- Tahminlerin kalitesini etiketler olmadan izleme (proxy metrikleri)
- Eşikler ve bildirimler içeren uyarı sistemi
- NLP tahminlerinin yapılandırılmış günlüğe kaydedilmesi
- Drift tabanlı tetikleyicilerle otomatik yeniden eğitim hattı
- Yeni şablon sürümleri için A/B testi
- Grafana ve Prometheus ile kontrol paneli
- Yeni modelleri hiçbir etki yaratmadan doğrulamak için gölge dağıtımı
1. NLP Modellerinde Kayma Türleri
NLP modelindeki "sürüklenme", her birinin nedenleri olan farklı şekillerde kendini gösterebilir. ve farklı çözümler.
Drift Taksonomisi
| Tip | Tanım | NLP örneği | Çözüm |
|---|---|---|---|
| Tarih Kayması | Giriş dağıtımı değişiklikleri | Twitter'da yeni argo | Yeni verilerle yeniden eğitin |
| Konsept Drift | Girdi-çıktı ilişkisi değişiklikleri | “Trump” = politika ve kişi | Sık sık yeniden eğitim |
| Etiket Kayması | Çıktı dağılımı değişiklikleri | Krizde daha olumsuz tahminler | Çıkış dağıtımı izleme |
| Özellik Kayması | Özellik istatistikleri değişikliği | Ortalama metin uzunluğu artar | Özellik izleme + uyarı |
2. Tahmin Kayıt Sistemi
Herhangi bir izleme sisteminin temeli, her tahminin yapılandırılmış olarak günlüğe kaydedilmesidir. Modelin davranışını analiz etmek için yeterli bilgiyi toplamamız gerekiyor zamanla.
import json
import time
import hashlib
import logging
from dataclasses import dataclass, asdict, field
from typing import Optional, List, Dict, Any
from datetime import datetime
import uuid
@dataclass
class NLPPredictionLog:
"""Schema di logging per predizioni NLP."""
prediction_id: str
timestamp: str
model_version: str
input_text: str
input_hash: str # hash del testo (non il testo per privacy)
input_length_chars: int
input_length_tokens: int
predicted_label: str
predicted_label_id: int
confidence_score: float
all_class_scores: Dict[str, float]
inference_latency_ms: float
true_label: Optional[str] = None # None se non disponibile
feedback: Optional[str] = None # feedback utente se disponibile
metadata: Dict[str, Any] = field(default_factory=dict)
class NLPPredictionLogger:
"""Logger strutturato per predizioni NLP."""
def __init__(self, model_version: str, log_path: str = "./prediction_logs"):
self.model_version = model_version
self.log_path = log_path
self.logger = logging.getLogger("nlp_predictions")
# Handler per file JSON Lines (JSONL)
handler = logging.FileHandler(f"{log_path}/predictions.jsonl")
handler.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
def log_prediction(self,
text: str,
predicted_label: str,
predicted_label_id: int,
confidence: float,
all_scores: Dict[str, float],
latency_ms: float,
num_tokens: int,
true_label: Optional[str] = None,
metadata: Optional[dict] = None) -> str:
"""Logga una singola predizione. Restituisce prediction_id."""
# Hash dell'input (non salvare il testo originale per GDPR)
input_hash = hashlib.sha256(text.encode()).hexdigest()[:16]
prediction_id = str(uuid.uuid4())
log_entry = NLPPredictionLog(
prediction_id=prediction_id,
timestamp=datetime.utcnow().isoformat(),
model_version=self.model_version,
input_text=text[:500], # troncato per storage
input_hash=input_hash,
input_length_chars=len(text),
input_length_tokens=num_tokens,
predicted_label=predicted_label,
predicted_label_id=predicted_label_id,
confidence_score=confidence,
all_class_scores=all_scores,
inference_latency_ms=latency_ms,
true_label=true_label,
metadata=metadata or {}
)
self.logger.info(json.dumps(asdict(log_entry)))
return prediction_id
# Uso nella pipeline di inferenza
class MonitoredSentimentClassifier:
def __init__(self, model_path: str, model_version: str):
from transformers import pipeline, AutoTokenizer
self.pipeline = pipeline("text-classification", model=model_path)
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.logger = NLPPredictionLogger(model_version)
self.model_version = model_version
def predict(self, text: str, metadata: dict = None) -> dict:
start = time.time()
# Inferenza
result = self.pipeline(text)[0]
# Calcola numero di token
num_tokens = len(self.tokenizer.tokenize(text)[:128])
latency_ms = (time.time() - start) * 1000
# Log
pred_id = self.logger.log_prediction(
text=text,
predicted_label=result['label'],
predicted_label_id=0 if result['label'] == 'NEGATIVE' else 1,
confidence=result['score'],
all_scores={result['label']: result['score']},
latency_ms=latency_ms,
num_tokens=num_tokens,
metadata=metadata or {}
)
return {
"prediction_id": pred_id,
"label": result['label'],
"confidence": result['score'],
"latency_ms": latency_ms
}3. Sürüklenme Algılama: Metin Gömme Yaklaşımı
Metindeki veri kaymasını tespit etmek ve verileri karşılaştırmak için en sağlam yöntem dağıtımı cümle yerleştirme eğitim setinde üretimde olanlarla.
import numpy as np
from sentence_transformers import SentenceTransformer
from scipy.stats import ks_2samp
from scipy.spatial.distance import jensenshannon
import warnings
class EmbeddingDriftDetector:
"""
Rileva data drift confrontando la distribuzione degli embedding.
Usa il test di Kolmogorov-Smirnov (KS) per ogni dimensione dell'embedding.
"""
def __init__(self, embedding_model: str = 'all-MiniLM-L6-v2',
ks_threshold: float = 0.1,
psi_threshold: float = 0.2):
self.model = SentenceTransformer(embedding_model)
self.ks_threshold = ks_threshold # soglia test KS
self.psi_threshold = psi_threshold # soglia PSI
self.reference_embeddings = None
self.reference_stats = None
def fit(self, reference_texts: List[str], batch_size: int = 64):
"""Calcola statistiche di riferimento dal training set."""
print(f"Computing reference embeddings for {len(reference_texts)} texts...")
self.reference_embeddings = self.model.encode(
reference_texts, batch_size=batch_size, show_progress_bar=True
)
self.reference_stats = {
'mean': self.reference_embeddings.mean(axis=0),
'std': self.reference_embeddings.std(axis=0),
'n': len(reference_texts)
}
print(f"Reference embeddings computed: shape={self.reference_embeddings.shape}")
def detect_drift(self, production_texts: List[str],
batch_size: int = 64) -> Dict[str, Any]:
"""Rileva drift confrontando produzione con riferimento."""
if self.reference_embeddings is None:
raise ValueError("Call fit() first with reference data")
prod_embeddings = self.model.encode(
production_texts, batch_size=batch_size, show_progress_bar=False
)
# Metodo 1: KS test per ogni dimensione dell'embedding
ks_stats = []
ks_pvalues = []
for dim in range(self.reference_embeddings.shape[1]):
stat, pvalue = ks_2samp(
self.reference_embeddings[:, dim],
prod_embeddings[:, dim]
)
ks_stats.append(stat)
ks_pvalues.append(pvalue)
avg_ks = np.mean(ks_stats)
max_ks = np.max(ks_stats)
# Metodo 2: Cosine distance media tra centroidi
ref_centroid = self.reference_embeddings.mean(axis=0)
prod_centroid = prod_embeddings.mean(axis=0)
centroid_distance = 1 - np.dot(ref_centroid, prod_centroid) / (
np.linalg.norm(ref_centroid) * np.linalg.norm(prod_centroid)
)
# Metodo 3: PSI (Population Stability Index)
psi = self._compute_psi(
self.reference_embeddings[:, :10], # prime 10 dim per PSI
prod_embeddings[:, :10]
)
drift_detected = (avg_ks > self.ks_threshold or
centroid_distance > 0.05)
return {
"drift_detected": drift_detected,
"avg_ks_statistic": float(avg_ks),
"max_ks_statistic": float(max_ks),
"centroid_cosine_distance": float(centroid_distance),
"psi": float(psi),
"n_production": len(production_texts),
"alert_level": "HIGH" if avg_ks > self.ks_threshold * 2
else "MEDIUM" if drift_detected
else "LOW"
}
def _compute_psi(self, reference: np.ndarray, production: np.ndarray,
n_bins: int = 10) -> float:
"""Population Stability Index: misura lo shift della distribuzione."""
psi_values = []
for dim in range(reference.shape[1]):
ref = reference[:, dim]
prod = production[:, dim]
bins = np.percentile(ref, np.linspace(0, 100, n_bins + 1))
bins[0] -= 0.001
bins[-1] += 0.001
ref_counts, _ = np.histogram(ref, bins=bins)
prod_counts, _ = np.histogram(prod, bins=bins)
ref_pct = (ref_counts / ref_counts.sum()) + 1e-10
prod_pct = (prod_counts / prod_counts.sum()) + 1e-10
psi = np.sum((prod_pct - ref_pct) * np.log(prod_pct / ref_pct))
psi_values.append(psi)
return float(np.mean(psi_values))4. Proxy Metrikleri: Etiketsiz Monitör
Üretimde genellikle doğruluğu hesaplamak için gerçek etiketlere sahip değiliz. hadi kullanalım proxy ölçümleri Bu da modelin kalitesiyle ilişkilidir.
import numpy as np
from collections import defaultdict
from datetime import datetime, timedelta
import pandas as pd
class NLPProxyMetricsMonitor:
"""
Monitora metriche proxy per modelli NLP senza label.
"""
def __init__(self, window_hours: int = 24):
self.window_hours = window_hours
self.predictions = []
def add_prediction(self, prediction: dict):
"""Aggiunge una predizione al log."""
prediction['timestamp'] = datetime.utcnow()
self.predictions.append(prediction)
def compute_proxy_metrics(self) -> dict:
"""Calcola metriche proxy dalla finestra temporale corrente."""
cutoff = datetime.utcnow() - timedelta(hours=self.window_hours)
recent = [p for p in self.predictions if p['timestamp'] > cutoff]
if not recent:
return {"error": "Nessuna predizione nella finestra temporale"}
confidences = [p['confidence'] for p in recent]
latencies = [p['latency_ms'] for p in recent]
labels = [p['predicted_label'] for p in recent]
# 1. Confidence distribution (bassa confidenza = modello incerto)
low_conf_pct = sum(1 for c in confidences if c < 0.7) / len(confidences)
avg_confidence = np.mean(confidences)
confidence_entropy = -np.sum(
[(c * np.log(c) + (1-c) * np.log(1-c + 1e-10)) for c in confidences]
) / len(confidences)
# 2. Label distribution (drift nelle predizioni)
label_counts = defaultdict(int)
for l in labels:
label_counts[l] += 1
label_distribution = {k: v/len(labels) for k, v in label_counts.items()}
# 3. Latency percentiles
p50 = np.percentile(latencies, 50)
p95 = np.percentile(latencies, 95)
p99 = np.percentile(latencies, 99)
# 4. Text length statistics
lengths = [p.get('input_length_chars', 0) for p in recent]
# 5. Refusal rate (se il modello ritorna "UNCERTAIN")
uncertain_pct = sum(1 for l in labels if l == 'UNCERTAIN') / len(labels)
return {
"window_hours": self.window_hours,
"n_predictions": len(recent),
"avg_confidence": round(avg_confidence, 4),
"low_confidence_pct": round(low_conf_pct, 4),
"confidence_entropy": round(float(confidence_entropy), 4),
"label_distribution": label_distribution,
"latency_p50_ms": round(p50, 1),
"latency_p95_ms": round(p95, 1),
"latency_p99_ms": round(p99, 1),
"avg_input_length": round(np.mean(lengths), 1),
"uncertain_pct": round(uncertain_pct, 4)
}
def check_alerts(self, thresholds: dict) -> list:
"""Verifica se le metriche proxy superano le soglie di alert."""
metrics = self.compute_proxy_metrics()
alerts = []
checks = {
"avg_confidence": ("<", thresholds.get("min_confidence", 0.75)),
"low_confidence_pct": (">", thresholds.get("max_low_conf_pct", 0.20)),
"latency_p95_ms": (">", thresholds.get("max_p95_latency_ms", 500)),
"uncertain_pct": (">", thresholds.get("max_uncertain_pct", 0.10)),
}
for metric_name, (op, threshold) in checks.items():
value = metrics.get(metric_name)
if value is None:
continue
triggered = (value < threshold if op == "<" else value > threshold)
if triggered:
alerts.append({
"metric": metric_name,
"value": value,
"threshold": threshold,
"severity": "HIGH" if abs(value - threshold) / threshold > 0.5 else "MEDIUM"
})
return alerts5. Otomatik Yeniden Eğitim Boru Hattı
import subprocess
from pathlib import Path
import json
from datetime import datetime
class AutoRetrainingPipeline:
"""
Pipeline di retraining automatico triggered dal drift detection.
"""
def __init__(self,
drift_detector: EmbeddingDriftDetector,
proxy_monitor: NLPProxyMetricsMonitor,
base_model_path: str,
data_path: str,
output_path: str):
self.drift_detector = drift_detector
self.proxy_monitor = proxy_monitor
self.base_model_path = base_model_path
self.data_path = data_path
self.output_path = output_path
self.retraining_history = []
def should_retrain(self,
production_texts: list,
drift_threshold: float = 0.1,
confidence_threshold: float = 0.75) -> dict:
"""
Determina se e necessario il retraining.
Ritorna {should_retrain: bool, reason: str, severity: str}
"""
# Check 1: Embedding drift
drift_report = self.drift_detector.detect_drift(production_texts)
if drift_report['drift_detected']:
return {
"should_retrain": True,
"reason": f"Embedding drift rilevato: KS={drift_report['avg_ks_statistic']:.4f}",
"severity": drift_report['alert_level'],
"drift_report": drift_report
}
# Check 2: Proxy metrics
metrics = self.proxy_monitor.compute_proxy_metrics()
alerts = self.proxy_monitor.check_alerts({
"min_confidence": confidence_threshold,
"max_low_conf_pct": 0.25
})
if any(a['severity'] == 'HIGH' for a in alerts):
return {
"should_retrain": True,
"reason": f"Metriche proxy critiche: {alerts}",
"severity": "HIGH",
"alerts": alerts
}
return {
"should_retrain": False,
"reason": "Tutte le metriche nella norma",
"severity": "LOW"
}
def trigger_retraining(self, trigger_reason: str, new_data_path: str):
"""Avvia il retraining con i nuovi dati."""
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
new_model_path = f"{self.output_path}/model_v{timestamp}"
print(f"Avvio retraining: {trigger_reason}")
print(f"Nuovo modello: {new_model_path}")
# Log del retraining
self.retraining_history.append({
"timestamp": timestamp,
"trigger_reason": trigger_reason,
"base_model": self.base_model_path,
"new_data": new_data_path,
"output_model": new_model_path,
"status": "started"
})
# In produzione: trigghera una pipeline CI/CD (Airflow, GitHub Actions, Kubeflow)
# Esempio con subprocess:
# subprocess.Popen([
# "python", "train.py",
# "--base-model", self.base_model_path,
# "--train-data", new_data_path,
# "--output", new_model_path,
# ])
return {
"retraining_id": timestamp,
"new_model_path": new_model_path,
"status": "triggered"
}6. Yeni Model Versiyonları için A/B Testi
import random
from typing import Callable
class ABTestingRouter:
"""
Router per A/B testing tra versioni del modello.
Splitta il traffico tra il modello corrente (A) e il nuovo (B).
"""
def __init__(self,
model_a: Callable,
model_b: Callable,
traffic_split_b: float = 0.1,
experiment_id: str = "exp_001"):
self.model_a = model_a
self.model_b = model_b
self.traffic_split_b = traffic_split_b
self.experiment_id = experiment_id
self.results = {"a": [], "b": []}
def predict(self, text: str, user_id: str = None) -> dict:
"""Instrada la richiesta al modello A o B in base al traffic split."""
# Instradamento deterministico basato su user_id (per coerenza)
if user_id:
use_b = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100 < (self.traffic_split_b * 100)
else:
use_b = random.random() < self.traffic_split_b
model_variant = "b" if use_b else "a"
model_fn = self.model_b if use_b else self.model_a
result = model_fn(text)
result["model_variant"] = model_variant
result["experiment_id"] = self.experiment_id
self.results[model_variant].append({
"confidence": result.get("confidence", 0),
"latency_ms": result.get("latency_ms", 0),
})
return result
def get_experiment_stats(self) -> dict:
"""Calcola statistiche dell'esperimento A/B."""
stats = {}
for variant in ["a", "b"]:
if self.results[variant]:
confs = [r["confidence"] for r in self.results[variant]]
lats = [r["latency_ms"] for r in self.results[variant]]
stats[variant] = {
"n_requests": len(self.results[variant]),
"avg_confidence": round(np.mean(confs), 4),
"avg_latency_ms": round(np.mean(lats), 1),
}
return {"experiment_id": self.experiment_id, "variants": stats}7. Prometheus ve Grafana'nın bulunduğu Kontrol Paneli
# monitoring_api.py
from fastapi import FastAPI
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response
import time
app = FastAPI()
# Metriche Prometheus per modelli NLP
PREDICTIONS_TOTAL = Counter(
"nlp_predictions_total",
"Numero totale di predizioni NLP",
["model_version", "predicted_label"]
)
CONFIDENCE_HISTOGRAM = Histogram(
"nlp_prediction_confidence",
"Distribuzione del confidence score",
["model_version"],
buckets=[0.5, 0.6, 0.7, 0.75, 0.8, 0.85, 0.9, 0.95, 0.99, 1.0]
)
LATENCY_HISTOGRAM = Histogram(
"nlp_inference_latency_seconds",
"Latenza dell'inferenza NLP",
["model_version"],
buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0]
)
DRIFT_SCORE = Gauge(
"nlp_embedding_drift_score",
"Score del drift degli embedding (0=no drift, 1=drift massimo)",
["model_version"]
)
INPUT_LENGTH_HISTOGRAM = Histogram(
"nlp_input_length_chars",
"Lunghezza dell'input in caratteri",
["model_version"],
buckets=[50, 100, 200, 500, 1000, 2000, 5000]
)
MODEL_VERSION = "v2.1.0"
@app.post("/predict")
def predict_with_monitoring(request: dict):
text = request["text"]
start = time.time()
# ... Inferenza ...
result = {"label": "POSITIVE", "score": 0.92}
latency = time.time() - start
# Aggiorna metriche Prometheus
PREDICTIONS_TOTAL.labels(
model_version=MODEL_VERSION,
predicted_label=result["label"]
).inc()
CONFIDENCE_HISTOGRAM.labels(model_version=MODEL_VERSION).observe(result["score"])
LATENCY_HISTOGRAM.labels(model_version=MODEL_VERSION).observe(latency)
INPUT_LENGTH_HISTOGRAM.labels(model_version=MODEL_VERSION).observe(len(text))
return {**result, "latency_ms": latency * 1000}
@app.get("/metrics")
def metrics():
"""Endpoint Prometheus per il scraping delle metriche."""
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
# docker-compose.yml per Prometheus + Grafana:
# services:
# prometheus:
# image: prom/prometheus
# volumes:
# - ./prometheus.yml:/etc/prometheus/prometheus.yml
# grafana:
# image: grafana/grafana
# ports:
# - "3000:3000"8. Gölge Dağıtımı ve Kademeli Dağıtım
Yeni bir modeli gerçek trafiğe sunmadan önce, gölge dağıtımı Kullanıcılar için herhangi bir risk olmadan gecikmeyi ve davranışı doğrulamanıza olanak tanır. Gölge modeli, üretim modeliyle aynı talepleri alır ancak kendi tahminler göz ardı edilir; yalnızca izleme amacıyla kullanılırlar.
import asyncio
import time
from typing import Callable, Dict, Any
class ShadowDeploymentManager:
"""
Gestisce il shadow deployment di un nuovo modello NLP.
Il modello shadow riceve tutto il traffico ma non risponde agli utenti.
"""
def __init__(self,
production_model: Callable,
shadow_model: Callable,
shadow_name: str = "shadow_v2"):
self.production_model = production_model
self.shadow_model = shadow_model
self.shadow_name = shadow_name
self.comparison_log: list = []
def predict(self, text: str, user_id: str = None) -> Dict[str, Any]:
"""
Esegue la predizione in produzione e in background quella shadow.
Restituisce solo il risultato del modello di produzione.
"""
# Predizione produzione (sincrona)
prod_start = time.time()
prod_result = self.production_model(text)
prod_latency = (time.time() - prod_start) * 1000
# Predizione shadow (asincrona, non blocca la risposta)
shadow_start = time.time()
try:
shadow_result = self.shadow_model(text)
shadow_latency = (time.time() - shadow_start) * 1000
shadow_error = None
except Exception as e:
shadow_result = None
shadow_latency = None
shadow_error = str(e)
# Log del confronto
self.comparison_log.append({
"text_hash": hash(text),
"prod_label": prod_result.get("label"),
"prod_confidence": prod_result.get("confidence"),
"prod_latency_ms": prod_latency,
"shadow_label": shadow_result.get("label") if shadow_result else None,
"shadow_confidence": shadow_result.get("confidence") if shadow_result else None,
"shadow_latency_ms": shadow_latency,
"shadow_error": shadow_error,
"agreement": prod_result.get("label") == (shadow_result or {}).get("label")
})
# Restituisce SOLO il risultato del modello di produzione
return prod_result
def get_shadow_stats(self) -> Dict[str, Any]:
"""Calcola statistiche di confronto tra produzione e shadow."""
if not self.comparison_log:
return {"error": "Nessun dato di confronto disponibile"}
agreement_rate = sum(1 for r in self.comparison_log if r["agreement"]) / len(self.comparison_log)
prod_latencies = [r["prod_latency_ms"] for r in self.comparison_log if r["prod_latency_ms"]]
shadow_latencies = [r["shadow_latency_ms"] for r in self.comparison_log if r["shadow_latency_ms"]]
error_rate = sum(1 for r in self.comparison_log if r["shadow_error"]) / len(self.comparison_log)
import numpy as np
return {
"n_requests": len(self.comparison_log),
"agreement_rate": round(agreement_rate, 4),
"prod_p95_latency_ms": round(np.percentile(prod_latencies, 95), 1) if prod_latencies else None,
"shadow_p95_latency_ms": round(np.percentile(shadow_latencies, 95), 1) if shadow_latencies else None,
"shadow_error_rate": round(error_rate, 4),
"ready_for_promotion": agreement_rate >= 0.95 and error_rate < 0.01
}
# Strategia di rollout graduale: 1% → 10% → 50% → 100%
ROLLOUT_STAGES = [
{"traffic_pct": 0.01, "min_requests": 500, "min_agreement": 0.95},
{"traffic_pct": 0.10, "min_requests": 2000, "min_agreement": 0.96},
{"traffic_pct": 0.50, "min_requests": 5000, "min_agreement": 0.97},
{"traffic_pct": 1.00, "min_requests": None, "min_agreement": None}, # full rollout
]9. NLP İzlemede Yaygın Anti-Paternler
Kaçınılması Gereken Anti-Desenler
- Yalnızca gecikmeyi izle: gecikme ve altyapı ölçümü, model kalitesi değil. Hızlı ama yanlış bir model ve yavaş doğru olandan daha kötü.
- Referans dağıtımı yok: sürüklenme tespiti anlamsızdır eğitim/doğrulama verileri üzerinden hesaplanan sağlam bir referans dağılımı olmadan.
- Uyarı yorgunluğu: Çok hassas eşikler, çağrı ekibini yanlış pozitiflerle doldurur. Muhafazakar eşiklerle başlayın ve gözlemlenen modellere göre kalibre edin.
- Tek sinyalle ilgili kararlar: Hiçbir zaman yeniden eğitimi aşağıdakilere dayalı olarak tetiklemeyin: tek bir göstergede. En az iki bağımsız, uyumlu sinyal gerektirir.
- Yukarı akış verilerinin kalitesini göz ardı edin: Modeli olmadan takip edin veri hattını izleyin ve eksik. Geçerli şema ve giriş verilerinin güncelliği.
- Çevrimdışı doğrulama olmadan yeniden eğitim: Otomatik olarak geri çekilen bir model Tetikleyici otomatik olsa bile, konuşlandırılmadan önce bir çevrimdışı test setini geçmelidir.
10. NLP İzleme için Tam Kontrol Listesi
Üretimde NLP İzleme Kontrol Listesi
- Günlüğe kaydetme: her tahmini metin (veya karma), güven, gecikme ile günlüğe kaydedin, JSONL formatında model sürümü ve zaman damgası
- Sürüklenme Algılama: 1.000 örnek pencerede haftalık olarak yerleştirme sürüklenmesini kontrol edin; KS > 0,15 veya PSI > 0,2 ise acil uyarı
- Proxy Metrikleri: güven dağılımını, etiket dağıtımını izleyin ve Prometheus aracılığıyla gerçek zamanlı gecikme
- Gerçek Gerçek Koleksiyonu: Kullanıcı geri bildirimleriyle gerçek etiketleri toplayın, ek açıklama ekibi veya rastgele örnekleme (trafiğin %1-5'i)
- Yeniden Eğitim Tetikleyicisi: Otomatik yeniden eğitim için net eşikler tanımlayın (örneğin sürüklenme puanı > 0,2 veya tahmini doğruluk < 0,85); 2+ uyumlu sinyal isteyin
- Gölge Dağıtımı: A/B testinden önce yeni modeli doğrulayın en az 24 saat gölge modunda
- A/B Testi: Modelin her yeni versiyonunu %10 ile doğrular tam kullanıma sunulmadan önce en az 48 saat boyunca trafik
- Uyarı: yüksek önem derecesine sahip uyarılar için bildirimleri (Slack, PagerDuty) yapılandırma yanıt runbook'larına bağlantılar içeren
- Veri Saklama: Geçmiş analiz için günlükleri en az 90 gün saklayın
- GDPR: üretim günlüklerindeki kullanıcı metinlerini anonimleştirin veya karma hale getirin; Kişisel bilgileri asla açık izin ve şifreleme olmadan saklamayın
Sonuçlar: Serinin Sonu
Bu yazıyla seriyi sonlandırıyoruz Modern NLP: BERT'ten Yüksek Lisans'a. Tam bir yolculuğa çıktık: tokenizasyon ve yerleştirmenin temellerinden, İtalyanca için duygu analizinden yerel yüksek lisans programlarının ince ayarına kadar BERT mimarisine kadar, Anlamsal benzerlikten üretimde izlemeye kadar.
Seri Özeti
| # | Öğe | Anahtar Kavramlar |
|---|---|---|
| 1 | NLP'nin Temelleri | Tokenleştirme, Word2Vec, GloVe, boru hattı |
| 2 | BERT ve Transformatörler | Mimari, MLM, NSP, ince ayar |
| 3 | Duygu Analizi | VADER, BERT, üretim, FastAPI |
| 4 | NLP İtalyanca | hisset, AlBERTo, spaCy, lehçeler |
| 5 | Adlandırılmış Varlık Tanıma | BIO formatı, spaCy, BERT NER, devamı |
| 6 | Metin Sınıflandırması | Çoklu etiket, sıfır atış, SetFit |
| 7 | SarılmaYüz Transformers | AutoClass, Trainer, PEFT, Hızlandırma |
| 8 | Yerel ince ayar | LoRA, QLoRA, DAPT, yıkıcı unutma |
| 9 | Anlamsal Benzerlik | SBERT, FAISS, çift kodlayıcı, çapraz kodlayıcı |
| 10 | NLP İzleme | Sapma tespiti, temsili ölçümler, yeniden eğitim |
Keşfedilecek İlgili Seriler
- Yapay Zeka Mühendisliği / RAG: Yerleştirmelerle eksiksiz RAG sistemleri oluşturun ve bu serideki anlamsal arama teknikleri
- Gelişmiş Derin Öğrenme: nicemleme, budama ve hakkında daha fazla bilgi edinin Büyük modeller için optimizasyon teknikleri
- MLOps: MLflow ile izlemeyi ve yeniden eğitimi otomatikleştirin, ML modelleri için DVC ve CI/CD ardışık düzenleri
- Bilgisayarla Görme: Bu serideki tekniklerin çoğu (BERT benzeri mimariler, ViT, ince ayar) CV için de geçerlidir