Cost Optimization for LLM Applications: Managing Token Budgets and Scaling Efficiently

# Anti-pattern: Sending full conversation history every message
def chat_with_context(messages, new_message):
    full_context = "\n".join([msg['content'] for msg in messages])
    # This grows linearly with conversation length
    prompt = f"{full_context}\n{new_message}"
    response = llm.complete(prompt)  # Costs increase by 2-3x over 10 messages

# Anti-pattern: Re-generating identical responses
def get_product_description(product_id):
    prompt = f"Generate a description for product {product_id}"
    # This same product description gets generated 1000x/day
    # Cost: 1000 × $0.002 = $2/day = $60/month for ONE product
    return llm.complete(prompt)

# Anti-pattern: Using GPT-4 for every task
def classify_sentiment(text):
    # GPT-4: $0.03/1K tokens
    # This task could use GPT-3.5-turbo: $0.001/1K tokens (30x cheaper)
    return gpt4.complete(f"Classify sentiment: {text}")

# Anti-pattern: Verbose instructions repeated every call
SYSTEM_PROMPT = """
You are a helpful assistant. Please be polite and professional.
Always provide accurate information. If you don't know something,
say so. Format responses in markdown. Use bullet points where
appropriate. Keep responses concise but comprehensive...
""" # 45 tokens × every single request

def ask_question(question):
    # This 45-token overhead costs $0.00135 per call
    # At 10K calls/month: $13.50/month just for instructions
    return llm.complete(f"{SYSTEM_PROMPT}\n{question}")

import tiktoken

class TokenBudgetManager:
    MODEL_PRICING = {
        'gpt-4': {'input': 0.03, 'output': 0.06},
        'gpt-3.5-turbo': {'input': 0.001, 'output': 0.002},
    }

    def __init__(self, model: str = 'gpt-3.5-turbo'):
        self.model = model
        self.encoding = tiktoken.encoding_for_model(model)

    def count_tokens(self, text: str) -> int:
        return len(self.encoding.encode(text))

    def estimate_cost(self, input_text: str, output_text: str) -> dict:
        input_tokens = self.count_tokens(input_text)
        output_tokens = self.count_tokens(output_text)
        pricing = self.MODEL_PRICING[self.model]

        return {
            'input_tokens': input_tokens,
            'output_tokens': output_tokens,
            'total_cost': (input_tokens/1000 * pricing['input']) +
                          (output_tokens/1000 * pricing['output'])
        }

    def truncate_to_budget(self, text: str, max_tokens: int) -> str:
        tokens = self.encoding.encode(text)
        if len(tokens) <= max_tokens:
            return text
        return self.encoding.decode(tokens[:max_tokens-3]) + "..."

# Usage
manager = TokenBudgetManager()
prompt = "Analyze this text..."
response = llm.complete(prompt)

cost = manager.estimate_cost(prompt, response)
print(f"Cost: ${cost['total_cost']:.4f}, Tokens: {cost['input_tokens']}+{cost['output_tokens']}")

# Anti-pattern: Exponential cost growth
conversation = [
    {"role": "user", "content": "Hello"},  # Turn 1: 1 message (2 tokens)
    {"role": "assistant", "content": "Hi there!"},  # Turn 2: 2 messages (6 tokens total)
    {"role": "user", "content": "How are you?"},  # Turn 3: 3 messages (12 tokens total)
    # By turn 10: sending 10 messages every time
    # Cumulative cost: 2 + 6 + 12 + 20 + 30 + 42 + 56 + 72 + 90 + 110 = 440 tokens
    # vs. optimal: 2 + 4 + 4 + 4 + 4 + 4 + 4 + 4 + 4 + 4 = 38 tokens (91% reduction)
]

class ConversationManager:
    """Manage conversation context with automatic cost optimization"""

    def __init__(self, max_history_tokens: int = 500):
        self.max_history_tokens = max_history_tokens
        self.budget_manager = TokenBudgetManager()
        self.full_history: list[Dict] = []
        self.summary: Optional[str] = None

    def add_message(self, role: str, content: str):
        """Add message to history"""
        self.full_history.append({"role": role, "content": content})

    def get_optimized_context(self) -> str:
        """Build context that fits within token budget"""
        # Always keep last 2 exchanges (immediate context)
        recent_messages = self.full_history[-4:]  # Last 2 user + 2 assistant
        recent_text = self._format_messages(recent_messages)
        recent_tokens = self.budget_manager.count_tokens(recent_text)

        # If recent history fits budget, use it
        if recent_tokens <= self.max_history_tokens:
            return recent_text

        # If recent history too large, summarize older context
        if len(self.full_history) > 4:
            older_messages = self.full_history[:-4]

            # Summarize older conversation (this itself costs tokens, but amortized)
            if not self.summary or len(older_messages) > 10:
                self.summary = self._summarize_conversation(older_messages)

            # Combine summary + recent context
            available_for_recent = self.max_history_tokens - self.budget_manager.count_tokens(self.summary)
            recent_text = self._format_messages(recent_messages, max_tokens=available_for_recent)

            return f"Previous conversation summary:\n{self.summary}\n\n{recent_text}"

        # If even recent history too large, truncate
        return self._format_messages(recent_messages, max_tokens=self.max_history_tokens)

    def _summarize_conversation(self, messages: list[Dict]) -> str:
        """Generate concise summary of conversation history"""
        conversation_text = self._format_messages(messages)

        prompt = f"""Summarize this conversation in 2-3 sentences, focusing on key facts and decisions:

{conversation_text}

Summary:"""

        # Use cheaper model for summarization
        summary_manager = TokenBudgetManager(model='gpt-3.5-turbo')
        response = call_llm_api(prompt, max_tokens=100, temperature=0.3)

        return response.strip()

    def _format_messages(self, messages: list[Dict], max_tokens: int = None) -> str:
        """Format messages into prompt text"""
        text = "\n".join([f"{m['role']}: {m['content']}" for m in messages])

        if max_tokens:
            text = self.budget_manager.truncate_to_budget(text, max_tokens)

        return text

    def get_cost_savings(self) -> Dict:
        """Calculate savings from context optimization"""
        # Compare naive vs optimized approach
        naive_tokens = sum(
            self.budget_manager.count_tokens(self._format_messages(self.full_history[:i+1]))
            for i in range(len(self.full_history))
        )

        optimized_tokens = len(self.full_history) * self.max_history_tokens  # Upper bound

        return {
            'naive_tokens': naive_tokens,
            'optimized_tokens': optimized_tokens,
            'tokens_saved': naive_tokens - optimized_tokens,
            'cost_saved': (naive_tokens - optimized_tokens) / 1000 * 0.001,  # Approximate
            'reduction_percent': ((naive_tokens - optimized_tokens) / naive_tokens * 100) if naive_tokens > 0 else 0
        }

# Usage example
conversation_mgr = ConversationManager(max_history_tokens=500)

# Simulate 10-turn conversation
for i in range(10):
    conversation_mgr.add_message("user", f"This is user message {i}")

    # Get optimized context for API call
    context = conversation_mgr.get_optimized_context()

    # Make API call with optimized context (50-70% fewer tokens)
    response = call_llm_api(f"{context}\n\nAssistant: ")
    conversation_mgr.add_message("assistant", response)

# Check savings
savings = conversation_mgr.get_cost_savings()
print(f"Tokens saved: {savings['tokens_saved']} ({savings['reduction_percent']:.1f}% reduction)")
print(f"Estimated cost saved: ${savings['cost_saved']:.4f}")

class PromptOptimizer:
    """Compress prompts without losing semantic meaning"""

    def __init__(self):
        self.budget_manager = TokenBudgetManager()

        # Common verbose patterns and their compressed equivalents
        self.compression_rules = {
            # Remove filler words
            r'\b(please|kindly|very|really|just|actually)\b': '',

            # Compress instructions
            'You are a helpful assistant': 'Assist:',
            'Generate a response': 'Generate:',
            'Based on the following': 'From:',
            'Please provide': 'Provide:',
            'Could you please': '',

            # Compress formatting instructions
            'Format your response in JSON': '[JSON]',
            'Use markdown formatting': '[MD]',
            'Keep it concise': '[CONCISE]',
            'Be professional': '[PRO]',
        }

    def compress_prompt(self, prompt: str, target_reduction: float = 0.3) -> tuple[str, Dict]:
        """Compress prompt by target percentage while preserving meaning"""
        original_tokens = self.budget_manager.count_tokens(prompt)
        target_tokens = int(original_tokens * (1 - target_reduction))

        compressed = prompt

        # Step 1: Apply rule-based compression
        import re
        for pattern, replacement in self.compression_rules.items():
            compressed = re.sub(pattern, replacement, compressed, flags=re.IGNORECASE)

        # Step 2: Remove extra whitespace
        compressed = ' '.join(compressed.split())

        # Step 3: Abbreviate common terms (preserve meaning)
        abbreviations = {
            'information': 'info',
            'documentation': 'docs',
            'description': 'desc',
            'example': 'ex',
            'following': 'ff:',
        }
        for full, abbrev in abbreviations.items():
            compressed = compressed.replace(full, abbrev)

        # Step 4: If still too long, use extractive summarization
        current_tokens = self.budget_manager.count_tokens(compressed)
        if current_tokens > target_tokens:
            # Extract most important sentences
            sentences = compressed.split('.')
            # Keep first and last sentences (usually most important)
            if len(sentences) > 3:
                compressed = f"{sentences[0]}.{sentences[-1]}"

        final_tokens = self.budget_manager.count_tokens(compressed)

        return compressed, {
            'original_tokens': original_tokens,
            'compressed_tokens': final_tokens,
            'tokens_saved': original_tokens - final_tokens,
            'reduction_percent': ((original_tokens - final_tokens) / original_tokens * 100),
            'original_length': len(prompt),
            'compressed_length': len(compressed)
        }

    def test_compression_quality(self, original_prompt: str,
                                compressed_prompt: str,
                                test_input: str) -> Dict:
        """Verify compressed prompt produces equivalent results"""
        # Get responses from both prompts
        original_response = call_llm_api(f"{original_prompt}\n{test_input}")
        compressed_response = call_llm_api(f"{compressed_prompt}\n{test_input}")

        # Compare semantic similarity (simplified - use embeddings in production)
        from difflib import SequenceMatcher
        similarity = SequenceMatcher(None, original_response, compressed_response).ratio()

        return {
            'semantic_similarity': similarity,
            'quality_preserved': similarity > 0.85,  # 85% similarity threshold
            'original_response': original_response,
            'compressed_response': compressed_response
        }

# Example: Compress a verbose prompt
verbose_prompt = """
You are a helpful AI assistant. Please help users with their questions.
Be very polite and professional in all your responses. If you don't know
something, please just say so honestly. Always format your responses using
markdown. Keep your responses concise but comprehensive. Please provide
examples where appropriate. Based on the following user input, generate a
helpful response:
"""

optimizer = PromptOptimizer()
compressed, stats = optimizer.compress_prompt(verbose_prompt, target_reduction=0.5)

print(f"Original: {stats['original_tokens']} tokens")
print(f"Compressed: {stats['compressed_tokens']} tokens")
print(f"Saved: {stats['tokens_saved']} tokens ({stats['reduction_percent']:.1f}%)")
print(f"\nCompressed prompt:\n{compressed}")

# Verify quality maintained
quality = optimizer.test_compression_quality(
    verbose_prompt,
    compressed,
    "What is the capital of France?"
)
print(f"\nQuality preserved: {quality['quality_preserved']}")
print(f"Semantic similarity: {quality['semantic_similarity']:.2%}")

import hashlib
import json
import time
from typing import Optional, Any
from functools import lru_cache
import redis

class LLMCacheManager:
    """Multi-layer caching for LLM responses"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.budget_manager = TokenBudgetManager()

        # Cache TTLs by use case
        self.ttls = {
            'exact_match': 86400 * 7,  # 7 days for exact query matches
            'semantic_similar': 3600,    # 1 hour for similar queries
            'static_content': 86400 * 30,  # 30 days for static content (product descriptions)
        }

    def generate_cache_key(self, prompt: str, model: str, **kwargs) -> str:
        """Generate deterministic cache key"""
        # Normalize prompt (remove extra whitespace, lowercase)
        normalized = ' '.join(prompt.lower().split())

        # Include model and key parameters in hash
        cache_data = {
            'prompt': normalized,
            'model': model,
            'temperature': kwargs.get('temperature', 0),
            'max_tokens': kwargs.get('max_tokens', 0)
        }

        cache_string = json.dumps(cache_data, sort_keys=True)
        return hashlib.sha256(cache_string.encode()).hexdigest()

    def get_cached_response(self, cache_key: str) -> Optional[Dict]:
        """Retrieve cached response with metadata"""
        cached = self.redis.get(f"llm:response:{cache_key}")

        if cached:
            data = json.loads(cached)

            # Update cache stats
            self.redis.hincrby(f"llm:stats:{cache_key}", 'hits', 1)

            return {
                'response': data['response'],
                'cached': True,
                'cache_timestamp': data['timestamp'],
                'cache_age_seconds': time.time() - data['timestamp'],
                'tokens_saved': data['tokens'],
                'cost_saved': data['cost']
            }

        return None

    def cache_response(self, cache_key: str, response: str,
                      tokens: int, cost: float, ttl: int):
        """Store response in cache"""
        cache_data = {
            'response': response,
            'timestamp': time.time(),
            'tokens': tokens,
            'cost': cost
        }

        self.redis.setex(
            f"llm:response:{cache_key}",
            ttl,
            json.dumps(cache_data)
        )

        # Initialize cache stats
        self.redis.hset(f"llm:stats:{cache_key}", mapping={
            'hits': 0,
            'created': time.time()
        })

    def get_cache_stats(self, hours: int = 24) -> Dict:
        """Analyze cache performance"""
        pattern = "llm:stats:*"
        cache_keys = self.redis.keys(pattern)

        total_hits = 0
        total_cost_saved = 0

        for key in cache_keys:
            stats = self.redis.hgetall(key)
            hits = int(stats.get(b'hits', 0))

            # Get original response cost from cached data
            response_key = key.decode().replace('stats', 'response')
            cached_data = self.redis.get(response_key)

            if cached_data:
                data = json.loads(cached_data)
                cost_saved = hits * data['cost']  # Cost saved by cache hits

                total_hits += hits
                total_cost_saved += cost_saved

        return {
            'cache_hits': total_hits,
            'cost_saved': total_cost_saved,
            'unique_cached_prompts': len(cache_keys),
            'avg_cost_per_hit': total_cost_saved / total_hits if total_hits > 0 else 0
        }

class SemanticCacheManager:
    """Cache similar prompts using embeddings"""

    def __init__(self, redis_client: redis.Redis, similarity_threshold: float = 0.95):
        self.redis = redis_client
        self.similarity_threshold = similarity_threshold
        self.embedding_model = 'text-embedding-ada-002'  # OpenAI embeddings

    def get_embedding(self, text: str) -> list[float]:
        """Get text embedding for similarity comparison"""
        # In production, call OpenAI embeddings API
        # For this example, we'll simulate
        import hashlib

        # Simulate embedding (in reality, call OpenAI API)
        # response = openai.Embedding.create(input=text, model=self.embedding_model)
        # return response['data'][0]['embedding']

        # Simulated embedding for demonstration
        hash_val = hashlib.md5(text.encode()).hexdigest()
        return [float(int(hash_val[i:i+2], 16)) / 255 for i in range(0, 32, 2)]

    def cosine_similarity(self, vec1: list[float], vec2: list[float]) -> float:
        """Calculate cosine similarity between embeddings"""
        import math

        dot_product = sum(a * b for a, b in zip(vec1, vec2))
        mag1 = math.sqrt(sum(a ** 2 for a in vec1))
        mag2 = math.sqrt(sum(b ** 2 for b in vec2))

        return dot_product / (mag1 * mag2) if mag1 and mag2 else 0

    def find_similar_cached_response(self, prompt: str) -> Optional[Dict]:
        """Find cached response for semantically similar prompt"""
        query_embedding = self.get_embedding(prompt)

        # Search through cached embeddings
        pattern = "llm:embedding:*"
        embedding_keys = self.redis.keys(pattern)

        best_match = None
        best_similarity = 0

        for key in embedding_keys:
            cached_embedding_json = self.redis.get(key)
            if not cached_embedding_json:
                continue

            cached_data = json.loads(cached_embedding_json)
            cached_embedding = cached_data['embedding']

            similarity = self.cosine_similarity(query_embedding, cached_embedding)

            if similarity > best_similarity and similarity >= self.similarity_threshold:
                best_similarity = similarity
                best_match = {
                    'cache_key': cached_data['cache_key'],
                    'original_prompt': cached_data['prompt'],
                    'similarity': similarity
                }

        if best_match:
            # Retrieve actual cached response
            cached_response = self.redis.get(f"llm:response:{best_match['cache_key']}")
            if cached_response:
                data = json.loads(cached_response)
                return {
                    'response': data['response'],
                    'cached': True,
                    'cache_type': 'semantic',
                    'similarity': best_similarity,
                    'original_prompt': best_match['original_prompt'],
                    'cost_saved': data['cost']
                }

        return None

    def cache_with_embedding(self, prompt: str, cache_key: str):
        """Store embedding for semantic lookup"""
        embedding = self.get_embedding(prompt)

        embedding_data = {
            'embedding': embedding,
            'cache_key': cache_key,
            'prompt': prompt,
            'timestamp': time.time()
        }

        self.redis.setex(
            f"llm:embedding:{cache_key}",
            3600,  # 1 hour TTL
            json.dumps(embedding_data)
        )

# Complete caching solution with both exact and semantic matching
class CompleteLLMCache:
    """Production-ready LLM caching system"""

    def __init__(self, redis_url: str = 'redis://localhost:6379'):
        self.redis = redis.from_url(redis_url)
        self.exact_cache = LLMCacheManager(self.redis)
        self.semantic_cache = SemanticCacheManager(self.redis)
        self.budget_manager = TokenBudgetManager()

    def get_or_generate(self, prompt: str, model: str = 'gpt-3.5-turbo',
                       use_semantic_cache: bool = True, **kwargs) -> Dict:
        """Get cached response or generate new one"""

        # Step 1: Try exact match cache
        cache_key = self.exact_cache.generate_cache_key(prompt, model, **kwargs)
        cached = self.exact_cache.get_cached_response(cache_key)

        if cached:
            return cached

        # Step 2: Try semantic similarity cache (for slight variations)
        if use_semantic_cache:
            semantic_match = self.semantic_cache.find_similar_cached_response(prompt)
            if semantic_match:
                return semantic_match

        # Step 3: No cache hit - generate new response
        response = call_llm_api(prompt, model=model, **kwargs)

        # Calculate cost
        metrics = self.budget_manager.estimate_cost(prompt, response)

        # Cache the response
        self.exact_cache.cache_response(
            cache_key,
            response,
            metrics.input_tokens + metrics.output_tokens,
            metrics.total_cost,
            ttl=self.exact_cache.ttls['exact_match']
        )

        # Cache embedding for semantic lookup
        if use_semantic_cache:
            self.semantic_cache.cache_with_embedding(prompt, cache_key)

        return {
            'response': response,
            'cached': False,
            'cost': metrics.total_cost,
            'tokens': metrics.input_tokens + metrics.output_tokens
        }

    def get_savings_report(self) -> Dict:
        """Generate comprehensive cache savings report"""
        stats = self.exact_cache.get_cache_stats()

        # Calculate theoretical cost without caching
        total_requests = stats['cache_hits']
        cost_without_cache = total_requests * stats['avg_cost_per_hit']

        return {
            **stats,
            'theoretical_cost_without_cache': cost_without_cache,
            'actual_cost_with_cache': cost_without_cache - stats['cost_saved'],
            'cost_reduction_percent': (stats['cost_saved'] / cost_without_cache * 100)
                                     if cost_without_cache > 0 else 0,
            'cache_hit_rate': (stats['cache_hits'] / total_requests * 100)
                             if total_requests > 0 else 0
        }

# Usage example showing dramatic cost savings
cache = CompleteLLMCache()

# Simulate repeated requests (common in production)
queries = [
    "What is the capital of France?",
    "what is the capital of france?",  # Case variation - exact cache hit
    "What's the capital city of France?",  # Semantic cache hit
    "Tell me France's capital",  # Semantic cache hit
]

for query in queries:
    result = cache.get_or_generate(query, use_semantic_cache=True)
    print(f"\nQuery: {query}")
    print(f"Cached: {result['cached']}")
    if 'cost_saved' in result:
        print(f"Cost saved: ${result['cost_saved']:.4f}")

# Check overall savings
report = cache.get_savings_report()
print(f"\n=== Cache Performance Report ===")
print(f"Cache hits: {report['cache_hits']}")
print(f"Cost saved: ${report['cost_saved']:.2f}")
print(f"Cost reduction: {report['cost_reduction_percent']:.1f}%")

from enum import Enum
from typing import Callable

class TaskComplexity(Enum):
    """Classification of task difficulty"""
    TRIVIAL = 1      # Simple classification, yes/no questions
    SIMPLE = 2       # Sentiment analysis, simple transformations
    MODERATE = 3     # Summarization, basic reasoning
    COMPLEX = 4      # Multi-step reasoning, code generation
    EXPERT = 5       # Complex analysis, creative writing

class ModelRouter:
    """Intelligently route tasks to cost-appropriate models"""

    # Model capabilities and pricing (updated 2024)
    MODEL_SPECS = {
        'gpt-4': {
            'capability_score': 10,
            'input_cost': 0.03,
            'output_cost': 0.06,
            'context_window': 8192,
            'best_for': [TaskComplexity.COMPLEX, TaskComplexity.EXPERT]
        },
        'gpt-4-turbo': {
            'capability_score': 10,
            'input_cost': 0.01,
            'output_cost': 0.03,
            'context_window': 128000,
            'best_for': [TaskComplexity.COMPLEX, TaskComplexity.EXPERT]
        },
        'gpt-3.5-turbo': {
            'capability_score': 7,
            'input_cost': 0.001,
            'output_cost': 0.002,
            'context_window': 16384,
            'best_for': [TaskComplexity.SIMPLE, TaskComplexity.MODERATE]
        },
        'gpt-3.5-turbo-instruct': {
            'capability_score': 6,
            'input_cost': 0.0015,
            'output_cost': 0.002,
            'context_window': 4096,
            'best_for': [TaskComplexity.TRIVIAL, TaskComplexity.SIMPLE]
        },
        'claude-2': {
            'capability_score': 9,
            'input_cost': 0.008,
            'output_cost': 0.024,
            'context_window': 100000,
            'best_for': [TaskComplexity.COMPLEX, TaskComplexity.EXPERT]
        },
        'claude-instant': {
            'capability_score': 6,
            'input_cost': 0.0008,
            'output_cost': 0.0024,
            'context_window': 100000,
            'best_for': [TaskComplexity.TRIVIAL, TaskComplexity.SIMPLE]
        }
    }

    def __init__(self):
        self.budget_manager = TokenBudgetManager()
        self.routing_stats = {model: {'requests': 0, 'cost': 0}
                             for model in self.MODEL_SPECS.keys()}

    def estimate_task_complexity(self, task_description: str) -> TaskComplexity:
        """Heuristically determine task complexity"""
        task_lower = task_description.lower()

        # Trivial indicators
        if any(keyword in task_lower for keyword in
               ['classify', 'yes or no', 'true or false', 'sentiment']):
            return TaskComplexity.TRIVIAL

        # Simple indicators
        if any(keyword in task_lower for keyword in
               ['extract', 'list', 'identify', 'count']):
            return TaskComplexity.SIMPLE

        # Moderate indicators
        if any(keyword in task_lower for keyword in
               ['summarize', 'explain', 'describe', 'translate']):
            return TaskComplexity.MODERATE

        # Complex indicators
        if any(keyword in task_lower for keyword in
               ['analyze', 'reason', 'solve', 'generate code']):
            return TaskComplexity.COMPLEX

        # Expert indicators
        if any(keyword in task_lower for keyword in
               ['creative writing', 'complex analysis', 'multi-step', 'research']):
            return TaskComplexity.EXPERT

        # Default to moderate if unclear
        return TaskComplexity.MODERATE

    def select_model(self, task_complexity: TaskComplexity,
                    input_tokens: int, max_budget: float = None) -> str:
        """Select most cost-effective model for task"""

        # Filter models capable of handling this complexity
        capable_models = [
            (model, specs) for model, specs in self.MODEL_SPECS.items()
            if task_complexity in specs['best_for'] and
               input_tokens <= specs['context_window']
        ]

        if not capable_models:
            # Fallback to most capable model
            return 'gpt-4-turbo'

        # If budget specified, filter by cost
        if max_budget:
            estimated_cost = lambda m: (input_tokens / 1000) * m[1]['input_cost']
            capable_models = [(m, s) for m, s in capable_models
                            if estimated_cost((m, s)) <= max_budget]

        # Select cheapest capable model
        capable_models.sort(key=lambda x: x[1]['input_cost'])
        return capable_models[0][0]

    def route_task(self, task_description: str, prompt: str,
                  max_budget: float = None, **kwargs) -> Dict:
        """Route task to optimal model and execute"""

        # Determine complexity
        complexity = self.estimate_task_complexity(task_description)

        # Count tokens
        input_tokens = self.budget_manager.count_tokens(prompt)

        # Select appropriate model
        selected_model = self.select_model(complexity, input_tokens, max_budget)
        model_specs = self.MODEL_SPECS[selected_model]

        # Execute task
        response = call_llm_api(prompt, model=selected_model, **kwargs)

        # Calculate actual cost
        output_tokens = self.budget_manager.count_tokens(response)
        cost = (input_tokens / 1000 * model_specs['input_cost'] +
                output_tokens / 1000 * model_specs['output_cost'])

        # Track routing stats
        self.routing_stats[selected_model]['requests'] += 1
        self.routing_stats[selected_model]['cost'] += cost

        return {
            'response': response,
            'model_used': selected_model,
            'task_complexity': complexity.name,
            'input_tokens': input_tokens,
            'output_tokens': output_tokens,
            'cost': cost,
            'cost_savings': self._calculate_savings(input_tokens, output_tokens,
                                                    selected_model)
        }

    def _calculate_savings(self, input_tokens: int, output_tokens: int,
                          selected_model: str) -> Dict:
        """Calculate cost savings vs. using GPT-4 for everything"""
        gpt4_cost = (input_tokens / 1000 * 0.03 + output_tokens / 1000 * 0.06)
        actual_cost = (input_tokens / 1000 * self.MODEL_SPECS[selected_model]['input_cost'] +
                      output_tokens / 1000 * self.MODEL_SPECS[selected_model]['output_cost'])

        return {
            'gpt4_cost': gpt4_cost,
            'actual_cost': actual_cost,
            'savings': gpt4_cost - actual_cost,
            'savings_percent': ((gpt4_cost - actual_cost) / gpt4_cost * 100)
                              if gpt4_cost > 0 else 0
        }

    def get_routing_summary(self) -> Dict:
        """Analyze routing decisions and savings"""
        total_cost = sum(stats['cost'] for stats in self.routing_stats.values())
        total_requests = sum(stats['requests'] for stats in self.routing_stats.values())

        # Calculate what cost would have been with GPT-4 only
        # Rough estimate: assume 70% cost increase
        gpt4_only_cost = total_cost * 1.7

        return {
            'total_requests': total_requests,
            'actual_cost': total_cost,
            'gpt4_only_cost': gpt4_only_cost,
            'cost_saved': gpt4_only_cost - total_cost,
            'savings_percent': ((gpt4_only_cost - total_cost) / gpt4_only_cost * 100),
            'by_model': self.routing_stats
        }

# Usage examples demonstrating dramatic cost savings
router = ModelRouter()

# Example 1: Simple classification (use cheap model)
result = router.route_task(
    task_description="Classify sentiment of customer review",
    prompt="Customer review: 'This product is amazing!' Sentiment:"
)
print(f"Task: Sentiment classification")
print(f"Model: {result['model_used']}")  # Will use gpt-3.5-turbo-instruct or claude-instant
print(f"Cost: ${result['cost']:.4f}")
print(f"Savings: ${result['cost_savings']['savings']:.4f} ({result['cost_savings']['savings_percent']:.0f}%)")

# Example 2: Complex reasoning (use powerful model)
result = router.route_task(
    task_description="Analyze complex business strategy",
    prompt="Given market data X, Y, Z, analyze competitive positioning and recommend strategy..."
)
print(f"\nTask: Strategic analysis")
print(f"Model: {result['model_used']}")  # Will use gpt-4-turbo or claude-2
print(f"Cost: ${result['cost']:.4f}")

# Example 3: Moderate task with budget constraint
result = router.route_task(
    task_description="Summarize customer feedback",
    prompt="Summarize these 10 customer reviews...",
    max_budget=0.005  # Force cheaper model
)
print(f"\nTask: Summarization")
print(f"Model: {result['model_used']}")  # Will use most capable model under budget
print(f"Cost: ${result['cost']:.4f}")

# Review routing performance
summary = router.get_routing_summary()
print(f"\n=== Routing Summary ===")
print(f"Total requests: {summary['total_requests']}")
print(f"Actual cost: ${summary['actual_cost']:.2f}")
print(f"Cost if GPT-4 only: ${summary['gpt4_only_cost']:.2f}")
print(f"Savings: ${summary['cost_saved']:.2f} ({summary['savings_percent']:.0f}%)")

import logging
from datetime import datetime, timedelta
from typing import List, Dict
from dataclasses import dataclass, asdict
import json

@dataclass
class LLMMetric:
    """Standard metric structure for LLM operations"""
    timestamp: float
    user_id: str
    feature: str
    model: str
    input_tokens: int
    output_tokens: int
    total_tokens: int
    cost: float
    latency_ms: float
    cached: bool
    cache_type: Optional[str]
    error: Optional[str]

class LLMObservability:
    """Complete observability system for LLM operations"""

    def __init__(self, alert_webhook: str = None):
        self.alert_webhook = alert_webhook
        self.metrics: List[LLMMetric] = []

        # Cost thresholds for alerting
        self.thresholds = {
            'hourly_cost': 10.0,      # $10/hour
            'daily_cost': 100.0,      # $100/day
            'per_request_cost': 0.50, # $0.50/request
            'error_rate': 0.05,       # 5% error rate
        }

        # Setup logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger('LLMObservability')

    def log_request(self, metric: LLMMetric):
        """Log individual request with full attribution"""
        self.metrics.append(metric)

        # Structured logging for aggregation (e.g., in Datadog, CloudWatch)
        self.logger.info(json.dumps({
            'event': 'llm_request',
            'timestamp': metric.timestamp,
            'user_id': metric.user_id,
            'feature': metric.feature,
            'model': metric.model,
            'tokens': metric.total_tokens,
            'cost': metric.cost,
            'latency_ms': metric.latency_ms,
            'cached': metric.cached,
            'error': metric.error
        }))

        # Check thresholds
        self._check_thresholds(metric)

    def _check_thresholds(self, metric: LLMMetric):
        """Real-time threshold monitoring with alerts"""

        # Check per-request cost
        if metric.cost > self.thresholds['per_request_cost']:
            self._send_alert(
                'high_cost_request',
                f"Request exceeded cost threshold: ${metric.cost:.2f}",
                metric
            )

        # Check hourly cost
        hour_ago = time.time() - 3600
        recent_cost = sum(m.cost for m in self.metrics if m.timestamp > hour_ago)

        if recent_cost > self.thresholds['hourly_cost']:
            self._send_alert(
                'high_hourly_cost',
                f"Hourly cost threshold exceeded: ${recent_cost:.2f}",
                {'hourly_cost': recent_cost, 'threshold': self.thresholds['hourly_cost']}
            )

        # Check error rate
        hour_metrics = [m for m in self.metrics if m.timestamp > hour_ago]
        if hour_metrics:
            error_rate = sum(1 for m in hour_metrics if m.error) / len(hour_metrics)

            if error_rate > self.thresholds['error_rate']:
                self._send_alert(
                    'high_error_rate',
                    f"Error rate threshold exceeded: {error_rate:.1%}",
                    {'error_rate': error_rate, 'threshold': self.thresholds['error_rate']}
                )

    def _send_alert(self, alert_type: str, message: str, data: Any):
        """Send alert to monitoring system"""
        self.logger.warning(f"ALERT [{alert_type}]: {message}")

        if self.alert_webhook:
            # Send to Slack/PagerDuty/etc
            import requests
            payload = {
                'alert_type': alert_type,
                'message': message,
                'data': data if isinstance(data, dict) else asdict(data),
                'timestamp': time.time()
            }
            try:
                requests.post(self.alert_webhook, json=payload, timeout=5)
            except Exception as e:
                self.logger.error(f"Failed to send alert: {e}")

    def get_cost_breakdown(self, hours: int = 24) -> Dict:
        """Detailed cost analysis for optimization"""
        cutoff = time.time() - (hours * 3600)
        recent = [m for m in self.metrics if m.timestamp > cutoff]

        if not recent:
            return {'error': 'No recent metrics'}

        # Aggregate by various dimensions
        by_feature = {}
        by_model = {}
        by_user = {}

        for m in recent:
            # By feature
            if m.feature not in by_feature:
                by_feature[m.feature] = {'cost': 0, 'requests': 0, 'tokens': 0}
            by_feature[m.feature]['cost'] += m.cost
            by_feature[m.feature]['requests'] += 1
            by_feature[m.feature]['tokens'] += m.total_tokens

            # By model
            if m.model not in by_model:
                by_model[m.model] = {'cost': 0, 'requests': 0}
            by_model[m.model]['cost'] += m.cost
            by_model[m.model]['requests'] += 1

            # By user (identify power users)
            if m.user_id not in by_user:
                by_user[m.user_id] = {'cost': 0, 'requests': 0}
            by_user[m.user_id]['cost'] += m.cost
            by_user[m.user_id]['requests'] += 1

        # Calculate cache effectiveness
        cached_requests = sum(1 for m in recent if m.cached)
        cache_hit_rate = cached_requests / len(recent) if recent else 0

        # Identify optimization opportunities
        optimizations = self._identify_optimizations(by_feature, by_model)

        return {
            'period_hours': hours,
            'total_requests': len(recent),
            'total_cost': sum(m.cost for m in recent),
            'total_tokens': sum(m.total_tokens for m in recent),
            'avg_latency_ms': sum(m.latency_ms for m in recent) / len(recent),
            'cache_hit_rate': cache_hit_rate,
            'error_rate': sum(1 for m in recent if m.error) / len(recent),
            'by_feature': by_feature,
            'by_model': by_model,
            'top_10_users': sorted(by_user.items(), key=lambda x: x[1]['cost'],
                                  reverse=True)[:10],
            'optimization_opportunities': optimizations
        }

    def _identify_optimizations(self, by_feature: Dict, by_model: Dict) -> List[Dict]:
        """Automatically identify cost optimization opportunities"""
        opportunities = []

        # Find expensive features
        for feature, stats in by_feature.items():
            avg_cost = stats['cost'] / stats['requests']

            if avg_cost > 0.10:  # >$0.10 per request
                opportunities.append({
                    'type': 'high_cost_feature',
                    'feature': feature,
                    'avg_cost': avg_cost,
                    'total_cost': stats['cost'],
                    'recommendation': 'Consider caching, prompt compression, or cheaper model'
                })

        # Find model overuse
        if 'gpt-4' in by_model or 'gpt-4-turbo' in by_model:
            gpt4_requests = by_model.get('gpt-4', {'requests': 0})['requests']
            gpt4_turbo_requests = by_model.get('gpt-4-turbo', {'requests': 0})['requests']
            total_gpt4 = gpt4_requests + gpt4_turbo_requests

            if total_gpt4 > len(self.metrics) * 0.5:  # >50% GPT-4 usage
                opportunities.append({
                    'type': 'overuse_expensive_model',
                    'model': 'gpt-4',
                    'percentage': (total_gpt4 / len(self.metrics) * 100),
                    'recommendation': 'Evaluate if simpler tasks can use GPT-3.5-turbo (30x cheaper)'
                })

        return opportunities

    def export_metrics(self, output_file: str):
        """Export metrics for external analysis (BI tools, spreadsheets)"""
        import csv

        with open(output_file, 'w', newline='') as f:
            if not self.metrics:
                return

            writer = csv.DictWriter(f, fieldnames=asdict(self.metrics[0]).keys())
            writer.writeheader()

            for metric in self.metrics:
                writer.writerow(asdict(metric))

        self.logger.info(f"Exported {len(self.metrics)} metrics to {output_file}")

# Integration example
observability = LLMObservability(alert_webhook='https://hooks.slack.com/your-webhook')

def monitored_llm_call(prompt: str, user_id: str, feature: str):
    """Wrapper that adds full observability"""
    start_time = time.time()

    try:
        # Make LLM call
        response = call_llm_api(prompt, model='gpt-3.5-turbo')

        # Calculate metrics
        latency_ms = (time.time() - start_time) * 1000
        budget_manager = TokenBudgetManager()
        input_tokens = budget_manager.count_tokens(prompt)
        output_tokens = budget_manager.count_tokens(response)
        cost = (input_tokens / 1000 * 0.001) + (output_tokens / 1000 * 0.002)

        # Log metrics
        metric = LLMMetric(
            timestamp=time.time(),
            user_id=user_id,
            feature=feature,
            model='gpt-3.5-turbo',
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            total_tokens=input_tokens + output_tokens,
            cost=cost,
            latency_ms=latency_ms,
            cached=False,
            cache_type=None,
            error=None
        )
        observability.log_request(metric)

        return response

    except Exception as e:
        # Log error metrics
        metric = LLMMetric(
            timestamp=time.time(),
            user_id=user_id,
            feature=feature,
            model='gpt-3.5-turbo',
            input_tokens=0,
            output_tokens=0,
            total_tokens=0,
            cost=0,
            latency_ms=(time.time() - start_time) * 1000,
            cached=False,
            cache_type=None,
            error=str(e)
        )
        observability.log_request(metric)
        raise

# Daily cost report
def generate_daily_report():
    """Generate comprehensive daily cost report"""
    breakdown = observability.get_cost_breakdown(hours=24)

    report = f"""
    📊 Daily LLM Cost Report
    ========================

    Total Cost: ${breakdown['total_cost']:.2f}
    Total Requests: {breakdown['total_requests']:,}
    Avg Latency: {breakdown['avg_latency_ms']:.0f}ms
    Cache Hit Rate: {breakdown['cache_hit_rate']:.1%}
    Error Rate: {breakdown['error_rate']:.1%}

    💰 Cost by Feature:
    {format_dict_table(breakdown['by_feature'])}

    🤖 Cost by Model:
    {format_dict_table(breakdown['by_model'])}

    ⚠️ Optimization Opportunities:
    {format_opportunities(breakdown['optimization_opportunities'])}

    📈 Top 10 Users by Cost:
    {format_user_costs(breakdown['top_10_users'])}
    """

    return report