When you integrate AI APIs into production systems, visibility is everything. Imagine this: it's 2 AM and your AI-powered customer service bot stops responding. You open your logs and see ConnectionError: timeout with no context—no request ID, no token count, no latency trace. You're flying blind. This guide shows you how to structure AI API logging so every request becomes a traceable, debuggable, and optimizable data point.
Why Structured Logging Matters for AI APIs
AI API calls are uniquely challenging: they involve network latency, token consumption (which directly impacts cost), model selection, and non-deterministic responses. Traditional print() statements won't cut it. You need structured logs that capture the full request-response lifecycle.
At HolySheheep AI, we process millions of API calls daily with sub-50ms latency and real-time observability built into our infrastructure. Our unified API supports 100+ models including GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, and DeepSeek V3.2—with prices starting at just $0.42 per million tokens for DeepSeek V3.2, compared to $8 for GPT-4.1.
Setting Up Structured Logging
1. Choose a Logging Format: JSON is Non-Negotiable
For AI API observability, JSON logs are essential because they are machine-parseable and work seamlessly with log aggregation tools like ELK Stack, Datadog, and Loki.
import logging
import json
from datetime import datetime
import uuid
class StructuredAIPLogger:
"""Structured logger for AI API requests with full observability."""
def __init__(self, service_name: str, log_level: int = logging.INFO):
self.logger = logging.getLogger(service_name)
self.logger.setLevel(log_level)
# Console handler with JSON output
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(handler)
def log_request(self,
request_id: str,
model: str,
base_url: str,
endpoint: str,
prompt_tokens: int,
completion_tokens: int,
latency_ms: float,
status_code: int,
error: str = None):
"""Log a complete AI API request with all metadata."""
log_entry = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"request_id": request_id,
"event_type": "ai_api_request",
"service": "holysheep-proxy",
"model": model,
"base_url": base_url,
"endpoint": endpoint,
"tokens": {
"prompt": prompt_tokens,
"completion": completion_tokens,
"total": prompt_tokens + completion_tokens
},
"latency_ms": latency_ms,
"status_code": status_code,
"error": error,
"cost_usd": self._calculate_cost(model, prompt_tokens, completion_tokens)
}
self.logger.info(json.dumps(log_entry))
def _calculate_cost(self, model: str, prompt: int, completion: int) -> float:
"""Calculate cost based on model pricing per million tokens."""
pricing = {
"gpt-4.1": {"prompt": 2.0, "completion": 8.0},
"claude-sonnet-4.5": {"prompt": 3.0, "completion": 15.0},
"gemini-2.5-flash": {"prompt": 0.35, "completion": 0.70},
"deepseek-v3.2": {"prompt": 0.14, "completion": 0.42}
}
if model not in pricing:
return 0.0
rates = pricing[model]
return ((prompt / 1_000_000) * rates["prompt"] +
(completion / 1_000_000) * rates["completion"])
Usage example
ai_logger = StructuredAIPLogger("customer-support-bot")
ai_logger.log_request(
request_id="req_abc123xyz",
model="deepseek-v3.2",
base_url="https://api.holysheep.ai/v1",
endpoint="/chat/completions",
prompt_tokens=150,
completion_tokens=85,
latency_ms=127.4,
status_code=200
)
2. Implement Request Tracing with Correlation IDs
Every AI API request should carry a unique correlation ID that propagates through your entire system. This enables end-to-end tracing from user action to AI response.
import httpx
import asyncio
from contextvars import ContextVar
from typing import Optional
import uuid
Thread-safe correlation ID storage
correlation_id: ContextVar[str] = ContextVar('correlation_id')
class HolySheepAIClient:
"""Production-ready AI API client with full observability."""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self._client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0, connect=5.0),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
)
async def chat_completions(
self,
model: str,
messages: list,
trace: bool = True
) -> dict:
"""Send chat completion request with full tracing."""
# Generate or extract correlation ID
req_id = correlation_id.get(str(uuid.uuid4()))
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Request-ID": req_id,
"X-Correlation-ID": req_id
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2000
}
start_time = asyncio.get_event_loop().time()
try:
response = await self._client.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
)
latency_ms = (asyncio.get_event_loop().time() - start_time) * 1000
# Structured log output
log_data = {
"request_id": req_id,
"model": model,
"latency_ms": round(latency_ms, 2),
"status_code": response.status_code,
"tokens_used": response.json().get("usage", {}),
"success": response.status_code == 200
}
print(f"[AI_REQUEST] {log_data}")
response.raise_for_status()
return response.json()
except httpx.TimeoutException as e:
print(f"[AI_ERROR] {req_id} - Timeout: {str(e)}")
raise
except httpx.HTTPStatusError as e:
print(f"[AI_ERROR] {req_id} - HTTP {e.response.status_code}: {e.response.text}")
raise
Example usage
async def main():
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Set correlation ID for this request chain
correlation_id.set("user_123_session_456")
result = await client.chat_completions(
model="deepseek-v3.2",
messages=[{"role": "user", "content": "Explain observability in 2 sentences."}]
)
print(f"Response: {result['choices'][0]['message']['content']}")
asyncio.run(main())
Building a Real-Time Observability Dashboard
Raw logs are valuable, but you need aggregated metrics for actionable insights. Here's a simple metrics collector that tracks the KPIs that matter most for AI API usage.
from dataclasses import dataclass, field
from collections import defaultdict
from datetime import datetime, timedelta
import threading
import time
@dataclass
class APIMetrics:
"""Aggregated metrics for AI API observability."""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_tokens: int = 0
total_latency_ms: float = 0.0
total_cost_usd: float = 0.0
error_counts: dict = field(default_factory=lambda: defaultdict(int))
model_usage: dict = field(default_factory=lambda: defaultdict(int))
def record_request(self, latency_ms: float, status: int,
tokens: int, cost: float, model: str, error: str = None):
self.total_requests += 1
self.total_tokens += tokens
self.total_latency_ms += latency_ms
self.total_cost_usd += cost
self.model_usage[model] += 1
if status == 200:
self.successful_requests += 1
else:
self.failed_requests += 1
if error:
self.error_counts[error] += 1
def get_report(self) -> dict:
success_rate = (self.successful_requests / self.total_requests * 100)
if self.total_requests > 0 else 0
avg_latency = self.total_latency_ms / self.total_requests
if self.total_requests > 0 else 0
return {
"period": "last_hour",
"total_requests": self.total_requests,
"success_rate_pct": round(success_rate, 2),
"avg_latency_ms": round(avg_latency, 2),
"total_tokens": self.total_tokens,
"total_cost_usd": round(self.total_cost_usd, 4),
"top_models": sorted(self.model_usage.items(),
key=lambda x: x[1], reverse=True)[:5],
"error_breakdown": dict(self.error_counts)
}
class MetricsCollector:
"""Thread-safe metrics collector for production monitoring."""
def __init__(self, flush_interval: int = 60):
self.metrics = APIMetrics()
self._lock = threading.Lock()
self._flush_interval = flush_interval
self._running = False
def record(self, latency_ms: float, status: int,
tokens: int, cost: float, model: str, error: str = None):
with self._lock:
self.metrics.record_request(latency_ms, status, tokens, cost, model, error)
def start_flush_loop(self):
"""Background thread to print periodic reports."""
self._running = True
def flush():
while self._running:
time.sleep(self._flush_interval)
with self._lock:
report = self.metrics.get_report()
print(f"[METRICS] {datetime.now().isoformat()} - {report}")
# Reset for next period
self.metrics = APIMetrics()
thread = threading.Thread(target=flush, daemon=True)
thread.start()
Initialize global collector
metrics = MetricsCollector(flush_interval=60)
metrics.start_flush_loop()
Record API calls
metrics.record(latency_ms=145.2, status=200, tokens=235, cost=0.0000987,
model="deepseek-v3.2")
metrics.record(latency_ms=89.7, status=200, tokens=156, cost=0.0000655,
model="gemini-2.5-flash")
metrics.record(latency_ms=2034.5, status=408, tokens=0, cost=0.0,
model="gpt-4.1", error="Request Timeout")
Common Errors & Fixes
Error 1: 401 Unauthorized — Invalid or Missing API Key
Symptom: HTTPError: 401 Client Error: Unauthorized
Cause: The most common cause is using an expired key, placing the API key in the wrong header, or copying incorrect whitespace around the key.
Fix:
# WRONG - extra spaces or wrong header format
headers = {"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY"} # May have spaces
headers = {"api-key": api_key} # Wrong header name
CORRECT
import os
api_key = os.environ.get("HOLYSHEEP_API_KEY") # Get from environment
headers = {
"Authorization": f"Bearer {api_key.strip()}", # Strip whitespace
"Content-Type": "application/json"
}
Verify key format (should start with 'hs_' for HolySheep)
if not api_key.startswith("hs_"):
raise ValueError(f"Invalid HolySheep API key format: {api_key[:5]}***")
Error 2: ConnectionError: timeout — Request Timeout
Symptom: ConnectError: [WinError 10060] or httpx.ConnectTimeout
Cause: Network connectivity issues, firewall blocking outbound HTTPS on port 443, or the request payload is too large causing slow processing.
Fix:
import httpx
Configure timeouts appropriately for AI API calls
client = httpx.AsyncClient(
timeout=httpx.Timeout(
timeout=60.0, # Total timeout (high for long completions)
connect=10.0, # Connection establishment timeout
read=30.0, # Response read timeout
write=10.0 # Request write timeout
),
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
Add retry logic with exponential backoff
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def resilient_request(url: str, **kwargs):
try:
response = await client.post(url, **kwargs)
response.raise_for_status()
return response.json()
except httpx.TimeoutException:
print("Request timed out - retrying with backoff...")
raise
except httpx.ConnectError as e:
print(f"Connection failed: {e} - check network/firewall settings")
raise
Error 3: 429 Too Many Requests — Rate Limit Exceeded
Symptom: HTTPError: 429 Client Error: Too Many Requests
Cause: You've exceeded your API rate limit. HolySheep AI offers competitive rate limits starting at ¥1=$1 with WeChat and Alipay support, but aggressive rate limiting can trigger this error.
Fix:
import asyncio
import time
class RateLimitedClient:
"""AI API client with automatic rate limiting."""
def __init__(self, requests_per_minute: int = 60):
self.rpm = requests_per_minute
self.min_interval = 60.0 / requests_per_minute
self.last_request_time = 0
self.queue = asyncio.Queue()
async def throttled_request(self, coro):
"""Wrap any async request with rate limiting."""
now = time.time()
time_since_last = now - self.last_request_time
if time_since_last < self.min_interval:
await asyncio.sleep(self.min_interval - time_since_last)
self.last_request_time = time.time()
return await coro
async def chat_completions(self, messages: list):
"""Rate-limited chat completion request."""
async def _make_request():
# Your actual API call here
return {"status": "success"}
return await self.throttled_request(_make_request())
Usage - respects 60 RPM limit
client = RateLimitedClient(requests_per_minute=60)
for batch in chunks(messages, 10):
await client.chat_completions(batch)
await asyncio.sleep(0.1) # Small delay between batches
Error 4: 500 Internal Server Error — Model Unavailable
Symptom: HTTPError: 500 Server Error: Internal Server Error
Cause: The AI model service is temporarily unavailable, undergoing maintenance, or experiencing high load.
Fix:
FALLBACK_MODELS = [
"deepseek-v3.2", # $0.42/MTok - Most cost-effective
"gemini-2.5-flash", # $2.50/MTok - Fast and reliable
"gpt-4.1", # $8.00/MTok - Premium option
]
async def resilient_completion(messages: list):
"""Automatically fallback to backup models on 500 errors."""
for model in FALLBACK_MODELS:
try:
response = await client.chat_completions(
model=model,
messages=messages
)
print(f"Success with model: {model}")
return response
except