Gemini API Caching in Production — Operational Notes from an Indie Mobile Developer

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
SYSTEM_DOC = "Long fixed content goes here..."  # Several thousand tokens in practice
 
# BAD: Variable content at the front
def bad_prompt_structure(user_message: str) -> str:
    """User message at the front blocks Implicit Caching."""
    prompt = f"{user_message}\n\nReference document below:\n{SYSTEM_DOC}"
    model = genai.GenerativeModel("gemini-2.0-flash")
    response = model.generate_content(prompt)
    return response.text
 
# GOOD: Fixed content first, variable content last
def good_prompt_structure(user_message: str) -> str:
    """Fixed prefix is more likely to hit Implicit Caching."""
    prompt = f"""Use the document below to answer the user's question.
 
===Document Start===
{SYSTEM_DOC}
===Document End===
 
User question: {user_message}"""
 
    model = genai.GenerativeModel("gemini-2.0-flash")
    response = model.generate_content(prompt)
 
    usage = response.usage_metadata
    cached = getattr(usage, "cached_content_token_count", 0)
    if cached > 0:
        print(f"HIT: Implicit cache saved {cached} tokens")
    else:
        print("MISS: full pricing")
 
    return response.text
 
# BEST: Use system_instruction
def use_system_instruction(user_message: str) -> str:
    """Putting fixed content into system_instruction maximises hit rate."""
    model = genai.GenerativeModel(
        "gemini-2.0-flash",
        system_instruction=SYSTEM_DOC,
    )
    response = model.generate_content(user_message)
 
    usage = response.usage_metadata
    cached = getattr(usage, "cached_content_token_count", 0)
    print(f"Input: {usage.prompt_token_count} / Cached: {cached}")
 
    return response.text

import google.generativeai as genai
from google.generativeai import caching
import datetime
import threading
from typing import Optional
import logging
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
logger = logging.getLogger(__name__)
 
class ProductionCacheStrategy:
    """
    Production cache manager.
    Combines Context Caching and Implicit Caching to minimise cost.
    """
 
    def __init__(
        self,
        model_name: str = "models/gemini-2.5-pro",
        system_prompt: str = "",
        cache_ttl_hours: int = 12,
        min_token_threshold: int = 2048,
        cache_display_name: str = "prod-cache",
    ):
        self.model_name = model_name
        self.system_prompt = system_prompt
        self.cache_ttl_hours = cache_ttl_hours
        self.min_token_threshold = min_token_threshold
        self.cache_display_name = cache_display_name
        self._cache: Optional[caching.CachedContent] = None
        self._lock = threading.Lock()
        self._request_count = 0
        self._cache_hit_count = 0
 
    def _estimate_tokens(self, text: str) -> int:
        """Rough token estimate. Use model.count_tokens() for accuracy."""
        return len(text) // 3
 
    def _should_use_context_cache(self) -> bool:
        """Is the system prompt long enough to bother with Context Caching?"""
        estimated = self._estimate_tokens(self.system_prompt)
        return estimated >= self.min_token_threshold
 
    def _get_or_create_cache(self) -> Optional[caching.CachedContent]:
        """Thread-safe get-or-create for the cache."""
        with self._lock:
            if self._cache is not None:
                now = datetime.datetime.now(datetime.timezone.utc)
                remaining_sec = (self._cache.expire_time - now).total_seconds()
                if remaining_sec > 300:
                    return self._cache
                logger.info("Cache near expiry, recreating")
 
            if not self._should_use_context_cache():
                logger.info("System prompt too short, skipping Context Caching")
                return None
 
            # Try to reuse an existing cache by name (survives process restarts)
            for existing in caching.CachedContent.list():
                if existing.display_name == self.cache_display_name:
                    now = datetime.datetime.now(datetime.timezone.utc)
                    remaining_h = (existing.expire_time - now).total_seconds() / 3600
                    if remaining_h > 0.5:
                        logger.info(f"Reusing existing cache: {existing.name} ({remaining_h:.1f}h left)")
                        self._cache = existing
                        return self._cache
 
            try:
                self._cache = caching.CachedContent.create(
                    model=self.model_name,
                    display_name=self.cache_display_name,
                    system_instruction=self.system_prompt,
                    ttl=datetime.timedelta(hours=self.cache_ttl_hours),
                )
                logger.info(f"Created new cache: {self._cache.name}")
                return self._cache
            except Exception as e:
                logger.error(f"Cache creation failed (falling back to Implicit Caching): {e}")
                return None
 
    def generate(self, user_message: str) -> dict:
        self._request_count += 1
        cache = self._get_or_create_cache()
 
        try:
            if cache is not None:
                model = genai.GenerativeModel.from_cached_content(cache)
                response = model.generate_content(user_message)
            else:
                model = genai.GenerativeModel(
                    self.model_name,
                    system_instruction=self.system_prompt,
                )
                response = model.generate_content(user_message)
 
            usage = response.usage_metadata
            cached_tokens = getattr(usage, "cached_content_token_count", 0)
 
            if cached_tokens > 0:
                self._cache_hit_count += 1
 
            return {
                "text": response.text,
                "cache_hit": cached_tokens > 0,
                "cached_tokens": cached_tokens,
                "input_tokens": usage.prompt_token_count,
                "output_tokens": usage.candidates_token_count,
            }
        except Exception as e:
            logger.error(f"Generation failed: {e}")
            raise
 
    def get_stats(self) -> dict:
        hit_rate = (
            self._cache_hit_count / self._request_count * 100
            if self._request_count > 0
            else 0
        )
        return {
            "total_requests": self._request_count,
            "cache_hits": self._cache_hit_count,
            "hit_rate_percent": round(hit_rate, 1),
        }
 
SYSTEM_PROMPT = """
[2,048+ token system prompt goes here]
Product manual, support policy, FAQ canonical answers, etc.
"""
 
strategy = ProductionCacheStrategy(
    model_name="models/gemini-2.5-pro",
    system_prompt=SYSTEM_PROMPT,
    cache_ttl_hours=12,
    cache_display_name="support-bot-v1",
)
 
result = strategy.generate("What's the return window?")
print(f"Answer: {result['text'][:200]}")
print(f"Cache hit: {result['cache_hit']}")
print(f"Cached tokens: {result['cached_tokens']}")
print(f"Stats: {strategy.get_stats()}")

import json
import pathlib
import hashlib
import google.generativeai as genai
 
CACHE_FILE = pathlib.Path("/tmp/.gemini_token_cache.json")
 
def count_tokens_cached(model_name: str, text: str) -> int:
    """Persist count_tokens results locally so identical text isn't re-billed."""
    key = hashlib.sha256(f"{model_name}::{text}".encode("utf-8")).hexdigest()
 
    if CACHE_FILE.exists():
        try:
            store = json.loads(CACHE_FILE.read_text())
        except json.JSONDecodeError:
            store = {}
    else:
        store = {}
 
    if key in store:
        return store[key]
 
    model = genai.GenerativeModel(model_name)
    count = model.count_tokens(text).total_tokens
    store[key] = count
    CACHE_FILE.write_text(json.dumps(store))
    return count

def cleanup_test_caches() -> int:
    """Delete every test-* cache at end of CI."""
    deleted = 0
    for c in caching.CachedContent.list():
        if c.display_name.startswith("test-"):
            try:
                c.delete()
                deleted += 1
            except Exception as e:
                print(f"WARN failed to delete {c.display_name}: {e}")
    print(f"CLEANUP: deleted {deleted} test caches")
    return deleted

import os
from datetime import date
 
def stable_display_name(base: str) -> str:
    """Build a display_name that includes environment and month."""
    env = os.environ.get("APP_ENV", "dev")
    # Month-key means the cache is recreated exactly once a month
    month_key = date.today().strftime("%Y-%m")
    return f"{env}-{base}-{month_key}"

import datetime
from google.generativeai import caching
 
def is_safely_usable(cache: caching.CachedContent, min_remaining_sec: int = 300) -> bool:
    """Avoid race conditions by treating near-expiry caches as unusable."""
    now = datetime.datetime.now(datetime.timezone.utc)
    remaining = (cache.expire_time - now).total_seconds()
    return remaining > min_remaining_sec

def cache_with_image(image_path: str, system_text: str) -> caching.CachedContent:
    """Cache content that includes an image, the right way."""
    print("Uploading via Files API...")
    uploaded_file = genai.upload_file(
        path=image_path,
        display_name="brand-palette-reference",
    )
    print(f"OK uploaded: {uploaded_file.uri}")
 
    return caching.CachedContent.create(
        model="models/gemini-2.5-pro",
        display_name="prod-brand-palette-2026-04",
        contents=[
            {
                "role": "user",
                "parts": [
                    {"text": system_text},
                    {"file_data": {"file_uri": uploaded_file.uri}},
                ],
            }
        ],
        ttl=datetime.timedelta(hours=6),
    )

def calculate_smart_ttl(hours_until_next_use: float, buffer_minutes: int = 30) -> int:
    """Compute TTL based on the next scheduled use."""
    raw = hours_until_next_use + (buffer_minutes / 60)
    return max(1, min(24, int(raw + 0.999)))
 
# Example: next batch is 4 hours out
ttl_hours = calculate_smart_ttl(4)
print(f"Recommended TTL: {ttl_hours}h")  # Recommended TTL: 5h

def daily_budget_alert(monthly_budget_usd: float) -> float:
    """Daily alert threshold: monthly budget divided by 30, then × 1.3."""
    return (monthly_budget_usd / 30) * 1.3
 
# An app with a $150/mo budget
threshold = daily_budget_alert(150)
print(f"Daily alert threshold: ${threshold:.2f}")  # $6.50

from prometheus_client import Counter, Gauge, Histogram
 
cache_hits_total = Counter(
    "gemini_cache_hits_total",
    "Cache hits",
    ["cache_type"],
)
cache_misses_total = Counter("gemini_cache_misses_total", "Cache misses")
token_savings_total = Counter(
    "gemini_token_savings_total", "Estimated tokens saved"
)
api_latency = Histogram(
    "gemini_api_latency_seconds",
    "API latency",
    buckets=[0.5, 1, 2, 5, 10, 30],
)
cache_hit_rate_gauge = Gauge("gemini_cache_hit_rate_percent", "Current cache hit rate")
 
class MonitoredCacheStrategy(ProductionCacheStrategy):
    """Production cache strategy with Prometheus metrics."""
 
    def generate(self, user_message: str) -> dict:
        import time
 
        start = time.time()
        result = super().generate(user_message)
        elapsed = time.time() - start
 
        api_latency.observe(elapsed)
 
        if result["cache_hit"]:
            cache_hits_total.labels(cache_type="context_or_implicit").inc()
            savings = int(result["cached_tokens"] * 0.75)
            token_savings_total.inc(savings)
        else:
            cache_misses_total.inc()
 
        stats = self.get_stats()
        cache_hit_rate_gauge.set(stats["hit_rate_percent"])
 
        return result