How to Reduce AI API Costs by 80% - Complete Developer Guide 2026

from openai import OpenAI

client = OpenAI(
    api_key="sk-your-api-key",
    base_url="https://crazyrouter.com/v1"
)

def get_optimal_model(task_complexity, budget_tier):
    """Select model based on task requirements"""

    if task_complexity == "simple":
        if budget_tier == "free":
            return "gemini-2.0-flash-exp"  # Free!
        return "deepseek-chat"  # $0.21/1M tokens

    elif task_complexity == "medium":
        return "claude-sonnet-4.5"  # $4.50/1M tokens

    else:  # complex
        return "claude-opus-4.5"  # $22.50/1M tokens

# Example usage
task = "Extract email from text"  # Simple task
model = get_optimal_model("simple", "free")

response = client.chat.completions.create(
    model=model,
    messages=[{"role": "user", "content": "Extract email: Contact us at hello@example.com"}]
)

print(f"Model used: {model}")
print(f"Result: {response.choices[0].message.content}")

# Inefficient: 150 tokens
prompt = """
I need you to analyze the following customer feedback and provide a detailed
summary of the main points, including sentiment analysis, key themes, and
actionable recommendations. Please be thorough and consider all aspects of
the feedback. Here is the feedback: "Great product but shipping was slow."
"""

# Efficient: 25 tokens
prompt = """
Analyze feedback: "Great product but shipping was slow."
Output: sentiment, themes, actions (brief)
"""

def optimize_prompt(user_input, task_type):
    """Generate optimized prompts"""

    templates = {
        "summarize": f"Summarize in 3 bullets: {user_input}",
        "extract": f"Extract {task_type}: {user_input}",
        "classify": f"Classify as [options]: {user_input}",
        "translate": f"Translate to {task_type}: {user_input}"
    }

    return templates.get(task_type, user_input)

# Example
original = "Please provide a comprehensive summary of the following article..."
optimized = optimize_prompt(article_text, "summarize")

# Original: ~50 tokens
# Optimized: ~10 tokens
# Savings: 80%

import hashlib
import json
from functools import lru_cache

class AICache:
    def __init__(self):
        self.cache = {}

    def get_cache_key(self, model, messages):
        """Generate cache key from request"""
        content = json.dumps({"model": model, "messages": messages}, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()

    def get(self, model, messages):
        """Get cached response"""
        key = self.get_cache_key(model, messages)
        return self.cache.get(key)

    def set(self, model, messages, response):
        """Cache response"""
        key = self.get_cache_key(model, messages)
        self.cache[key] = response

# Usage
cache = AICache()

def get_ai_response(model, messages):
    # Check cache first
    cached = cache.get(model, messages)
    if cached:
        print("Cache hit! Saved API call")
        return cached

    # Make API call
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )

    # Cache result
    cache.set(model, messages, response)
    return response

# Example: Same question asked twice
messages = [{"role": "user", "content": "What is Python?"}]

response1 = get_ai_response("deepseek-chat", messages)  # API call
response2 = get_ai_response("deepseek-chat", messages)  # Cache hit!

import redis
import json

class RedisAICache:
    def __init__(self, redis_url="redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.ttl = 3600  # 1 hour

    def get(self, key):
        data = self.redis.get(key)
        return json.loads(data) if data else None

    def set(self, key, value):
        self.redis.setex(key, self.ttl, json.dumps(value))

# Usage
cache = RedisAICache()

def cached_completion(model, messages):
    cache_key = f"ai:{model}:{hash(str(messages))}"

    # Try cache
    cached = cache.get(cache_key)
    if cached:
        return cached

    # API call
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )

    # Cache for 1 hour
    cache.set(cache_key, response.model_dump())
    return response

def stream_with_timeout(model, messages, max_tokens=500):
    """Stream with token limit to control costs"""

    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=max_tokens,  # Hard limit
        stream=True
    )

    tokens_used = 0
    for chunk in stream:
        if chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            tokens_used += len(content.split())  # Approximate

            # Stop if approaching limit
            if tokens_used > max_tokens * 0.9:
                break

            yield content

# Usage
for text in stream_with_timeout("claude-sonnet-4.5", messages):
    print(text, end="", flush=True)

async def batch_process(items, model="deepseek-chat"):
    """Process multiple items efficiently"""

    import asyncio

    async def process_one(item):
        response = await client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": f"Summarize: {item}"}],
            max_tokens=50  # Limit output
        )
        return response.choices[0].message.content

    # Process in parallel (but respect rate limits)
    results = await asyncio.gather(*[process_one(item) for item in items])
    return results

# Example: Process 100 items
items = ["Article 1...", "Article 2...", ...]  # 100 items
summaries = await batch_process(items)

# Cost comparison:
# Sequential with gpt-5: $5.00
# Batch with deepseek: $0.21
# Savings: 95.8%

def safe_completion(model, messages, max_input_tokens=1000, max_output_tokens=500):
    """Completion with token limits"""

    # Truncate input if needed
    input_text = messages[-1]["content"]
    if len(input_text.split()) > max_input_tokens:
        words = input_text.split()[:max_input_tokens]
        messages[-1]["content"] = " ".join(words) + "..."

    # Set output limit
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=max_output_tokens
    )

    return response

# Usage
response = safe_completion(
    "claude-sonnet-4.5",
    [{"role": "user", "content": very_long_text}],
    max_input_tokens=500,
    max_output_tokens=200
)

# Inefficient: Model generates verbose JSON
response = client.chat.completions.create(
    model="gpt-5",
    messages=[{
        "role": "user",
        "content": "Extract name, email, phone from: John Doe, john@example.com, 555-1234"
    }]
)

# Output: ~100 tokens of explanation + JSON

# Efficient: Structured output only
tools = [{
    "type": "function",
    "function": {
        "name": "extract_contact",
        "parameters": {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "email": {"type": "string"},
                "phone": {"type": "string"}
            }
        }
    }
}]

response = client.chat.completions.create(
    model="gpt-5",
    messages=[{"role": "user", "content": "John Doe, john@example.com, 555-1234"}],
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "extract_contact"}}
)

# Output: ~20 tokens (just the data)
# Savings: 80%

class CostMonitor:
    def __init__(self, daily_budget=100):
        self.daily_budget = daily_budget
        self.daily_spend = 0

    def estimate_cost(self, model, input_tokens, output_tokens):
        """Estimate cost for a request"""

        pricing = {
            "gpt-5": {"input": 5.00, "output": 25.00},
            "claude-opus-4.5": {"input": 7.50, "output": 37.50},
            "claude-sonnet-4.5": {"input": 1.50, "output": 7.50},
            "deepseek-chat": {"input": 0.14, "output": 0.28}
        }

        rates = pricing.get(model, {"input": 1.0, "output": 1.0})
        cost = (input_tokens * rates["input"] + output_tokens * rates["output"]) / 1_000_000

        return cost

    def check_budget(self, estimated_cost):
        """Check if request fits budget"""

        if self.daily_spend + estimated_cost > self.daily_budget:
            raise Exception(f"Daily budget exceeded! Spent: ${self.daily_spend:.2f}")

        return True

    def record_usage(self, cost):
        """Record actual usage"""
        self.daily_spend += cost

# Usage
monitor = CostMonitor(daily_budget=50)

def monitored_completion(model, messages):
    # Estimate cost
    input_tokens = sum(len(m["content"].split()) for m in messages) * 1.3
    estimated_output = 500
    estimated_cost = monitor.estimate_cost(model, input_tokens, estimated_output)

    # Check budget
    monitor.check_budget(estimated_cost)

    # Make request
    response = client.chat.completions.create(model=model, messages=messages)

    # Record actual cost
    actual_cost = monitor.estimate_cost(
        model,
        response.usage.prompt_tokens,
        response.usage.completion_tokens
    )
    monitor.record_usage(actual_cost)

    return response

from openai import OpenAI
import hashlib
import json

class CostOptimizedAI:
    def __init__(self, api_key, daily_budget=100):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://crazyrouter.com/v1"
        )
        self.cache = {}
        self.daily_spend = 0
        self.daily_budget = daily_budget

    def get_optimal_model(self, task_complexity):
        """Select cheapest model for task"""
        models = {
            "simple": "deepseek-chat",      # $0.21/1M
            "medium": "claude-sonnet-4.5",  # $4.50/1M
            "complex": "claude-opus-4.5"    # $22.50/1M
        }
        return models.get(task_complexity, "deepseek-chat")

    def optimize_prompt(self, prompt):
        """Shorten prompt while preserving meaning"""
        # Remove unnecessary words
        prompt = prompt.replace("please", "").replace("kindly", "")
        prompt = prompt.replace("I would like you to", "")
        return prompt.strip()

    def get_cache_key(self, model, prompt):
        """Generate cache key"""
        return hashlib.md5(f"{model}:{prompt}".encode()).hexdigest()

    def complete(self, prompt, task_complexity="simple", max_tokens=500):
        """Cost-optimized completion"""

        # 1. Optimize prompt
        prompt = self.optimize_prompt(prompt)

        # 2. Select optimal model
        model = self.get_optimal_model(task_complexity)

        # 3. Check cache
        cache_key = self.get_cache_key(model, prompt)
        if cache_key in self.cache:
            print(f"Cache hit! Saved ${self.estimate_cost(model, len(prompt.split()), max_tokens):.4f}")
            return self.cache[cache_key]

        # 4. Check budget
        estimated_cost = self.estimate_cost(model, len(prompt.split()), max_tokens)
        if self.daily_spend + estimated_cost > self.daily_budget:
            raise Exception("Daily budget exceeded!")

        # 5. Make API call
        response = self.client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=max_tokens
        )

        # 6. Cache result
        result = response.choices[0].message.content
        self.cache[cache_key] = result

        # 7. Track spending
        actual_cost = self.estimate_cost(
            model,
            response.usage.prompt_tokens,
            response.usage.completion_tokens
        )
        self.daily_spend += actual_cost

        print(f"Cost: ${actual_cost:.4f} | Daily total: ${self.daily_spend:.2f}")

        return result

    def estimate_cost(self, model, input_tokens, output_tokens):
        """Estimate cost"""
        pricing = {
            "deepseek-chat": {"input": 0.14, "output": 0.28},
            "claude-sonnet-4.5": {"input": 1.50, "output": 7.50},
            "claude-opus-4.5": {"input": 7.50, "output": 37.50}
        }
        rates = pricing.get(model, {"input": 1.0, "output": 1.0})
        return (input_tokens * rates["input"] + output_tokens * rates["output"]) / 1_000_000

# Usage
ai = CostOptimizedAI("sk-your-api-key", daily_budget=10)

# Simple task - uses cheapest model
result1 = ai.complete("Summarize: AI is transforming industries", task_complexity="simple")

# Same query - uses cache
result2 = ai.complete("Summarize: AI is transforming industries", task_complexity="simple")

# Complex task - uses better model
result3 = ai.complete("Analyze the philosophical implications...", task_complexity="complex")