Cutting Gemini API Costs by 80%: Context Caching and Implicit Caching

# ❌ BAD: Dynamic content at the front breaks prefix matching
def bad_prompt(user_id: str, user_message: str, large_context: str) -> str:
    return f"""
User ID: {user_id}
Timestamp: {datetime.now().isoformat()}
Session: {session_id}
 
{large_context}
 
User question: {user_message}
"""
 
# ✅ GOOD: Static content first, dynamic content last
def good_prompt(large_context: str, user_id: str, user_message: str) -> str:
    return f"""
{large_context}
 
---
User ID: {user_id}
User question: {user_message}
"""

import google.generativeai as genai
from typing import NamedTuple
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro")
 
# Large static context — coding guidelines, ~35,000 tokens
CODING_GUIDELINES = """
[Comprehensive Python Coding Guidelines]
1. Use snake_case for variable and function names
2. Every function must have type hints
3. Maximum function length: 30 lines
...
[35,000 tokens of guidelines continue]
"""
 
class UsageStats(NamedTuple):
    total_input_tokens: int
    cached_tokens: int
    output_tokens: int
    hit_rate_pct: float
    estimated_savings_usd: float
 
PRICING_PER_TOKEN = {
    "standard_input": 0.00000125,
    "cached_input": 0.000000313,
    "output": 0.000005,
}
 
def review_code(code_snippet: str) -> tuple[str, UsageStats]:
    """
    Code review using Implicit Caching.
    The large guidelines block is always at the front.
    """
    prompt = f"{CODING_GUIDELINES}\n\nPlease review this code:\n\n```python\n{code_snippet}\n```"
    
    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(temperature=0)
    )
    
    usage = response.usage_metadata
    total_input = usage.prompt_token_count or 0
    cached = usage.cached_content_token_count or 0
    output = usage.candidates_token_count or 0
    non_cached = total_input - cached
    
    actual_cost = (
        non_cached * PRICING_PER_TOKEN["standard_input"] +
        cached * PRICING_PER_TOKEN["cached_input"] +
        output * PRICING_PER_TOKEN["output"]
    )
    hypothetical_cost = (
        total_input * PRICING_PER_TOKEN["standard_input"] +
        output * PRICING_PER_TOKEN["output"]
    )
    
    stats = UsageStats(
        total_input_tokens=total_input,
        cached_tokens=cached,
        output_tokens=output,
        hit_rate_pct=cached / max(total_input, 1) * 100,
        estimated_savings_usd=hypothetical_cost - actual_cost,
    )
    
    return response.text, stats
 
# Test with consecutive requests
snippets = [
    "def get_user(id):\n    return db.query(f'SELECT * FROM users WHERE id={id}')",
    "class Config:\n    DEBUG = True\n    SECRET_KEY = 'hardcoded-secret'",
    "import requests\nresponse = requests.get(url, verify=False)",
]
 
total_savings = 0.0
for i, snippet in enumerate(snippets, 1):
    review, stats = review_code(snippet)
    total_savings += stats.estimated_savings_usd
    print(f"Request {i}: {stats.cached_tokens} cached tokens ({stats.hit_rate_pct:.0f}% hit), "
          f"saved ${stats.estimated_savings_usd:.5f}")
 
print(f"\nEstimated total savings across {len(snippets)} requests: ${total_savings:.4f}")

# Cache created with one model string
cache = caching.CachedContent.create(
    model="models/gemini-2.5-pro",
    ...
)
 
# Request sent with a slightly different string
model = genai.GenerativeModel("gemini-2.5-pro-latest")  # Different alias!

GEMINI_MODEL = "models/gemini-2.5-pro"  # Single source of truth

def safe_generate(cache_name: str, message: str, fallback_content: str) -> str:
    """Always handle expired caches gracefully."""
    max_retries = 2
    for attempt in range(max_retries):
        try:
            cache = caching.CachedContent.get(cache_name)
            model = genai.GenerativeModel.from_cached_content(cache)
            return model.generate_content(message).text
        except Exception as e:
            error_str = str(e).lower()
            if any(kw in error_str for kw in ["not found", "invalid_argument", "expired"]):
                if attempt < max_retries - 1:
                    print(f"Cache expired or invalid, recreating (attempt {attempt + 1})...")
                    new_cache = get_or_create_cache(fallback_content)
                    cache_name = new_cache.name
                else:
                    raise RuntimeError(f"Cache recovery failed after {max_retries} attempts") from e
            else:
                raise

def is_cache_eligible(content: str, model_name: str = "models/gemini-2.5-pro") -> bool:
    """Check if content meets minimum token threshold for Context Caching."""
    count_response = genai.GenerativeModel(model_name).count_tokens(content)
    token_count = count_response.total_tokens
    MIN_CONTEXT_CACHE_TOKENS = 32768
    print(f"Content token count: {token_count}")
    if token_count < MIN_CONTEXT_CACHE_TOKENS:
        print(f"Too short for Context Caching. Need at least {MIN_CONTEXT_CACHE_TOKENS}, got {token_count}.")
        return False
    return True

from dataclasses import dataclass, field
from collections import defaultdict
import json
 
@dataclass
class CostMonitor:
    requests: int = 0
    total_input_tokens: int = 0
    total_cached_tokens: int = 0
    total_output_tokens: int = 0
    actual_cost_usd: float = 0.0
    hypothetical_cost_usd: float = 0.0
 
    def record(self, usage_metadata) -> None:
        total_in = usage_metadata.prompt_token_count or 0
        cached = usage_metadata.cached_content_token_count or 0
        output = usage_metadata.candidates_token_count or 0
        non_cached = total_in - cached
 
        actual = (
            non_cached * 0.00000125 +
            cached * 0.000000313 +
            output * 0.000005
        )
        hypothetical = total_in * 0.00000125 + output * 0.000005
 
        self.requests += 1
        self.total_input_tokens += total_in
        self.total_cached_tokens += cached
        self.total_output_tokens += output
        self.actual_cost_usd += actual
        self.hypothetical_cost_usd += hypothetical
 
    @property
    def cache_hit_rate(self) -> float:
        return self.total_cached_tokens / max(self.total_input_tokens, 1) * 100
 
    @property
    def savings_rate(self) -> float:
        saved = self.hypothetical_cost_usd - self.actual_cost_usd
        return saved / max(self.hypothetical_cost_usd, 0.0001) * 100
 
    def summary(self) -> str:
        return (
            f"Requests: {self.requests:,} | "
            f"Cache hit rate: {self.cache_hit_rate:.1f}% | "
            f"Actual cost: ${self.actual_cost_usd:.4f} | "
            f"Savings: ${self.hypothetical_cost_usd - self.actual_cost_usd:.4f} "
            f"({self.savings_rate:.1f}%)"
        )
 
# Global monitor instance — log this to your observability platform
monitor = CostMonitor()