How to Set Up Alerting for LLM Latency Spikes

How to Set Up Alerting for LLM Latency Spikes

Large Language Models (LLMs) have become the backbone of many modern applications, from customer support chatbots to code assistants. While accuracy and cost often dominate the conversation, latency is the metric that users feel most directly. When an LLM endpoint suddenly takes 30 seconds to respond instead of the usual 2 seconds, users abandon the session, downstream services time out, and trust evaporates. This tutorial walks you through everything you need to know to detect, alert on, and respond to LLM latency spikes before they become incidents.

What Is an LLM Latency Spike?

An LLM latency spike is a sudden, statistically significant increase in the time it takes for a model to process a request and return a response. Unlike a slow drift caused by gradual traffic growth, a spike is abrupt. It typically manifests as a sharp jump in metrics such as time-to-first-token (TTFT), total generation time, or end-to-end request duration measured at the application layer.

Latency in LLM systems is multi-layered. A single request passes through your application code, a load balancer, an API gateway, the model provider's inference infrastructure, and back. Each hop contributes to total latency. A spike could originate at any of these layers, which is why alerting must be granular enough to point you toward the root cause.

Why Alerting on Latency Spikes Matters

• User experience degradation: Studies show that response times above 3 seconds lead to measurable drop-off in engagement. LLM chat interfaces are particularly sensitive because users expect conversational pacing.
• Cascading failures: Many applications hold open connections while waiting for LLM responses. A latency spike can exhaust connection pools, thread pools, and memory, taking down unrelated services.
• Cost amplification: When requests slow down, retry logic and timeout-based fallbacks often fire duplicate requests, multiplying your token spend and API costs.
• SLA and contractual obligations: If you offer an LLM-powered product with uptime or latency guarantees, unmonitored spikes can result in SLA breaches and financial penalties.
• Early warning for broader incidents: Latency spikes frequently precede full outages. A provider experiencing capacity issues will often slow down before going completely dark.

Key Metrics to Monitor

Before configuring alerts, you need to instrument your LLM calls to capture the right metrics. The four most important measurements are:

• Time to First Token (TTFT): The delay between sending the request and receiving the first token of the response. High TTFT usually indicates provider-side queuing or model loading delays.
• Token Generation Rate: Tokens per second during streaming. A drop here suggests the inference server is under load.
• Total Request Duration: End-to-end time from request dispatch to full response receipt. This is what the user actually experiences.
• Queue Depth: If you batch or queue requests internally, a growing backlog is a leading indicator of latency problems.

You should track these as both averages and percentiles. Averages hide tail latency, which is where spikes live. Focus on p95 and p99 values for alerting thresholds.

Instrumenting Your LLM Calls

The first step is wrapping your LLM client calls with timing instrumentation. Below is a Python example using a lightweight metrics collector that captures the key latency metrics for each request.

import time
import statistics
from dataclasses import dataclass, field
from collections import deque
from threading import Lock

@dataclass
class LatencyMetrics:
    ttft_samples: deque = field(default_factory=lambda: deque(maxlen=1000))
    total_duration_samples: deque = field(default_factory=lambda: deque(maxlen=1000))
    token_rate_samples: deque = field(default_factory=lambda: deque(maxlen=1000))
    _lock: Lock = field(default_factory=Lock)

    def record(self, ttft: float, total_duration: float, tokens_generated: int):
        with self._lock:
            self.ttft_samples.append(ttft)
            self.total_duration_samples.append(total_duration)
            if total_duration > ttft and tokens_generated > 0:
                rate = tokens_generated / (total_duration - ttft)
                self.token_rate_samples.append(rate)

    def percentile(self, samples: deque, p: float) -> float:
        if not samples:
            return 0.0
        sorted_samples = sorted(samples)
        index = int(len(sorted_samples) * p / 100)
        index = min(index, len(sorted_samples) - 1)
        return sorted_samples[index]

    def get_snapshot(self) -> dict:
        with self._lock:
            return {
                "ttft_p95": self.percentile(self.ttft_samples, 95),
                "ttft_p99": self.percentile(self.ttft_samples, 99),
                "duration_p95": self.percentile(self.total_duration_samples, 95),
                "duration_p99": self.percentile(self.total_duration_samples, 99),
                "token_rate_p5": self.percentile(self.token_rate_samples, 5),
                "sample_count": len(self.total_duration_samples),
            }

# Global metrics instance
metrics = LatencyMetrics()

def call_llm_with_metrics(client, prompt: str, model: str = "gpt-4") -> str:
    start_time = time.monotonic()
    first_token_time = None
    token_count = 0

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )

    full_response = []
    for chunk in response:
        if chunk.choices and chunk.choices[0].delta.content:
            if first_token_time is None:
                first_token_time = time.monotonic()
            content = chunk.choices[0].delta.content
            full_response.append(content)
            token_count += 1

    total_duration = time.monotonic() - start_time
    ttft = (first_token_time - start_time) if first_token_time else total_duration

    metrics.record(
        ttft=ttft,
        total_duration=total_duration,
        tokens_generated=token_count,
    )

    return "".join(full_response)

This instrumentation captures per-request timing and maintains a rolling window of samples. The get_snapshot method gives you a point-in-time view of your latency percentiles, which is what your alerting logic will evaluate.

Building the Alerting Engine

With metrics in place, you need an alerting engine that evaluates thresholds and triggers notifications. The design below uses a sliding window approach with configurable thresholds and cooldown periods to prevent alert fatigue.

import time
import smtplib
from email.mime.text import MIMEText
from dataclasses import dataclass
from typing import Callable, List

@dataclass
class AlertRule:
    name: str
    metric_key: str
    threshold: float
    comparison: str  # "gt" or "lt"
    window_seconds: int
    min_samples: int
    cooldown_seconds: int
    last_fired: float = 0.0

@dataclass
class Alert:
    rule_name: str
    metric_value: float
    threshold: float
    timestamp: float
    message: str

class LatencyAlerter:
    def __init__(self, metrics_collector):
        self.metrics = metrics_collector
        self.rules: List[AlertRule] = []
        self.alert_history: List[Alert] = []
        self.notification_channels: List[Callable] = []

    def add_rule(self, rule: AlertRule):
        self.rules.append(rule)

    def add_notification_channel(self, channel: Callable[[Alert], None]):
        self.notification_channels.append(channel)

    def evaluate(self) -> List[Alert]:
        snapshot = self.metrics.get_snapshot()
        fired_alerts = []
        now = time.time()

        for rule in self.rules:
            if rule.metric_key not in snapshot:
                continue

            value = snapshot[rule.metric_key]
            if snapshot.get("sample_count", 0) < rule.min_samples:
                continue

            condition_met = (
                value > rule.threshold if rule.comparison == "gt"
                else value < rule.threshold
            )

            if condition_met and (now - rule.last_fired) > rule.cooldown_seconds:
                alert = Alert(
                    rule_name=rule.name,
                    metric_value=value,
                    threshold=rule.threshold,
                    timestamp=now,
                    message=(
                        f"[LATENCY ALERT] {rule.name}: "
                        f"{rule.metric_key} = {value:.2f}s "
                        f"(threshold: {rule.threshold:.2f}s)"
                    ),
                )
                fired_alerts.append(alert)
                self.alert_history.append(alert)
                rule.last_fired = now

                for channel in self.notification_channels:
                    try:
                        channel(alert)
                    except Exception as e:
                        print(f"Notification channel failed: {e}")

        return fired_alerts

The alerter evaluates each rule against the current metrics snapshot. The min_samples field prevents false alerts when the system has only processed a handful of requests, and the cooldown_seconds field ensures you do not get spammed with repeated alerts for the same ongoing condition.

Defining Alert Rules

Choosing the right thresholds is critical. Set them too low and you will drown in noise. Set them too high and you will miss real problems. A good starting strategy is to baseline your metrics for one to two weeks, then set thresholds at roughly 2x your p95 values.

alerter = LatencyAlerter(metrics)

# Alert if p99 time-to-first-token exceeds 5 seconds
alerter.add_rule(AlertRule(
    name="TTFT Spike",
    metric_key="ttft_p99",
    threshold=5.0,
    comparison="gt",
    window_seconds=300,
    min_samples=20,
    cooldown_seconds=300,
))

# Alert if p95 total duration exceeds 15 seconds
alerter.add_rule(AlertRule(
    name="Total Duration Spike",
    metric_key="duration_p95",
    threshold=15.0,
    comparison="gt",
    window_seconds=300,
    min_samples=20,
    cooldown_seconds=300,
))

# Alert if token generation rate drops below 10 tokens/sec
alerter.add_rule(AlertRule(
    name="Slow Token Generation",
    metric_key="token_rate_p5",
    threshold=10.0,
    comparison="lt",
    window_seconds=300,
    min_samples=20,
    cooldown_seconds=600,
))

Setting Up Notification Channels

An alert that nobody sees is useless. You should route latency alerts to the channels your team actively monitors. Below are implementations for two common channels: Slack webhooks and email.

Slack Notifications

import json
import urllib.request

def make_slack_notifier(webhook_url: str):
    def notifier(alert: Alert):
        payload = {
            "text": alert.message,
            "attachments": [{
                "color": "danger",
                "fields": [
                    {"title": "Rule", "value": alert.rule_name, "short": True},
                    {"title": "Value", "value": f"{alert.metric_value:.2f}s", "short": True},
                    {"title": "Threshold", "value": f"{alert.threshold:.2f}s", "short": True},
                    {"title": "Timestamp", "value": str(alert.timestamp), "short": True},
                ],
            }],
        }
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(
            webhook_url,
            data=data,
            headers={"Content-Type": "application/json"},
        )
        urllib.request.urlopen(req)
    return notifier

slack_notifier = make_slack_notifier("https://hooks.slack.com/services/YOUR/WEBHOOK/URL")
alerter.add_notification_channel(slack_notifier)

Email Notifications

def make_email_notifier(smtp_host: str, smtp_port: int, sender: str, recipients: list):
    def notifier(alert: Alert):
        msg = MIMEText(alert.message)
        msg["Subject"] = f"LLM Latency Alert: {alert.rule_name}"
        msg["From"] = sender
        msg["To"] = ", ".join(recipients)

        with smtplib.SMTP(smtp_host, smtp_port) as server:
            server.sendmail(sender, recipients, msg.as_string())
    return notifier

email_notifier = make_email_notifier(
    smtp_host="smtp.yourcompany.com",
    smtp_port=587,
    sender="alerts@yourcompany.com",
    recipients=["oncall@yourcompany.com"],
)
alerter.add_notification_channel(email_notifier)

Running the Alerting Loop

The alerting engine needs to run on a schedule. A common pattern is a background thread that evaluates rules every 30 to 60 seconds. Here is a complete loop that ties everything together:

import threading
import time

def alerting_loop(alerter: LatencyAlerter, interval: int = 60):
    def run():
        while True:
            try:
                alerts = alerter.evaluate()
                for alert in alerts:
                    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {alert.message}")
            except Exception as e:
                print(f"Alerting loop error: {e}")
            time.sleep(interval)

    thread = threading.Thread(target=run, daemon=True)
    thread.start()
    return thread

# Start the alerting loop
alert_thread = alerting_loop(alerter, interval=60)

# Simulate traffic
from openai import OpenAI
client = OpenAI()

while True:
    try:
        call_llm_with_metrics(client, "Summarize the benefits of monitoring.")
    except Exception as e:
        print(f"LLM call failed: {e}")
    time.sleep(1)

Integrating with Observability Platforms

While a custom alerting engine is useful for understanding the mechanics, most production systems integrate with established observability platforms. Below is an example of exporting your LLM latency metrics to Prometheus, which pairs naturally with Alertmanager for routing.

from prometheus_client import CollectorRegistry, Gauge, start_http_server

registry = CollectorRegistry()

ttft_p95 = Gauge(
    "llm_ttft_p95_seconds",
    "95th percentile time to first token",
    registry=registry,
)
ttft_p99 = Gauge(
    "llm_ttft_p99_seconds",
    "99th percentile time to first token",
    registry=registry,
)
duration_p95 = Gauge(
    "llm_duration_p95_seconds",
    "95th percentile total request duration",
    registry=registry,
)
duration_p99 = Gauge(
    "llm_duration_p99_seconds",
    "99th percentile total request duration",
    registry=registry,
)
token_rate_p5 = Gauge(
    "llm_token_rate_p5",
    "5th percentile token generation rate (tokens/sec)",
    registry=registry,
)

def export_metrics_to_prometheus(metrics_collector):
    snapshot = metrics_collector.get_snapshot()
    ttft_p95.set(snapshot["ttft_p95"])
    ttft_p99.set(snapshot["ttft_p99"])
    duration_p95.set(snapshot["duration_p95"])
    duration_p99.set(snapshot["duration_p99"])
    token_rate_p5.set(snapshot["token_rate_p5"])

# Start Prometheus metrics endpoint
start_http_server(8000, registry=registry)

# In your alerting loop, also export metrics
def alerting_loop_with_prometheus(alerter, metrics, interval=60):
    while True:
        export_metrics_to_prometheus(metrics)
        alerter.evaluate()
        time.sleep(interval)

With metrics exported to Prometheus, you define alerting rules in a YAML configuration file that Alertmanager evaluates:

# prometheus_alerts.yml
groups:
  - name: llm_latency_alerts
    rules:
      - alert: LLMTTFTSpike
        expr: llm_ttft_p99_seconds > 5
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "LLM time-to-first-token p99 exceeds 5 seconds"
          description: "TTFT p99 is {{ $value }}s on {{ $labels.instance }}"

      - alert: LLMTotalDurationSpike
        expr: llm_duration_p95_seconds > 15
        for: 3m
        labels:
          severity: critical
        annotations:
          summary: "LLM total duration p95 exceeds 15 seconds"
          description: "Duration p95 is {{ $value }}s"

      - alert: LLMSlowTokenGeneration
        expr: llm_token_rate_p5 < 10
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "LLM token generation rate dropped below 10 tokens/sec"

The for field in each rule adds a sustained-condition requirement, meaning the threshold must be breached continuously for the specified duration before the alert fires. This filters out transient blips.

Best Practices

• Use percentile-based thresholds, not averages. Averages mask tail latency. If 95% of your requests complete in 1 second but 5% take 30 seconds, your average looks healthy while a significant portion of users suffer.
• Require sustained breaches before firing. A single slow request should not trigger a page. Use a for duration or a minimum sample count to ensure the spike represents a real trend.
• Set tiered severity levels. A p95 breach might be a warning sent to a Slack channel, while a p99 breach sustained for 5 minutes pages the on-call engineer. This prevents alert fatigue while ensuring serious issues get immediate attention.
• Include context in alert messages. Every alert should contain the metric value, the threshold, the time window, and a link to the relevant dashboard. Responders should not have to hunt for context.
• Monitor per-model and per-provider. If you use multiple models or providers, track latency separately. A spike in one provider should not be masked by healthy metrics from another.
• Track error rates alongside latency. Latency spikes and error spikes often correlate. An alerting system that considers both provides a more complete picture of system health.
• Implement automatic fallback. When latency alerts fire, your application should have a fallback path, such as switching to a smaller model, a cached response, or a different provider. Alerting is most valuable when paired with automated remediation.
• Regularly review and tune thresholds. As your traffic patterns and model usage evolve, your baseline latency will shift. Review threshold settings monthly and adjust based on observed data.
• Test your alerting pipeline. Periodically inject artificial latency to verify that alerts fire and notifications reach the right people. An untested alerting system is a liability.

Conclusion

Setting up alerting for LLM latency spikes is a multi-step process that begins with proper instrumentation, moves through thoughtful threshold design, and culminates in a reliable notification pipeline. By tracking the right metrics — time to first token, total duration, and token generation rate — at the percentile level, you gain visibility into the tail latency that actually affects users. Combining a custom alerting engine with established platforms like Prometheus and Alertmanager gives you both flexibility and production-grade reliability. The key is to treat latency as a first-class signal, tune your thresholds against real baselines, and ensure that every alert is actionable, contextual, and routed to someone who can respond. With these practices in place, you can catch latency spikes early, protect your user experience, and prevent minor slowdowns from escalating into full-scale incidents.