LLM Observability: Monitoring, Logging, and Tracing AI APIs in Production (2026)

August 14, 2026 · 18 min read

Running AI APIs in production without observability is like flying blind. Latency spikes, cost overruns, quality degradation, and errors can go unnoticed for hours. This guide covers production-tested observability patterns for Chinese AI APIs — from basic logging to distributed tracing with OpenTelemetry.

The Three Pillars of LLM Observability

1. Metrics — Quantitative data: latency, throughput, error rates, token usage, costs
2. Logs — Event records: requests, responses, errors, model selections
3. Traces — Request flows: end-to-end latency breakdown across services

Key Metrics to Monitor

1. Request Logging and Cost Tracking

import time
import json
import uuid
from datetime import datetime
import requests

class LLMObserver:
    def __init__(self, base_url, token, log_file="/var/log/llm_requests.jsonl"):
        self.base_url = base_url
        self.token = token
        self.log_file = log_file
        self.metrics = {
            "total_requests": 0,
            "total_tokens": 0,
            "total_cost": 0.0,
            "errors": 0,
            "latencies": []
        }
    
    def call(self, model, messages, **kwargs):
        """Tracked API call with full observability"""
        request_id = str(uuid.uuid4())
        start_time = time.time()
        
        log_entry = {
            "request_id": request_id,
            "timestamp": datetime.utcnow().isoformat(),
            "model": model,
            "messages_count": len(messages),
            "input_tokens": self.estimate_tokens(messages),
            "status": "pending"
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers={"Authorization": f"Bearer {self.token}"},
                json={
                    "model": model,
                    "messages": messages,
                    **kwargs
                }
            )
            
            latency = time.time() - start_time
            data = response.json()
            
            # Extract usage
            usage = data.get("usage", {})
            output_tokens = usage.get("completion_tokens", 0)
            total_tokens = usage.get("total_tokens", 0)
            
            # Calculate cost
            cost = self.calculate_cost(model, total_tokens)
            
            # Update log entry
            log_entry.update({
                "status": "success",
                "latency_ms": round(latency * 1000, 2),
                "output_tokens": output_tokens,
                "total_tokens": total_tokens,
                "cost_usd": cost,
                "http_status": response.status_code,
                "finish_reason": data["choices"][0].get("finish_reason")
            })
            
            # Update metrics
            self.metrics["total_requests"] += 1
            self.metrics["total_tokens"] += total_tokens
            self.metrics["total_cost"] += cost
            self.metrics["latencies"].append(latency)
            
        except Exception as e:
            latency = time.time() - start_time
            log_entry.update({
                "status": "error",
                "latency_ms": round(latency * 1000, 2),
                "error": str(e),
                "error_type": type(e).__name__
            })
            self.metrics["errors"] += 1
        
        # Write to log file
        with open(self.log_file, "a") as f:
            f.write(json.dumps(log_entry) + "\n")
        
        return response
    
    def estimate_tokens(self, messages):
        """Rough token estimation"""
        total = 0
        for msg in messages:
            total += len(msg.get("content", "").split()) * 1.3
        return int(total)
    
    def calculate_cost(self, model, tokens):
        """Calculate cost based on model pricing"""
        pricing = {
            "deepseek-v4": 0.0005,
            "glm-4": 0.0007,
            "kimi-k2": 0.0003,
            "qwen-max": 0.0005,
            "doubao-pro": 0.0004
        }
        return tokens / 1000 * pricing.get(model, 0.0005)
    
    def get_summary(self):
        """Get observability summary"""
        latencies = self.metrics["latencies"]
        if not latencies:
            return {"status": "no_data"}
        
        latencies.sort()
        n = len(latencies)
        
        return {
            "total_requests": self.metrics["total_requests"],
            "total_tokens": self.metrics["total_tokens"],
            "total_cost_usd": round(self.metrics["total_cost"], 4),
            "error_rate": round(self.metrics["errors"] / n * 100, 2),
            "latency_ms": {
                "p50": round(latencies[n // 2] * 1000, 2),
                "p95": round(latencies[int(n * 0.95)] * 1000, 2),
                "p99": round(latencies[int(n * 0.99)] * 1000, 2)
            }
        }

# Usage
observer = LLMObserver("https://tokenease.io/v1", "your_api_key")

response = observer.call(
    model="deepseek-v4",
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    max_tokens=500
)

print(observer.get_summary())

2. Prometheus Metrics Export

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# Define metrics
llm_requests_total = Counter(
    "llm_requests_total",
    "Total LLM API requests",
    ["model", "status"]
)

llm_latency_seconds = Histogram(
    "llm_latency_seconds",
    "Request latency in seconds",
    ["model"],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0]
)

llm_tokens_total = Counter(
    "llm_tokens_total",
    "Total tokens processed",
    ["model", "token_type"]
)

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

llm_active_requests = Gauge(
    "llm_active_requests",
    "Currently active requests",
    ["model"]
)

def tracked_call(model, messages):
    """Call with Prometheus metrics"""
    llm_active_requests.labels(model=model).inc()
    start = time.time()
    
    try:
        response = call_llm_api(model, messages)
        status = "success"
        
        # Record token usage
        usage = response.json().get("usage", {})
        llm_tokens_total.labels(model=model, token_type="input").inc(
            usage.get("prompt_tokens", 0)
        )
        llm_tokens_total.labels(model=model, token_type="output").inc(
            usage.get("completion_tokens", 0)
        )
        
        # Record cost
        cost = calculate_cost(model, usage.get("total_tokens", 0))
        llm_cost_usd.labels(model=model).inc(cost)
        
    except Exception as e:
        status = "error"
        raise
    finally:
        latency = time.time() - start
        llm_latency_seconds.labels(model=model).observe(latency)
        llm_requests_total.labels(model=model, status=status).inc()
        llm_active_requests.labels(model=model).dec()
    
    return response

# Start metrics server
start_http_server(9090)  # Prometheus scrapes localhost:9090/metrics

3. Distributed Tracing with OpenTelemetry

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# Setup tracing
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer("llm-app")

otlp_exporter = OTLPSpanExporter(endpoint="otel-collector:4317")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

# Auto-instrument HTTP requests
RequestsInstrumentor().instrument()

@tracer.start_as_current_span("llm_chat_completion")
def chat_with_tracing(model, messages):
    """Chat completion with distributed tracing"""
    span = trace.get_current_span()
    
    # Add span attributes
    span.set_attribute("llm.model", model)
    span.set_attribute("llm.messages.count", len(messages))
    span.set_attribute("llm.input_tokens", estimate_tokens(messages))
    
    start = time.time()
    
    with tracer.start_as_current_span("llm.api_call") as api_span:
        response = requests.post(
            "https://tokenease.io/v1/chat/completions",
            headers={"Authorization": "Bearer token"},
            json={"model": model, "messages": messages}
        )
        api_span.set_attribute("http.status_code", response.status_code)
    
    latency = time.time() - start
    data = response.json()
    
    # Record output metrics
    usage = data.get("usage", {})
    span.set_attribute("llm.output_tokens", usage.get("completion_tokens", 0))
    span.set_attribute("llm.total_tokens", usage.get("total_tokens", 0))
    span.set_attribute("llm.latency_ms", latency * 1000)
    span.set_attribute("llm.finish_reason", data["choices"][0].get("finish_reason"))
    
    return response

# View traces in Jaeger/Tempo
# Each request shows: total latency, API call latency, token counts, model used

4. Quality Metrics and Evaluation

def evaluate_response_quality(query, response, model):
    """Track response quality metrics"""
    
    metrics = {
        "response_length": len(response),
        "word_count": len(response.split()),
        "estimated_read_time_seconds": len(response.split()) / 200 * 60,
    }
    
    # Check for common quality issues
    metrics["has_refusal"] = any(word in response.lower() for word in 
        ["cannot", "unable", "sorry", "i cannot", "i'm sorry"])
    
    metrics["has_code_blocks"] = "```" in response
    metrics["has_lists"] = any(line.strip().startswith(("-", "*", "1.")) 
                               for line in response.split("\n"))
    
    # Relevance score (using another LLM call)
    relevance_prompt = f"""Rate the relevance of this response to the query (0-10):
Query: {query}
Response: {response[:500]}
Score:"""
    
    try:
        relevance_response = call_llm_api("kimi-k2", [
            {"role": "user", "content": relevance_prompt}
        ])
        metrics["relevance_score"] = float(relevance_response.strip())
    except:
        metrics["relevance_score"] = None
    
    return metrics

# Log quality metrics with each request
quality = evaluate_response_quality(query, response_text, model)
log_entry["quality"] = quality

5. Alerting Rules

# Prometheus alerting rules
# /etc/prometheus/rules/llm.yml

groups:
  - name: llm_alerts
    rules:
      - alert: HighLLMLatency
        expr: histogram_quantile(0.99, llm_latency_seconds_bucket) > 10
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "LLM p99 latency > 10s"
          
      - alert: HighErrorRate
        expr: rate(llm_requests_total{status="error"}[5m]) / rate(llm_requests_total[5m]) > 0.05
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "LLM error rate > 5%"
          
      - alert: CostSpike
        expr: increase(llm_cost_usd_total[1h]) > 100
        for: 0m
        labels:
          severity: warning
        annotations:
          summary: "LLM cost > $100/hour"
          
      - alert: ModelDegraded
        expr: llm_tokens_per_second < 5
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Model throughput degraded"

6. Dashboard Configuration (Grafana)

# Key panels for LLM observability dashboard

# Panel 1: Request Rate by Model
# Query: sum(rate(llm_requests_total[5m])) by (model)
# Type: Time series

# Panel 2: Latency Percentiles
# Query: histogram_quantile(0.50, sum(rate(llm_latency_seconds_bucket[5m])) by (le, model))
# Type: Time series (add p95, p99 as additional queries)

# Panel 3: Token Usage
# Query: sum(rate(llm_tokens_total[5m])) by (token_type)
# Type: Stacked area

# Panel 4: Cost Over Time
# Query: sum(increase(llm_cost_usd_total[1h]))
# Type: Stat (current) + Time series (trend)

# Panel 5: Error Rate
# Query: rate(llm_requests_total{status="error"}[5m]) / rate(llm_requests_total[5m])
# Type: Gauge (threshold: green < 1%, yellow < 5%, red > 5%)

# Panel 6: Active Requests
# Query: sum(llm_active_requests) by (model)
# Type: Stat

7. Cost Anomaly Detection

import numpy as np
from scipy import stats

def detect_cost_anomaly(hourly_costs, threshold_z=3.0):
    """Detect unusual cost spikes using Z-score"""
    
    if len(hourly_costs) < 24:
        return None  # Need at least 24 hours of data
    
    # Calculate rolling statistics
    mean = np.mean(hourly_costs[-168:])  # 7-day average
    std = np.std(hourly_costs[-168:])
    
    current = hourly_costs[-1]
    z_score = (current - mean) / std if std > 0 else 0
    
    if z_score > threshold_z:
        return {
            "anomaly": True,
            "current_cost": current,
            "expected_cost": mean,
            "z_score": z_score,
            "severity": "high" if z_score > 5 else "medium"
        }
    
    return {"anomaly": False}

# Run every hour
hourly_costs = get_hourly_cost_history()
alert = detect_cost_anomaly(hourly_costs)

if alert and alert["anomaly"]:
    send_alert(f"Cost anomaly detected: ${alert['current_cost']:.2f} "
               f"(expected: ${alert['expected_cost']:.2f}, z={alert['z_score']:.1f})")

TokenEase Built-in Observability

TokenEase provides built-in observability features:

• Usage Dashboard — Real-time token consumption and cost tracking per API key
• Model Performance — Latency and throughput metrics per model
• Error Tracking — Automatic classification of API errors
• Cost Alerts — Configurable spend thresholds with email notifications
• Request Logs — 30-day retention of all API calls with full metadata

Next Steps

1. Start with basic request logging to understand your baseline
2. Add Prometheus metrics for real-time monitoring
3. Implement OpenTelemetry tracing for multi-service architectures
4. Set up Grafana dashboards for team visibility
5. Configure alerts for latency, errors, and cost anomalies
6. Get your TokenEase API key with built-in usage analytics

For production patterns, see our guides on API security and failover and load balancing.

Last updated: August 2026. Observability tooling evolves rapidly.