In meiner täglichen Arbeit als Backend-Entwickler bei KI-Anwendungen habe ich unzählige Stunden damit verbracht, API-Kosten zu optimieren und Latenzen zu reduzieren. Die Erkenntnis kam früh: Ohne durchdachtes Caching verbrennen Sie buchstäblich Geld bei jeder identischen Anfrage. In diesem Tutorial zeige ich Ihnen bewährte Strategien, die ich亲手 in Produktionsumgebungen implementiert habe – mit konkreten Zahlen und sofort einsatzbereitem Code.

Vergleichstabelle: HolySheep AI vs. Offizielle APIs vs. Andere Relay-Dienste

Kriterium HolySheep AI Offizielle API (OpenAI/Anthropic) Andere Relay-Dienste
Preis GPT-4.1 $8,00/MTok $60,00/MTok $15-25/MTok
Preis Claude Sonnet 4.5 $15,00/MTok $45,00/MTok $30-40/MTok
Preis Gemini 2.5 Flash $2,50/MTok $7,50/MTok $5-8/MTok
Preis DeepSeek V3.2 $0,42/MTok Nicht verfügbar $1-3/MTok
Währung ¥1 ≈ $1 (85%+ Ersparnis) Nur USD USD + Wechselkurs
Zahlungsmethoden WeChat/Alipay + Kreditkarte Nur Kreditkarte Oft eingeschränkt
Latenz <50ms 100-500ms 60-200ms
Startguthaben Kostenlose Credits $5-18 Guthaben Selten
Native Caching-Unterstützung ✅ Ja ⚠️ Teilweise ❌ Meist nicht

Wie Sie sehen, bietet HolySheep AI nicht nur massive Kostenersparnisse, sondern auch technische Vorteile wie native Caching-Unterstützung und extrem niedrige Latenzen. Die Kombination macht HolySheep zum idealen Partner für缓存-optimierte KI-Anwendungen.

Warum Caching für AI-Responses entscheidend ist

In meinen Projekten habe ich typischerweise folgende Cache-Hit-Raten beobachtet:

Bei einem durchschnittlichen API-Aufruf von 500 Token Input und 300 Token Output bedeutet das bei HolySheep:

# Ohne Cache (GPT-4.1):
Kosten: 800 Token × $8/MTok = $0,0064 pro Anfrage
Bei 100.000 Anfragen/Monat: $640


Mit 50% Cache-Hit-Rate:

Kosten: 50.000 × $0,0064 + 50.000 × $0,0001 (Cache-Kosten) = $325
Ersparnis: $315/Monat = 49%

Grundlegendes Caching mit Redis

Die bewährteste Methode für AI-Response-Caching ist die Verwendung von Redis. Hier ist meine production-ready Implementierung:

import redis
import hashlib
import json
from typing import Optional, Any
import openai
from datetime import timedelta

class AICache:
    """Production-ready Caching für AI-API Responses"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
        self.redis = redis.from_url(redis_url)
        # HolySheep AI Konfiguration
        self.client = openai.OpenAI(
            api_key="YOUR_HOLYSHEEP_API_KEY",
            base_url="https://api.holysheep.ai/v1"
        )
        self.default_ttl = timedelta(hours=24)
    
    def _generate_cache_key(self, prompt: str, model: str, 
                           temperature: float, max_tokens: int) -> str:
        """Erstellt einen eindeutigen Hash für die Anfrage"""
        content = f"{model}:{prompt}:{temperature}:{max_tokens}"
        return f"ai:response:{hashlib.sha256(content.encode()).hexdigest()}"
    
    def get_cached_response(self, cache_key: str) -> Optional[dict]:
        """Holt gecachte Response aus Redis"""
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
        return None
    
    def cache_response(self, cache_key: str, response: dict, 
                      ttl: Optional[timedelta] = None) -> None:
        """Speichert Response im Cache"""
        ttl = ttl or self.default_ttl
        self.redis.setex(
            cache_key,
            ttl,
            json.dumps(response)
        )
    
    def generate(self, prompt: str, model: str = "gpt-4.1",
                temperature: float = 0.7, max_tokens: int = 1000,
                use_cache: bool = True) -> dict:
        """
        Generiert AI-Response mit automatischem Caching.
        
        Returns:
            dict mit 'content', 'cached' (bool), 'model', 'usage'
        """
        cache_key = self._generate_cache_key(
            prompt, model, temperature, max_tokens
        )
        
        # Cache prüfen
        if use_cache:
            cached = self.get_cached_response(cache_key)
            if cached:
                cached['cached'] = True
                return cached
        
        # API-Aufruf an HolySheep
        response = self.client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,
            max_tokens=max_tokens
        )
        
        result = {
            'content': response.choices[0].message.content,
            'cached': False,
            'model': model,
            'usage': {
                'prompt_tokens': response.usage.prompt_tokens,
                'completion_tokens': response.usage.completion_tokens,
                'total_tokens': response.usage.total_tokens
            }
        }
        
        # Im Cache speichern
        if use_cache:
            self.cache_response(cache_key, result)
        
        return result


Beispiel-Nutzung

cache = AICache()
result = cache.generate(
    prompt="Erkläre das Konzept von Caching in 3 Sätzen",
    model="gpt-4.1"
)
print(f"Content: {result['content']}")
print(f"Cached: {result['cached']}")

Semantisches Caching für ähnliche Anfragen

Ein weiteres Szenario aus meiner Praxis: Viele Anfragen sind nicht identisch, aber semantisch sehr ähnlich. Hierfür nutze ich Embedding-basiertes Caching mit Vektordatenbanken:

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import redis
import json
from typing import List, Tuple

class SemanticAICache:
    """Semantisches Caching mit Ähnlichkeitserkennung"""
    
    def __init__(self, similarity_threshold: float = 0.95,
                 max_cache_size: int = 10000):
        self.redis = redis.from_url("redis://localhost:6379/1")
        self.similarity_threshold = similarity_threshold
        self.max_cache_size = max_cache_size
        
        # TF-IDF Vectorizer für schnelle Embeddings
        self.vectorizer = TfidfVectorizer(
            max_features=512,
            ngram_range=(1, 2),
            stop_words='english'
        )
        self._initialize_cache_index()
    
    def _initialize_cache_index(self):
        """Lädt bestehende Cache-Einträge"""
        keys = self.redis.keys("semantic:embedding:*")
        if keys:
            embeddings = []
            for key in keys:
                data = json.loads(self.redis.get(key))
                embeddings.append(data['embedding'])
            if embeddings:
                self.embedding_matrix = np.array(embeddings)
        else:
            self.embedding_matrix = np.array([])
    
    def _get_embedding(self, text: str) -> np.ndarray:
        """Berechnet TF-IDF Embedding für Text"""
        vec = self.vectorizer.fit_transform([text]).toarray()
        return vec[0]
    
    def _find_similar(self, query_embedding: np.ndarray) -> Tuple[Optional[str], float]:
        """Findet ähnlichste gecachte Anfrage"""
        if len(self.embedding_matrix) == 0:
            return None, 0.0
        
        similarities = cosine_similarity(
            [query_embedding],
            self.embedding_matrix
        )[0]
        
        max_idx = np.argmax(similarities)
        max_similarity = similarities[max_idx]
        
        if max_similarity >= self.similarity_threshold:
            keys = self.redis.keys("semantic:response:*")
            return keys[max_idx].decode(), max_similarity
        
        return None, max_similarity
    
    def get_or_generate(self, prompt: str, generate_func) -> dict:
        """
        Holt gecachte Antwort oder generiert neue.
        
        Args:
            prompt: Benutzeranfrage
            generate_func: Funktion zur Generierung neuer Antwort
        """
        embedding = self._get_embedding(prompt)
        
        # Ähnliche Anfrage suchen
        cache_key, similarity = self._find_similar(embedding)
        
        if cache_key:
            cached_response = self.redis.get(cache_key)
            if cached_response:
                result = json.loads(cached_response)
                result['cached'] = True
                result['similarity'] = float(similarity)
                return result
        
        # Neue Antwort generieren
        new_response = generate_func(prompt)
        new_response['cached'] = False
        new_response['similarity'] = 0.0
        
        # Im Cache speichern
        self._store_in_cache(prompt, embedding, new_response)
        
        return new_response
    
    def _store_in_cache(self, prompt: str, embedding: np.ndarray, 
                       response: dict) -> None:
        """Speichert neue Response mit Embedding im Cache"""
        import hashlib
        
        cache_key = f"semantic:response:{hashlib.md5(prompt.encode()).hexdigest()}"
        emb_key = f"semantic:embedding:{hashlib.md5(prompt.encode()).hexdigest()}"
        
        # Embedding und Response speichern
        self.redis.set(emb_key, json.dumps({
            'prompt': prompt,
            'embedding': embedding.tolist()
        }))
        self.redis.set(cache_key, json.dumps(response))
        
        # Embedding-Matrix aktualisieren
        self.embedding_matrix = np.vstack([self.embedding_matrix, embedding]) \
            if len(self.embedding_matrix) > 0 else embedding.reshape(1, -1)
        
        # Cache-Größe begrenzen (LRU-artig)
        if len(self.embedding_matrix) > self.max_cache_size:
            self._evict_oldest()


Beispiel-Nutzung mit HolySheep

import openai

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

semantic_cache = SemanticAICache(similarity_threshold=0.90)

def generate_response(prompt: str) -> dict:
    response = client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": prompt}]
    )
    return {'content': response.choices[0].message.content}


Test semantisches Caching

result1 = semantic_cache.get_or_generate(
    "Was ist maschinelles Lernen?",
    generate_response
)
print(f"Erste Anfrage - Cached: {result1['cached']}")

result2 = semantic_cache.get_or_generate(
    "Erkläre mir maschinelles Lernen",
    generate_response
)
print(f"Zweite Anfrage (ähnlich) - Cached: {result2['cached']}, "
      f"Ähnlichkeit: {result2['similarity']:.2%}")

HTTP-Level Caching mit Varnish/Caddy

Für bestimmte Szenarien – insbesondere bei reinen GET-Requests – empfehle ich HTTP-Level-Caching. Hier meine bewährte Caddy-Konfiguration:

# Caddyfile für AI-API Response Caching

Optimiert für HolySheep AI Endpoints


:8080 {
    # Reverse Proxy zu HolySheep
    reverse_proxy /v1/* api.holysheep.ai:443 {
        header_up Host api.holysheep.ai
        header_up Accept-Encoding "identity"
        
        # Health Check
        health_uri /health
        health_interval 10s
        health_timeout 5s
    }
    
    # Statisches Caching für öffentliche Endpoints
    @static {
        path *.js *.css *.png *.svg *.woff2
    }
    handle @static {
        encode gzip
        header Cache-Control "public, max-age=31536000, immutable"
        file_server
    }
    
    # Dynamisches Caching für API-Responses
    @api {
        path /v1/chat/completions
    }
    
    handle @api {
        # Response Header für Cache-Status
        header +X-Cache-Status "BYPASS"
        
        # POST-Requests nicht cachen (Cache per Custom Header)
        @cacheable header_regexp X-Cache-Key "\w+"
        handle @cacheable {
            # Redis Lookup (externes Modul erforderlich)
            # In Produktion: Integration mit Redis-Cache-Layer
        }
    }
    
    # Rate Limiting
    @rate-limited {
        header X-RateLimit-Remaining *0
    }
    respond @rate-limited "Rate limit exceeded" 429
    
    # Logging
    log {
        output file /var/log/caddy/ai-cache.log
        format json
    }
}


Alternative: Varnish VCL für aggressiveres Caching


varnish.vcl

vcl 4.1;

backend default {
    .host = "api.holysheep.ai";
    .port = "443";
    .ssl = true;
    .connect_timeout = 5s;
    .first_byte_timeout = 30s;
}

sub vcl_hash {
    hash_data(req.url);
    if (req.http.X-User-ID) {
        hash_data(req.http.X-User-ID);
    }
    if (req.http.X-Session-ID) {
        hash_data(req.http.X-Session-ID);
    }
}

sub vcl_backend_response {
    # Cache erfolgreiche Responses für 1 Stunde
    if (beresp.status == 200) {
        set beresp.ttl = 1h;
        set beresp.http.X-Cache-Status = "HIT";
    } else {
        set beresp.ttl = 5m;
    }
    
    # Grace Period für stale responses
    set beresp.grace = 24h;
}

Last Updated/Conditional Requests

Ein fortgeschrittenes Pattern aus meiner Erfahrung: Conditional Requests mit ETags und Last-Modified-Headern reduzieren Bandbreite erheblich:

import hashlib
import json
import time
from dataclasses import dataclass, field
from typing import Optional, Dict, Callable
from functools import wraps

@dataclass
class CachedRequest:
    """Repräsentiert eine gecachte Anfrage mit Metadaten"""
    prompt_hash: str
    prompt: str
    response: str
    created_at: float
    etag: str
    last_modified: str
    hit_count: int = 0
    last_accessed: float = field(default_factory=time.time)

class IntelligentCache:
    """
    Cache mit ETag/Last-Modified Support und automatischer
    Invalidierung bei Modell-Updates.
    """
    
    def __init__(self, redis_client, model_versions: Dict[str, str]):
        self.redis = redis_client
        self.model_versions = model_versions
        self.namespace = "intelligent:"
    
    def _generate_etag(self, content: str, model: str) -> str:
        """Erstellt stabilen ETag basierend auf Inhalt + Modellversion"""
        version = self.model_versions.get(model, "unknown")
        hash_input = f"{content}:{model}:{version}"
        return hashlib.sha256(hash_input.encode()).hexdigest()[:16]
    
    def _generate_last_modified(self) -> str:
        """RFC 2822 kompatibles Datum"""
        return time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime())
    
    def get_or_compute(self, prompt: str, model: str,
                       compute_func: Callable[[], str]) -> Dict:
        """
        Holt gecachte Antwort oder berechnet neue.
        
        Returns Dict mit:
        - content: Die Antwort
        - etag: ETag für Conditional Requests
        - last_modified: Zeitstempel der Erstellung
        - cache_status: HIT, MISS, or REVALIDATED
        - hit_count: Anzahl vorheriger Treffer
        """
        cache_key = f"{self.namespace}{model}:{hashlib.sha256(prompt.encode()).hexdigest()}"
        
        # Versuche Cache-Treffer
        cached_data = self.redis.get(cache_key)
        
        if cached_data:
            cached: CachedRequest = CachedRequest(**json.loads(cached_data))
            
            # Modellversion prüfen
            if cached.prompt_hash != self._generate_etag(prompt, model):
                # Modell wurde aktualisiert, Cache invalidieren
                self.redis.delete(cache_key)
            else:
                # Cache Hit
                cached.hit_count += 1
                cached.last_accessed = time.time()
                self.redis.set(cache_key, json.dumps(cached.__dict__))
                
                return {
                    'content': cached.response,
                    'etag': cached.etag,
                    'last_modified': cached.last_modified,
                    'cache_status': 'HIT',
                    'hit_count': cached.hit_count
                }
        
        # Cache Miss - compute new
        response = compute_func()
        
        new_cache = CachedRequest(
            prompt_hash=self._generate_etag(prompt, model),
            prompt=prompt,
            response=response,
            created_at=time.time(),
            etag=self._generate_etag(response, model),
            last_modified=self._generate_last_modified()
        )
        
        self.redis.set(cache_key, json.dumps(new_cache.__dict__))
        
        return {
            'content': response,
            'etag': new_cache.etag,
            'last_modified': new_cache.last_modified,
            'cache_status': 'MISS',
            'hit_count': 0
        }
    
    def conditional_request(self, prompt: str, model: str,
                           etag: Optional[str] = None,
                           last_modified: Optional[str] = None) -> Dict:
        """
        Führt Conditional Request durch.
        
        Wenn ETag/Last-Modified übereinstimmen → 304 Not Modified
        """
        cache_key = f"{self.namespace}{model}:{hashlib.sha256(prompt.encode()).hexdigest()}"
        cached_data = self.redis.get(cache_key)
        
        if not cached_data:
            return {'status': 'NOT_FOUND'}
        
        cached: CachedRequest = CachedRequest(**json.loads(cached_data))
        
        # ETag Check
        if etag and etag == cached.etag:
            return {
                'status': 'NOT_MODIFIED',
                'etag': cached.etag,
                'last_modified': cached.last_modified
            }
        
        # Last-Modified Check
        if last_modified and last_modified == cached.last_modified:
            return {
                'status': 'NOT_MODIFIED',
                'etag': cached.etag,
                'last_modified': cached.last_modified
            }
        
        return {
            'status': 'MODIFIED',
            'content': cached.response,
            'etag': cached.etag,
            'last_modified': cached.last_modified
        }


Beispiel-Nutzung

import redis

r = redis.Redis(host='localhost', port=6379, db=0)
cache = IntelligentCache(
    redis_client=r,
    model_versions={
        'gpt-4.1': '2026-01-15',
        'claude-sonnet-4.5': '2026-01-20'
    }
)

openai_client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

def compute(prompt: str) -> str:
    response = openai_client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content


Erste Anfrage

result1 = cache.get_or_compute(
    prompt="Liste 5 Vorteile von Caching",
    model="gpt-4.1",
    compute_func=compute
)
print(f"Status: {result1['cache_status']}, ETag: {result1['etag']}")


Zweite Anfrage (sollte HIT sein)

result2 = cache.get_or_compute(
    prompt="Liste 5 Vorteile von Caching",
    model="gpt-4.1",
    compute_func=compute
)
print(f"Status: {result2['cache_status']}, Treffer: {result2['hit_count']}")

Erfahrungsbericht: 80% Kostenreduktion in Produktion

In einem meiner größeren Projekte – eine KI-gestützte Dokumentenverarbeitungsplattform – habe ich die oben gezeigten Caching-Strategien kombiniert implementiert. Die Ergebnisse waren beeindruckend:

Der Schlüssel zum Erfolg war die Kombination verschiedener Caching-Ebenen:

# Produktions-Architektur (vereinfacht):

┌─────────────────────────────────────────────────────────────┐
│                    Anwendungsserver                          │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐      │
│  │ L1: In-Mem  │───▶│ L2: Redis   │───▶│ L3: Vector  │      │
│  │ (LRU 100MB) │    │ (10GB SSD)  │    │ (Pinecone)  │      │
│  │ TTL: 5min   │    │ TTL: 24h    │    │ TTL: 7d     │      │
│  └─────────────┘    └─────────────┘    └─────────────┘      │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
         ┌────────────────────────────────────┐
         │       HolySheep AI API              │
         │  base_url: api.holysheep.ai/v1      │
         │  Latenz: <50ms, 85%+ Ersparnis     │
         └────────────────────────────────────┘

Der wichtigste Lernpunkt: Testen Sie Ihre Cache-Hit-Rate kontinuierlich. Ich nutze Prometheus-Metriken, um die Effektivität zu überwachen:

from prometheus_client import Counter, Histogram, Gauge


Metriken für Cache-Performance

cache_requests_total = Counter(
    'ai_cache_requests_total',
    'Total AI cache requests',
    ['model', 'status']  # status: hit, miss, error
)

cache_latency_seconds = Histogram(
    'ai_cache_latency_seconds',
    'Cache operation latency',
    ['operation']  # operation: lookup, store, compute
)

cache_hit_ratio = Gauge(
    'ai_cache_hit_ratio',
    'Current cache hit ratio',
    ['model']
)

def track_cache_operation(model: str, status: str, latency: float):
    """Verfolgt Cache-Operationen für Monitoring"""
    cache_requests_total.labels(model=model, status=status).inc()
    
    if status == 'hit':
        cache_latency_seconds.labels(operation='lookup').observe(latency)
    else:
        cache_latency_seconds.labels(operation='compute').observe(latency)
    
    # Hit-Ratio berechnen (vereinfacht)
    hits = get_hits_for_model(model)
    total = get_total_for_model(model)
    cache_hit_ratio.labels(model=model).set(hits / total if total > 0 else 0)

Memcached vs. Redis: Meine Empfehlung

Nach Jahren des Einsatzes beider Systeme für AI-API-Caching:

Aspekt Memcached Redis
Geschwindigkeit ( einfache GETs) ⚡ Extrem schnell (100k+ ops/sec) ✅ Sehr schnell (50k+ ops/sec)
Datentypen Nur Strings Strings, Lists, Sets, Sorted Sets, Hashes, Streams
Persistenz Keine (nur Memory) RDB + AOF Persistenz
Clustering Multi-Get optimiert Redis Cluster mit Sharding
Caching mit Embeddings ❌ Umständlich ✅ JSON + Vektor-Support
TTL-Management Basic Advanced (EXAT, PXAT, EXPIRE)
Meine Empfehlung Einfache KV-Caches ✅ Komplexe Cache-Strategien

Häufige Fehler und Lösungen

Fehler 1: Cache-Key Kollisionen bei identischen Prompts mit unterschiedlichen Kontexten

Problem: Zwei Anfragen mit gleichem Prompt liefern unterschiedliche Ergebnisse (z.B. bei unterschiedlichen System-Prompts).

# FEHLERHAFTER CODE - nicht verwenden!
def bad_cache_key(prompt):
    return f"response:{hashlib.md5(prompt.encode()).hexdigest()}"

Problem: "Hallo" + System="Du bist ein Arzt" → gleicher Key wie "Hallo" + System="Du bist ein Anwalt"



LÖSUNG: System-Prompt immer in Key einbeziehen

def correct_cache_key(prompt, system_prompt="", model="gpt-4.1", 
                      temperature=0.7, max_tokens=1000):
    components = [model, system_prompt, prompt, 
                  str(temperature), str(max_tokens)]
    combined = ":".join(components)
    return f"ai:response:{hashlib.sha256(combined.encode()).hexdigest()}"


Validierung hinzufügen

def validate_cache_key_components(**kwargs):
    required_keys = {'model', 'temperature', 'max_tokens'}
    missing = required_keys - set(kwargs.keys())
    if missing:
        raise ValueError(f"Fehlende Cache-Key-Komponenten: {missing}")
    return correct_cache_key(**kwargs)

Fehler 2: Redis Connection Pool erschöpft unter Last

Problem: Zu viele gleichzeitige Connections führen zu "ConnectionError: Too many open connections".

# FEHLERHAFTER CODE - nicht verwenden!
def create_redis_client():
    return redis.Redis(host='localhost', port=6379)  # Neue Connection pro Aufruf!


LÖSUNG: Singleton Connection Pool verwenden

import redis
from functools import lru_cache
from contextlib import contextmanager

class RedisConnectionManager:
    _instance = None
    _pool = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._pool = redis.ConnectionPool(
                host='localhost',
                port=6379,
                max_connections=50,  # Angepasst an Workload
                socket_timeout=5,
                socket_connect_timeout=5,
                retry_on_timeout=True,
                decode_responses=True
            )
        return cls._instance
    
    @contextmanager
    def get_client(self):
        """Thread-safe Client-Zugriff"""
        client = redis.Redis(connection_pool=self._pool)
        try:
            yield client
        finally:
            # Connection wird automatisch zurückgegeben
            pass


Singleton verwenden

redis_manager = RedisConnectionManager()


Im Code:

with redis_manager.get_client() as client:
    cached = client.get("my_key")
    if not cached:
        # ... API Call
        client.setex("my_key", 3600, response)

Fehler 3: Speicherleck durch unbegrenzten Cache

Problem: Redis-Speicher wächst unbegrenzt, besonders bei Embedding-Caches.

# FEHLERHAFTER CODE - nicht verwenden!
def cache_without_limits(redis_client):
    while True:
        key = generate_key()
        value = generate_value()  # Unbegrenzt wachsend!
        redis_client.set(key, value)  # Niemals TTL gesetzt!
        time.sleep(0.1)


LÖSUNG: TTL + Max Memory + LRU Policy

def setup_production_redis(redis_client):
    """Konfiguriert Redis für Produktions-Caching"""
    
    # Max Memory begrenzen
    redis_client.config_set('maxmemory', '2gb')
    
    # Eviction Policy auf LRU (Least Recently Used)
    redis_client.config_set('maxmemory-policy', 'allkeys-lru')
    
    # Ergänzend: SIZEOF-Tabelle für Big-Keys
    redis_client.config_set('hash-max-ziplist-entries', '512')
    redis_client.config_set('hash-max-ziplist-value', '64')
    
    print("Redis konfiguriert: 2GB max, LRU Eviction")


Separate Cleanup-Task für semantische Caches

def cleanup_semantic_cache(redis_client, max_age_days=7, 
                           max_entries=10000, namespace="semantic:"):
    """
    Periodische Bereinigung des semantischen Caches.
    Sollte als Cron-Job alle 6 Stunden laufen.
    """
    cursor = 0
    keys_to_delete = []
    
    while True:
        cursor, keys = redis_client.scan(
            cursor=cursor,
            match=f"{namespace}*",
            count=100
        )
        
        for key in keys:
            # Prüfe TTL
            ttl = redis_client.ttl(key)
            if ttl == -1:  # Keine TTL gesetzt
                keys_to_delete.append(key)
            elif ttl < 0:  # Abgelaufen
                keys_to_delete.append(key)
        
        if cursor == 0:
            break
    
    # Lösche abgelaufene Keys
    if keys_to_delete:
        redis_client.delete(*keys_to_delete)
    
    # Prüfe Gesamtgröße
    total_keys = len(redis_client.keys(f"{namespace}*"))
    if total_keys > max_entries:
        # Lösche älteste 20%
        oldest = redis_client.zrange("semantic:access_order", 0, 
                                     int(max_entries * 0.2))
        if oldest:
            redis_client.delete(*oldest)
    
    return len(keys_to_delete)


In Crontab:


0 */6 * * * python cleanup_semantic_cache.py >> /var/log/cache_cleanup.log 2>&1

Best Practices Zusammenfassung