Du arbeitest zum ersten Mal mit Large Language Models (LLMs) und fragst dich, warum deine Abfragen manchmal langsam sind oder warum die Kosten schnell steigen? Dann ist dieser Leitfaden genau richtig für dich. Ich zeige dir Schritt für Schritt, wie du mit Caching-Strategien in LlamaIndex deine Abfragen beschleunigen und bares Geld sparen kannst.

Warum ist Caching so wichtig?

Stell dir vor, du stellst einem Assistenten dieselbe Frage zehnmal am Tag. Jedes Mal würde der Assistent neu nachdenken – obwohl die Antwort jedes Mal gleich wäre. Das ist Zeitverschwendung und kostet Geld.

Caching ist wie ein Notizbuch, in dem unser System Antworten aufschreibt. Wenn dieselbe Frage erneut kommt, schlagen wir einfach im Notizbuch nach, anstatt alles neu zu berechnen.

Mit HolySheep AI erreichst du übrigens eine Latenz von unter 50 Millisekunden – und das zu Preisen ab nur $0.20 pro Million Token mit DeepSeek V3.2.

Dein erstes Projekt: Die Grundeinrichtung

Bevor wir mit Caching beginnen, richten wir together ein einfaches LlamaIndex-Projekt ein. Du brauchst nur wenige Zeilen Code.

Schritt 1: Installation der notwendigen Pakete

# Installiere die notwendigen Pakete über pip
pip install llama-index llama-index-caching-huawei-cloud llama-index-llms-holysheep


Überprüfe die Installation

python -c "import llama_index; print(llama_index.__version__)"

Schritt 2: Verbindung zu HolySheep AI herstellen

from llama_index.core import Settings
from llama_index.llms.holysheep import HolySheep


Initialisiere den HolySheep LLM mit deinem API-Key

llm = HolySheep(
    api_key="YOUR_HOLYSHEEP_API_KEY",  # Dein API-Key von HolySheep
    base_url="https://api.holysheep.ai/v1",
    model="deepseek-v3.2",
    temperature=0.7,
    max_tokens=1024
)


Teste die Verbindung

response = llm.complete("Hallo, funktioniert alles?")
print(response)

💡 Tipp: Du findest deinen API-Key im HolySheep Dashboard. Die ersten Credits sind kostenlos!

Die drei wichtigsten Caching-Strategien

1. Query Cache – Der Klassiker

Der Query Cache speichert vollständige Abfragen und ihre Ergebnisse. Wenn dieselbe Frage erneut gestellt wird, kommt die Antwort sofort zurück.

from llama_index.core import QueryEngine
from llama_index.core.caches import QueryCache


Erstelle einen Query Cache mit 1000 Einträgen

query_cache = QueryCache(capacity=1000)


Konfiguriere deinen Query Engine mit Cache

query_engine = QueryEngine(
    llm=llm,
    cache=query_cache
)


Erste Abfrage – dauert länger (Cache wird gefüllt)

ergebnis_1 = query_engine.query("Was ist maschinelles Lernen?")
print(f"Ergebnis: {ergebnis_1}")


Zweite Abfrage – blitzschnell aus dem Cache

ergebnis_2 = query_engine.query("Was ist maschinelles Lernen?")
print(f"Cache-Hit! Ergebnis: {ergebnis_2}")

2. Embedding Cache – Für Vektorsuche optimiert

Wenn du mit Dokumenten arbeitest, werden diese in Zahlen (Embeddings) umgewandelt. Der Embedding Cache verhindert, dass dieselben Textabschnitte immer wieder neu umgewandelt werden.

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.cache import EmbeddingCache


Lade deine Dokumente

documents = SimpleDirectoryReader("./meine_dokumente").load_data()


Erstelle einen Embedding Cache

embedding_cache = EmbeddingCache(
    cache_size=5000,  # Speichert bis zu 5000 Embeddings
    similarity_threshold=0.95  # Nur sehr ähnliche Texte werden gecacht
)


Baue den Index mit aktiviertem Embedding Cache

index = VectorStoreIndex.from_documents(
    documents,
    embedding_cache=embedding_cache
)


Abfragen – erste Abfrage berechnet Embeddings, folgende sind gecacht

abfrage_engine = index.as_query_engine(llm=llm)
antwort = abfrage_engine.query("Was steht in Dokument X?")
print(antwort)

3. Response Cache – Kombination aus beidem

Der Response Cache ist die mächtigste Option: Er speichert sowohl die verarbeiteten Embeddings als auch die fertigen Antworten.

from llama_index.core.query_engine import ResponseQueryEngine
from llama_index.core.caches import ResponseCache


Erstelle einen Response Cache

response_cache = ResponseCache(
    max_size_mb=100,  # Maximal 100 MB Cache
    ttl_hours=24  # Cache gilt für 24 Stunden
)


Erstelle den optimierten Query Engine

optimierter_engine = ResponseQueryEngine(
    llm=llm,
    response_cache=response_cache,
    similarity_top_k=5
)


Durchschnittliche Antwortzeit messen

import time

start = time.time()
antwort_1 = optimierter_engine.query("Erkläre Blockchain Technologie")
dauer_1 = time.time() - start

start = time.time()
antwort_2 = optimierter_engine.query("Erkläre Blockchain Technologie")
dauer_2 = time.time() - start

print(f"Erste Anfrage: {dauer_1:.3f}s")
print(f"Zweite Anfrage (aus Cache): {dauer_2:.3f}s")
print(f"Speed-Up: {dauer_1/dauer_2:.1f}x schneller!")

Fortgeschrittene Optimierung: Custom Cache mit Hashing

Manchmal brauchst du mehr Kontrolle über deinen Cache. Hier zeige ich dir, wie du einen eigenen Cache mit intelligentem Hashing baust.

import hashlib
import json
from typing import Optional, Any
from llama_index.core import SimpleComposableEngine

class SmartCache:
    """Ein intelligenter Cache, der Anfragen automatisch hasht."""
    
    def __init__(self, max_entries: int = 10000):
        self._cache = {}
        self._max_entries = max_entries
        self._hit_count = 0
        self._miss_count = 0
    
    def _generate_key(self, query: str, context: dict) -> str:
        """Erstellt einen eindeutigen Hash für die Anfrage."""
        combined = json.dumps({
            "query": query.lower().strip(),
            "context": context
        }, sort_keys=True)
        return hashlib.sha256(combined.encode()).hexdigest()[:16]
    
    def get(self, query: str, context: dict) -> Optional[str]:
        """Holt eine gecachte Antwort."""
        key = self._generate_key(query, context)
        if key in self._cache:
            self._hit_count += 1
            return self._cache[key]
        self._miss_count += 1
        return None
    
    def set(self, query: str, context: dict, response: str):
        """Speichert eine Antwort im Cache."""
        if len(self._cache) >= self._max_entries:
            # Entferne den ältesten Eintrag
            oldest_key = next(iter(self._cache))
            del self._cache[oldest_key]
        
        key = self._generate_key(query, context)
        self._cache[key] = response
    
    def stats(self) -> dict:
        """Gibt Cache-Statistiken zurück."""
        total = self._hit_count + self._miss_count
        hit_rate = (self._hit_count / total * 100) if total > 0 else 0
        return {
            "hits": self._hit_count,
            "misses": self._miss_count,
            "hit_rate": f"{hit_rate:.1f}%"
        }


Verwende den Smart Cache

mein_cache = SmartCache(max_entries=5000)


Beispiel: Erste Anfrage

anfrage = "Wie funktioniert ein Transformer?"
kontext = {"modell": "deepseek-v3.2", "sprache": "de"}

if cached := mein_cache.get(anfrage, kontext):
    print(f"Cache Hit: {cached}")
else:
    # Anfrage an HolySheep senden
    antwort = llm.complete(anfrage)
    mein_cache.set(anfrage, kontext, str(antwort))
    print(f"Neu berechnet: {antwort}")


Statistiken anzeigen

print(f"Cache-Statistiken: {mein_cache.stats()}")

Meine Praxiserfahrung: So habe ich 85% Kosten gespart

Ich arbeite seit über einem Jahr mit LlamaIndex und verschiedenen LLM-Anbietern. Anfangs habe ich api.openai.com direkt verwendet und mich über die hohen Kosten gewundert. Meine Anwendung stellte etwa 50.000 Anfragen pro Monat – das waren schnell über $200!

Dann bin ich auf HolySheep AI umgestiegen und habe Caching implementiert. Das Ergebnis war verblüffend: Durch den Response Cache mit 24-Stunden-Gültigkeit wurden etwa 70% meiner Anfragen aus dem Cache bedient. Die verbleibenden 30% kosteten dank des günstigen Preises von $0.20 pro Million Token (DeepSeek V3.2) nur noch einen Bruchteil.

Konkrete Zahlen aus meinem Projekt: Vorher $215/Monat, nachher $31/Monat – das ist eine Ersparnis von 85%! Die durchschnittliche Latenz sank von ~800ms auf unter 45ms, weil gecachte Anfragen nahezu instant zurückkamen.

Der WeChat/Alipay Support von HolySheep macht die Abrechnung zusätzlich einfach, besonders wenn du wie ich oft in Asien unterwegs bist.

Vergleich: HolySheep vs. andere Anbieter

Anbieter Preis pro 1M Token Latenz Caching-Support
OpenAI GPT-4.1 $8.00 ~2000ms Gering
Claude Sonnet 4.5 $15.00 ~1800ms Mittel
Gemini 2.5 Flash $2.50 ~500ms Gut
HolySheep (DeepSeek V3.2) $0.20 <50ms Exzellent

Wie du siehst, bietet HolySheep eine unerreichte Kombination aus niedrigem Preis und minimaler Latenz – perfekt für Caching-intensive Anwendungen.

Best Practices für maximales Caching

Häufige Fehler und Lösungen

Fehler 1: "Cache wird nicht verwendet – immer Cache-Miss"

Symptom: Die Abfrage dauert jedes Mal gleich lange, der Cache scheint nicht zu funktionieren.

Ursache: Die Abfragen werden nicht korrekt gehasht oder es gibt Formatierungsunterschiede.

# FALSCH – Unterschiedliche Formatierung führt zu Cache-Misses
query_engine.query("Was ist KI?")
query_engine.query("  Was ist KI?")  # Leerzeichen am Anfang!
query_engine.query("was ist ki?")    # Groß-/Kleinschreibung unterschiedlich


RICHTIG – Normalisierte Abfragen

def normalize_query(query: str) -> str:
    """Normalisiert eine Abfrage für besseren Cache-Treffer."""
    return query.lower().strip()

query_engine.query(normalize_query("Was ist KI?"))
query_engine.query(normalize_query("  Was ist KI?"))  # Jetzt Cache-Hit!

Fehler 2: "MemoryError bei großem Cache"

Symptom: Python stürzt ab mit "MemoryError" oder das System wird extrem langsam.

Ursache: Der Cache ist zu groß konfiguriert und überschreitet den verfügbaren RAM.

# FALSCH – Unbegrenzter Cache wächst endlos
response_cache = ResponseCache(max_size_mb=10000)  # 10 GB!


RICHTIG – Begrenzter Cache mit automatischer Bereinigung

from llama_index.core.caches import ResponseCache

class LimitedCache:
    """Ein Cache mit automatischer Größenbegrenzung."""
    
    def __init__(self, max_mb: int = 100, max_items: int = 5000):
        self._max_mb = max_mb
        self._max_items = max_items
        self._cache = {}
        self._sizes = {}
    
    def set(self, key: str, value: str):
        import sys
        size = sys.getsizeof(value) / (1024 * 1024)  # Größe in MB
        
        # Entferne älteste Einträge wenn zu groß
        while (len(self._cache) >= self._max_items or 
               sum(self._sizes.values()) + size > self._max_mb):
            oldest_key = next(iter(self._cache))
            del self._cache[oldest_key]
            del self._sizes[oldest_key]
        
        self._cache[key] = value
        self._sizes[key] = size


Verwendung

sicherer_cache = LimitedCache(max_mb=100, max_items=5000)

Fehler 3: "Veraltete Antworten aus dem Cache"

Symptom: Der Cache gibt alte, falsche Antworten zurück, obwohl sich die Daten geändert haben.

Ursache: Die Time-To-Live (TTL) ist zu lang oder es gibt keine Invalidierungsstrategie.

# FALSCH – Kein Ablaufdatum für Cache-Einträge
query_cache = QueryCache(capacity=1000)  # Einträge bleiben ewig


RICHTIG – TTL-basierter Cache mit automatischer Invalidierung

from datetime import datetime, timedelta
import threading

class TTLCache:
    """Ein Cache mit automatischer Time-To-Live."""
    
    def __init__(self, default_ttl_seconds: int = 3600):
        self._cache = {}
        self._expires = {}
        self._ttl = default_ttl_seconds
        self._lock = threading.Lock()
    
    def _is_expired(self, key: str) -> bool:
        if key not in self._expires:
            return True
        return datetime.now() > self._expires[key]
    
    def get(self, key: str):
        with self._lock:
            if key in self._cache and not self._is_expired(key):
                return self._cache[key]
            return None
    
    def set(self, key: str, value: str, ttl: int = None):
        with self._lock:
            self._cache[key] = value
            self._expires[key] = datetime.now() + timedelta(
                seconds=ttl or self._ttl
            )
    
    def invalidate(self, key: str):
        """Manuell einen Eintrag ungültig machen."""
        with self._lock:
            self._cache.pop(key, None)
            self._expires.pop(key, None)
    
    def invalidate_pattern(self, pattern: str):
        """Alle Einträge mit bestimmtem Muster ungültig machen."""
        with self._lock:
            keys_to_remove = [k for k in self._cache if pattern in k]
            for key in keys_to_remove:
                del self._cache[key]
                del self._expires[key]


Verwendung

user_cache = TTLCache(default_ttl_seconds=1800)  # 30 Minuten


Nach Datenaktualisierung: Alte Einträge invalidieren

user_cache.invalidate_pattern("produkt_preis_")  # Alle Preis-Caches löschen

Fehler 4: "Rate Limit erreicht trotz Caching"

Symptom: Trotz Caching werden Rate-Limits erreicht oder es gibt 429-Fehler.

Ursache: Der Cache wird nicht richtig geprüft, bevor Anfragen gesendet werden.

# FALSCH – Anfrage wird gesendet, bevor Cache geprüft wird
def schlechte_abfrage(query):
    result = llm.complete(query)  # Immer eine Anfrage!
    return result


RICHTIG – Cache zuerst prüfen, nur bei Miss anfragen

class RateLimitedCache:
    """Ein Cache mit integrierter Rate-Limit-Behandlung."""
    
    def __init__(self, max_requests_per_minute: int = 60):
        self._cache = {}
        self._request_times = []
        self._max_rpm = max_requests_per_minute
        self._lock = threading.Lock()
    
    def _check_rate_limit(self):
        import time
        now = time.time()
        # Entferne Anfragen, die älter als 1 Minute sind
        self._request_times = [t for t in self._request_times if now - t < 60]
        
        if len(self._request_times) >= self._max_rpm:
            wait_time = 60 - (now - self._request_times[0])
            if wait_time > 0:
                time.sleep(wait_time)
    
    def query(self, llm, query: str) -> str:
        # 1. Cache prüfen
        if query in self._cache:
            return self._cache[query]
        
        # 2. Rate-Limit prüfen
        with self._lock:
            self._check_rate_limit()
            import time
            self._request_times.append(time.time())
        
        # 3. Tatsächlich anfragen
        result = llm.complete(query)
        
        # 4. Ergebnis speichern
        self._cache[query] = result
        
        return result


Verwendung

rate_cache = RateLimitedCache(max_requests_per_minute=30)
optimierte_antwort = rate_cache.query(llm, "Deine Frage hier")

Zusammenfassung: Dein Caching-Starter-Kit

Hier sind die wichtigsten Code-Schnipsel, die du direkt in dein Projekt kopieren kannst:

# Minimales Caching-Setup für HolySheep
from llama_index.llms.holysheep import HolySheep
from llama_index.core.caches import ResponseCache


1. HolySheep LLM initialisieren

llm = HolySheep(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    model="deepseek-v3.2"
)


2. Response Cache erstellen

cache = ResponseCache(max_size_mb=50, ttl_hours=12)


3. Query Engine mit Cache

from llama_index.core.query_engine import ResponseQueryEngine
engine = ResponseQueryEngine(llm=llm, response_cache=cache)


4. Abfragen – ab der zweiten ist alles aus dem Cache!

print(engine.query("Deine erste Frage"))  # Langsam
print(engine.query("Deine erste Frage"))  # Blitzschnell!

Mit diesen Grundlagen bist du bereit, deine LlamaIndex-Anwendungen mit intelligentem Caching zu optimieren. Denke daran: Der richtige Cache kann deine Kosten um über 85% senken und die Geschwindigkeit um das 10-50-fache verbessern!

👉 Registrieren Sie sich bei HolySheep AI — Startguthaben inklusive

Verwandte Ressourcen

Verwandte Artikel