AI API Production Best Practices

1. Error Handling & Retries

AI APIs fail. Networks timeout. Rate limits hit. Your code must handle this gracefully.

Exponential Backoff with Jitter

import random
import time
from openai import OpenAI

client = OpenAI(base_url="https://tokenease.io/v1", api_key="your_key")

def call_with_retry(messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model="deepseek",
                messages=messages,
                max_tokens=1000
            )
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            
            # Exponential backoff: 1s, 2s, 4s + random jitter
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait:.1f}s...")
            time.sleep(wait)

# Usage
response = call_with_retry([{"role": "user", "content": "Hello"}])

Handle Specific Error Types

from openai import RateLimitError, APIError, APITimeoutError

def robust_call(messages):
    try:
        return client.chat.completions.create(
            model="deepseek",
            messages=messages
        )
    except RateLimitError:
        # Wait longer and retry
        time.sleep(10)
        return robust_call(messages)
    except APITimeoutError:
        # Retry immediately (might be transient)
        return robust_call(messages)
    except APIError as e:
        if e.code == "insufficient_quota":
            # Switch to cheaper model
            return client.chat.completions.create(
                model="doubao",  # Cheaper fallback
                messages=messages
            )
        raise

2. Smart Caching

Cache repeated queries to save money and improve speed:

import hashlib
import json
from functools import lru_cache

class AIQueryCache:
    def __init__(self, ttl_seconds=3600):
        self.cache = {}
        self.ttl = ttl_seconds
    
    def _hash(self, messages, model):
        content = json.dumps({"messages": messages, "model": model})
        return hashlib.md5(content.encode()).hexdigest()
    
    def get(self, messages, model):
        key = self._hash(messages, model)
        if key in self.cache:
            result, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return result
            del self.cache[key]
        return None
    
    def set(self, messages, model, result):
        key = self._hash(messages, model)
        self.cache[key] = (result, time.time())

# Usage
cache = AIQueryCache(ttl_seconds=1800)  # 30 min cache

def cached_chat(messages, model="deepseek"):
    cached = cache.get(messages, model)
    if cached:
        return cached
    
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    cache.set(messages, model, response)
    return response

3. Timeouts & Circuit Breakers

Don't let slow AI responses cascade into system failures:

import asyncio
from asyncio import TimeoutError

async def call_with_timeout(messages, timeout=10):
    try:
        return await asyncio.wait_for(
            asyncio.to_thread(
                client.chat.completions.create,
                model="deepseek",
                messages=messages,
                max_tokens=500
            ),
            timeout=timeout
        )
    except TimeoutError:
        return {"error": "Request timeout", "fallback": "Please try again"}

# Circuit breaker pattern
class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_time=60):
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.recovery_time = recovery_time
        self.last_failure = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure > self.recovery_time:
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit breaker OPEN")
        
        try:
            result = func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "OPEN"
            raise

breaker = CircuitBreaker()
# Usage: breaker.call(client.chat.completions.create, ...)

4. Monitoring & Alerting

Track what matters:

Metric	Target	Alert If
Latency (P95)	< 2s	> 5s
Error rate	< 1%	> 5%
Daily cost	On budget	> 120% of budget
Token usage	Growing steadily	Spike > 200%
Cache hit rate	> 30%	< 10%

import time

class APIMonitor:
    def __init__(self):
        self.stats = {
            "requests": 0,
            "errors": 0,
            "total_latency": 0,
            "total_tokens": 0
        }
    
    def record(self, latency, tokens, error=False):
        self.stats["requests"] += 1
        self.stats["total_latency"] += latency
        self.stats["total_tokens"] += tokens
        if error:
            self.stats["errors"] += 1
    
    def report(self):
        if self.stats["requests"] == 0:
            return "No requests yet"
        
        avg_latency = self.stats["total_latency"] / self.stats["requests"]
        error_rate = self.stats["errors"] / self.stats["requests"]
        
        return {
            "avg_latency_ms": round(avg_latency * 1000, 1),
            "error_rate": f"{error_rate:.1%}",
            "total_tokens": self.stats["total_tokens"],
            "total_requests": self.stats["requests"]
        }

# Usage
monitor = APIMonitor()

start = time.time()
try:
    response = client.chat.completions.create(...)
    latency = time.time() - start
    tokens = response.usage.total_tokens
    monitor.record(latency, tokens)
except Exception as e:
    latency = time.time() - start
    monitor.record(latency, 0, error=True)

print(monitor.report())

5. Cost Control

Set Hard Limits

class CostController:
    def __init__(self, daily_budget_usd=50):
        self.daily_budget = daily_budget_usd
        self.daily_spend = 0
        self.last_reset = time.time()
    
    def can_spend(self, estimated_cost):
        # Reset daily counter
        if time.time() - self.last_reset > 86400:
            self.daily_spend = 0
            self.last_reset = time.time()
        
        return (self.daily_spend + estimated_cost) <= self.daily_budget
    
    def record_spend(self, tokens, model="deepseek"):
        # Approximate cost per 1K tokens
        rates = {"deepseek": 0.0015, "doubao": 0.0008, "zhipu": 0.0015}
        cost = (tokens / 1000) * rates.get(model, 0.0015)
        self.daily_spend += cost
        return cost

# Usage
controller = CostController(daily_budget_usd=20)

if not controller.can_spend(estimated_cost=0.50):
    raise Exception("Daily budget exceeded")

response = client.chat.completions.create(...)
cost = controller.record_spend(response.usage.total_tokens)

Model Tiering by Cost

def smart_model_select(query_complexity, budget_remaining):
    """Choose model based on complexity and budget"""
    if budget_remaining < 5:
        return "doubao"  # Cheapest
    elif query_complexity == "simple":
        return "doubao"
    elif query_complexity == "medium":
        return "qwen"
    else:
        return "deepseek"  # Best quality

6. Security Best Practices

• Never commit API keys - Use environment variables
• Validate inputs - Sanitize user data before sending to API
• Rate limit per user - Prevent abuse and cost spikes
• Log without PII - Don't log sensitive user data
• Use HTTPS only - TokenEase enforces this
• Rotate keys regularly - Set calendar reminders

# Secure API key management
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("TOKENEASE_API_KEY")
if not API_KEY:
    raise ValueError("TOKENEASE_API_KEY not set")

# Input validation
def validate_input(user_input, max_length=4000):
    if len(user_input) > max_length:
        raise ValueError(f"Input too long: {len(user_input)} chars")
    # Strip potential prompt injection attempts
    forbidden = ["ignore previous", "system prompt", "ignore all"]
    for f in forbidden:
        if f in user_input.lower():
            raise ValueError("Potential prompt injection detected")
    return user_input