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")