AI API Cost Management: How to Cut LLM Costs by 80% Without Losing Quality

import json
import time
from openai import OpenAI
from dataclasses import dataclass

client = OpenAI()

PRICING = {
    "gpt-4o":        {"input": 5.00, "output": 15.00},
    "gpt-4o-mini":   {"input": 0.15, "output": 0.60},
    "gpt-3.5-turbo": {"input": 0.50, "output": 1.50},
}

@dataclass
class APICallMetrics:
    model: str
    feature: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    latency_ms: float

def tracked_completion(
    messages: list,
    model: str = "gpt-4o-mini",
    feature: str = "unknown",
    **kwargs
) -> tuple[str, APICallMetrics]:
    """Wrapper that tracks cost and latency."""
    
    start = time.time()
    
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        **kwargs
    )
    
    latency_ms = (time.time() - start) * 1000
    usage = response.usage
    
    rates = PRICING.get(model, {"input": 0, "output": 0})
    cost = (
        usage.prompt_tokens / 1_000_000 * rates["input"] +
        usage.completion_tokens / 1_000_000 * rates["output"]
    )
    
    metrics = APICallMetrics(
        model=model,
        feature=feature,
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
        cost_usd=cost,
        latency_ms=latency_ms
    )
    
    # Log to database, DataDog, CloudWatch, etc.
    log_metrics(metrics)
    
    return response.choices[0].message.content, metrics

def log_metrics(metrics: APICallMetrics):
    """Log to your preferred backend."""
    print(f"[{metrics.feature}] ${metrics.cost_usd:.5f} | "
          f"{metrics.input_tokens}+{metrics.output_tokens} tokens | "
          f"{metrics.latency_ms:.0f}ms | {metrics.model}")

from enum import Enum

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

MODEL_MAP = {
    TaskComplexity.SIMPLE:  "gpt-4o-mini",   # $0.15/$0.60 per M
    TaskComplexity.MEDIUM:  "gpt-4o-mini",   # Test if sufficient
    TaskComplexity.COMPLEX: "gpt-4o",        # $5/$15 per M (33× more expensive)
}

SIMPLE_TASKS = [
    "classify", "categorize", "extract", "is this", "yes or no",
    "list the", "summarize in one sentence"
]
COMPLEX_TASKS = [
    "analyze", "debug", "explain why", "compare", "architect",
    "write a detailed", "multi-step", "reason through"
]

def classify_task(prompt: str) -> TaskComplexity:
    prompt_lower = prompt.lower()
    
    if any(indicator in prompt_lower for indicator in COMPLEX_TASKS):
        return TaskComplexity.COMPLEX
    if any(indicator in prompt_lower for indicator in SIMPLE_TASKS):
        return TaskComplexity.SIMPLE
    return TaskComplexity.MEDIUM

def smart_complete(messages: list, task: str) -> str:
    complexity = classify_task(task)
    model = MODEL_MAP[complexity]
    
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    return response.choices[0].message.content

# Result: 70-80% of typical workloads route to gpt-4o-mini
# Cost reduction: 25-30× for those queries

import hashlib
import json
import redis
from typing import Optional

# pip install redis
cache = redis.Redis(host="localhost", port=6379)

def get_cache_key(model: str, messages: list) -> str:
    """Create a deterministic cache key."""
    content = json.dumps({"model": model, "messages": messages}, sort_keys=True)
    return f"llm_cache:{hashlib.sha256(content.encode()).hexdigest()}"

def cached_completion(
    messages: list,
    model: str = "gpt-4o-mini",
    ttl: int = 3600,  # 1 hour cache
) -> tuple[str, bool]:
    """Returns (response, was_cached)."""
    
    key = get_cache_key(model, messages)
    cached = cache.get(key)
    
    if cached:
        return json.loads(cached), True
    
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0  # Deterministic output for caching
    )
    
    result = response.choices[0].message.content
    cache.setex(key, ttl, json.dumps(result))
    
    return result, False

# Semantic caching for paraphrases
import numpy as np

class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.95):
        self.threshold = similarity_threshold
        self.cache_entries = []  # (embedding, response)
    
    def embed(self, text: str) -> list[float]:
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=[text]
        )
        return response.data[0].embedding
    
    def get(self, query: str) -> Optional[str]:
        if not self.cache_entries:
            return None
        
        query_emb = np.array(self.embed(query))
        
        for cached_emb, response in self.cache_entries:
            similarity = np.dot(query_emb, cached_emb) / (
                np.linalg.norm(query_emb) * np.linalg.norm(cached_emb)
            )
            if similarity >= self.threshold:
                return response
        
        return None
    
    def set(self, query: str, response: str):
        embedding = np.array(self.embed(query))
        self.cache_entries.append((embedding, response))

import json
from openai import OpenAI

client = OpenAI()

def process_batch(items: list[dict], output_path: str = "batch_output.jsonl") -> str:
    """
    Process items in batch for 50% discount.
    items: list of {"id": "...", "prompt": "..."}
    Returns batch job ID.
    """
    
    # Create JSONL input file
    requests = [
        {
            "custom_id": item["id"],
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4o",      # Full model at half price!
                "messages": [
                    {"role": "user", "content": item["prompt"]}
                ],
                "max_tokens": 500
            }
        }
        for item in items
    ]
    
    with open("batch_input.jsonl", "w") as f:
        for req in requests:
            f.write(json.dumps(req) + "\n")
    
    # Upload and create batch
    with open("batch_input.jsonl", "rb") as f:
        batch_file = client.files.create(file=f, purpose="batch")
    
    batch = client.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h"
    )
    
    print(f"Batch created: {batch.id}")
    print(f"Items: {len(requests)}")
    print(f"Estimated cost (50% discount applies): calculate based on tokens")
    
    return batch.id

def retrieve_batch_results(batch_id: str) -> list[dict]:
    """Retrieve completed batch results."""
    
    batch = client.batches.retrieve(batch_id)
    
    if batch.status != "completed":
        print(f"Status: {batch.status}")
        return []
    
    results_file = client.files.content(batch.output_file_id)
    
    results = []
    for line in results_file.text.splitlines():
        result = json.loads(line)
        results.append({
            "id": result["custom_id"],
            "response": result["response"]["body"]["choices"][0]["message"]["content"]
        })
    
    return results

# Example: Process 1,000 product descriptions overnight
products = [
    {"id": f"product_{i}", "prompt": f"Write a 100-word product description for SKU-{i}"}
    for i in range(1000)
]

batch_id = process_batch(products)
# Come back tomorrow, retrieve results at half price
results = retrieve_batch_results(batch_id)

def compress_prompt(prompt: str, max_tokens: int = 500) -> str:
    """Use a cheap model to compress a long prompt."""
    
    if len(prompt.split()) < 200:  # Already short
        return prompt
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Cheap for compression
        messages=[
            {
                "role": "system",
                "content": "Compress the following text to essential information only. "
                           "Remove redundancy and verbose language. "
                           "Preserve all specific facts, numbers, and key points."
            },
            {"role": "user", "content": prompt}
        ],
        max_tokens=max_tokens
    )
    
    return response.choices[0].message.content

# Before: 5,000 token document as context
# After: 500 token compressed summary
# Cost reduction: 10× on input tokens

# Combined with RAG: send only relevant chunks, not full document
# Combined with compression: 50-500× cost reduction on context

def calculate_monthly_costs(scenarios: list[dict]) -> None:
    """Calculate and compare costs across scenarios."""
    
    print("Monthly Cost Comparison")
    print("=" * 60)
    
    for scenario in scenarios:
        model = scenario["model"]
        daily_queries = scenario["daily_queries"]
        avg_input = scenario["avg_input_tokens"]
        avg_output = scenario["avg_output_tokens"]
        cache_hit_rate = scenario.get("cache_hit_rate", 0)
        batch_fraction = scenario.get("batch_fraction", 0)
        
        rates = PRICING.get(model, {"input": 0, "output": 0})
        
        # Effective queries after caching
        billable_queries = daily_queries * (1 - cache_hit_rate)
        
        # Real-time portion
        rt_queries = billable_queries * (1 - batch_fraction)
        batch_queries = billable_queries * batch_fraction
        
        # Costs
        rt_cost = rt_queries * (
            avg_input / 1_000_000 * rates["input"] +
            avg_output / 1_000_000 * rates["output"]
        ) * 30
        
        batch_cost = batch_queries * (
            avg_input / 1_000_000 * rates["input"] * 0.5 +  # 50% discount
            avg_output / 1_000_000 * rates["output"] * 0.5
        ) * 30
        
        total = rt_cost + batch_cost
        
        print(f"\n{scenario['name']}:")
        print(f"  Model: {model}")
        print(f"  Cache hit rate: {cache_hit_rate*100:.0f}%")
        print(f"  Monthly cost: ${total:.2f}")

scenarios = [
    {
        "name": "Before optimization",
        "model": "gpt-4o",
        "daily_queries": 1000,
        "avg_input_tokens": 8000,
        "avg_output_tokens": 500,
        "cache_hit_rate": 0,
        "batch_fraction": 0
    },
    {
        "name": "After optimization",
        "model": "gpt-4o-mini",
        "daily_queries": 1000,
        "avg_input_tokens": 1500,  # RAG: only relevant chunks
        "avg_output_tokens": 300,  # Concise prompts
        "cache_hit_rate": 0.25,    # 25% cache hit rate
        "batch_fraction": 0.3      # 30% batch processing
    }
]

calculate_monthly_costs(scenarios)
# Before: ~$2,475/month
# After:  ~$52/month