Production AI Engineering

MODEL_COSTS = { "gpt-4o": (5.00, 15.00), # (input per 1M, output per 1M) "gpt-4o-mini": (0.15, 0.60), "claude-3-5-sonnet": (3.00, 15.00), "claude-3-haiku": (0.25, 1.25), } def estimate_cost(in_tokens: int, out_tokens: int, model: str) -> float: i, o = MODEL_COSTS.get(model, (0.15, 0.60)) return round((in_tokens * i + out_tokens * o) / 1_000_000, 6) # After every API call, calculate and log cost: response = client.chat.completions.create(...) cost = estimate_cost( response.usage.prompt_tokens, response.usage.completion_tokens, "gpt-4o-mini" ) logger.info(f"cost=${cost:.6f} tokens_in={response.usage.prompt_tokens} " f"tokens_out={response.usage.completion_tokens}")

import hashlib, json, redis r = redis.Redis(host="localhost", port=6379, decode_responses=True) CACHE_TTL = 3600 # 1 hour def cache_key(messages: list, model: str) -> str: payload = json.dumps({"messages": messages, "model": model}, sort_keys=True) return "llm:" + hashlib.sha256(payload.encode()).hexdigest() def cached_llm_call(messages: list, model: str = "gpt-4o-mini") -> str: key = cache_key(messages, model) # Check cache first cached = r.get(key) if cached: return cached # free — no API call # Cache miss — call the API response = client.chat.completions.create( model=model, messages=messages, max_tokens=500 ) result = response.choices[0].message.content # Store in cache r.setex(key, CACHE_TTL, result) return result

import redis, time from fastapi import HTTPException r = redis.Redis(host="localhost", port=6379, decode_responses=True) def rate_limit(user_id: str, max_req_per_min: int = 20): """Sliding window rate limiter — max N requests per minute per user.""" key = f"rl:{user_id}:{int(time.time() // 60)}" # key changes every minute count = r.incr(key) # atomic increment r.expire(key, 60) # expire after 60s if count > max_req_per_min: raise HTTPException( status_code=429, detail="Rate limit exceeded. Please wait before sending another request.", headers={"Retry-After": "60"} ) # In your FastAPI endpoint: @app.post("/chat") async def chat(request: ChatRequest, user_id: str = Depends(get_current_user)): rate_limit(user_id, max_req_per_min=10) # ... rest of endpoint

import uuid, time, logging, hashlib logger = logging.getLogger("ai_service") async def tracked_llm_call(messages: list, model: str, user_id: str) -> dict: request_id = str(uuid.uuid4())[:8] prompt_hash = hashlib.md5(str(messages).encode()).hexdigest()[:8] start = time.time() try: response = await async_client.chat.completions.create( model=model, messages=messages, max_tokens=500 ) latency_ms = round((time.time() - start) * 1000) cost = estimate_cost( response.usage.prompt_tokens, response.usage.completion_tokens, model ) logger.info( f"rid={request_id} user={user_id} model={model} " f"prompt_hash={prompt_hash} " f"tokens_in={response.usage.prompt_tokens} " f"tokens_out={response.usage.completion_tokens} " f"cost=${cost:.6f} latency={latency_ms}ms " f"finish={response.choices[0].finish_reason}" ) return {"content": response.choices[0].message.content, "cost": cost, "latency_ms": latency_ms} except Exception as e: latency_ms = round((time.time() - start) * 1000) logger.error(f"rid={request_id} user={user_id} error={type(e).__name__} " f"latency={latency_ms}ms msg={str(e)}") raise

class CircuitBreaker: """Simple circuit breaker for LLM API calls. States: CLOSED (normal) → OPEN (failing) → HALF_OPEN (testing recovery) """ def __init__(self, failure_threshold=5, recovery_timeout=60): self.failures = 0 self.threshold = failure_threshold self.recovery_timeout = recovery_timeout self.state = "CLOSED" # CLOSED | OPEN | HALF_OPEN self.last_failure_time = None def call(self, func, *args, **kwargs): if self.state == "OPEN": # Check if recovery timeout has passed if time.time() - self.last_failure_time > self.recovery_timeout: self.state = "HALF_OPEN" else: raise Exception("Circuit breaker is OPEN — API temporarily unavailable") try: result = func(*args, **kwargs) if self.state == "HALF_OPEN": self.state = "CLOSED" # Recovery confirmed self.failures = 0 return result except (openai.APIConnectionError, openai.RateLimitError) as e: self.failures += 1 self.last_failure_time = time.time() if self.failures >= self.threshold: self.state = "OPEN" logger.warning(f"Circuit breaker OPENED after {self.failures} failures") raise breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60) def safe_llm_call(messages): try: return breaker.call(client.chat.completions.create, model="gpt-4o-mini", messages=messages) except Exception: return FALLBACK_RESPONSE # cached or static response

from openai import AsyncOpenAI import anthropic, asyncio async_openai = AsyncOpenAI() async def with_fallback(messages: list, preferred: str = "gpt-4o-mini") -> tuple[str, str]: """Try preferred model, fall back through providers on failure.""" providers = [ ("gpt-4o-mini", lambda: async_openai.chat.completions.create(model="gpt-4o-mini", messages=messages)), ("gpt-3.5-turbo", lambda: async_openai.chat.completions.create(model="gpt-3.5-turbo", messages=messages)), ] for model_name, call in providers: try: response = await asyncio.wait_for(call(), timeout=10.0) return response.choices[0].message.content, model_name except (openai.APIConnectionError, openai.RateLimitError, asyncio.TimeoutError) as e: logger.warning(f"Provider {model_name} failed: {e}. Trying next...") continue raise Exception("All providers failed")

from starlette.middleware.base import BaseHTTPMiddleware import uuid, time, logging logger = logging.getLogger("ai_service") class LoggingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): rid = str(uuid.uuid4())[:8] request.state.request_id = rid t = time.time() response = await call_next(request) latency_ms = round((time.time() - t) * 1000) logger.info( f"rid={rid} {request.method} {request.url.path} " f"status={response.status_code} latency={latency_ms}ms " f"user_agent={request.headers.get('user-agent','unknown')[:50]}" ) response.headers["X-Request-ID"] = rid return response app.add_middleware(LoggingMiddleware)

import re INJECTION_PATTERNS = [ r"ignore.*instructions", r"you are now", r"new instructions", r"forget everything", r"reveal.*system.*prompt", r"ignore.*previous", r"pretend.*you.*are", ] def is_injection_attempt(text: str) -> bool: text_lower = text.lower() return any(re.search(pattern, text_lower) for pattern in INJECTION_PATTERNS) def wrap_user_input(text: str) -> str: """Wrap user input in delimiters to reduce injection effectiveness.""" return f"<user_input>\n{text}\n</user_input>" # In your endpoint: if is_injection_attempt(request.text): logger.warning(f"Injection attempt detected from user={user_id}") raise HTTPException(400, "Invalid input detected.")

from unittest.mock import MagicMock, patch from fastapi.testclient import TestClient test_client = TestClient(app) def mock_llm_response(content: str, tokens_in: int = 100, tokens_out: int = 50): m = MagicMock() m.choices[0].message.content = content m.choices[0].finish_reason = "stop" m.usage.prompt_tokens = tokens_in m.usage.completion_tokens = tokens_out m.usage.total_tokens = tokens_in + tokens_out m.model = "gpt-4o-mini" return m @patch("main.client.chat.completions.create") def test_chat_endpoint(mock_create): mock_create.return_value = mock_llm_response("Hello, I can help with that!") resp = test_client.post("/chat", json={"message": "Hello", "user_id": "test123"}) assert resp.status_code == 200 assert resp.json()["content"] == "Hello, I can help with that!" # Verify model used (cost guard) assert mock_create.call_args[1]["model"] == "gpt-4o-mini"