Caching-strategie voor de Gemini API

De drie lagen van Gemini-caching

Efficiente Gemini-applicaties combineren drie onafhankelijke caching-lagen:

1. Context caching: sla grote, vaste inhoud op (handleidingen, knowledge base) zodat je die niet bij elke call opnieuw hoeft te sturen.
2. Response caching: bewaar antwoorden op identieke vragen zodat je de API helemaal niet hoeft te bellen.
3. Embedding caching: sla gegenereerde embeddings op zodat je niet steeds dezelfde tekst opnieuw hoeft te embedden.

Elke laag heeft een eigen use case en een eigen optimalisatiepotentieel.

Laag 1: Response caching

De eenvoudigste en goedkoopste cache: sla het volledige API-antwoord op voor identieke vragen, zodat je de API helemaal overslaat.

import hashlib
import json
import time
import os
from typing import Optional
from google import genai

client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

class ResponseCache:
    def __init__(self, cache_dir: str = "/tmp/gemini_cache", ttl: int = 3600):
        self.cache_dir = cache_dir
        self.ttl = ttl
        os.makedirs(cache_dir, exist_ok=True)

    def _cache_key(self, prompt: str, model: str) -> str:
        content = f"{model}:{prompt}"
        return hashlib.sha256(content.encode()).hexdigest()

    def get(self, prompt: str, model: str) -> Optional[str]:
        key = self._cache_key(prompt, model)
        cache_file = os.path.join(self.cache_dir, f"{key}.json")

        if not os.path.exists(cache_file):
            return None

        with open(cache_file) as f:
            data = json.load(f)

        if time.time() - data["timestamp"] > self.ttl:
            os.remove(cache_file)
            return None

        return data["response"]

    def set(self, prompt: str, model: str, response: str) -> None:
        key = self._cache_key(prompt, model)
        cache_file = os.path.join(self.cache_dir, f"{key}.json")

        with open(cache_file, "w") as f:
            json.dump({
                "timestamp": time.time(),
                "prompt": prompt[:100],
                "response": response
            }, f)

cache = ResponseCache(ttl=3600)
DEFAULT_MODEL = "gemini-2.5-flash"

def cached_generate(prompt: str, model_name: str = DEFAULT_MODEL) -> tuple:
    cached = cache.get(prompt, model_name)
    if cached:
        return cached, True

    response = client.models.generate_content(
        model=model_name,
        contents=prompt,
    )
    text = response.text
    cache.set(prompt, model_name, text)
    return text, False

Laag 2: Context caching

Voor grote vaste content (handboeken, KB-artikelen) maak je een expliciete context cache aan met client.caches.create. De TTL geef je op als een string in seconden, bijvoorbeeld "86400s" voor 24 uur.

from google.genai import types

CONTEXT_MODEL = "gemini-2.5-flash"

class ContextCacheManager:
    def __init__(self):
        self._caches = {}

    def get_or_create(self, cache_id: str, content: str, ttl_seconds: int = 86400):
        if cache_id in self._caches:
            cached = self._caches[cache_id]
            try:
                client.caches.get(name=cached.name)
                return cached
            except Exception:
                del self._caches[cache_id]

        cached = client.caches.create(
            model=CONTEXT_MODEL,
            config=types.CreateCachedContentConfig(
                contents=[
                    types.Content(
                        role="user",
                        parts=[types.Part(text=content)],
                    )
                ],
                ttl=f"{ttl_seconds}s",
                display_name=cache_id,
            ),
        )
        self._caches[cache_id] = cached
        return cached

context_manager = ContextCacheManager()

def query_with_context_cache(context: str, vraag: str, cache_id: str) -> str:
    cached = context_manager.get_or_create(cache_id, context)
    response = client.models.generate_content(
        model=CONTEXT_MODEL,
        contents=vraag,
        config=types.GenerateContentConfig(cached_content=cached.name),
    )
    return response.text

Laag 3: Embedding caching

Het huidige aanbevolen embedding-model is gemini-embedding-001. Sla berekende embeddings lokaal op zodat je dezelfde tekst niet steeds opnieuw embedt.

import pickle
from google.genai import types

EMBED_MODEL = "gemini-embedding-001"

class EmbeddingCache:
    def __init__(self, cache_path: str = "/tmp/embeddings.pkl"):
        self.cache_path = cache_path
        self._cache = self._load()

    def _load(self) -> dict:
        if os.path.exists(self.cache_path):
            with open(self.cache_path, "rb") as f:
                return pickle.load(f)
        return {}

    def _save(self) -> None:
        with open(self.cache_path, "wb") as f:
            pickle.dump(self._cache, f)

    def _key(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()

    def get(self, text: str) -> Optional[list]:
        return self._cache.get(self._key(text))

    def set(self, text: str, embedding: list) -> None:
        self._cache[self._key(text)] = embedding
        self._save()

    def batch_embed(self, teksten: list) -> list:
        resultaten = []
        niet_gecached = []
        indices = []

        for i, tekst in enumerate(teksten):
            cached = self.get(tekst)
            if cached is not None:
                resultaten.append((i, cached))
            else:
                niet_gecached.append(tekst)
                indices.append(i)

        if niet_gecached:
            result = client.models.embed_content(
                model=EMBED_MODEL,
                contents=niet_gecached,
                config=types.EmbedContentConfig(task_type="RETRIEVAL_DOCUMENT"),
            )
            for idx, tekst, item in zip(indices, niet_gecached, result.embeddings):
                vector = item.values
                self.set(tekst, vector)
                resultaten.append((idx, vector))

        resultaten.sort(key=lambda x: x[0])
        return [emb for _, emb in resultaten]

embedding_cache = EmbeddingCache()

Gecombineerde strategie

Onderstaande client brengt de drie lagen samen en houdt eenvoudige statistieken bij over je hit-rate.

class OptimalGeminiClient:
    def __init__(self):
        self.response_cache = ResponseCache(ttl=1800)
        self.embedding_cache = EmbeddingCache()
        self.context_manager = ContextCacheManager()
        self.stats = {"cache_hits": 0, "api_calls": 0}

    def ask(self, vraag: str) -> str:
        antwoord, hit = cached_generate(vraag)
        if hit:
            self.stats["cache_hits"] += 1
        else:
            self.stats["api_calls"] += 1
        return antwoord

    def ask_about_document(self, document: str, vraag: str, doc_id: str) -> str:
        response_key = f"{doc_id}:{vraag}"
        antwoord, hit = cached_generate(response_key)
        if hit:
            self.stats["cache_hits"] += 1
            return antwoord

        self.stats["api_calls"] += 1
        return query_with_context_cache(document, vraag, doc_id)

    def embed(self, teksten: list) -> list:
        return self.embedding_cache.batch_embed(teksten)

    def cache_stats(self) -> dict:
        totaal = self.stats["cache_hits"] + self.stats["api_calls"]
        hit_rate = self.stats["cache_hits"] / totaal if totaal > 0 else 0
        return {**self.stats, "hit_rate": f"{hit_rate:.1%}"}

Let op een subtiel maar belangrijk detail: in ask_about_document cachen we eerst op de gecombineerde sleutel doc_id:vraag (response caching), en pas bij een misser vallen we terug op context caching voor het hele document. Zo betaal je alleen voor verwerking als de vraag echt nieuw is.

Praktische volgorde van besparen

Begin bij de goedkoopste laag en werk omhoog:

1. Zet response caching aan voor alle volledig identieke vragen. Dit voorkomt API-calls en is bijna gratis.
2. Voeg context caching toe zodra je dezelfde grote context (een handleiding, een dossier) meerdere keren bevraagt.
3. Cache embeddings als je een vector-zoekindex of RAG-pijplijn bouwt en documenten zelden wijzigen.

Meet daarna je hit-rate met cache_stats(). Een lage hit-rate betekent meestal dat je sleutels te veel variabele velden bevatten (denk aan tijdstempels of gebruikers-ID's) die je eruit kunt laten.