How to Reduce AI API Costs by 70%: Caching, Batching, Model Routing, and Prompt Optimization With Code

# Semantic caching: serves cached responses for similar queries
# Reduces API calls by 30-50% in FAQ/support applications

import redis
import numpy as np
from openai import OpenAI
import json
import hashlib

client       = OpenAI()
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)

CACHE_SIMILARITY_THRESHOLD = 0.92  # Queries this similar get cached response
CACHE_TTL_SECONDS          = 3600  # Cache responses for 1 hour

def get_embedding(text: str) -> list[float]:
    response = client.embeddings.create(
        model="text-embedding-3-small",  # Cheap: $0.02 per million tokens
        input=text
    )
    return response.data[0].embedding

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

def semantic_cache_lookup(query: str) -> dict | None:
    """
    Check if a semantically similar query has a cached response.
    Returns cached response dict or None if cache miss.
    """
    query_embedding = get_embedding(query)
    
    # Get all cached query keys
    cache_keys = redis_client.keys("cache:query:*")
    
    best_match_key   = None
    best_similarity  = 0.0
    
    for key in cache_keys:
        cached_data = json.loads(redis_client.get(key))
        similarity  = cosine_similarity(
            query_embedding, cached_data["embedding"]
        )
        if similarity > best_similarity:
            best_similarity  = similarity
            best_match_key   = key
    
    if best_similarity >= CACHE_SIMILARITY_THRESHOLD and best_match_key:
        cached_data = json.loads(redis_client.get(best_match_key))
        return {
            "response":   cached_data["response"],
            "cache_hit":  True,
            "similarity": round(best_similarity, 3)
        }
    return None

def cached_api_call(
    query: str,
    system_prompt: str = "You are a helpful assistant.",
    model: str = "gpt-4o-mini"
) -> dict:
    """
    API call with semantic caching.
    Cache hit: ~0.001 cents per request (embedding only)
    Cache miss: normal API cost
    """
    # Check cache first
    cache_result = semantic_cache_lookup(query)
    if cache_result:
        return cache_result
    
    # Cache miss: call API
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user",   "content": query}
        ]
    )
    answer = response.choices[0].message.content
    
    # Store in cache with embedding
    query_embedding = get_embedding(query)
    cache_key       = f"cache:query:{hashlib.md5(query.encode()).hexdigest()}"
    cache_data      = {"embedding": query_embedding, "response": answer, "query": query}
    
    redis_client.setex(cache_key, CACHE_TTL_SECONDS, json.dumps(cache_data))
    
    return {"response": answer, "cache_hit": False, "similarity": 0.0}

# Model router: sends simple tasks to cheap models, complex to expensive
# Typical cost reduction: 60-80% vs routing everything to GPT-4o

import tiktoken

# Approximate costs per 1K tokens (input/output combined, 2026 rates)
MODEL_COSTS = {
    "gpt-4o-mini":    0.000150,  # $0.15 per 1M tokens
    "gpt-4o":         0.002500,  # $2.50 per 1M tokens
    "claude-haiku":   0.000250,  # $0.25 per 1M tokens  
    "claude-sonnet":  0.003000,  # $3.00 per 1M tokens
}

def classify_query_complexity(query: str, context: str = "") -> str:
    """
    Classifies query complexity to route to appropriate model.
    Returns: 'simple', 'medium', or 'complex'
    """
    enc              = tiktoken.get_encoding("cl100k_base")
    total_tokens     = len(enc.encode(query + context))
    
    # Heuristics for complexity classification
    complexity_signals = {
        "multi_step":    any(w in query.lower() for w in [
            "analyze", "compare", "evaluate", "synthesize", "design"
        ]),
        "long_context":  total_tokens > 2000,
        "code_complex":  "implement" in query.lower() and "algorithm" in query.lower(),
        "simple_lookup": any(w in query.lower() for w in [
            "what is", "when was", "who is", "define", "list"
        ]),
        "classification": any(w in query.lower() for w in [
            "is this", "does this", "classify", "categorize"
        ])
    }
    
    if complexity_signals["simple_lookup"] or complexity_signals["classification"]:
        return "simple"
    elif complexity_signals["multi_step"] or complexity_signals["long_context"]:
        return "complex"
    else:
        return "medium"

def route_to_model(query: str, context: str = "") -> str:
    """
    Returns the optimal model name for a given query.
    """
    complexity = classify_query_complexity(query, context)
    
    routing = {
        "simple":  "gpt-4o-mini",     # Simple Q&A, classification
        "medium":  "gpt-4o-mini",     # Most tasks — mini handles well
        "complex": "gpt-4o"           # Multi-step reasoning, code generation
    }
    return routing[complexity]

# Cost estimation before calling
def estimate_cost(prompt: str, model: str, expected_output_tokens: int = 200) -> float:
    enc          = tiktoken.get_encoding("cl100k_base")
    input_tokens = len(enc.encode(prompt))
    total_tokens = input_tokens + expected_output_tokens
    
    return (total_tokens / 1000) * MODEL_COSTS.get(model, 0.0025)

# Example: classifying 1000 support tickets
# Without routing: 1000 × GPT-4o = $2.50 per 1K tokens × avg 500 tokens = $1.25
# With routing: 800 simple (mini) + 200 complex (4o)
#   = (800 × 0.15 × 500/1000) + (200 × 2.50 × 500/1000) = $0.06 + $0.25 = $0.31
# Cost reduction: 75%

# Prompt compression: reduces token count before API call
# LLMLingua approach: remove low-importance tokens from long context

import tiktoken
from openai import OpenAI

client = OpenAI()
enc    = tiktoken.get_encoding("cl100k_base")

def compress_system_prompt(verbose_prompt: str) -> str:
    """
    Compresses a verbose system prompt.
    Reduces tokens while preserving semantic meaning.
    Run once and cache the result — compress once, use thousands of times.
    """
    compression_prompt = f"""
    Compress the following system prompt to use as few tokens as possible
    while preserving all instructions and their meaning exactly.
    Remove:
    - Redundant phrases
    - Filler words ('Please', 'You should', 'Make sure to')
    - Repeated concepts
    - Verbose explanations where a directive word suffices
    
    Original prompt:
    {verbose_prompt}
    
    Respond with ONLY the compressed prompt, nothing else.
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": compression_prompt}],
        temperature=0
    )
    compressed = response.choices[0].message.content
    
    original_tokens   = len(enc.encode(verbose_prompt))
    compressed_tokens = len(enc.encode(compressed))
    reduction         = (1 - compressed_tokens / original_tokens) * 100
    
    print(f"Token reduction: {original_tokens} → {compressed_tokens} ({reduction:.1f}% saved)")
    return compressed

# Apply max_tokens to limit output length
def capped_api_call(
    prompt: str,
    max_output_tokens: int = 300,
    model: str = "gpt-4o-mini"
) -> str:
    """
    Most applications do not need open-ended responses.
    Setting max_tokens prevents expensive long outputs.
    For summaries: 150-200 tokens
    For answers: 200-400 tokens
    For code snippets: 500-1000 tokens
    """
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_output_tokens  # Hard cap on output cost
    )
    return response.choices[0].message.content