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")