How to Reduce AI API Costs by 40% Without Sacrificing Quality

def route_to_model(task_type, complexity_score): """ Route requests to appropriate model tier based on task requirements """ if complexity_score < 3 or task_type in ['classify', 'sentiment', 'simple_qa']: return { 'model': 'gpt-3.5-turbo', 'max_tokens': 150, 'temperature': 0.3 } elif complexity_score < 7 or task_type in ['summarize', 'review', 'reasoning']: return { 'model': 'gpt-4o-mini', 'max_tokens': 500, 'temperature': 0.5 } else: return { 'model': 'gpt-4', 'max_tokens': 1500, 'temperature': 0.7 } # Example usage task = { 'type': 'classify', 'complexity': 2, 'prompt': 'Is this email spam: "Great deal on watches..."' } config = route_to_model(task['type'], task['complexity']) response = openai.ChatCompletion.create( model=config['model'], messages=[{"role": "user", "content": task['prompt']}], max_tokens=config['max_tokens'], temperature=config['temperature'] ) 

prompt = """ I need you to analyze the following customer feedback and provide me with a detailed sentiment analysis. Please consider the emotional tone, the specific issues mentioned, and whether the customer seems satisfied or unsatisfied with our service. Here is the feedback: "{customer_feedback}" Please provide your analysis in a structured format with clear sections for sentiment, key issues, and recommendations. """ 

prompt = f"""Analyze sentiment: Feedback: {customer_feedback} Output: sentiment (positive/negative/neutral), key issues (list), recommendation (1 sentence)""" 

def build_efficient_prompt(task, input_data, output_format): """ Build token-efficient prompts using structured templates """ return f"""{task} Input: {input_data} Format: {output_format}""" # Example prompt = build_efficient_prompt( task="Extract entities", input_data="Apple announced iPhone 15 in California", output_format="JSON: {company, product, location}" ) 

class ContextManager { constructor(maxContextTokens = 2000) { this.messages = []; this.maxTokens = maxContextTokens; } addMessage(role, content) { this.messages.push({ role, content }); this.pruneIfNeeded(); } pruneIfNeeded() { // Estimate tokens (rough: 1 token ≈ 4 characters) let estimatedTokens = this.messages.reduce( (sum, msg) => sum + msg.content.length / 4, 0 ); // Keep system message + last N messages under limit while (estimatedTokens > this.maxTokens && this.messages.length > 2) { // Remove oldest non-system message const indexToRemove = this.messages.findIndex( msg => msg.role !== 'system' ); if (indexToRemove !== -1) { const removed = this.messages.splice(indexToRemove, 1)[0]; estimatedTokens -= removed.content.length / 4; } } } getContext() { return this.messages; } } // Usage const context = new ContextManager(2000); context.addMessage('system', 'You are a helpful assistant'); context.addMessage('user', 'What is AI?'); context.addMessage('assistant', 'AI is...'); // Automatically prunes old messages when limit is reached 

# This can run forever and cost you a fortune response = openai.ChatCompletion.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], stream=True ) for chunk in response: print(chunk) # No stop condition! 

def stream_with_limit(prompt, max_tokens=500, stop_patterns=None): """ Stream responses with hard token limits and early stopping """ response = openai.ChatCompletion.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], max_tokens=max_tokens, # Hard limit  stop=stop_patterns or ["\n\n---\n\n", "END"], # Stop sequences  stream=True ) accumulated = "" token_count = 0 for chunk in response: if chunk.choices[0].delta.get('content'): content = chunk.choices[0].delta.content accumulated += content token_count += len(content) / 4 # Rough estimate  # Additional safeguard  if token_count > max_tokens * 1.1: break yield content return accumulated # Usage with early stopping result = stream_with_limit( "Explain quantum computing", max_tokens=300, stop_patterns=["\n\nIn conclusion", "Summary:"] ) 

import redis import hashlib import json class LLMCache: def __init__(self, redis_client, ttl=3600): self.redis = redis_client self.ttl = ttl def _generate_key(self, model, prompt, params): """Generate deterministic cache key""" cache_input = f"{model}:{prompt}:{json.dumps(params, sort_keys=True)}" return f"llm_cache:{hashlib.sha256(cache_input.encode()).hexdigest()}" def get(self, model, prompt, params): """Try to retrieve cached response""" key = self._generate_key(model, prompt, params) cached = self.redis.get(key) if cached: return json.loads(cached) return None def set(self, model, prompt, params, response): """Cache the response""" key = self._generate_key(model, prompt, params) self.redis.setex(key, self.ttl, json.dumps(response)) # Usage cache = LLMCache(redis.Redis(host='localhost', port=6379)) def cached_completion(model, prompt, **params): # Check cache first  cached_response = cache.get(model, prompt, params) if cached_response: return cached_response # Make API call  response = openai.ChatCompletion.create( model=model, messages=[{"role": "user", "content": prompt}], **params ) # Cache for future requests  cache.set(model, prompt, params, response) return response 

results = [] for item in data_items: # 100 items  response = call_api(item) # 100 API calls  results.append(response) 

def batch_process(items, batch_size=10): """Process multiple items in a single API call""" results = [] for i in range(0, len(items), batch_size): batch = items[i:i+batch_size] # Create batched prompt  prompt = "Process each item:\n" for idx, item in enumerate(batch): prompt += f"{idx+1}. {item}\n" prompt += "\nReturn JSON array with results in same order." response = call_api(prompt) batch_results = json.loads(response) results.extend(batch_results) return results # 100 items -> 10 API calls (90% reduction) 

class MultiProviderRouter: def __init__(self): self.providers = { 'openai': {'cost_per_1k': 0.03, 'speed': 'fast'}, 'anthropic': {'cost_per_1k': 0.025, 'speed': 'medium'}, 'google': {'cost_per_1k': 0.02, 'speed': 'fast'} } def select_provider(self, task_priority='cost'): """Route based on cost or speed priority""" if task_priority == 'cost': return min(self.providers.items(), key=lambda x: x[1]['cost_per_1k']) elif task_priority == 'speed': speed_order = {'fast': 0, 'medium': 1, 'slow': 2} return min(self.providers.items(), key=lambda x: speed_order[x[1]['speed']]) 

# Simple cost tracking decorator def track_cost(func): def wrapper(*args, **kwargs): start_time = time.time() response = func(*args, **kwargs) duration = time.time() - start_time # Log metrics  log_metric({ 'function': func.__name__, 'model': kwargs.get('model'), 'tokens': response.get('usage', {}).get('total_tokens', 0), 'cost': calculate_cost(response), 'duration': duration }) return response return wrapper