0Pricing
AI Engineering Academy · Lesson

Alerting on Latency, Cost, and Quality Degradation

Define alert thresholds on p99 latency, per-request cost, and automated quality scores, and route alerts to Slack or PagerDuty when your LLM pipeline degrades.

Alerting on Latency, Cost, and Quality Degradation is a free AI Engineering Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Alerting Matters for LLM Systems

LLM applications fail in ways that are subtle and gradual. A prompt change might increase average latency by 30%, a retrieval upgrade might slightly reduce answer quality, or a surge in usage might push daily costs 5x above budget. Without proactive alerting, you discover these problems only when users complain or the monthly bill arrives. Alerts transform reactive fire-fighting into proactive operations.

The Three Alert Categories

LLM application alerts fall into three categories. Latency alerts fire when response time exceeds a user experience threshold (e.g., p99 > 10 seconds). Cost alerts fire when per-request cost or daily spend exceeds budget thresholds, preventing bill shock. Quality alerts fire when automated quality metrics (LLM-as-judge scores, user satisfaction rates, faithfulness scores) drop below an acceptable floor. Each category requires different instrumentation and different alert channels.

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()

Collecting Latency Metrics

To alert on latency, you first need to collect it consistently. Measure three latency components separately: time to first token (TTFT, critical for streaming UX), total response time, and per-step latencies for retrieval and generation. Store these as time-series data (one data point per request) so you can compute percentiles and rolling averages over the past N minutes or hours.

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()

Tracking and Alerting on Cost

Cost monitoring requires tracking spend at multiple granularities: per-request cost (to catch expensive outlier queries), hourly and daily totals (to catch usage spikes), and per-feature cost (to identify which part of your application is most expensive). Alert immediately on per-request cost anomalies, and use scheduled checks for daily budget limits.

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()

Quality Score Alerting

Quality alerting is the most complex category because quality cannot be measured from infrastructure metrics alone — it requires semantic evaluation. The most scalable approach is sampled automated evaluation: for a random sample of production requests (5-10%), run an LLM-as-judge evaluator asynchronously, store the scores, and compute a rolling average. Alert when the rolling average drops below your quality floor.

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 None

Rolling Averages for Trend Detection

A single bad response does not indicate a systemic problem. Use rolling averages over a time window (e.g., the last 1 hour or last 100 requests) to detect trends. A rolling average that crosses a threshold and stays there for 10+ minutes indicates a real problem, while a brief spike might be noise. Exponentially weighted moving averages (EWMA) react faster to recent changes than simple rolling averages.

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()

Alert Routing: Slack and PagerDuty

Alerts are only useful if they reach the right person through the right channel. Route alerts by severity: quality degradation and cost budget alerts go to a Slack channel where your team monitors in business hours. Latency alerts above critical thresholds (e.g., p99 > 30 seconds) and sudden cost spikes (e.g., 10x above normal in 5 minutes) go to PagerDuty for immediate on-call escalation.

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)

The Alert Evaluation Loop

Alerting logic should run on a scheduled loop, not inline with request processing. Running alert checks asynchronously prevents them from adding latency to production requests. A background thread or scheduled job that runs every 60 seconds is sufficient for most alerting needs. Collect metrics in the request path, evaluate thresholds in the background loop.

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()

Alert Fatigue and Threshold Tuning

Alert fatigue occurs when alerts fire so frequently that the team starts ignoring them. Prevent this by: starting with conservative (high) thresholds and tightening them as you understand your baseline, requiring alerts to sustain for multiple evaluation cycles before firing, grouping related alerts into single notifications, and regularly reviewing and retiring alerts that never lead to meaningful action.

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)

Monitoring Cost Anomalies with Statistical Methods

Simple threshold alerts miss nuanced cost anomalies like a gradual 50% cost increase over two days. Use statistical anomaly detection: compute the rolling mean and standard deviation of your hourly costs, and alert when the current hour's cost is more than N standard deviations above the mean (Z-score alerting). This adapts automatically to natural usage patterns like weekday vs. weekend traffic.

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 None

Building an Ops Dashboard

Alerts are the reactive layer; a live operations dashboard is the proactive layer. Build a simple dashboard that shows real-time metrics: current p50/p99 latency, requests per minute, current quality score (rolling EWMA), daily cost to date vs. budget, and top-5 most expensive requests in the last hour. This gives your team at-a-glance visibility into system health without waiting for an alert to trigger.

Quick Check

Test your understanding of alerting on LLM application metrics from this lesson.

Lesson Recap

In this lesson you learned: three alert categories cover LLM systems — latency, cost, and quality — each requiring different instrumentation, rolling averages and EWMA detect gradual quality degradation better than threshold checks on individual requests, and alert debouncing prevents alert fatigue by requiring conditions to sustain across multiple evaluation cycles before firing. Next up we dive into prompt injection attacks and AI security.

Frequently asked questions

Is the “Alerting on Latency, Cost, and Quality Degradation” lesson free?

Yes — the full text of “Alerting on Latency, Cost, and Quality Degradation” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.

What will I learn in “Alerting on Latency, Cost, and Quality Degradation”?

Define alert thresholds on p99 latency, per-request cost, and automated quality scores, and route alerts to Slack or PagerDuty when your LLM pipeline degrades. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Engineering Academy?

No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Alerting on Latency, Cost, and Quality Degradation” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Engineering Academy lesson?

Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Why LLM Apps Are Hard to Debug
  2. Tracing with LangSmith
  3. Langfuse for Model-Agnostic Observability
  4. Alerting on Latency, Cost, and Quality Degradation
← Back to AI Engineering Academy