6. Monitoring and Observability

Overview

Tracking the health of your LLM system in production: latency (p50/p95/p99), error rate, cost per request, and output quality. Automatic alerts, dashboards, and observability tools built for LLMs.

Why Monitoring Is Non-Negotiable

Without monitoring, you're the last to know when something breaks:

LLM system WITHOUT monitoring:
──────────────────────────
Day 1:  New model deployed, the prompt changed slightly
Day 3:  Accuracy dropped from 94% to 71% — nobody knows
Day 7:  A user reports it as "weird answers"
Day 8:  You investigate — the problem has been live for 7 days, thousands of users affected

LLM system WITH monitoring:
──────────────────────────
Day 1:  New model deployed
Day 1, 15 minutes later:  Alert: accuracy dropped from 94% to 71%
Day 1, 20 minutes later:  Rollback executed
Day 1: Impact: 15 minutes, a few hundred requests

The 4 Golden Signals (Google SRE)

Applied to LLM systems:

For LLMs, we add a fifth signal:

Implementation: Core Metrics

import time
import threading
from collections import defaultdict, deque
from datetime import datetime
from typing import Optional
from openai import OpenAI

client = OpenAI()


class MetricsCollector:
    """
    In-memory metrics collector for LLM systems.
    Thread-safe, with sliding time windows.
    """
    
    def __init__(self, window_seconds: int = 300):  # 5 minutes default
        self.window_seconds = window_seconds
        self._lock = threading.Lock()
        
        # Metrics storage with timestamps
        self._latencies: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        self._errors: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        self._requests: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        self._tokens: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        self._costs: dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
    
    def _clean_window(self, queue: deque) -> list:
        """Returns only the elements inside the time window."""
        now = time.time()
        limit = now - self.window_seconds
        return [(ts, val) for ts, val in queue if ts > limit]
    
    def record_request(
        self,
        prompt_name: str,
        latency_ms: float,
        success: bool,
        tokens: int,
        cost: float
    ) -> None:
        """Records the metrics of one request."""
        now = time.time()
        
        with self._lock:
            self._latencies[prompt_name].append((now, latency_ms))
            self._requests[prompt_name].append((now, 1))
            self._tokens[prompt_name].append((now, tokens))
            self._costs[prompt_name].append((now, cost))
            
            if not success:
                self._errors[prompt_name].append((now, 1))
    
    def calculate_percentile(self, values: list[float], percentile: float) -> float:
        """Computes the P percentile of a list of values."""
        if not values:
            return 0.0
        
        sorted_values = sorted(values)
        idx = int(len(sorted_values) * percentile / 100)
        idx = min(idx, len(sorted_values) - 1)
        return sorted_values[idx]
    
    def current_metrics(self, prompt_name: str) -> dict:
        """Returns the metrics computed for the current time window."""
        with self._lock:
            recent_latencies = self._clean_window(self._latencies[prompt_name])
            recent_requests = self._clean_window(self._requests[prompt_name])
            recent_errors = self._clean_window(self._errors[prompt_name])
            recent_tokens = self._clean_window(self._tokens[prompt_name])
            recent_costs = self._clean_window(self._costs[prompt_name])
        
        latency_values = [v for _, v in recent_latencies]
        n_requests = len(recent_requests)
        n_errors = len(recent_errors)
        
        return {
            "prompt_name": prompt_name,
            "window_seconds": self.window_seconds,
            "n_requests": n_requests,
            "error_rate": n_errors / n_requests if n_requests > 0 else 0.0,
            "latency_p50": self.calculate_percentile(latency_values, 50),
            "latency_p95": self.calculate_percentile(latency_values, 95),
            "latency_p99": self.calculate_percentile(latency_values, 99),
            "avg_latency": sum(latency_values) / len(latency_values) if latency_values else 0,
            "total_tokens": sum(v for _, v in recent_tokens),
            "total_cost": sum(v for _, v in recent_costs),
            "cost_per_request": (
                sum(v for _, v in recent_costs) / n_requests
                if n_requests > 0 else 0.0
            ),
            "rps": n_requests / self.window_seconds  # Requests per second
        }
    
    def global_summary(self) -> dict:
        """Metrics for every prompt."""
        return {
            name: self.current_metrics(name)
            for name in self._requests.keys()
        }


# Global singleton
_metrics = MetricsCollector(window_seconds=300)


def tracked_call(
    prompt_name: str,
    prompt_template: str,
    input_text: str,
    model: str = "gpt-4o-mini"
) -> dict:
    """
    Wrapper that runs a prompt and records the metrics automatically.
    """
    start = time.time()
    success = True
    tokens = 0
    cost = 0.0
    output = ""
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{
                "role": "user",
                "content": prompt_template.format(input=input_text)
            }],
            temperature=0
        )
        
        output = response.choices[0].message.content
        tokens = response.usage.total_tokens
        
        # Cost (GPT-4o-mini)
        input_price = 0.15 / 1_000_000
        output_price = 0.60 / 1_000_000
        cost = (
            response.usage.prompt_tokens * input_price +
            response.usage.completion_tokens * output_price
        )
    
    except Exception as e:
        success = False
        output = f"ERROR: {str(e)}"
    
    finally:
        latency_ms = (time.time() - start) * 1000
        _metrics.record_request(
            prompt_name=prompt_name,
            latency_ms=latency_ms,
            success=success,
            tokens=tokens,
            cost=cost
        )
    
    return {
        "output": output,
        "latency_ms": latency_ms,
        "success": success,
        "tokens": tokens,
        "cost": cost
    }

Alerting System

from dataclasses import dataclass


@dataclass
class AlertConfig:
    """Threshold configuration for alerts."""
    # Latency
    latency_p95_max_ms: float = 3000.0    # 3 seconds
    latency_p99_max_ms: float = 5000.0    # 5 seconds
    
    # Errors
    error_rate_max: float = 0.01            # 1%
    
    # Cost
    cost_per_request_max: float = 0.005   # $0.005 per request
    
    # Quality (needs a separate evaluation)
    accuracy_min: float = 0.85
    accuracy_degradation_max: float = 0.05  # 5% degradation vs baseline


class AlertManager:
    """Manages monitoring alerts with deduplication."""
    
    def __init__(
        self,
        config: AlertConfig,
        cooldown_seconds: int = 300  # Don't repeat the same alert for 5 minutes
    ):
        self.config = config
        self.cooldown = cooldown_seconds
        self._last_alert: dict[str, float] = {}
        self._handlers: list[callable] = []
    
    def add_handler(self, handler: callable) -> None:
        """Adds a handler to process alerts (Slack, email, log, etc.)."""
        self._handlers.append(handler)
    
    def _can_alert(self, key: str) -> bool:
        """Checks whether we can send an alert (deduplication)."""
        now = time.time()
        last = self._last_alert.get(key, 0)
        
        if now - last > self.cooldown:
            self._last_alert[key] = now
            return True
        return False
    
    def verify(self, metrics: dict) -> list[dict]:
        """Checks the metrics and raises alerts if needed."""
        prompt_name = metrics.get("prompt_name", "unknown")
        alerts = []
        
        # Latency p95
        p95 = metrics.get("latency_p95", 0)
        if p95 > self.config.latency_p95_max_ms:
            key = f"{prompt_name}:latency_p95"
            if self._can_alert(key):
                alert = {
                    "type": "HIGH_LATENCY",
                    "severity": "WARNING",
                    "prompt": prompt_name,
                    "message": f"Latency p95={p95:.0f}ms is over the threshold {self.config.latency_p95_max_ms:.0f}ms",
                    "value": p95,
                    "threshold": self.config.latency_p95_max_ms
                }
                alerts.append(alert)
        
        # Error rate
        error_rate = metrics.get("error_rate", 0)
        if error_rate > self.config.error_rate_max:
            key = f"{prompt_name}:error_rate"
            if self._can_alert(key):
                alerts.append({
                    "type": "HIGH_ERROR_RATE",
                    "severity": "CRITICAL" if error_rate > 0.05 else "WARNING",
                    "prompt": prompt_name,
                    "message": f"Error rate={error_rate:.1%} is over the threshold {self.config.error_rate_max:.1%}",
                    "value": error_rate,
                    "threshold": self.config.error_rate_max
                })
        
        # Cost per request
        request_cost = metrics.get("cost_per_request", 0)
        if request_cost > self.config.cost_per_request_max:
            key = f"{prompt_name}:cost"
            if self._can_alert(key):
                alerts.append({
                    "type": "HIGH_COST",
                    "severity": "WARNING",
                    "prompt": prompt_name,
                    "message": f"Cost=${request_cost:.6f}/request is over the threshold ${self.config.cost_per_request_max:.6f}",
                    "value": request_cost,
                    "threshold": self.config.cost_per_request_max
                })
        
        # Fire the handlers
        for alert in alerts:
            for handler in self._handlers:
                try:
                    handler(alert)
                except Exception as e:
                    print(f"Error in the alert handler: {e}")
        
        return alerts
    
    def verify_periodically(self, metrics_collector: MetricsCollector, interval: int = 60) -> None:
        """Checks the metrics every N seconds in the background."""
        def loop():
            while True:
                for prompt_name in metrics_collector._requests.keys():
                    metrics = metrics_collector.current_metrics(prompt_name)
                    self.verify(metrics)
                time.sleep(interval)
        
        thread = threading.Thread(target=loop, daemon=True)
        thread.start()


# Example handlers:
def log_handler(alert: dict) -> None:
    """Logs the alert to the console."""
    severity = alert["severity"]
    emoji = "🚨" if severity == "CRITICAL" else "⚠️"
    print(f"{emoji} [{severity}] {alert['message']}")

def slack_handler(alert: dict, webhook_url: str) -> None:
    """Sends the alert to Slack."""
    import urllib.request
    import json
    
    text = f"{'🚨' if alert['severity'] == 'CRITICAL' else '⚠️'} *{alert['type']}*\n{alert['message']}"
    payload = {"text": text}
    
    req = urllib.request.Request(
        webhook_url,
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req, timeout=5)


# Usage:
config = AlertConfig(latency_p95_max_ms=2000, error_rate_max=0.02)
alert_mgr = AlertManager(config)
alert_mgr.add_handler(log_handler)
# alert_mgr.add_handler(lambda a: slack_handler(a, SLACK_WEBHOOK))
alert_mgr.verify_periodically(_metrics, interval=60)

Prometheus + Grafana Integration

For teams with monitoring infrastructure:

from prometheus_client import (
    Counter, Histogram, Gauge,
    start_http_server, REGISTRY
)

# Define Prometheus metrics
latency_histogram = Histogram(
    "llm_request_duration_milliseconds",
    "Latency of LLM requests",
    ["prompt_name", "model", "status"],
    buckets=[100, 250, 500, 1000, 2000, 3000, 5000, 10000]
)

requests_counter = Counter(
    "llm_requests_total",
    "Total LLM requests",
    ["prompt_name", "model", "status"]
)

tokens_counter = Counter(
    "llm_tokens_total",
    "Total tokens processed",
    ["prompt_name", "model", "token_type"]  # token_type: prompt | completion
)

cost_counter = Counter(
    "llm_cost_usd_total",
    "Total cost in USD",
    ["prompt_name", "model"]
)

quality_gauge = Gauge(
    "llm_quality_score",
    "Quality score (accuracy) from the last evaluation run",
    ["prompt_name"]
)


def tracked_call_prometheus(
    prompt_name: str,
    prompt_template: str,
    input_text: str,
    model: str = "gpt-4o-mini"
) -> dict:
    """LLM call with Prometheus metrics."""
    start = time.time()
    status = "success"
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt_template.format(input=input_text)}],
            temperature=0
        )
        output = response.choices[0].message.content
        
        # Record tokens
        tokens_counter.labels(
            prompt_name=prompt_name, model=model, token_type="prompt"
        ).inc(response.usage.prompt_tokens)
        
        tokens_counter.labels(
            prompt_name=prompt_name, model=model, token_type="completion"
        ).inc(response.usage.completion_tokens)
        
        # Cost
        cost = (
            response.usage.prompt_tokens * 0.15 / 1e6 +
            response.usage.completion_tokens * 0.60 / 1e6
        )
        cost_counter.labels(prompt_name=prompt_name, model=model).inc(cost)
        
    except Exception as e:
        status = "error"
        output = f"ERROR: {e}"
    
    finally:
        latency_ms = (time.time() - start) * 1000
        
        latency_histogram.labels(
            prompt_name=prompt_name, model=model, status=status
        ).observe(latency_ms)
        
        requests_counter.labels(
            prompt_name=prompt_name, model=model, status=status
        ).inc()
    
    return {"output": output, "status": status, "latency_ms": latency_ms}


# Expose the metrics at /metrics for Prometheus
def start_metrics_server(port: int = 8080) -> None:
    """Starts an HTTP server for Prometheus metrics."""
    start_http_server(port)
    print(f"Metrics available at http://localhost:{port}/metrics")

LangSmith for Observability

LangSmith (from LangChain) offers native tracing for LLMs:

import os
from langsmith import Client
from langsmith.wrappers import wrap_openai

# Configure LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__xxx"
os.environ["LANGCHAIN_PROJECT"] = "my-llm-project"

# Wrap the OpenAI client for automatic tracing
from openai import OpenAI
wrapped_client = wrap_openai(OpenAI())

def call_with_tracing(prompt: str, input_text: str) -> str:
    """
    Call with automatic tracing in LangSmith.
    Every call shows up in the dashboard with: latency, tokens, cost, input/output.
    """
    response = wrapped_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt.format(input=input_text)}],
        temperature=0
    )
    return response.choices[0].message.content


# Evaluation built into LangSmith:
def evaluate_with_langsmith(
    prompt_name: str,
    golden_set: list[dict],
    prompt_template: str
) -> None:
    """
    Creates a dataset in LangSmith and evaluates the prompt.
    The results show up in the dashboard with a historical comparison.
    """
    langsmith_client = Client()
    
    # Create the dataset if it doesn't exist
    dataset_name = f"golden_set_{prompt_name}"
    
    try:
        dataset = langsmith_client.create_dataset(dataset_name)
        for ex in golden_set:
            langsmith_client.create_example(
                inputs={"input": ex["input"]},
                outputs={"expected": ex["expected_output"]},
                dataset_id=dataset.id
            )
    except Exception:
        pass  # The dataset already exists
    
    # The automatic evaluation shows up in LangSmith with metrics and traces

Quality Monitoring: Continuous Evaluation in Production

import random
import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

class QualityMonitor:
    """
    Monitors LLM quality in production through sampling.
    
    Instead of evaluating every request (expensive), it evaluates a random sample.
    """
    
    def __init__(
        self,
        baseline_accuracy: float,
        sample_rate: float = 0.05,  # Evaluate 5% of traffic
        min_sample_size: int = 50   # Minimum requests needed to evaluate
    ):
        self.baseline = baseline_accuracy
        self.sample_rate = sample_rate
        self.min_sample_size = min_sample_size
        self._buffer: list[dict] = []
        self._quality_scores: list[float] = []
    
    def should_sample(self) -> bool:
        """Decides whether this request should be evaluated."""
        return random.random() < self.sample_rate
    
    def add_to_buffer(self, input_text: str, output: str, expected: str | None = None) -> None:
        """Adds a request to the evaluation buffer."""
        if self.should_sample():
            self._buffer.append({
                "input": input_text,
                "output": output,
                "expected": expected,
                "timestamp": time.time()
            })
    
    async def evaluate_buffer(self, judge_prompt: str) -> float:
        """
        Evaluates the accumulated buffer with LLM-as-judge.
        Returns the average quality score.
        """
        if len(self._buffer) < self.min_sample_size:
            return None  # Not enough of a sample
        
        semaphore = asyncio.Semaphore(5)
        
        async def evaluate_one(item):
            async with semaphore:
                if item.get("expected"):
                    # Exact match if there's an expected value
                    return 1.0 if item["output"].strip().lower() == item["expected"].strip().lower() else 0.0
                else:
                    # LLM-as-judge if there's no expected value
                    prompt = judge_prompt.format(
                        input=item["input"],
                        output=item["output"]
                    )
                    r = await async_client.chat.completions.create(
                        model="gpt-4o-mini",
                        messages=[{"role": "user", "content": prompt}],
                        temperature=0,
                        max_tokens=5
                    )
                    raw = r.choices[0].message.content.strip()
                    try:
                        return float(raw.split()[0]) / 10.0
                    except:
                        return 0.5
        
        scores = await asyncio.gather(*[evaluate_one(item) for item in self._buffer])
        quality = sum(scores) / len(scores)
        
        self._quality_scores.append(quality)
        self._buffer.clear()
        
        # Check degradation vs baseline
        degradation = self.baseline - quality
        if degradation > 0.05:  # > 5% degradation
            return quality, f"⚠️ Quality degradation: {quality:.2%} vs baseline {self.baseline:.2%}"
        
        return quality, "✅ Quality within the normal range"

Text Dashboard for the Terminal

def generate_terminal_dashboard(
    metrics_collector: MetricsCollector,
    prompt_names: list[str]
) -> str:
    """
    Generates a text dashboard to display in the terminal.
    Handy for quick debugging.
    """
    lines = [
        "╔══════════════════════════════════════════════════════════╗",
        "║          LLM MONITORING DASHBOARD                        ║",
        f"║  Updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}                             ║",
        "╠══════════════════════════════════════════════════════════╣",
    ]
    
    for prompt_name in prompt_names:
        m = metrics_collector.current_metrics(prompt_name)
        
        error_rate = m['error_rate']
        p95 = m['latency_p95']
        
        # Status indicators
        latency_ok = "✅" if p95 < 2000 else "⚠️" if p95 < 4000 else "❌"
        error_ok = "✅" if error_rate < 0.01 else "⚠️" if error_rate < 0.05 else "❌"
        
        lines.extend([
            f"║  📊 {prompt_name[:30]:30s}                ║",
            f"║     Requests: {m['n_requests']:6d}  |  RPS: {m['rps']:.2f}              ║",
            f"║     {latency_ok} Latency p50/p95/p99: {m['latency_p50']:.0f}/{p95:.0f}/{m['latency_p99']:.0f}ms     ║",
            f"║     {error_ok} Error rate: {error_rate:.2%}                        ║",
            f"║     💰 Cost/req: ${m['cost_per_request']:.6f}                     ║",
            "║  ─────────────────────────────────────────────────────║",
        ])
    
    lines.append("╚══════════════════════════════════════════════════════════╝")
    return "\n".join(lines)


# Auto-refresh every 30 seconds:
def dashboard_loop(metrics_collector: MetricsCollector, prompt_names: list[str]) -> None:
    """Loop that refreshes the dashboard in the terminal."""
    import os
    
    while True:
        os.system("clear")  # Clear the terminal
        print(generate_terminal_dashboard(metrics_collector, prompt_names))
        time.sleep(30)

Structured Logging

import logging
import json

class LLMLogger:
    """Structured logger for LLMs. Compatible with ELK, Datadog, CloudWatch."""
    
    def __init__(self, name: str = "llm_system"):
        self.logger = logging.getLogger(name)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    
    def log_request(
        self,
        prompt_name: str,
        prompt_version: str,
        input_text: str,
        output: str,
        latency_ms: float,
        tokens: int,
        cost: float,
        success: bool,
        request_id: str | None = None
    ) -> None:
        """Logs a request in structured JSON format."""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "request_id": request_id,
            "prompt_name": prompt_name,
            "prompt_version": prompt_version,
            "input_preview": input_text[:100],
            "output_preview": output[:200],
            "latency_ms": latency_ms,
            "tokens": tokens,
            "cost_usd": cost,
            "success": success,
            "level": "INFO" if success else "ERROR"
        }
        
        self.logger.info(json.dumps(log_entry))
    
    def log_alert(self, alert_type: str, prompt_name: str, message: str, details: dict) -> None:
        """Logs an alert in structured format."""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "level": "ALERT",
            "type": alert_type,
            "prompt_name": prompt_name,
            "message": message,
            **details
        }
        self.logger.warning(json.dumps(entry))


llm_logger = LLMLogger()

Troubleshooting

Problem 1: Excessive monitoring overhead

Symptom: Monitoring adds 50ms of latency to every request.

Cause: Metrics processed in the request path.

Solution:

# Use async/queue to process metrics outside the request path
import asyncio
from queue import Queue

metrics_queue = Queue(maxsize=10000)

def record_async(metrics_data: dict) -> None:
    """Pushes metrics onto the queue without blocking."""
    try:
        metrics_queue.put_nowait(metrics_data)
    except:
        pass  # If the queue is full, drop it (prefer losing metrics over latency)

def processor_loop():
    """Processes metrics in the background."""
    while True:
        try:
            data = metrics_queue.get(timeout=1)
            _metrics.record_request(**data)
        except:
            pass

import threading
threading.Thread(target=processor_loop, daemon=True).start()

Problem 2: Noisy alerts

Symptom: 50+ alerts a day from momentary spikes.

Cause: Thresholds evaluated over too short a window.

Solution:

# Use a longer sliding window and a percentage of time exceeded
def evaluate_robust_threshold(
    values: list[float],
    threshold: float,
    max_pct_time: float = 0.05  # Alert if it's over the threshold more than 5% of the time
) -> bool:
    """Only alerts if the threshold is exceeded more than max_pct_time."""
    if not values:
        return False
    
    exceeded = sum(1 for v in values if v > threshold)
    return exceeded / len(values) > max_pct_time

Problem 3: Quality monitoring is expensive

Symptom: The quality evaluation sample doubles the cost.

Solution:

# Lower the sample rate or use exact match when there's an expected_output
def efficient_quality(buffer: list[dict]) -> float:
    """Prefers exact match (free) over LLM-as-judge (expensive)."""
    with_expected = [b for b in buffer if b.get("expected")]
    without_expected = [b for b in buffer if not b.get("expected")]
    
    scores = []
    
    # Exact match for the ones with an expected value (free)
    for item in with_expected:
        correct = item["output"].strip().lower() == item["expected"].strip().lower()
        scores.append(1.0 if correct else 0.0)
    
    # LLM-as-judge only for the ones without an expected value, on a 10% sample
    sample_without_expected = random.sample(without_expected, max(1, int(len(without_expected) * 0.1)))
    # ... evaluate with the LLM
    
    return sum(scores) / len(scores) if scores else 0.5

Exercises

Exercise 1: Implement a basic latency tracker

Create a class that tracks the latency of LLM calls and computes p50 and p95:

from collections import deque
from openai import OpenAI
import time

client = OpenAI()

class LatencyTracker:
    def __init__(self, max_samples: int = 1000):
        self._latencies = deque(maxlen=max_samples)
    
    def record(self, latency_ms: float):
        self._latencies.append(latency_ms)
    
    def percentile(self, p: float) -> float:
        if not self._latencies:
            return 0.0
        sorted_latencies = sorted(self._latencies)
        idx = min(int(len(sorted_latencies) * p / 100), len(sorted_latencies) - 1)
        return sorted_latencies[idx]
    
    def summary(self) -> dict:
        return {
            "n": len(self._latencies),
            "p50": f"{self.percentile(50):.0f}ms",
            "p95": f"{self.percentile(95):.0f}ms",
            "p99": f"{self.percentile(99):.0f}ms",
            "average": f"{sum(self._latencies)/len(self._latencies):.0f}ms" if self._latencies else "0ms"
        }

tracker = LatencyTracker()

def call_tracked(prompt: str) -> str:
    start = time.time()
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    tracker.record((time.time() - start) * 1000)
    return r.choices[0].message.content

# Make 5 calls and look at the statistics
for i in range(5):
    call_tracked(f"Say the number {i}")

print(tracker.summary())

Exercise 2: Error rate alert

Create a system that alerts when the error rate goes over 5% in the last 10 calls:

from collections import deque
import time

class ErrorRateAlert:
    def __init__(self, window: int = 10, threshold: float = 0.05):
        self._results = deque(maxlen=window)
        self.threshold = threshold
        self._alerts_sent = 0
    
    def record(self, success: bool):
        self._results.append(1 if success else 0)
        
        if len(self._results) >= 5:  # At least 5 to compute
            error_rate = 1 - sum(self._results) / len(self._results)
            
            if error_rate > self.threshold:
                self._alerts_sent += 1
                print(f"🚨 ALERT #{self._alerts_sent}: Error rate={error_rate:.1%} > threshold {self.threshold:.1%}")
                print(f"   Last {len(self._results)} calls: {list(self._results)}")

alert = ErrorRateAlert(window=10, threshold=0.05)

# Simulate calls with a few errors
import random
for i in range(20):
    success = random.random() > 0.15  # 15% error rate (over the 5% threshold)
    alert.record(success)
    print(f"Request {i+1}: {'OK' if success else 'ERROR'}")

Summary

• 4+1 signals: Latency, traffic, errors, saturation + quality (LLM-specific)
• MetricsCollector: Thread-safe, with sliding windows for real-time metrics
• Alerts: Thresholds with a cooldown to avoid noise — only alert when it matters
• Prometheus: For teams with Grafana — histograms, counters, gauges
• LangSmith: Automatic tracing by wrapping the OpenAI client
• Quality monitoring: Sampling 5% of traffic to evaluate quality without excessive cost
• Structured logging: JSON compatible with ELK, Datadog, CloudWatch

Additional resources

1. LangSmith — Observability and tracing for LLMs
2. Prometheus — Open source monitoring system
3. OpenTelemetry — Instrumentation standard
4. Datadog LLM Observability — LLM-specific monitoring
5. Grafana — Dashboards for Prometheus
6. Langfuse — Open source LLM observability with evaluation