Avvisi su latenza, costi e peggioramento della qualità
Definisca soglie di avviso per la latenza p99, il costo per richiesta e i punteggi di qualità automatizzati, quindi indirizzi gli avvisi a Slack o PagerDuty quando la pipeline LLM peggiora.
Avvisi su latenza, costi e peggioramento della qualità è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.
Perché gli alert sono importanti per i sistemi LLM
Le applicazioni LLM possono presentare problemi difficili da individuare, che si manifestano gradualmente. Una modifica al prompt potrebbe aumentare la latenza media del 30%, un miglioramento del recupero potrebbe ridurre leggermente la qualità delle risposte oppure un picco di utilizzo potrebbe portare i costi giornalieri a un valore cinque volte superiore al budget. Senza alert proattivi, scoprireste questi problemi solo quando gli utenti iniziano a lamentarsi o arriva la fattura mensile. Gli alert trasformano la gestione reattiva delle emergenze in operazioni proattive.
Le tre categorie di alert
Gli alert per le applicazioni LLM rientrano in tre categorie. Gli alert sulla latenza si attivano quando il tempo di risposta supera una soglia relativa all’esperienza utente (ad esempio, p99 > 10 secondi). Gli alert sui costi si attivano quando il costo per richiesta o la spesa giornaliera supera le soglie di budget, prevenendo sorprese in fattura. Gli alert sulla qualità si attivano quando le metriche automatiche della qualità (punteggi LLM-as-judge, tassi di soddisfazione degli utenti, punteggi di fedeltà) scendono al di sotto di un limite accettabile. Ogni categoria richiede una strumentazione e canali di alert diversi.
from dataclasses import dataclass
@dataclass
class AlertThresholds:
# Latency (milliseconds)
p50_latency_ms: int = 2000 # median should be under 2s
p99_latency_ms: int = 10000 # 99th percentile under 10s
# Cost (USD)
max_cost_per_request: float = 0.05 # alert if one request costs > 5 cents
max_daily_spend: float = 50.00 # alert if daily spend exceeds $50
# Quality (0.0 to 1.0 scale)
min_quality_score: float = 0.75 # alert if rolling avg drops below 75%
min_faithfulness: float = 0.80 # alert if RAGAS faithfulness drops below 80%
DEFAULT_THRESHOLDS = AlertThresholds()Raccolta delle metriche di latenza
Per creare alert sulla latenza, dovete innanzitutto raccoglierla in modo coerente. Misurate separatamente tre componenti della latenza: il tempo al primo token (TTFT, fondamentale per l’esperienza utente con lo streaming), il tempo di risposta totale e la latenza di ogni passaggio per il recupero e la generazione. Memorizzate questi dati come serie temporali (un punto dati per richiesta), così potrete calcolare percentili e medie mobili sugli ultimi N minuti o ore.
import time
from collections import deque
from statistics import quantiles
class LatencyTracker:
def __init__(self, window_size=100):
self.window = deque(maxlen=window_size) # rolling window of latencies
def record(self, latency_ms: float):
self.window.append(latency_ms)
def p50(self) -> float:
if not self.window:
return 0
return quantiles(self.window, n=100)[49]
def p99(self) -> float:
if not self.window:
return 0
return quantiles(self.window, n=100)[98]
def check_alert(self, thresholds: AlertThresholds) -> list[str]:
alerts = []
if self.p99() > thresholds.p99_latency_ms:
alerts.append(f'p99 latency {self.p99():.0f}ms exceeds {thresholds.p99_latency_ms}ms')
if self.p50() > thresholds.p50_latency_ms:
alerts.append(f'p50 latency {self.p50():.0f}ms exceeds {thresholds.p50_latency_ms}ms')
return alerts
latency_tracker = LatencyTracker()Monitoraggio e alert sui costi
Il monitoraggio dei costi richiede di tenere traccia della spesa a più livelli di granularità: costo per richiesta (per individuare le query anomale più costose), totali orari e giornalieri (per rilevare i picchi di utilizzo) e costo per funzionalità (per identificare la parte più costosa dell’applicazione). Attivate immediatamente un alert per le anomalie del costo per richiesta e utilizzate controlli pianificati per i limiti di budget giornalieri.
from datetime import datetime, date
import threading
class CostTracker:
def __init__(self):
self._lock = threading.Lock()
self._daily_spend = {} # date -> total USD
self._per_request = [] # list of (timestamp, cost)
def record(self, cost_usd: float) -> list[str]:
today = date.today().isoformat()
alerts = []
with self._lock:
# Track daily spend
self._daily_spend[today] = self._daily_spend.get(today, 0) + cost_usd
self._per_request.append((datetime.now(), cost_usd))
# Alert on expensive single requests
if cost_usd > DEFAULT_THRESHOLDS.max_cost_per_request:
alerts.append(f'Expensive request: ${cost_usd:.4f} (threshold: ${DEFAULT_THRESHOLDS.max_cost_per_request})')
# Alert on daily budget exceeded
daily_total = self._daily_spend[today]
if daily_total > DEFAULT_THRESHOLDS.max_daily_spend:
alerts.append(f'Daily budget exceeded: ${daily_total:.2f} > ${DEFAULT_THRESHOLDS.max_daily_spend}')
return alerts
cost_tracker = CostTracker()Alert sui punteggi di qualità
La gestione degli alert sulla qualità è la categoria più complessa, perché la qualità non può essere misurata solo tramite le metriche dell’infrastruttura: richiede una valutazione semantica. L’approccio più scalabile è la valutazione automatica su un campione: per un campione casuale di richieste in produzione (5-10%), eseguite in modo asincrono un valutatore LLM-as-judge, memorizzate i punteggi e calcolate una media mobile. Attivate un alert quando la media mobile scende al di sotto del limite di qualità.
import random
from openai import OpenAI
client = OpenAI()
def evaluate_response_quality(question: str, answer: str) -> float:
judge_prompt = f'''Rate the quality of this AI assistant response on a scale from 0 to 1.
Question: {question}
Answer: {answer}
Return only a JSON object: {{"score": 0.85, "reason": "brief reason"}}
Scoring guide:
1.0 = Perfect, accurate, helpful
0.75 = Good, minor issues
0.5 = Acceptable but incomplete
0.25 = Poor, major gaps
0.0 = Completely wrong or harmful'''
response = client.chat.completions.create(
model='gpt-4o-mini', # cheaper model for evaluation
messages=[{'role': 'user', 'content': judge_prompt}],
response_format={'type': 'json_object'}
)
import json
result = json.loads(response.choices[0].message.content)
return float(result['score'])
def maybe_evaluate(question: str, answer: str, sample_rate=0.1):
if random.random() < sample_rate:
score = evaluate_response_quality(question, answer)
quality_tracker.record(score)
return score
return NoneMedie mobili per rilevare le tendenze
Una singola risposta errata non indica un problema sistemico. Utilizzate le medie mobili su una finestra temporale (ad esempio, l’ultima ora o le ultime 100 richieste) per rilevare le tendenze. Una media mobile che supera una soglia e vi rimane per più di 10 minuti indica un problema reale, mentre un picco breve potrebbe essere rumore. Le medie mobili esponenzialmente ponderate (EWMA) reagiscono più rapidamente ai cambiamenti recenti rispetto alle semplici medie mobili.
class RollingQualityMonitor:
def __init__(self, window=50, alert_threshold=0.75, ewma_alpha=0.1):
self.scores = []
self.window = window
self.alert_threshold = alert_threshold
self.alpha = ewma_alpha # EWMA decay factor
self.ewma_score = None
def record(self, score: float):
self.scores.append(score)
if len(self.scores) > self.window:
self.scores.pop(0)
# Update exponentially weighted moving average
if self.ewma_score is None:
self.ewma_score = score
else:
self.ewma_score = self.alpha * score + (1 - self.alpha) * self.ewma_score
def should_alert(self) -> bool:
if len(self.scores) < 10: # not enough data yet
return False
return self.ewma_score < self.alert_threshold
def summary(self) -> dict:
if not self.scores:
return {}
return {
'rolling_avg': sum(self.scores) / len(self.scores),
'ewma': self.ewma_score,
'sample_count': len(self.scores),
'alert': self.should_alert()
}
quality_tracker = RollingQualityMonitor()Instradamento degli alert: Slack e PagerDuty
Gli alert sono utili solo se raggiungono la persona giusta attraverso il canale giusto. Instradate gli alert in base alla gravità: gli alert sul peggioramento della qualità e sul budget dei costi devono essere inviati a un canale Slack monitorato dal team durante l’orario lavorativo. Gli alert sulla latenza oltre le soglie critiche (ad esempio, p99 > 30 secondi) e i picchi improvvisi dei costi (ad esempio, 10 volte il valore normale in 5 minuti) devono essere inviati a PagerDuty per l’escalation immediata al personale di turno.
import requests
def send_slack_alert(message: str, severity: str, webhook_url: str):
color = {'critical': '#ff0000', 'warning': '#ff9900', 'info': '#36a64f'}[severity]
payload = {
'attachments': [{
'color': color,
'title': f'LLM Alert [{severity.upper()}]',
'text': message,
'footer': 'AI Pipeline Monitor'
}]
}
requests.post(webhook_url, json=payload)
def send_pagerduty_alert(title: str, details: str, routing_key: str):
payload = {
'routing_key': routing_key,
'event_action': 'trigger',
'payload': {
'summary': title,
'severity': 'critical',
'source': 'ai-pipeline-monitor',
'custom_details': {'details': details}
}
}
requests.post('https://events.pagerduty.com/v2/enqueue', json=payload)
def route_alert(alert_text: str, severity: str):
send_slack_alert(alert_text, severity, SLACK_WEBHOOK_URL)
if severity == 'critical':
send_pagerduty_alert(alert_text, alert_text, PAGERDUTY_ROUTING_KEY)Il ciclo di valutazione degli alert
La logica degli alert dovrebbe essere eseguita in un ciclo pianificato, non inline con l’elaborazione delle richieste. Eseguire i controlli degli alert in modo asincrono impedisce che aggiungano latenza alle richieste in produzione. Per la maggior parte delle esigenze di alert è sufficiente un thread in background o un job pianificato eseguito ogni 60 secondi. Raccogliete le metriche nel percorso della richiesta e valutate le soglie nel ciclo in background.
import threading
import time
def alert_evaluation_loop(interval_seconds=60):
while True:
try:
# Check latency
latency_alerts = latency_tracker.check_alert(DEFAULT_THRESHOLDS)
for alert in latency_alerts:
route_alert(f'LATENCY: {alert}', 'warning')
# Check quality
quality_summary = quality_tracker.summary()
if quality_summary.get('alert'):
msg = f'QUALITY DEGRADATION: Rolling avg {quality_summary["ewma"]:.2f} below threshold {DEFAULT_THRESHOLDS.min_quality_score}'
route_alert(msg, 'warning')
print(f'Alert check complete. Quality: {quality_summary.get("ewma", "N/A")}')
except Exception as e:
print(f'Alert evaluation error: {e}')
time.sleep(interval_seconds)
# Start in background thread
alert_thread = threading.Thread(target=alert_evaluation_loop, daemon=True)
alert_thread.start()Affaticamento da alert e regolazione delle soglie
Il fenomeno dell’affaticamento da alert si verifica quando gli alert si attivano così frequentemente che il team inizia a ignorarli. Per prevenirlo: iniziate con soglie conservative (alte) e restringetele man mano che comprendete il vostro comportamento di base; richiedete che le condizioni di alert persistano per più cicli di valutazione prima di attivare una notifica; raggruppate gli alert correlati in un’unica notifica; e rivedete regolarmente gli alert che non portano mai ad azioni significative, disattivandoli.
class AlertDebouncer:
def __init__(self, required_consecutive_fires=3):
self.required = required_consecutive_fires
self.fire_counts = {} # alert_name -> consecutive fires
self.already_firing = set() # alert_names currently active
def should_fire(self, alert_name: str, condition: bool) -> bool:
if condition:
self.fire_counts[alert_name] = self.fire_counts.get(alert_name, 0) + 1
if self.fire_counts[alert_name] >= self.required and alert_name not in self.already_firing:
self.already_firing.add(alert_name)
return True # fire the alert (first time condition sustained)
else:
if self.fire_counts.get(alert_name, 0) > 0:
self.fire_counts[alert_name] = 0
self.already_firing.discard(alert_name) # condition cleared
return False # either not sustained or already firing (no duplicate alert)
debouncer = AlertDebouncer(required_consecutive_fires=3)Monitoraggio delle anomalie dei costi con metodi statistici
I semplici alert basati su soglie non rilevano anomalie dei costi più sfumate, come un aumento graduale del 50% nell’arco di due giorni. Utilizzate il rilevamento statistico delle anomalie: calcolate la media mobile e la deviazione standard dei costi orari e attivate un alert quando il costo dell’ora corrente supera la media di un numero N di deviazioni standard (alert basati sullo Z-score). In questo modo il sistema si adatta automaticamente ai normali modelli di utilizzo, come il traffico dei giorni feriali rispetto a quello del fine settimana.
import statistics
def compute_z_score(recent_value: float, historical_values: list[float]) -> float:
if len(historical_values) < 5:
return 0 # not enough data
mean = statistics.mean(historical_values)
stdev = statistics.stdev(historical_values)
if stdev == 0:
return 0
return (recent_value - mean) / stdev
def check_cost_anomaly(current_hour_cost: float, historical_hourly_costs: list[float]) -> str | None:
z = compute_z_score(current_hour_cost, historical_hourly_costs)
if z > 3.0: # more than 3 standard deviations above mean
mean = statistics.mean(historical_hourly_costs)
return f'Cost anomaly: ${current_hour_cost:.2f} this hour (normal: ${mean:.2f}, z={z:.1f})'
return NoneCreazione di una dashboard operativa
Gli alert sono il livello reattivo; una dashboard operativa aggiornata in tempo reale è il livello proattivo. Create una dashboard semplice che mostri le metriche in tempo reale: latenza p50/p99 corrente, richieste al minuto, punteggio di qualità corrente (EWMA mobile), costo giornaliero accumulato rispetto al budget e le 5 richieste più costose dell’ultima ora. In questo modo il team può avere una visione immediata dello stato del sistema senza attendere l’attivazione di un alert.
Verifica rapida
Verificate la vostra comprensione degli alert sulle metriche delle applicazioni LLM trattati in questa lezione.
Riepilogo della lezione
In questa lezione ha imparato che: tre categorie di avvisi coprono i sistemi LLM — latenza, costi e qualità — e ciascuna richiede una strumentazione diversa; medie mobili ed EWMA rilevano il peggioramento graduale della qualità meglio dei controlli delle soglie sulle singole richieste; inoltre, il debouncing degli avvisi previene l'affaticamento da avvisi richiedendo che le condizioni persistano per più cicli di valutazione prima di attivare un avviso. Ora approfondiremo gli attacchi di prompt injection e la sicurezza dell'IA.
Domande Frequenti
La lezione «Avvisi su latenza, costi e peggioramento della qualità» è gratuita?
Sì — il testo completo di «Avvisi su latenza, costi e peggioramento della qualità» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.
Cosa imparerò in «Avvisi su latenza, costi e peggioramento della qualità»?
Definisca soglie di avviso per la latenza p99, il costo per richiesta e i punteggi di qualità automatizzati, quindi indirizzi gli avvisi a Slack o PagerDuty quando la pipeline LLM peggiora. Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Engineering Academy?
Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.
Quanto tempo richiede la lezione «Avvisi su latenza, costi e peggioramento della qualità»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?
Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Perché le app LLM sono difficili da sottoporre a debug
- Tracing con LangSmith
- Langfuse per l'observability indipendente dal modello
- Avvisi su latenza, costi e peggioramento della qualità