Agent Cost Optimization
What is Agent Cost Optimization?
Cost optimization reduces LLM API expenses while maintaining quality. The main cost drivers are: token count (input + output), model selection (GPT-4 vs GPT-3.5), request frequency, and caching effectiveness.
Optimization strategies include: semantic caching (avoid duplicate LLM calls), model routing (use cheaper models for simple tasks), token optimization (compress prompts, limit output), batching (amortize overhead), and prompt engineering (reduce iterations).
Why This Matters
LLM API costs can quickly spiral out of control. A single GPT-4 call costs 3,000-15,000/month. Optimization can reduce these costs by 50-80% while maintaining 95%+ quality, saving thousands of dollars monthly.
Real-World Analogy
Cost optimization is like fuel-efficient driving. You can drive anywhere with a gas-guzzler (unoptimized agent), but techniques like route planning (model routing), carpooling (batch processing), and maintaining proper tire pressure (prompt compression) can cut fuel costs dramatically without reaching your destination any slower.
Project Overview
We will build a cost optimization layer that:
- Implements semantic caching with Redis and embedding similarity
- Routes requests to optimal models based on query complexity
- Optimizes token usage through prompt compression
- Tracks costs per request, user, and model
- Implements budget alerts and hard limits
- Provides cost analytics and optimization recommendations
Expected outcome: An agent with 50-80% cost reduction.
Difficulty: Advanced (requires understanding of LLM pricing, caching strategies, and optimization)
Tools & Setup
| Tool | Version | Purpose |
|---|---|---|
| Python | 3.11+ | Core language |
| redis | 5.0+ | Caching |
| openai | 1.0+ | LLM backbone |
| tiktoken | 0.5+ | Token counting |
| numpy | 1.24+ | Similarity computation |
Step 1: Environment Setup
python -m venv venv
source venv/bin/activate
pip install redis openai tiktoken numpy
export OPENAI_API_KEY="sk-your-key"
export REDIS_URL="redis://localhost:6379"
Step 2: Semantic Cache
# caching/semantic_cache.py
import redis
import hashlib
import json
from typing import Optional, Dict, Any
import numpy as np
import logging
import time
logger = logging.getLogger(__name__)
class SemanticCache:
"""Redis-based semantic cache using embedding similarity."""
def __init__(
self,
redis_url: str = "redis://localhost:6379",
threshold: float = 0.92,
default_ttl: int = 3600,
):
self.client = redis.from_url(redis_url, decode_responses=True)
self.threshold = threshold
self.default_ttl = default_ttl
self._hit_count = 0
self._miss_count = 0
def _hash_query(self, query: str) -> str:
return hashlib.md5(query.encode()).hexdigest()
def _get_embedding(self, text: str) -> list:
import openai
client = openai.OpenAI()
response = client.embeddings.create(model="text-embedding-3-small", input=text)
return response.data[0].embedding
def _cosine_similarity(self, a: list, b: list) -> float:
a_arr, b_arr = np.array(a), np.array(b)
norm_a = np.linalg.norm(a_arr)
norm_b = np.linalg.norm(b_arr)
if norm_a == 0 or norm_b == 0:
return 0.0
return float(np.dot(a_arr, b_arr) / (norm_a * norm_b))
def get(self, query: str) -> Optional[Dict[str, Any]]:
exact_key = f"cache:exact:{self._hash_query(query)}"
cached = self.client.get(exact_key)
if cached:
self._hit_count += 1
logger.info("Exact cache hit for query")
return json.loads(cached)
try:
query_embedding = self._get_embedding(query)
keys = self.client.keys("cache:semantic:*")
best_similarity = 0.0
best_response = None
for key in keys:
data = json.loads(self.client.get(key))
embedding = data.get("embedding", [])
if embedding:
similarity = self._cosine_similarity(query_embedding, embedding)
if similarity > best_similarity:
best_similarity = similarity
best_response = data.get("response")
if best_similarity >= self.threshold:
self._hit_count += 1
logger.info("Semantic cache hit: similarity=%.3f", best_similarity)
return best_response
except Exception as e:
logger.warning("Semantic cache lookup failed: %s", e)
self._miss_count += 1
return None
def set(self, query: str, response: Dict[str, Any], ttl: Optional[int] = None) -> None:
ttl = ttl or self.default_ttl
exact_key = f"cache:exact:{self._hash_query(query)}"
self.client.setex(exact_key, ttl, json.dumps(response))
try:
semantic_key = f"cache:semantic:{self._hash_query(query)}"
embedding = self._get_embedding(query)
self.client.setex(semantic_key, ttl, json.dumps({
"embedding": embedding,
"response": response,
"query": query[:100],
}))
except Exception as e:
logger.warning("Failed to store semantic embedding: %s", e)
def stats(self) -> Dict[str, Any]:
total = self._hit_count + self._miss_count
return {
"total_lookups": total,
"hits": self._hit_count,
"misses": self._miss_count,
"hit_rate": round(self._hit_count / total * 100, 2) if total > 0 else 0,
"threshold": self.threshold,
}
Step 3: Model Router and Token Optimizer
# routing/model_router.py
from openai import OpenAI
from typing import Dict, Optional
import logging
logger = logging.getLogger(__name__)
class ModelRouter:
"""Route queries to optimal model based on complexity."""
MODEL_TIERS = {
"simple": {"model": "gpt-3.5-turbo", "cost_per_1k_input": 0.0005, "cost_per_1k_output": 0.0015},
"moderate": {"model": "gpt-4o-mini", "cost_per_1k_input": 0.00015, "cost_per_1k_output": 0.0006},
"complex": {"model": "gpt-4o", "cost_per_1k_input": 0.0025, "cost_per_1k_output": 0.01},
}
def __init__(self, classifier_model: str = "gpt-3.5-turbo"):
self.client = OpenAI()
self.classifier_model = classifier_model
self._routing_count: Dict[str, int] = {"simple": 0, "moderate": 0, "complex": 0}
def classify_complexity(self, query: str) -> str:
try:
response = self.client.chat.completions.create(
model=self.classifier_model,
messages=[
{
"role": "system",
"content": """Classify query complexity. Return ONLY one word: simple, moderate, or complex.
Simple: factual questions, simple calculations, basic lookups
Moderate: analysis, comparisons, explanations, multi-step
Complex: creative tasks, code generation, complex reasoning, multi-domain""",
},
{"role": "user", "content": query},
],
temperature=0.0,
max_tokens=10,
)
complexity = response.choices[0].message.content.strip().lower()
if complexity in self.MODEL_TIERS:
self._routing_count[complexity] += 1
return complexity
except Exception as e:
logger.warning("Complexity classification failed: %s", e)
self._routing_count["moderate"] += 1
return "moderate"
def route(self, query: str) -> Dict[str, str]:
complexity = self.classify_complexity(query)
tier = self.MODEL_TIERS[complexity]
return {
"model": tier["model"],
"complexity": complexity,
"cost_per_1k_input": tier["cost_per_1k_input"],
"cost_per_1k_output": tier["cost_per_1k_output"],
}
def get_routing_stats(self) -> Dict[str, int]:
return self._routing_count.copy()
# optimization/token_optimizer.py
import tiktoken
from typing import Dict
import logging
logger = logging.getLogger(__name__)
class TokenOptimizer:
"""Optimize token usage through compression and truncation."""
def __init__(self, encoding_name: str = "cl100k_base"):
self.enc = tiktoken.get_encoding(encoding_name)
self._total_saved = 0
def count_tokens(self, text: str) -> int:
return len(self.enc.encode(text))
def compress_prompt(
self,
system_prompt: str,
user_input: str,
max_tokens: int = 4000,
) -> Dict[str, any]:
system_tokens = self.count_tokens(system_prompt)
user_tokens = self.count_tokens(user_input)
total = system_tokens + user_tokens
if total <= max_tokens:
return {
"system": system_prompt,
"user": user_input,
"tokens_saved": 0,
"original_tokens": total,
"compressed_tokens": total,
}
ratio = max_tokens / total
compressed_system = self._truncate_to_tokens(system_prompt, int(system_tokens * ratio))
compressed_user = self._truncate_to_tokens(user_input, int(user_tokens * ratio))
new_total = self.count_tokens(compressed_system) + self.count_tokens(compressed_user)
saved = total - new_total
self._total_saved += saved
logger.info("Compressed prompt: %d -> %d tokens (saved %d)", total, new_total, saved)
return {
"system": compressed_system,
"user": compressed_user,
"tokens_saved": saved,
"original_tokens": total,
"compressed_tokens": new_total,
}
def _truncate_to_tokens(self, text: str, max_tokens: int) -> str:
tokens = self.enc.encode(text)
return self.enc.decode(tokens[:max_tokens])
def get_stats(self) -> Dict[str, int]:
return {"total_tokens_saved": self._total_saved}
Step 4: Cost Tracker and Complete Optimizer
# tracking/cost_tracker.py
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import json
import logging
logger = logging.getLogger(__name__)
class CostTracker:
"""Track and analyze LLM API costs with budget management."""
PRICING = {
"gpt-4o": {"input": 0.0025, "output": 0.01},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
"gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
"text-embedding-3-small": {"input": 0.00002, "output": 0},
}
def __init__(self):
self.records: List[Dict] = []
self.budgets: Dict[str, float] = {}
def record(
self,
model: str,
input_tokens: int,
output_tokens: int,
user_id: str = "system",
request_id: str = "",
) -> float:
pricing = self.PRICING.get(model, {"input": 0.01, "output": 0.03})
cost = (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1000
record = {
"timestamp": datetime.now().isoformat(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": cost,
"user_id": user_id,
"request_id": request_id,
}
self.records.append(record)
return cost
def get_total_cost(self, period: timedelta = timedelta(hours=24)) -> float:
cutoff = datetime.now() - period
return sum(
r["cost"] for r in self.records
if datetime.fromisoformat(r["timestamp"]) > cutoff
)
def get_cost_by_model(self) -> Dict[str, float]:
costs: Dict[str, float] = {}
for r in self.records:
costs[r["model"]] = costs.get(r["model"], 0) + r["cost"]
return costs
def get_cost_by_user(self) -> Dict[str, float]:
costs: Dict[str, float] = {}
for r in self.records:
costs[r["user_id"]] = costs.get(r["user_id"], 0) + r["cost"]
return costs
def set_budget(self, name: str, amount: float) -> None:
self.budgets[name] = amount
def check_budget(self, name: str, period: timedelta = timedelta(hours=24)) -> Dict:
budget = self.budgets.get(name, float("inf"))
current = self.get_total_cost(period)
return {
"budget": budget,
"current": round(current, 4),
"remaining": round(budget - current, 4),
"utilization_pct": round(current / budget * 100, 2) if budget > 0 else 0,
"alert": current > budget * 0.8,
"exceeded": current > budget,
}
# optimizer.py
from caching.semantic_cache import SemanticCache
from routing.model_router import ModelRouter
from optimization.token_optimizer import TokenOptimizer
from tracking.cost_tracker import CostTracker
from openai import OpenAI
from typing import Dict, Any
import logging
logger = logging.getLogger(__name__)
class CostOptimizedAgent:
"""Complete cost-optimized agent with caching, routing, and tracking."""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.cache = SemanticCache(redis_url)
self.router = ModelRouter()
self.optimizer = TokenOptimizer()
self.tracker = CostTracker()
self.client = OpenAI()
async def process(self, query: str, user_id: str = "system") -> Dict[str, Any]:
cached = self.cache.get(query)
if cached:
return {**cached, "cache_hit": True}
route = self.router.route(query)
compressed = self.optimizer.compress_prompt("You are a helpful assistant.", query)
try:
response = self.client.chat.completions.create(
model=route["model"],
messages=[
{"role": "system", "content": compressed["system"]},
{"role": "user", "content": compressed["user"]},
],
temperature=0.7,
max_tokens=1000,
)
answer = response.choices[0].message.content
cost = self.tracker.record(
route["model"],
response.usage.prompt_tokens,
response.usage.completion_tokens,
user_id,
)
result = {
"answer": answer,
"model": route["model"],
"complexity": route["complexity"],
"tokens_used": response.usage.total_tokens,
"cost": cost,
"cache_hit": False,
"tokens_saved": compressed["tokens_saved"],
}
self.cache.set(query, result)
return result
except Exception as e:
logger.error("LLM call failed: %s", e)
return {"answer": "Error processing request", "error": str(e)}
def get_analytics(self) -> Dict[str, Any]:
return {
"total_cost_24h": round(self.tracker.get_total_cost(), 4),
"cost_by_model": self.tracker.get_cost_by_model(),
"cost_by_user": self.tracker.get_cost_by_user(),
"cache_stats": self.cache.stats(),
"routing_stats": self.router.get_routing_stats(),
"token_savings": self.optimizer.get_stats(),
}
Why This Matters
LLM API costs are usage-based and can spike unexpectedly. Without optimization, a viral feature or increased usage can cause bill shock. Cost optimization provides predictable, manageable expenses while maintaining quality.
Real-World Analogy
Cost optimization is like managing a household budget. You need to track spending (cost tracking), use coupons when available (caching), buy generic brands for simple needs (model routing), and set spending limits (budget management). The goal is getting the same quality of life for less money.
Mathematical Foundation
Cost Savings:
Intuition: Percentage reduction in costs after optimization. Target 50-80% savings with full optimization stack.
Cache Hit Rate:
Intuition: Percentage of requests served from cache without LLM calls. Higher is better; 30-60% is typical for conversational agents.
Token Efficiency:
Intuition: Percentage of tokens that contribute to the final answer. Prompt compression increases efficiency.
Performance Considerations
| Metric | Value | Notes |
|---|---|---|
| Cache Hit Rate | 30-60% | Depends on query patterns |
| Cost Reduction | 50-80% | With full optimization |
| Model Routing Accuracy | 85%+ | Complexity classification |
| Token Savings | 20-40% | Prompt compression |
| Cache Lookup Time | 10-50ms | Redis exact match |
| Semantic Cache Lookup | 100-500ms | Includes embedding computation |
| Budget Check | <5ms | In-memory |
Security Considerations
- Cache Data Privacy: Encrypt cached queries and responses; implement TTL for automatic expiry
- Budget Security: Prevent budget manipulation by validating user IDs
- API Key Protection: Never log API keys in cost tracking
- Cost Allocation: Ensure accurate user attribution to prevent abuse
- Rate Limiting: Prevent cost amplification through excessive requests
- Cache Invalidation: Implement proper invalidation when data changes
Testing & Evaluation
import pytest
from optimization.token_optimizer import TokenOptimizer
from tracking.cost_tracker import CostTracker
def test_token_counting():
optimizer = TokenOptimizer()
count = optimizer.count_tokens("Hello world")
assert count > 0
def test_prompt_compression():
optimizer = TokenOptimizer()
result = optimizer.compress_prompt("System prompt", "User input " * 1000, max_tokens=100)
assert result["tokens_saved"] > 0
def test_cost_tracking():
tracker = CostTracker()
cost = tracker.record("gpt-4o", 100, 50)
assert cost > 0
total = tracker.get_total_cost()
assert total > 0
def test_budget_check():
tracker = CostTracker()
tracker.set_budget("test", 1.0)
tracker.record("gpt-4o", 1000, 500)
status = tracker.check_budget("test")
assert "utilization_pct" in status
Interview Q&A
Q1: How does semantic caching differ from exact caching? A: Exact caching matches queries character-for-character. Semantic caching uses embedding similarity to match queries with similar meaning (e.g., "What is Python?" matches "Tell me about Python"). This dramatically increases cache hit rates for natural language queries.
Q2: What is model routing and how does it save costs? A: Model routing classifies query complexity and routes simple queries to cheaper models (GPT-3.5) and complex queries to expensive models (GPT-4). Since 60-70% of queries are simple, this saves significant costs while maintaining quality.
Q3: How would you calculate the ROI of cost optimization? A: Track costs before and after optimization, calculate: . Include engineering time, infrastructure costs (Redis), and quality impact.
Q4: What is prompt compression and when should it be used? A: Prompt compression reduces token count by removing redundant instructions, abbreviating examples, and truncating context. Use when approaching token limits or to reduce costs. Must balance compression with response quality.
Q5: How do you prevent cost overruns in production? A: Implement per-user and per-organization budgets, set daily/monthly spending limits, send alerts at 80% budget utilization, block requests when budget exceeded, and monitor real-time cost dashboards.
Q6: What is the cost difference between GPT-4 and GPT-3.5? A: GPT-4o: ~10/1M output tokens. GPT-3.5-turbo: ~1.50/1M output. GPT-4 is ~5-7x more expensive. Model routing saves by using GPT-3.5 for 60-70% of queries.
Q7: How would you optimize costs for a high-volume API? A: Implement semantic caching (30-60% hit rate), model routing (use GPT-3.5 for simple queries), prompt compression (20-40% token savings), request batching (amortize overhead), and result streaming (reduce timeout costs).
Q8: What metrics should be tracked for cost optimization? A: Cost per request, cost per user, cost per task type, cache hit rate, model distribution (GPT-4 vs GPT-3.5 usage), token usage trends, and budget utilization percentage.
Common Pitfalls & Solutions
| Pitfall | Impact | Solution |
|---|---|---|
| Stale cache | Incorrect responses | Implement TTL and semantic invalidation |
| Quality degradation | Poor user experience | A/B test model routing; maintain quality benchmarks |
| Cache poisoning | Wrong cached responses | Validate cached responses before storage |
| Cost overruns | Budget exceeded | Implement hard budget limits and real-time alerts |
| Memory overhead | Redis bloat | Monitor cache size, use LRU eviction |
| Incorrect routing | Quality loss | Regularly validate complexity classifier accuracy |
| Over-compression | Degraded responses | Test quality impact of prompt compression |
| Attribution errors | Wrong user charged | Validate user IDs, implement idempotency keys |
Summary with Key Takeaways
- Semantic caching can reduce costs by 30-60% by avoiding duplicate LLM calls
- Model routing saves money by using cheaper models for simple tasks (60-70% of queries)
- Token optimization reduces input costs through prompt compression (20-40% savings)
- Cost tracking enables visibility and budget management per user and per task
- Budget alerts prevent unexpected cost overruns in production
- Regular optimization reviews maintain cost efficiency as usage patterns change