Caching-Strategien zur Reduzierung von Latenz und RAG-Kosten

Warum einen Cache verwenden ?

Ohne Cache :

• Jede Anfrage → API-Aufruf ($$$)
• Latenz von 500ms+
• Durchsatzbeschränkungen

Mit Cache :

• Kostensenkung um 80%
• Antworten in 10ms aus dem Cache
• Keine Durchsatzbeschränkungen

1. Semantischer Anfrage-Cache

Verstecken Sie nicht exakte Übereinstimmungen — cachen Sie ähnliche Anfragen:

DEVELOPERpython
import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')
cache = {}  # {embedding: response}

def semantic_cache_lookup(query, threshold=0.95):
    query_emb = model.encode(query)

    # Prüfen, ob eine ähnliche Anfrage im Cache vorhanden ist
    for cached_emb, response in cache.items():
        similarity = np.dot(query_emb, cached_emb)
        if similarity > threshold:
            return response  # Cache hit!

    return None  # Cache miss

def rag_with_cache(query):
    # Zuerst den Cache prüfen
    cached = semantic_cache_lookup(query)
    if cached:
        return cached

    # Cache miss - kompletten RAG-Pipeline ausführen
    response = full_rag_pipeline(query)

    # Im Cache speichern
    cache[model.encode(query)] = response

    return response

2. Embedding-Cache

Cachen Sie embeddings, um Neuberechnungen zu vermeiden :

DEVELOPERpython
import hashlib
import redis

redis_client = redis.Redis(host='localhost', port=6379)

def get_embedding_cached(text):
    # Erstelle einen Cache-Schlüssel
    cache_key = f"emb:{hashlib.md5(text.encode()).hexdigest()}"

    # Cache überprüfen
    cached = redis_client.get(cache_key)
    if cached:
        return np.frombuffer(cached, dtype=np.float32)

    # Embedding berechnen
    embedding = openai_embed(text)

    # Im Cache speichern (läuft nach 7 Tagen ab)
    redis_client.setex(
        cache_key,
        604800,  # 7 jours
        embedding.tobytes()
    )

    return embedding

3. Integration GPTCache

DEVELOPERpython
from gptcache import Cache
from gptcache.embedding import OpenAI
from gptcache.similarity_evaluation import SearchDistanceEvaluation

cache = Cache()
cache.init(
    embedding_func=OpenAI().to_embeddings,
    similarity_evaluation=SearchDistanceEvaluation(),
)

def cached_llm_call(prompt):
    # Cache prüfen
    cached_response = cache.get(prompt)
    if cached_response:
        return cached_response

    # LLM aufrufen
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

    # Antwort cachen
    cache.set(prompt, response)

    return response

4. Zwei-Ebenen-Cache

Schneller Speicher + persistentes Redis :

DEVELOPERpython
from functools import lru_cache
import redis

redis_client = redis.Redis()

@lru_cache(maxsize=1000)
def l1_cache(query):
    # L2-Cache (Redis)
    cached = redis_client.get(f"rag:{query}")
    if cached:
        return cached.decode()

    # Cache miss - berechnen
    result = rag_pipeline(query)

    # In L2 speichern
    redis_client.setex(f"rag:{query}", 3600, result)

    return result

5. Cache-Invalidierung

DEVELOPERpython
import time

cache_with_ttl = {}

def get_with_ttl(key, ttl=3600):
    if key in cache_with_ttl:
        value, timestamp = cache_with_ttl[key]
        if time.time() - timestamp < ttl:
            return value
        else:
            del cache_with_ttl[key]  # Abgelaufen

    return None

def set_with_ttl(key, value):
    cache_with_ttl[key] = (value, time.time())

Kostenanalyse

Ohne Cache (1M Anfragen/Monat) :

• Embeddings : 100$
• LLM : 3000$
• Total : 3100$

Mit Cache (Trefferquote 80%) :

• Embeddings : 20$
• LLM : 600$
• Redis : 50$
• Total : 670$ (Einsparung von 78%)

Caching ist der am leichtesten zugängliche Hebel zur RAG-Optimierung. Implementieren Sie ihn von Anfang an.