API Engineering Best Practices

# Use LiteLLM as a unified interface across multiple LLM providers
# A single completion() call works with OpenAI, Anthropic, and others
import litellm
# Same function call, different providers
# LiteLLM translates automatically
# OpenAI
response_openai = litellm.completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Explain caching in one sentence."}],
    max_tokens=50
    )
# Anthropic (note the provider prefix)
response_anthropic = litellm.completion(
    model="anthropic/claude-sonnet-4-20250514",
    messages=[{"role": "user", "content": "Explain caching in one sentence."}],
    max_tokens=50
    )
# Google Gemini
response_gemini = litellm.completion(
    model="gemini/gemini-2.5-flash",
    messages=[{"role": "user", "content": "Explain caching in one sentence."}],
    max_tokens=50
    )
for name, resp in [("OpenAI", response_openai),
    ("Anthropic", response_anthropic),
    ("Gemini", response_gemini)]:
    print(f"{name}: {resp.choices[0].message.content[:80]}...")
    print(f" Cost: ${litellm.completion_cost(resp):.6f}")

# LiteLLM Router with built-in failure isolation
from litellm import Router
router = Router(
    model_list=[
    {"model_name": "gpt-4o", "litellm_params": {"model": "gpt-4o"}},
    {"model_name": "claude", "litellm_params": {"model": "anthropic/claude-sonnet-4-20250514"}},
    ],
    allowed_fails=3, # Cooldown after 3 consecutive failures
    cooldown_time=60, # Wait 60s before retrying failed model
    num_retries=2, # Retry each request up to 2 times
    timeout=30, # 30s timeout per request
    fallbacks=[{"gpt-4o": ["claude"]}], # Fall back to Claude if GPT-4o fails
)
response = router.completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Explain caching in one sentence."}]
)
print(response.choices[0].message.content)

# Implement a circuit breaker pattern for resilient LLM calls
# Combines budget checks, caching, model fallback, and static responses
import time
from dataclasses import dataclass, field
from enum import Enum
class CircuitState(Enum):
    CLOSED = "closed" # Normal operation
    OPEN = "open" # Provider is down, use fallback
    HALF_OPEN = "half_open" # Testing if provider recovered
@dataclass
class CircuitBreaker:
    failure_threshold: int = 5 # Failures before opening
    recovery_timeout: float = 60.0 # Seconds before testing recovery
    failure_count: int = field(default=0, init=False)
    state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    last_failure_time: float = field(default=0.0, init=False)
    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            # Check if recovery timeout has elapsed
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                return True # Allow one test request
            return False
        if self.state == CircuitState.HALF_OPEN:
            return True
        return False
    def record_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"Circuit OPENED after {self.failure_count} failures. "
                f"Routing to fallback for {self.recovery_timeout}s.")
# Usage
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30.0)
def call_with_circuit_breaker(messages, primary_fn, fallback_fn):
    if breaker.can_execute():
        try:
            result = primary_fn(messages)
            breaker.record_success()
            return result
        except Exception as e:
            breaker.record_failure()
            if breaker.state == CircuitState.OPEN:
                return fallback_fn(messages)
            raise
        else:
            return fallback_fn(messages)
        print(f"Circuit state: {breaker.state.value}")
        print(f"Ready: {breaker.can_execute()}")

# Build a semantic cache that hashes prompts to avoid redundant API calls
# Cache hits skip the network round-trip entirely, saving cost and latency
import hashlib
import json
import time
from functools import lru_cache
class LLMCache:
    """Simple in-memory cache for LLM responses."""
    def __init__(self, max_size=1000, ttl_seconds=3600):
        self.cache = {}
        self.max_size = max_size
        self.ttl = ttl_seconds
        self.hits = 0
        self.misses = 0
    def _make_key(self, model, messages, **kwargs):
        """Create a deterministic cache key from request parameters."""
        key_data = json.dumps({
            "model": model,
            "messages": messages,
            "params": {k: v for k, v in sorted(kwargs.items())}
            }, sort_keys=True)
        return hashlib.sha256(key_data.encode()).hexdigest()
    def get(self, model, messages, **kwargs):
        key = self._make_key(model, messages, **kwargs)
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                self.hits += 1
                return entry["response"]
            else:
                del self.cache[key] # Expired
                self.misses += 1
                return None
        def put(self, model, messages, response, **kwargs):
            if len(self.cache) >= self.max_size:
                # Evict oldest entry
                oldest = min(self.cache, key=lambda k: self.cache[k]["timestamp"])
                del self.cache[oldest]
                key = self._make_key(model, messages, **kwargs)
                self.cache[key] = {"response": response, "timestamp": time.time()}
                @property
                def hit_rate(self):
                    total = self.hits + self.misses
                    return self.hits / total if total > 0 else 0.0
                    cache = LLMCache(ttl_seconds=3600)
                    print(f"Cache initialized. Hit rate: {cache.hit_rate:.1%}")
                    print(f"Max size: {cache.max_size}, TTL: {cache.ttl}s")

# Build a semantic cache that hashes prompts to avoid redundant API calls
# Cache hits skip the network round-trip entirely, saving cost and latency
import numpy as np
from dataclasses import dataclass
@dataclass
class CacheEntry:
    query_text: str
    query_embedding: np.ndarray
    response: str
    timestamp: float
class SemanticCache:
    """Semantic cache using embedding similarity."""
    def __init__(self, similarity_threshold=0.95, max_entries=500):
        self.entries: list[CacheEntry] = []
        self.threshold = similarity_threshold
        self.max_entries = max_entries
    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
    def search(self, query_embedding: np.ndarray) -> str | None:
        """Find a cached response if a similar query exists."""
        best_score = 0.0
        best_response = None
        for entry in self.entries:
            score = self._cosine_similarity(query_embedding, entry.query_embedding)
            if score > best_score:
                best_score = score
                best_response = entry.response
                if best_score >= self.threshold:
                    return best_response
                    return None
        def store(self, query_text, query_embedding, response):
            import time
            if len(self.entries) >= self.max_entries:
                self.entries.pop(0) # Remove oldest
                self.entries.append(CacheEntry(
                    query_text=query_text,
                    query_embedding=query_embedding,
                    response=response,
                    timestamp=time.time()
                    ))
                # Example: two semantically identical queries
                cache = SemanticCache(similarity_threshold=0.95)
                # Simulate embeddings (in production, use an embedding model)
                emb1 = np.random.randn(1536)
                emb1 = emb1 / np.linalg.norm(emb1)
                # A very similar embedding (simulating a semantically close query)
                noise = np.random.randn(1536) * 0.02
                emb2 = emb1 + noise
                emb2 = emb2 / np.linalg.norm(emb2)
                cache.store("What is the capital of France?", emb1, "The capital of France is Paris.")
                result = cache.search(emb2)
                similarity = float(np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2)))
                print(f"Similarity: {similarity:.4f}")
                print(f"Cache hit: {result is not None}")
                print(f"Result: {result}")

# Route requests through the Portkey AI gateway for fallback and caching
# Portkey handles provider failover and semantic response caching
from openai import OpenAI
from portkey_ai import PORTKEY_GATEWAY_URL, createHeaders

# Configure Portkey with routing and caching
client = OpenAI(
    base_url=PORTKEY_GATEWAY_URL,
    default_headers=createHeaders(
        api_key="your-portkey-key",
        config={
            "strategy": {
                "mode": "fallback",  # Try models in order
            },
            "targets": [
                {
                    "provider": "openai",
                    "api_key": "sk-...",
                    "override_params": {"model": "gpt-4o"},
                },
                {
                    "provider": "anthropic",
                    "api_key": "sk-ant-...",
                    "override_params": {"model": "claude-sonnet-4-20250514"},
                },
            ],
            "cache": {"mode": "semantic", "max_age": 3600},
        },
    ),
)
# Your code is unchanged; Portkey handles routing and caching
response = client.chat.completions.create(
    model="gpt-4o", messages=[{"role": "user", "content": "Explain caching briefly."}]
)
# Extract the generated message from the API response
print(response.choices[0].message.content[:100])

# Build a semantic cache that hashes prompts to avoid redundant API calls
# Cache hits skip the network round-trip entirely, saving cost and latency
from openai import OpenAI
# Route through Helicone for observability
client = OpenAI(
    base_url="https://oai.helicone.ai/v1",
    default_headers={
    "Helicone-Auth": "Bearer your-helicone-key",
    "Helicone-Cache-Enabled": "true", # Enable response caching
    "Helicone-Property-Environment": "production",
    "Helicone-Property-Feature": "customer-support", # Tag for cost attribution
    }
)
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "How does Helicone work?"}]
)
# Helicone dashboard now shows:
# - Request latency, token count, estimated cost
# - Cache hit/miss status
# - Custom properties for filtering and grouping
print(f"Response received. Tokens: {response.usage.total_tokens}")
print("Check Helicone dashboard for detailed analytics.")

# Enforce per-user token budgets to prevent runaway API costs
# Track usage across configurable time windows (hourly, daily, monthly)
import time
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class TokenBudget:
    """Track and enforce token spending limits."""
    limits: dict = field(default_factory=dict) # entity -> max tokens per period
    usage: dict = field(default_factory=lambda: defaultdict(int))
    period_start: dict = field(default_factory=dict)
    period_seconds: float = 86400 # Default: daily budget
    def set_limit(self, entity: str, max_tokens: int):
        self.limits[entity] = max_tokens
        self.period_start[entity] = time.time()
    def check_budget(self, entity: str, estimated_tokens: int) -> bool:
        """Return True if the request is within budget."""
        if entity not in self.limits:
            return True # No limit set
            # Reset period if expired
            if time.time() - self.period_start.get(entity, 0) > self.period_seconds:
                self.usage[entity] = 0
                self.period_start[entity] = time.time()
                return self.usage[entity] + estimated_tokens <= self.limits[entity]
        def record_usage(self, entity: str, tokens_used: int):
            self.usage[entity] += tokens_used
        def remaining(self, entity: str) -> int:
            if entity not in self.limits:
                return float('inf')
                return max(0, self.limits[entity] - self.usage[entity])
                # Example: per-user budget enforcement
                budget = TokenBudget(period_seconds=86400) # Daily budgets
                budget.set_limit("user:alice", max_tokens=100_000)
                budget.set_limit("user:bob", max_tokens=50_000)
                budget.set_limit("feature:chat", max_tokens=1_000_000)
                # Check before making API call
                user = "user:alice"
                estimated = 500 # Estimated tokens for this request
                if budget.check_budget(user, estimated):
                    # Make the API call
                    budget.record_usage(user, 480) # Actual tokens used
                    print(f"Request allowed. {user} remaining: {budget.remaining(user):,} tokens")
                else:
                    print(f"Budget exceeded for {user}. Remaining: {budget.remaining(user):,} tokens")
                    # Simulate heavy usage
                    budget.record_usage("user:bob", 48_000)
                    print(f"user:bob remaining: {budget.remaining('user:bob'):,} tokens")
                    print(f"user:bob can make 500-token request: {budget.check_budget('user:bob', 500)}")

# Build a semantic cache that hashes prompts to avoid redundant API calls
# Cache hits skip the network round-trip entirely, saving cost and latency
from dataclasses import dataclass
from enum import Enum
import time
import json
class FallbackLevel(Enum):
    PRIMARY = "primary"
    CACHE = "cache"
    SIMPLE_MODEL = "simple_model"
    STATIC = "static"
    ERROR = "error"
@dataclass
class ResilientResponse:
    content: str
    fallback_level: FallbackLevel
    latency_ms: float
    model_used: str | None = None
def resilient_llm_call(messages, cache, circuit_breaker, budget,
    user_id="default") -> ResilientResponse:
    """Production-grade LLM call with full resilience stack."""
    start = time.time()
    # Step 1: Check budget
    if not budget.check_budget(user_id, estimated_tokens=500):
        return ResilientResponse(
            content="You have reached your daily usage limit. Please try again tomorrow.",
            fallback_level=FallbackLevel.ERROR,
            latency_ms=(time.time() - start) * 1000
            )
    # Step 2: Check cache
    cached = cache.get("gpt-4o", messages, temperature=0)
    if cached:
        return ResilientResponse(
            content=cached,
            fallback_level=FallbackLevel.CACHE,
            latency_ms=(time.time() - start) * 1000,
            model_used="cache"
            )
    # Step 3: Try primary model (with circuit breaker)
    if circuit_breaker.can_execute():
        try:
            response = call_primary_model(messages)
            circuit_breaker.record_success()
            cache.put("gpt-4o", messages, response)
            budget.record_usage(user_id, 480)
            return ResilientResponse(
                content=response,
                fallback_level=FallbackLevel.PRIMARY,
                latency_ms=(time.time() - start) * 1000,
                model_used="gpt-4o"
                )
        except Exception:
            circuit_breaker.record_failure()
            # Step 4: Try simpler model
            try:
                response = call_simple_model(messages)
                return ResilientResponse(
                    content=response,
                    fallback_level=FallbackLevel.SIMPLE_MODEL,
                    latency_ms=(time.time() - start) * 1000,
                    model_used="gpt-4o-mini"
                    )
            except Exception:
                pass
                # Step 5: Static fallback
                return ResilientResponse(
                    content="I'm currently experiencing high demand. Please try again shortly.",
                    fallback_level=FallbackLevel.STATIC,
                    latency_ms=(time.time() - start) * 1000
                    )
# These would be real API calls in production
def call_primary_model(messages):
    return "Response from GPT-4o"
def call_simple_model(messages):
    return "Response from GPT-4o-mini"
print("Resilient LLM call pattern configured.")
print("Fallback order: Primary -> Cache -> Simple Model -> Static -> Error")