Managing and Reducing AI Agent Costs: Complete Guide to Cost Optimization Strategies

def calculate_interaction_cost(input_tokens, output_tokens, model="claude-sonnet-4-5"):
    """Calculate the cost of a single model interaction."""
    # Pricing per million tokens (November 2025 rates)
    pricing = {
        "claude-sonnet-4-5": {"input": 3.00, "output": 15.00},
        "gpt-5": {"input": 2.50, "output": 10.00},
        "gemini-2.5-flash": {"input": 0.40, "output": 1.20},
        "gemini-2.5-pro": {"input": 1.25, "output": 5.00}
    }
    
    rates = pricing[model]
    
    # Calculate cost (rates are per million tokens)
    input_cost = (input_tokens / 1_000_000) * rates["input"]
    output_cost = (output_tokens / 1_000_000) * rates["output"]
    total_cost = input_cost + output_cost
    
    return {
        "input_cost": input_cost,
        "output_cost": output_cost,
        "total_cost": total_cost
    }

## Example: A conversation with context
input_tokens = 1500  # System prompt + conversation history + query
output_tokens = 500  # Agent's response

cost = calculate_interaction_cost(input_tokens, output_tokens, "claude-sonnet-4-5")
print(f"Input cost: ${cost['input_cost']:.6f}")
print(f"Output cost: ${cost['output_cost']:.6f}")
print(f"Total cost: ${cost['total_cost']:.6f}")
print(f"\nCost per 1000 interactions: ${cost['total_cost'] * 1000:.2f}")

import os
from anthropic import Anthropic
from datetime import datetime

class CostTrackingAgent:
    """Agent that tracks API costs for monitoring and optimization."""
    
    def __init__(self):
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.cost_log = []
        
        # Pricing per million tokens
        self.pricing = {
            "claude-sonnet-4-5": {"input": 3.00, "output": 15.00}
        }
    
    def _calculate_cost(self, usage, model):
        """Calculate cost from token usage."""
        rates = self.pricing[model]
        input_cost = (usage.input_tokens / 1_000_000) * rates["input"]
        output_cost = (usage.output_tokens / 1_000_000) * rates["output"]
        return input_cost + output_cost
    
    def respond(self, query):
        """Generate response and track costs."""
        model = "claude-sonnet-4-5"
        
        response = self.client.messages.create(
            model=model,
            max_tokens=1024,
            messages=[{"role": "user", "content": query}]
        )
        
        # Calculate and log cost
        cost = self._calculate_cost(response.usage, model)
        
        self.cost_log.append({
            "timestamp": datetime.now(),
            "query": query[:50] + "..." if len(query) > 50 else query,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "cost": cost,
            "model": model
        })
        
        return response.content[0].text
    
    def get_cost_summary(self):
        """Get summary of costs."""
        if not self.cost_log:
            return "No interactions yet."
        
        total_cost = sum(entry["cost"] for entry in self.cost_log)
        total_tokens = sum(
            entry["input_tokens"] + entry["output_tokens"] 
            for entry in self.cost_log
        )
        
        return {
            "total_interactions": len(self.cost_log),
            "total_cost": total_cost,
            "total_tokens": total_tokens,
            "average_cost_per_interaction": total_cost / len(self.cost_log),
            "most_expensive": max(self.cost_log, key=lambda x: x["cost"])
        }

## Test the tracking
agent = CostTrackingAgent()

agent.respond("What is Python?")
agent.respond("Explain machine learning in simple terms.")
agent.respond("How do neural networks work?")

summary = agent.get_cost_summary()
print(f"Total interactions: {summary['total_interactions']}")
print(f"Total cost: ${summary['total_cost']:.4f}")
print(f"Average cost: ${summary['average_cost_per_interaction']:.4f}")
print(f"\nMost expensive query: {summary['most_expensive']['query']}")
print(f"Cost: ${summary['most_expensive']['cost']:.4f}")

import os
from anthropic import Anthropic
from openai import OpenAI
from google import genai

class CostOptimizedAgent:
    """Agent that chooses the most cost-effective model for each task."""
    
    def __init__(self):
        # Initialize clients for different providers
        self.anthropic = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.gemini = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
    
    def _classify_task_complexity(self, query):
        """Determine what level of model capability is needed."""
        # High complexity: needs reasoning, tool use, or complex understanding
        high_complexity_indicators = [
            "explain why", "analyze", "compare and contrast",
            "step by step", "reasoning", "pros and cons",
            "evaluate", "critique"
        ]
        
        # Medium complexity: straightforward questions or tasks
        medium_complexity_indicators = [
            "how to", "what is", "describe", "summarize"
        ]
        
        query_lower = query.lower()
        
        if any(ind in query_lower for ind in high_complexity_indicators):
            return "high"
        elif any(ind in query_lower for ind in medium_complexity_indicators):
            return "medium"
        else:
            return "low"
    
    def respond(self, query):
        """Route to the most cost-effective model."""
        complexity = self._classify_task_complexity(query)
        
        if complexity == "high":
            # Use Claude Sonnet 4.5 for complex reasoning
            # Cost: ~$0.012 per interaction
            response = self.anthropic.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=1024,
                messages=[{"role": "user", "content": query}]
            )
            return response.content[0].text, "claude-sonnet-4-5", "high"
        
        elif complexity == "medium":
            # Use GPT-5 for general tasks
            # Cost: ~$0.008 per interaction (33% savings)
            response = self.openai.chat.completions.create(
                model="gpt-5",
                max_completion_tokens=512,
                messages=[{"role": "user", "content": query}]
            )
            return response.choices[0].message.content, "gpt-5", "medium"
        
        else:
            # Use Gemini 2.5 Flash for simple queries
            # Cost: ~$0.001 per interaction (92% savings!)
            response = self.gemini.models.generate_content(
                model="gemini-2.5-flash",
                contents=query
            )
            return response.text, "gemini-2.5-flash", "low"

## Test with different complexity levels
agent = CostOptimizedAgent()

queries = [
    ("What's the capital of France?", "low"),
    ("What is machine learning?", "medium"),
    ("Analyze the pros and cons of different database architectures", "high")
]

for query, expected in queries:
    result, model, complexity = agent.respond(query)
    print(f"Query: {query}")
    print(f"Complexity: {complexity} (expected: {expected})")
    print(f"Model: {model}")
    print(f"Response: {result[:100]}...")
    print()

import os
from anthropic import Anthropic

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def compare_response_costs(query):
    """Compare costs of verbose vs concise responses."""
    
    # Verbose response (default behavior)
    verbose_response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": query}]
    )
    
    # Concise response (optimized)
    concise_system = """You are a helpful assistant. Provide concise, direct answers.
    Use 1-2 sentences for simple questions. Avoid unnecessary elaboration."""
    
    concise_response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=200,  # Hard limit
        system=concise_system,
        messages=[{"role": "user", "content": query}]
    )
    
    # Calculate costs
    def calc_cost(usage):
        input_cost = (usage.input_tokens / 1_000_000) * 3.00
        output_cost = (usage.output_tokens / 1_000_000) * 15.00
        return input_cost + output_cost
    
    verbose_cost = calc_cost(verbose_response.usage)
    concise_cost = calc_cost(concise_response.usage)
    savings = ((verbose_cost - concise_cost) / verbose_cost) * 100
    
    return {
        "verbose": {
            "response": verbose_response.content[0].text,
            "tokens": verbose_response.usage.output_tokens,
            "cost": verbose_cost
        },
        "concise": {
            "response": concise_response.content[0].text,
            "tokens": concise_response.usage.output_tokens,
            "cost": concise_cost
        },
        "savings_percent": savings
    }

## Test with a simple query
result = compare_response_costs("What is Python?")

print("Verbose response:")
print(f"Tokens: {result['verbose']['tokens']}")
print(f"Cost: ${result['verbose']['cost']:.6f}")
print(f"Response: {result['verbose']['response'][:150]}...")
print()

print("Concise response:")
print(f"Tokens: {result['concise']['tokens']}")
print(f"Cost: ${result['concise']['cost']:.6f}")
print(f"Response: {result['concise']['response']}")
print()

print(f"Cost savings: {result['savings_percent']:.1f}%")

from google import genai
import hashlib
import time
from datetime import datetime, timedelta

gemini_client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))

class CachingAgent:
    """Agent with intelligent caching to minimize API calls."""
    
    def __init__(self):
        self.client = gemini_client
        
        # Short-term cache: exact query matches
        self.exact_cache = {}
        
        # Long-term cache: common queries that rarely change
        self.persistent_cache = {
            "what is python": "Python is a high-level programming language...",
            "what is machine learning": "Machine learning is a subset of AI...",
            # Preload common queries
        }
        
        # Cache metadata
        self.cache_stats = {
            "hits": 0,
            "misses": 0,
            "api_calls": 0,
            "cost_saved": 0.0
        }
    
    def _hash_query(self, query):
        """Create cache key from query."""
        return hashlib.md5(query.lower().strip().encode()).hexdigest()
    
    def _estimate_cost_saved(self, query):
        """Estimate cost saved by cache hit."""
        # Rough estimate: 100 input tokens + 200 output tokens
        input_cost = (100 / 1_000_000) * 0.40
        output_cost = (200 / 1_000_000) * 1.20
        return input_cost + output_cost
    
    def respond(self, query):
        """Get response, using cache when possible."""
        cache_key = self._hash_query(query)
        query_normalized = query.lower().strip()
        
        # Check persistent cache first (common queries)
        if query_normalized in self.persistent_cache:
            self.cache_stats["hits"] += 1
            self.cache_stats["cost_saved"] += self._estimate_cost_saved(query)
            return self.persistent_cache[query_normalized], "persistent_cache"
        
        # Check exact cache (recent queries)
        if cache_key in self.exact_cache:
            self.cache_stats["hits"] += 1
            self.cache_stats["cost_saved"] += self._estimate_cost_saved(query)
            return self.exact_cache[cache_key], "exact_cache"
        
        # Cache miss: call the model
        self.cache_stats["misses"] += 1
        self.cache_stats["api_calls"] += 1
        
        response = self.client.models.generate_content(
            model="gemini-2.5-flash",
            contents=query
        )
        result = response.text
        
        # Store in exact cache
        self.exact_cache[cache_key] = result
        
        # If cache is getting large, prune old entries
        if len(self.exact_cache) > 1000:
            # Keep only the most recent 500
            keys_to_remove = list(self.exact_cache.keys())[:-500]
            for key in keys_to_remove:
                del self.exact_cache[key]
        
        return result, "api_call"
    
    def get_cache_stats(self):
        """Get caching performance metrics."""
        total_requests = self.cache_stats["hits"] + self.cache_stats["misses"]
        hit_rate = (self.cache_stats["hits"] / total_requests * 100) if total_requests > 0 else 0
        
        return {
            "total_requests": total_requests,
            "cache_hits": self.cache_stats["hits"],
            "cache_misses": self.cache_stats["misses"],
            "hit_rate": hit_rate,
            "api_calls": self.cache_stats["api_calls"],
            "estimated_cost_saved": self.cache_stats["cost_saved"]
        }

## Test the caching agent
agent = CachingAgent()

## Simulate user queries (with some repetition)
queries = [
    "What is Python?",
    "What is machine learning?",
    "How do I learn programming?",
    "What is Python?",  # Duplicate
    "What is machine learning?",  # Duplicate
    "How do I learn programming?",  # Duplicate
    "What are data structures?",
    "What is Python?",  # Duplicate again
]

for query in queries:
    result, source = agent.respond(query)
    print(f"Q: {query}")
    print(f"Source: {source}")
    print()

## Show cache performance
stats = agent.get_cache_stats()
print("Cache Performance:")
print(f"Total requests: {stats['total_requests']}")
print(f"Cache hits: {stats['cache_hits']} ({stats['hit_rate']:.1f}%)")
print(f"API calls: {stats['api_calls']}")
print(f"Estimated cost saved: ${stats['estimated_cost_saved']:.4f}")

from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def process_individually(queries):
    """Process each query separately (expensive)."""
    results = []
    total_cost = 0.0
    
    for query in queries:
        response = client.chat.completions.create(
            model="gpt-5",
            max_completion_tokens=100,
            messages=[{"role": "user", "content": query}]
        )
        
        # Estimate cost (rough approximation)
        tokens = response.usage.total_tokens
        cost = (tokens / 1_000_000) * 6.25  # Average of input/output rates
        total_cost += cost
        
        results.append(response.choices[0].message.content)
    
    return results, total_cost

def process_batched(queries):
    """Process all queries in a single call (cheaper)."""
    # Combine queries into a single prompt
    batch_prompt = "Answer each of the following questions concisely:\n\n"
    for i, query in enumerate(queries, 1):
        batch_prompt += f"{i}. {query}\n"
    
    response = client.chat.completions.create(
        model="gpt-5",
        max_completion_tokens=500,
        messages=[{"role": "user", "content": batch_prompt}]
    )
    
    # Estimate cost
    tokens = response.usage.total_tokens
    cost = (tokens / 1_000_000) * 6.25
    
    # Parse the batched response
    result = response.choices[0].message.content
    
    return result, cost

## Test both approaches
queries = [
    "What is Python?",
    "What is JavaScript?",
    "What is Ruby?",
    "What is Go?",
    "What is Rust?"
]

print("Individual processing:")
results_individual, cost_individual = process_individually(queries)
print(f"Cost: ${cost_individual:.4f}")
print()

print("Batched processing:")
result_batched, cost_batched = process_batched(queries)
print(f"Cost: ${cost_batched:.4f}")
print(f"Savings: ${cost_individual - cost_batched:.4f} ({((cost_individual - cost_batched) / cost_individual * 100):.1f}%)")
print()
print("Batched response:")
print(result_batched)

import os
from anthropic import Anthropic

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

class HistoryOptimizedAgent:
    """Agent that manages conversation history efficiently."""
    
    def __init__(self):
        self.conversation_history = []
        self.max_history_messages = 6  # Keep last 3 exchanges
    
    def _trim_history(self):
        """Keep only recent messages to reduce token costs."""
        if len(self.conversation_history) > self.max_history_messages:
            # Keep only the most recent messages
            self.conversation_history = self.conversation_history[-self.max_history_messages:]
    
    def _estimate_tokens(self, text):
        """Rough token estimate (4 chars ≈ 1 token)."""
        return len(text) // 4
    
    def respond(self, user_message):
        """Generate response with optimized history."""
        # Add user message to history
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        
        # Trim history before sending
        self._trim_history()
        
        # Calculate token usage
        total_input_chars = sum(
            len(msg["content"]) for msg in self.conversation_history
        )
        estimated_input_tokens = total_input_chars // 4
        
        # Make the call
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=512,
            messages=self.conversation_history
        )
        
        # Add assistant response to history
        assistant_message = response.content[0].text
        self.conversation_history.append({
            "role": "assistant",
            "content": assistant_message
        })
        
        # Calculate cost
        input_cost = (response.usage.input_tokens / 1_000_000) * 3.00
        output_cost = (response.usage.output_tokens / 1_000_000) * 15.00
        total_cost = input_cost + output_cost
        
        return {
            "response": assistant_message,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "cost": total_cost,
            "history_length": len(self.conversation_history)
        }

## Test with a multi-turn conversation
agent = HistoryOptimizedAgent()

queries = [
    "What is Python?",
    "What are its main features?",
    "How does it compare to Java?",
    "What about performance?",
    "Should I learn it?",
    "What resources do you recommend?",
    "How long will it take?",
    "What projects should I build?"
]

total_cost = 0.0

for query in queries:
    result = agent.respond(query)
    total_cost += result["cost"]
    
    print(f"Q: {query}")
    print(f"Input tokens: {result['input_tokens']}")
    print(f"History length: {result['history_length']} messages")
    print(f"Cost: ${result['cost']:.6f}")
    print()

print(f"Total conversation cost: ${total_cost:.4f}")
print("\nNote: Without trimming, costs would be ~40% higher")

import os
from anthropic import Anthropic

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def summarize_context(long_context, max_length=500):
    """Compress long context into a shorter summary."""
    if len(long_context) <= max_length:
        return long_context
    
    # Use the model to create a concise summary
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": f"Summarize this in {max_length//4} words or less:\n\n{long_context}"
        }]
    )
    
    return response.content[0].text

def respond_with_context(query, long_context):
    """Answer query using compressed context."""
    # Compress the context first
    compressed_context = summarize_context(long_context, max_length=500)
    
    # Use compressed context in the actual query
    full_prompt = f"Context: {compressed_context}\n\nQuestion: {query}"
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        messages=[{"role": "user", "content": full_prompt}]
    )
    
    return response.content[0].text

## Example: Long document that needs to be included
long_document = """
[Imagine a 5000-word document about machine learning here...]
Machine learning is a field of artificial intelligence that focuses on...
[... many more paragraphs ...]
"""

## Without compression: ~5000 tokens input
## With compression: ~500 tokens input
## Savings: ~90% on input tokens for this context

query = "What are the key concepts in this document?"
answer = respond_with_context(query, long_document)
print(answer)

import os
from anthropic import Anthropic
from datetime import datetime, timedelta

class BudgetControlledAgent:
    """Agent with built-in budget limits and alerts."""
    
    def __init__(self, daily_budget=10.0):
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.daily_budget = daily_budget
        self.current_day = datetime.now().date()
        self.daily_spending = 0.0
        self.total_spending = 0.0
    
    def _reset_daily_budget_if_needed(self):
        """Reset daily spending counter at midnight."""
        today = datetime.now().date()
        if today != self.current_day:
            self.current_day = today
            self.daily_spending = 0.0
    
    def _calculate_cost(self, usage):
        """Calculate cost from token usage."""
        input_cost = (usage.input_tokens / 1_000_000) * 3.00
        output_cost = (usage.output_tokens / 1_000_000) * 15.00
        return input_cost + output_cost
    
    def respond(self, query):
        """Generate response if within budget."""
        self._reset_daily_budget_if_needed()
        
        # Check if we're over budget
        if self.daily_spending >= self.daily_budget:
            return {
                "response": None,
                "error": f"Daily budget of ${self.daily_budget:.2f} exceeded. Current spending: ${self.daily_spending:.2f}",
                "budget_remaining": 0.0
            }
        
        # Make the call
        response = self.client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=512,
            messages=[{"role": "user", "content": query}]
        )
        
        # Track spending
        cost = self._calculate_cost(response.usage)
        self.daily_spending += cost
        self.total_spending += cost
        
        # Check if approaching budget limit
        budget_remaining = self.daily_budget - self.daily_spending
        warning = None
        
        if budget_remaining < self.daily_budget * 0.2:  # Less than 20% remaining
            warning = f"Warning: Only ${budget_remaining:.2f} remaining in daily budget"
        
        return {
            "response": response.content[0].text,
            "cost": cost,
            "daily_spending": self.daily_spending,
            "budget_remaining": budget_remaining,
            "warning": warning
        }
    
    def get_spending_summary(self):
        """Get spending statistics."""
        return {
            "daily_spending": self.daily_spending,
            "daily_budget": self.daily_budget,
            "budget_used_percent": (self.daily_spending / self.daily_budget) * 100,
            "total_spending": self.total_spending
        }

## Test budget controls
agent = BudgetControlledAgent(daily_budget=0.10)  # $0.10 daily limit

queries = [
    "What is Python?",
    "Explain machine learning.",
    "How do neural networks work?",
    "What is deep learning?",
    "Describe reinforcement learning.",
    "What are transformers?",
    "Explain attention mechanisms.",
    "What is GPT?",
    "How does BERT work?",
    "What is transfer learning?"
]

for i, query in enumerate(queries, 1):
    print(f"\n--- Query {i} ---")
    result = agent.respond(query)
    
    if result["response"]:
        print(f"Response: {result['response'][:100]}...")
        print(f"Cost: ${result['cost']:.6f}")
        print(f"Daily spending: ${result['daily_spending']:.4f}")
        
        if result["warning"]:
            print(f"⚠️  {result['warning']}")
    else:
        print(f"❌ {result['error']}")
        break

summary = agent.get_spending_summary()
print(f"\n=== Spending Summary ===")
print(f"Daily budget: ${summary['daily_budget']:.2f}")
print(f"Daily spending: ${summary['daily_spending']:.4f}")
print(f"Budget used: {summary['budget_used_percent']:.1f}%")

import os
from anthropic import Anthropic
from datetime import datetime
import statistics

class CostAnalyzer:
    """Analyze and compare costs across different optimization strategies."""
    
    def __init__(self):
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.baseline_costs = []
        self.optimized_costs = []
    
    def _calculate_cost(self, usage):
        """Calculate cost from token usage."""
        input_cost = (usage.input_tokens / 1_000_000) * 3.00
        output_cost = (usage.output_tokens / 1_000_000) * 15.00
        return input_cost + output_cost
    
    def run_baseline(self, queries):
        """Run queries without optimization."""
        print("Running baseline (no optimization)...")
        
        for query in queries:
            response = self.client.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=1024,  # No limits
                messages=[{"role": "user", "content": query}]
            )
            
            cost = self._calculate_cost(response.usage)
            self.baseline_costs.append(cost)
    
    def run_optimized(self, queries):
        """Run queries with optimization."""
        print("Running optimized version...")
        
        system_prompt = """You are a helpful assistant. Provide concise, direct answers.
        Use 1-2 sentences for simple questions."""
        
        for query in queries:
            response = self.client.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=200,  # Limited
                system=system_prompt,
                messages=[{"role": "user", "content": query}]
            )
            
            cost = self._calculate_cost(response.usage)
            self.optimized_costs.append(cost)
    
    def generate_report(self):
        """Generate cost comparison report."""
        baseline_total = sum(self.baseline_costs)
        optimized_total = sum(self.optimized_costs)
        savings = baseline_total - optimized_total
        savings_percent = (savings / baseline_total) * 100
        
        baseline_avg = statistics.mean(self.baseline_costs)
        optimized_avg = statistics.mean(self.optimized_costs)
        
        report = f"""
=== Cost Optimization Report ===

Baseline (No Optimization):
  Total cost: ${baseline_total:.4f}
  Average per query: ${baseline_avg:.6f}
  Number of queries: {len(self.baseline_costs)}

Optimized:
  Total cost: ${optimized_total:.4f}
  Average per query: ${optimized_avg:.6f}
  Number of queries: {len(self.optimized_costs)}

Savings:
  Total saved: ${savings:.4f}
  Percentage saved: {savings_percent:.1f}%
  
Projected Monthly Savings (at 10,000 queries/month):
  ${savings * (10000 / len(self.baseline_costs)):.2f}
"""
        return report

## Run the analysis
analyzer = CostAnalyzer()

test_queries = [
    "What is Python?",
    "What is JavaScript?",
    "What is machine learning?",
    "What are neural networks?",
    "What is deep learning?",
    "What is natural language processing?",
    "What are transformers?",
    "What is computer vision?",
    "What is reinforcement learning?",
    "What is data science?"
]

analyzer.run_baseline(test_queries)
analyzer.run_optimized(test_queries)

print(analyzer.generate_report())