API Monitoring and Observability for AI Applications

You can't manage what you don't measure. AI APIs are expensive, unpredictable, and failure-prone. Without proper monitoring, costs spiral out of control, latency degrades silently, and outages surprise you at 3 AM. This guide covers the essential metrics, dashboards, and alerting patterns for production AI API monitoring.

The Four Golden Signals for AI APIs

Signal	What to Track	Why It Matters
Latency	Time to first token, total generation time	User experience, timeout risks
Traffic	Requests/min, tokens/min, active users	Capacity planning, cost forecasting
Errors	Rate of 4xx/5xx, timeout rate	Reliability, user trust
Saturation	Queue depth, rate limit proximity	Preventing cascading failures

Essential Metrics to Track

1. Cost Metrics

class CostTracker:
    def __init__(self):
        self.daily_cost = 0
        self.model_costs = {}
        self.hourly_history = []
    
    def record_request(self, model, input_tokens, output_tokens):
        # Pricing per 1K tokens (example rates)
        pricing = {
            "deepseek": {"input": 0.001, "output": 0.002},
            "zhipu": {"input": 0.001, "output": 0.002},
            "qwen": {"input": 0.0005, "output": 0.001},
            "kimi": {"input": 0.001, "output": 0.002},
            "doubao": {"input": 0.0008, "output": 0.0015}
        }
        
        cost = (input_tokens * pricing[model]["input"] + 
                output_tokens * pricing[model]["output"]) / 1000
        
        self.daily_cost += cost
        self.model_costs[model] = self.model_costs.get(model, 0) + cost
        
        return cost
    
    def get_daily_summary(self):
        return {
            "total": round(self.daily_cost, 4),
            "by_model": {k: round(v, 4) for k, v in self.model_costs.items()},
            "projected_monthly": round(self.daily_cost * 30, 2)
        }

# Usage
tracker = CostTracker()
cost = tracker.record_request("qwen", input_tokens=500, output_tokens=200)
print(f"Request cost: ${cost:.4f}")
print(f"Daily total: ${tracker.get_daily_summary()['total']:.4f}")

2. Latency Metrics

import time
from collections import deque

class LatencyTracker:
    def __init__(self, window_size=1000):
        self.latencies = deque(maxlen=window_size)
        self.ttfb_times = deque(maxlen=window_size)  # Time to first byte
    
    def record(self, start_time, first_token_time, end_time):
        ttfb = first_token_time - start_time
        total = end_time - start_time
        
        self.ttfb_times.append(ttfb)
        self.latencies.append(total)
    
    def get_stats(self):
        if not self.latencies:
            return {}
        
        sorted_lat = sorted(self.latencies)
        sorted_ttfb = sorted(self.ttfb_times)
        
        return {
            "ttfb_p50": sorted_ttfb[len(sorted_ttfb)//2],
            "ttfb_p99": sorted_ttfb[int(len(sorted_ttfb)*0.99)],
            "total_p50": sorted_lat[len(sorted_lat)//2],
            "total_p99": sorted_lat[int(len(sorted_lat)*0.99)],
            "total_avg": sum(self.latencies) / len(self.latencies)
        }

# Wrap your API calls
start = time.time()
response = call_api(...)
first_token = time.time()  # For streaming: when first chunk arrives
end = time.time()

latency_tracker.record(start, first_token, end)
print(latency_tracker.get_stats())

3. Token Usage Metrics

class TokenTracker:
    def __init__(self):
        self.total_input = 0
        self.total_output = 0
        self.by_model = {}
    
    def record(self, model, input_tokens, output_tokens):
        self.total_input += input_tokens
        self.total_output += output_tokens
        
        if model not in self.by_model:
            self.by_model[model] = {"input": 0, "output": 0}
        
        self.by_model[model]["input"] += input_tokens
        self.by_model[model]["output"] += output_tokens
    
    def get_efficiency_ratio(self):
        # Output/Input ratio: higher = more efficient (more output per input)
        if self.total_input == 0:
            return 0
        return self.total_output / self.total_input
    
    def get_report(self):
        return {
            "total_tokens": self.total_input + self.total_output,
            "input_tokens": self.total_input,
            "output_tokens": self.total_output,
            "efficiency_ratio": round(self.get_efficiency_ratio(), 2),
            "by_model": self.by_model
        }

Building a Monitoring Dashboard

Export metrics to Prometheus and visualize with Grafana:

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# Define metrics
REQUEST_COUNT = Counter('ai_api_requests_total', 'Total requests', ['model', 'status'])
REQUEST_LATENCY = Histogram('ai_api_latency_seconds', 'Request latency', ['model'])
TOKEN_USAGE = Counter('ai_api_tokens_total', 'Token usage', ['model', 'type'])
COST_DOLLARS = Counter('ai_api_cost_dollars_total', 'Total cost', ['model'])
ACTIVE_REQUESTS = Gauge('ai_api_active_requests', 'Active requests', ['model'])

def monitored_api_call(model, messages):
    ACTIVE_REQUESTS.labels(model=model).inc()
    start = time.time()
    
    try:
        response = call_api(model, messages)
        status = "success"
        
        # Record tokens
        usage = response.get("usage", {})
        TOKEN_USAGE.labels(model=model, type="input").inc(usage.get("prompt_tokens", 0))
        TOKEN_USAGE.labels(model=model, type="output").inc(usage.get("completion_tokens", 0))
        
    except Exception as e:
        status = "error"
        raise
    finally:
        REQUEST_COUNT.labels(model=model, status=status).inc()
        REQUEST_LATENCY.labels(model=model).observe(time.time() - start)
        ACTIVE_REQUESTS.labels(model=model).dec()
    
    return response

# Start metrics server on port 8001
start_http_server(8001)

Alerting Rules

Condition	Severity	Action
Error rate > 5% for 5 minutes	Critical	Page on-call, enable fallback models
p99 latency > 10s	Warning	Investigate model/provider issues
Daily cost > 2x projected	Warning	Check for runaway requests
Rate limit errors > 10/min	Warning	Reduce concurrency or upgrade plan
Zero requests for 1 hour (business hours)	Warning	Check if service is down
Token efficiency ratio < 0.3	Info	Review prompt design for verbosity

Log Aggregation

Structured logging makes debugging API issues much easier:

import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ai_api")

def log_request(model, messages, response, duration, error=None):
    log_entry = {
        "timestamp": time.time(),
        "model": model,
        "duration_ms": round(duration * 1000, 2),
        "success": error is None,
        "error": str(error) if error else None,
        "input_tokens": len(str(messages)) // 4,  # Rough estimate
        "output_tokens": len(response.get("content", "")) // 4 if response else 0,
        "trace_id": generate_trace_id()  # For distributed tracing
    }
    
    if error:
        logger.error(json.dumps(log_entry))
    else:
        logger.info(json.dumps(log_entry))

# Query logs for debugging
# cat app.log | jq 'select(.model == "deepseek" and .duration_ms > 5000)'

Health Checks and Synthetic Monitoring

Proactively detect issues before users complain:

def synthetic_health_check():
    test_prompts = [
        {"model": "deepseek", "prompt": "Say 'healthy'", "expected": "healthy"},
        {"model": "qwen", "prompt": "Say 'healthy'", "expected": "healthy"},
        {"model": "zhipu", "prompt": "Say 'healthy'", "expected": "healthy"}
    ]
    
    results = {}
    for test in test_prompts:
        try:
            start = time.time()
            response = call_api(test["model"], [{"role": "user", "content": test["prompt"]}])
            latency = time.time() - start
            
            results[test["model"]] = {
                "healthy": test["expected"] in response["content"].lower(),
                "latency_ms": round(latency * 1000, 2)
            }
        except Exception as e:
            results[test["model"]] = {"healthy": False, "error": str(e)}
    
    return results

# Run every 60 seconds
# If any model fails 3 consecutive checks, alert and switch to fallback

Cost Anomaly Detection

class CostAnomalyDetector:
    def __init__(self, window_hours=24):
        self.hourly_costs = deque(maxlen=window_hours)
    
    def add_hourly_cost(self, cost):
        self.hourly_costs.append(cost)
    
    def is_anomaly(self, current_cost):
        if len(self.hourly_costs) < 6:
            return False
        
        mean = sum(self.hourly_costs) / len(self.hourly_costs)
        std = (sum((x - mean) ** 2 for x in self.hourly_costs) / len(self.hourly_costs)) ** 0.5
        
        # Flag if current cost is 3 standard deviations above mean
        return current_cost > mean + 3 * std

# Usage
detector = CostAnomalyDetector()
hourly_cost = calculate_last_hour_cost()
if detector.is_anomaly(hourly_cost):
    send_alert(f"Cost anomaly detected: ${hourly_cost:.2f}/hour")
detector.add_hourly_cost(hourly_cost)

Conclusion

Monitoring AI APIs requires tracking cost, latency, errors, and token usage simultaneously. Start with basic counters and logs, add Prometheus metrics for dashboards, and implement alerting for critical thresholds.

The most important metric is cost per successful request — it combines efficiency, reliability, and pricing into a single business-relevant number. Track it daily, alert on anomalies, and optimize relentlessly.