Managing and Reducing Costs › Putting It All Together

import os
from anthropic import Anthropic
from google import genai
import hashlib
from datetime import datetime

class ProductionCostOptimizedAgent:
    """Production agent with comprehensive cost optimization."""
    
    def __init__(self, daily_budget=50.0):
        # Initialize clients
        self.anthropic = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
        
        # Budget tracking
        self.daily_budget = daily_budget
        self.daily_spending = 0.0
        self.current_day = datetime.now().date()
        
        # Caching
        self.cache = {}
        self.cache_hits = 0
        self.cache_misses = 0
        
        # Conversation history (trimmed)
        self.history = []
        self.max_history = 6
    
    def _reset_if_new_day(self):
        """Reset daily counters."""
        today = datetime.now().date()
        if today != self.current_day:
            self.current_day = today
            self.daily_spending = 0.0
    
    def _hash_query(self, query):
        """Create cache key."""
        return hashlib.md5(query.lower().strip().encode()).hexdigest()
    
    def _classify_complexity(self, query):
        """Determine query complexity."""
        high_complexity = ["explain why", "analyze", "compare", "evaluate"]
        simple_patterns = ["what is", "who is", "when is"]
        
        query_lower = query.lower()
        
        if any(p in query_lower for p in high_complexity):
            return "high"
        elif any(p in query_lower for p in simple_patterns):
            return "low"
        else:
            return "medium"
    
    def respond(self, query):
        """Generate optimized response."""
        self._reset_if_new_day()
        
        # Check budget
        if self.daily_spending >= self.daily_budget:
            return {
                "response": "Daily budget exceeded. Please try again tomorrow.",
                "source": "budget_limit",
                "cost": 0.0
            }
        
        # Check cache
        cache_key = self._hash_query(query)
        if cache_key in self.cache:
            self.cache_hits += 1
            return {
                "response": self.cache[cache_key],
                "source": "cache",
                "cost": 0.0
            }
        
        self.cache_misses += 1
        
        # Route to appropriate model
        complexity = self._classify_complexity(query)
        
        if complexity == "low":
            # Use cheapest model for simple queries
            response_text = self.gemini_client.models.generate_content(
                model="gemini-2.5-flash",
                contents=query
            ).text
            cost = 0.0008  # Estimated cost for Gemini Flash
            source = "gemini-2.5-flash"
        
        else:
            # Use Claude for complex queries, with concise prompts
            system = "Provide concise, direct answers. Be brief but complete."
            
            response = self.anthropic.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=300 if complexity == "medium" else 512,
                system=system,
                messages=[{"role": "user", "content": query}]
            )
            
            response_text = response.content[0].text
            
            # Calculate cost
            input_cost = (response.usage.input_tokens / 1_000_000) * 3.00
            output_cost = (response.usage.output_tokens / 1_000_000) * 15.00
            cost = input_cost + output_cost
            source = "claude-sonnet-4-5"
        
        # Update spending
        self.daily_spending += cost
        
        # Cache the response
        self.cache[cache_key] = response_text
        
        return {
            "response": response_text,
            "source": source,
            "cost": cost,
            "daily_spending": self.daily_spending
        }
    
    def get_stats(self):
        """Get performance statistics."""
        total_requests = self.cache_hits + self.cache_misses
        hit_rate = (self.cache_hits / total_requests * 100) if total_requests > 0 else 0
        
        return {
            "total_requests": total_requests,
            "cache_hit_rate": hit_rate,
            "daily_spending": self.daily_spending,
            "budget_remaining": self.daily_budget - self.daily_spending
        }

## Test the production agent
agent = ProductionCostOptimizedAgent(daily_budget=1.0)

test_queries = [
    "What is Python?",
    "What is Python?",  # Should hit cache
    "Analyze the trade-offs between microservices and monolithic architectures",
    "What is JavaScript?",
    "What is Python?",  # Should hit cache again
]

for query in test_queries:
    result = agent.respond(query)
    print(f"Q: {query}")
    print(f"Source: {result['source']}")
    print(f"Cost: ${result['cost']:.6f}")
    if 'daily_spending' in result:
        print(f"Daily spending: ${result['daily_spending']:.4f}")
    print()

stats = agent.get_stats()
print("=== Agent Statistics ===")
print(f"Total requests: {stats['total_requests']}")
print(f"Cache hit rate: {stats['cache_hit_rate']:.1f}%")
print(f"Daily spending: ${stats['daily_spending']:.4f}")
print(f"Budget remaining: ${stats['budget_remaining']:.4f}")