AI Cost Optimization: Cutting Your LLM Bill by 70%

from enum import Enum
from typing import Tuple

class QueryComplexity(Enum):
    SIMPLE = "simple"
    MEDIUM = "medium"
    COMPLEX = "complex"

class ModelRouter:
    def __init__(self):
        self.models = {
            QueryComplexity.SIMPLE: "gpt-3.5-turbo",
            QueryComplexity.MEDIUM: "claude-3-sonnet",
            QueryComplexity.COMPLEX: "gpt-4-turbo",
        }
        
        # Keywords that indicate complexity
        self.complex_indicators = [
            "analyze", "compare", "synthesize", "evaluate",
            "code review", "architecture", "design system"
        ]
        self.simple_indicators = [
            "what is", "define", "list", "when was",
            "how many", "yes or no"
        ]
    
    def classify(self, query: str) -> Tuple[QueryComplexity, str]:
        query_lower = query.lower()
        
        # Rule-based fast path
        if any(ind in query_lower for ind in self.simple_indicators):
            if len(query) < 100:
                return QueryComplexity.SIMPLE, self.models[QueryComplexity.SIMPLE]
        
        if any(ind in query_lower for ind in self.complex_indicators):
            return QueryComplexity.COMPLEX, self.models[QueryComplexity.COMPLEX]
        
        # Fallback: Use token count as heuristic
        token_estimate = len(query.split())
        if token_estimate < 20:
            return QueryComplexity.SIMPLE, self.models[QueryComplexity.SIMPLE]
        elif token_estimate < 100:
            return QueryComplexity.MEDIUM, self.models[QueryComplexity.MEDIUM]
        else:
            return QueryComplexity.COMPLEX, self.models[QueryComplexity.COMPLEX]
    
    async def route_and_call(self, query: str, messages: list) -> dict:
        complexity, model = self.classify(query)
        
        response = await self.llm_client.chat(
            model=model,
            messages=messages
        )
        
        # Track for analytics
        self.metrics.record_routing(complexity, model, response.usage)
        
        return {
            "response": response.content,
            "model_used": model,
            "complexity": complexity.value,
            "cost": self.calculate_cost(model, response.usage)
        }

import hashlib
import numpy as np
from typing import Optional, Tuple

class SemanticCache:
    def __init__(self, redis_client, vector_db, embedding_model):
        self.redis = redis_client
        self.vector_db = vector_db
        self.embedder = embedding_model
        self.similarity_threshold = 0.95
        self.ttl_seconds = 86400  # 24 hours
    
    async def get(self, query: str) -> Optional[Tuple[str, float]]:
        # Layer 1: Exact match (fastest)
        cache_key = self._hash_query(query)
        exact_hit = await self.redis.get(cache_key)
        if exact_hit:
            return exact_hit, 1.0
        
        # Layer 2: Semantic match (slower but more hits)
        query_embedding = await self.embedder.embed(query)
        
        results = await self.vector_db.search(
            vector=query_embedding,
            top_k=1,
            include_metadata=True
        )
        
        if results and results[0].score >= self.similarity_threshold:
            cached_response = results[0].metadata["response"]
            # Warm exact cache for next time
            await self.redis.setex(cache_key, self.ttl_seconds, cached_response)
            return cached_response, results[0].score
        
        return None
    
    async def set(self, query: str, response: str):
        # Store in exact cache
        cache_key = self._hash_query(query)
        await self.redis.setex(cache_key, self.ttl_seconds, response)
        
        # Store in semantic cache
        query_embedding = await self.embedder.embed(query)
        await self.vector_db.upsert(
            id=cache_key,
            vector=query_embedding,
            metadata={"query": query, "response": response}
        )
    
    def _hash_query(self, query: str) -> str:
        normalized = query.lower().strip()
        return hashlib.sha256(normalized.encode()).hexdigest()[:16]


# Usage in API endpoint
cache = SemanticCache(redis, pinecone, openai_embedder)

async def chat_with_cache(query: str) -> dict:
    # Check cache first
    cached = await cache.get(query)
    if cached:
        response, similarity = cached
        return {
            "response": response,
            "cached": True,
            "similarity": similarity,
            "cost": 0
        }
    
    # Cache miss - call LLM
    response = await llm.chat(query)
    
    # Store for future
    await cache.set(query, response.content)
    
    return {
        "response": response.content,
        "cached": False,
        "cost": response.usage.total_cost
    }