Exaktes Caching mit Redis
Cachen Sie LLM-Antworten, indem Sie den vollständigen Prompt hashen und das Ergebnis mit einer TTL in Redis speichern. So liefern Sie identische Anfragen sofort aus, ohne einen API-Aufruf durchzuführen.
Exaktes Caching mit Redis ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 1 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Warum LLM-Antworten cachen?
LLM-API-Aufrufe sind teuer: Eine einzelne GPT-4o-Anfrage kann je nach Tokenanzahl 0,005–0,15 $ kosten. In vielen Anwendungen ist ein erheblicher Anteil der eingehenden Anfragen identisch oder nahezu identisch mit früheren Anfragen – etwa bei FAQ-Bots, Kundensupportsystemen oder Code-Review-Tools, bei denen Benutzer wiederholt dieselben Fragen stellen. Caching kann in diesen Anwendungsfällen 20–50 Prozent der API-Aufrufe vermeiden, wodurch Kosten direkt sinken und die Latenz reduziert wird.
Exakter Cache: Cache-Key-Design
Ein exakter Cache speichert LLM-Antworten anhand eines deterministischen Hashes der Eingabe. Der Cache-Key muss jede Eingabe erfassen, die das Ergebnis beeinflusst: das Nachrichten-Array, den Modellnamen, die Temperatur und alle weiteren Parameter, die die Antwort verändern. Wenn einer dieser Werte im Key fehlt, entstehen Cache-Kollisionen, bei denen eine zwischengespeicherte Antwort für eine andere effektive Anfrage zurückgegeben wird.
import hashlib
import json
def make_cache_key(messages: list[dict], model: str, temperature: float) -> str:
# Create a canonical, order-stable representation
key_data = {
'model': model,
'temperature': temperature,
'messages': messages, # list order matters
}
# Serialize to JSON with sorted keys for determinism
serialized = json.dumps(key_data, sort_keys=True, ensure_ascii=False)
# Hash to a fixed-length key safe for Redis
return 'llm_cache:' + hashlib.sha256(serialized.encode()).hexdigest()Mit Redis verbinden
Redis ist aufgrund seiner Lese-Latenz im Sub-Millisekundenbereich und der integrierten TTL-Unterstützung die Standardwahl für das Caching von LLM-Antworten. Verwenden Sie die Bibliothek redis-py für den synchronen Zugriff oder aioredis (jetzt als redis.asyncio in redis-py integriert) für den asynchronen Zugriff in FastAPI-Anwendungen. Speichern Sie die Redis-Verbindung als Singleton, um eine Erschöpfung des Verbindungspools zu vermeiden.
import redis
import redis.asyncio as aioredis
# Synchronous Redis client
r = redis.Redis(
host='localhost',
port=6379,
db=0,
decode_responses=True, # return str instead of bytes
)
# Async Redis client (for FastAPI)
async_r = aioredis.Redis(
host='localhost',
port=6379,
db=0,
decode_responses=True,
)
# Test connection
print(r.ping()) # True if Redis is runningDas Cache-Aside-Muster implementieren
Das Cache-Aside-Muster ist die Standardstrategie für das Caching von LLM-APIs. Bei jeder Anfrage: (1) den Cache-Key berechnen, (2) Redis auf eine zwischengespeicherte Antwort prüfen, (3) bei einem Treffer (Cache Hit) diese sofort zurückgeben, (4) bei einem Fehlschlag (Cache Miss) die LLM-API aufrufen, (5) die Antwort mit einer TTL in Redis speichern, (6) die Antwort zurückgeben. Bei diesem Muster bleibt die Caching-Logik vom eigentlichen LLM-Aufruf getrennt.
import json
from openai import OpenAI
client = OpenAI()
def cached_completion(
messages: list[dict],
model: str = 'gpt-4o-mini',
temperature: float = 0.7,
ttl_seconds: int = 3600,
) -> str:
cache_key = make_cache_key(messages, model, temperature)
# Cache hit?
cached = r.get(cache_key)
if cached is not None:
print('[CACHE HIT]')
return json.loads(cached)
# Cache miss: call API
print('[CACHE MISS]')
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
)
result = response.choices[0].message.content
# Store in cache with TTL
r.setex(cache_key, ttl_seconds, json.dumps(result))
return resultAsynchrones Cache-Aside für FastAPI
Verwenden Sie in einer asynchronen FastAPI-Anwendung den asynchronen Redis-Client, damit Cache-Abfragen die Ereignisschleife nicht blockieren. Das Muster ist identisch mit der synchronen Variante, verwendet jedoch für alle Redis-Operationen await. Dadurch bleibt die Caching-Schicht vollständig nicht blockierend und mit dem asynchronen LLM-Client kompatibel.
from openai import AsyncOpenAI
import redis.asyncio as aioredis
import json
async_client = AsyncOpenAI()
async_r = aioredis.Redis(host='localhost', port=6379, decode_responses=True)
async def async_cached_completion(
messages: list[dict],
model: str = 'gpt-4o-mini',
temperature: float = 0.0,
ttl: int = 86400,
) -> str:
key = make_cache_key(messages, model, temperature)
cached = await async_r.get(key)
if cached:
return json.loads(cached)
response = await async_client.chat.completions.create(
model=model, messages=messages, temperature=temperature
)
result = response.choices[0].message.content
await async_r.setex(key, ttl, json.dumps(result))
return resultDie richtige TTL auswählen
Die TTL (Time To Live) legt fest, wie lange zwischengespeicherte Antworten gültig bleiben. Für faktische Fragen und Antworten mit stabilen Wissensdatenbanken maximieren lange TTLs (24–72 Stunden) die Cache-Trefferquote. Für Antworten, die aktuelle Daten widerspiegeln sollen (Nachrichten-Zusammenfassungen, Live-Preise), sind kurze TTLs (5–15 Minuten) oder vollständiger Verzicht auf Caching angemessen. Bei kreativen Aufgaben mit einer Temperatur ungleich null können zwischengespeicherte Antworten veralten – erwägen Sie Caching nur für temperature=0.
# TTL strategy by use case
TTL_STRATEGY = {
'faq_answering': 86400 * 7, # 7 days — stable facts
'code_explanation': 86400, # 1 day — code rarely changes
'document_summarization': 3600 * 6, # 6 hours
'news_analysis': 300, # 5 minutes — stale quickly
'creative_writing': 0, # 0 = don't cache (non-deterministic)
}
def get_ttl_for_use_case(use_case: str) -> int:
return TTL_STRATEGY.get(use_case, 3600) # default 1 hourCaching-Metriken und Überwachung
Verfolgen Sie die Cache-Trefferquote als zentrale Metrik zur Kostensenkung. Eine Cache-Trefferquote von 30 Prozent bedeutet, dass 30 Prozent der API-Aufrufe vermieden werden. Speichern Sie Treffer- und Fehlversuchszähler direkt in Redis, indem Sie INCR-Befehle auf separaten Zählern verwenden. Stellen Sie in Ihrer FastAPI-Anwendung einen /metrics-Endpunkt bereit, der die aktuelle Trefferquote, die Gesamtzahl der Anfragen und die geschätzten Kosteneinsparungen meldet, um den ROI des Cachings zu quantifizieren.
CACHE_HITS_KEY = 'llm_cache_metrics:hits'
CACHE_MISSES_KEY = 'llm_cache_metrics:misses'
async def async_cached_completion_instrumented(messages, model, temperature=0.0):
key = make_cache_key(messages, model, temperature)
cached = await async_r.get(key)
if cached:
await async_r.incr(CACHE_HITS_KEY)
return json.loads(cached)
await async_r.incr(CACHE_MISSES_KEY)
response = await async_client.chat.completions.create(
model=model, messages=messages, temperature=temperature
)
result = response.choices[0].message.content
await async_r.setex(key, 3600, json.dumps(result))
return result
async def get_cache_stats():
hits = int(await async_r.get(CACHE_HITS_KEY) or 0)
misses = int(await async_r.get(CACHE_MISSES_KEY) or 0)
total = hits + misses
return {'hit_rate': hits / total if total > 0 else 0, 'total': total}Strategien zur Cache-Invalidierung
Die Invalidierung eines exakten Caches ist unkompliziert, da die Keys deterministisch sind. Um einen bestimmten Eintrag zu invalidieren, berechnen Sie seinen Key erneut und rufen Sie r.delete(key) auf. Um alle Einträge für ein bestimmtes Prompt-Muster zu invalidieren, verwenden Sie Redis-Key-Präfixe mit einer Wildcard-Suche. Um den gesamten Cache nach einer größeren Aktualisierung der Wissensdatenbank zu invalidieren, rufen Sie r.flushdb() auf (mit Vorsicht verwenden – dadurch werden alle Keys in der Datenbank gelöscht).
async def invalidate_cache_entry(messages, model, temperature):
key = make_cache_key(messages, model, temperature)
deleted = await async_r.delete(key)
print(f'Deleted {deleted} cache entries')
async def invalidate_all_llm_cache():
# Scan for all keys with prefix 'llm_cache:'
keys_to_delete = []
async for key in async_r.scan_iter(match='llm_cache:*'):
keys_to_delete.append(key)
if keys_to_delete:
await async_r.delete(*keys_to_delete)
print(f'Invalidated {len(keys_to_delete)} cache entries')Komplexe Antworten serialisieren
Wenn Ihre Anwendung vollständige API-Antwortobjekte (nicht nur den Textinhalt) cached, serialisieren Sie sie sorgfältig. Das vollständige ChatCompletion-Objekt enthält die Token-Nutzung, die Modellversion und den Beendigungsgrund – nützlich für Protokollierung und Kostenverfolgung. Verwenden Sie die SDK-Methode .model_dump_json(), um Pydantic-Antwortobjekte in JSON-Strings zu serialisieren, und stellen Sie sie beim Abruf aus dem Cache mit ChatCompletion.model_validate_json() wieder her.
from openai.types.chat import ChatCompletion
async def cached_completion_full_response(
messages, model='gpt-4o-mini', temperature=0.0
):
key = make_cache_key(messages, model, temperature) + ':full'
cached = await async_r.get(key)
if cached:
return ChatCompletion.model_validate_json(cached) # reconstruct object
response = await async_client.chat.completions.create(
model=model, messages=messages, temperature=temperature
)
# Serialize Pydantic model to JSON
await async_r.setex(key, 3600, response.model_dump_json())
return responseCaching und Nichtdeterminismus
Exaktes Caching ist nur für deterministische oder nahezu deterministische Anfragen sinnvoll. Bei temperature=0 und top_p=1.0 erzeugen die meisten LLMs für dieselbe Eingabe dieselbe Ausgabe (aufgrund von Nichtdeterminismus bei Gleitkommazahlen ist dies jedoch nicht garantiert). Bei höheren Temperaturen veralten zwischengespeicherte Antworten, da das Modell andere Ausgaben erzeugt hätte. Cachen Sie immer bei temperature=0 oder dokumentieren Sie im Cache-Key eindeutig, dass Antworten variieren können.
Redis-Cluster und produktives Setup
Verwenden Sie für Produktivbereitstellungen mit großen Cache-Volumen Redis Cluster zur horizontalen Verteilung auf mehrere Knoten oder einen verwalteten Redis-Dienst wie AWS ElastiCache oder Redis Cloud. Legen Sie eine maxmemory-Richtlinie fest (typischerweise allkeys-lru, um bei vollem Speicher die am längsten nicht verwendeten Einträge zu entfernen), damit Redis nicht mit Arbeitsspeicher voll läuft und die Cache-Größe automatisch verwaltet wird.
# Redis configuration for production LLM caching
# In redis.conf:
# maxmemory 2gb
# maxmemory-policy allkeys-lru
# Connection with retry and connection pool
import redis
from redis.retry import Retry
from redis.backoff import ExponentialBackoff
retry = Retry(ExponentialBackoff(base=0.1), 3)
production_redis = redis.Redis(
host='your-redis-host.cache.amazonaws.com',
port=6379,
ssl=True,
decode_responses=True,
max_connections=50,
retry=retry,
retry_on_error=[redis.ConnectionError, redis.TimeoutError],
)Schnelltest
Testen Sie Ihr Verständnis des exakten Cachings von LLM-Antworten mit Redis aus dieser Lektion.
Zusammenfassung der Lektion
In dieser Lektion haben Sie gelernt: exaktes Caching hasht alle LLM-Eingaben, um einen deterministischen Cache-Schlüssel zu erzeugen, das Cache-Aside-Muster prüft Redis vor dem API-Aufruf und speichert Ergebnisse nach einem Cache-Miss, und die TTL-Auswahl sollte widerspiegeln, wie häufig sich Ihre Inhalte ändern — länger für stabiles Wissen, kürzer für dynamische Daten. Überwachen Sie die Cache-Trefferquote als zentrale Kennzahl zur Kostensenkung. Als Nächstes erstellen wir einen semantischen Cache für ähnliche, aber nicht identische Abfragen.
Häufig gestellte Fragen
Ist die Lektion „Exaktes Caching mit Redis“ kostenlos?
Ja — der vollständige Text von „Exaktes Caching mit Redis“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Exaktes Caching mit Redis“?
Cachen Sie LLM-Antworten, indem Sie den vollständigen Prompt hashen und das Ergebnis mit einer TTL in Redis speichern. So liefern Sie identische Anfragen sofort aus, ohne einen API-Aufruf durchzuführ… Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Engineering Academy zu starten?
Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 1 von 4.
Wie lange dauert die Lektion „Exaktes Caching mit Redis“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?
Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- Exaktes Caching mit Redis
- Semantisches Caching mit Embeddings
- Prompt-Prefix-Caching von OpenAI
- Batching, Model Routing und Kosten-Dashboards