Produktion: Skalierung und Caching

Lernziele

• Optimierung der Leistung eines Embedding-Systems in der Produktion
• Implementierung von Caching zur Reduzierung von Kosten und Latenz
• Verwaltung von Skalierung und hoher Verfügbarkeit

Herausforderungen in der Produktion

Während der Entwicklung ist es kein Problem, für jede Anfrage die OpenAI API aufzurufen: Sie könnten nur zehn Aufrufe pro Tag tätigen. In der Produktion stößt derselbe Code jedoch gleichzeitig an drei Wänden. Erstens Kosten: Jeder Aufruf wird abgerechnet, und eine vollständige Neuindizierung eines Korpus wird jedes Mal in vollem Umfang berechnet. Zweitens Latenz: Eine Netzwerkrundreise von 100 bis 300 ms pro Anfrage addiert sich zur Such- und Generierungszeit, und der Nutzer spürt die Summe aller drei Faktoren. Schließlich Ratenlimits: OpenAI begrenzt die Anzahl der Aufrufe pro Minute – ein Limit, das Sie in der Entwicklung nie sehen werden, aber am Starttag erreichen werden.

Alle drei Probleme haben im Wesentlichen dieselbe Lösung: Rufen Sie die API nicht auf, wenn es nicht notwendig ist.

Caching von Embeddings

Ein Embedding ist deterministisch: Der gleiche Text mit demselben Modell liefert immer denselben Vektor. Dies macht Caching hier hochwirksam, weit mehr als bei einem generativen LLM. Für eine Anwendung mit moderatem Verkehr ist der Standard-lru_cache-Decorator der Standardbibliothek ausreichend und benötigt nur zwei Zeilen. Der Rückgabewert wird in ein Tupel umgewandelt, da lru_cache hashbare Werte erfordert; das ist die einzige Nuance, die man kennen muss.

from functools import lru_cache
from openai import OpenAI
import hashlib
import json

client = OpenAI()

@lru_cache(maxsize=10_000)
def embedding_cache(text: str, model: str = "text-embedding-3-large") -> tuple:
    """LRU-Cache für Embeddings (Tupel, da es hashbar ist)."""
    response = client.embeddings.create(input=text, model=model)
    return tuple(response.data[0].embedding)

# Nutzung - der 2. Aufruf ist sofortig
emb1 = embedding_cache("Hello world")  # API-Aufruf
emb2 = embedding_cache("Hello world")  # Aus Cache

Dieser Cache verschwindet bei jedem Neustart und wird nicht über Prozesse hinweg geteilt: Sobald Sie zwei Instanzen hinter einem Load Balancer ausführen, baut jede ihre eigene neu auf. Redis löst beide Probleme. Der Schlüssel ist ein Hash des Modell-Text-Paares, der verhindert, dass eine Modelländerung alte Vektoren bereitstellt, und eine TTL von 30 Tagen begrenzt das Speicherwachstum.

Die Methode get_embeddings_batch verdient besondere Aufmerksamkeit, da sie den Unterschied bei der Ingestion ausmacht. Sie prüft den Cache nicht nur Text für Text: Sie trennt die Treffer, sendet sie in einem einzigen API-Aufruf und fügt jedes Ergebnis dann an seiner ursprünglichen Position wieder ein, unter Verwendung von indices_a_calculer. Bei einem Batch von tausend Dokumenten, von denen neunhundert bereits bekannt sind, gehen Sie von einem Aufruf pro Dokument auf einen einzigen Aufruf für die verbleibenden hundert.

import redis
import numpy as np
import json

class EmbeddingCacheRedis:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.client = OpenAI()
        self.ttl = 86400 * 30  # 30 Tage

    def _cache_key(self, text: str, model: str) -> str:
        h = hashlib.md5(f"{model}:{text}".encode()).hexdigest()
        return f"emb:{h}"

    def get_embedding(
        self, text: str, model: str = "text-embedding-3-large"
    ) -> list[float]:
        key = self._cache_key(text, model)

        # Cache prüfen
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)

        # Generieren und cachen
        response = self.client.embeddings.create(
            input=text, model=model
        )
        embedding = response.data[0].embedding

        self.redis.setex(key, self.ttl, json.dumps(embedding))
        return embedding

    def get_embeddings_batch(
        self, texts: list[str], model: str = "text-embedding-3-large"
    ) -> list[list[float]]:
        """Batch mit teilweisem Caching."""
        results = [None] * len(texts)
        to_calculate = []
        indices_to_calculate = []

        # Cache für jeden Text prüfen
        for i, text in enumerate(texts):
            key = self._cache_key(text, model)
            cached = self.redis.get(key)
            if cached:
                results[i] = json.loads(cached)
            else:
                to_calculate.append(text)
                indices_to_calculate.append(i)

        # Treffer in einem einzigen Aufruf berechnen
        if to_calculate:
            response = self.client.embeddings.create(
                input=to_calculate, model=model
            )
            for resp_item in sorted(response.data, key=lambda x: x.index):
                idx = indices_to_calculate[resp_item.index]
                text = to_calculate[resp_item.index]
                embedding = resp_item.embedding

                results[idx] = embedding
                key = self._cache_key(text, model)
                self.redis.setex(key, self.ttl, json.dumps(embedding))

        print(f"Cache: {len(texts) - len(to_calculate)}/{len(texts)} Treffer")
        return results

# Nutzung
cache = EmbeddingCacheRedis()
emb = cache.get_embedding("cache test")

Umgang mit Ratenlimits

Caching reduziert den Verkehr, eliminiert ihn aber nicht, und ein Verkehrsspitze wird schließlich einen RateLimitError auslösen. Die richtige Reaktion besteht nicht darin, sofort erneut zu versuchen – dies verschlimmert die Verstopfung –, sondern zu warten und die Verzögerung bei jedem Versuch zu verdoppeln: eine Sekunde, zwei, vier, acht, sechzehn. Fünf Versuche decken somit etwa dreißig Sekunden ab, was die vasten meisten vorübergehenden Spitzen absorbiert, und der letzte Versuch wirft die Ausnahme erneut auf, sodass das Vorfall gemeldet wird, anstatt stumm verschluckt zu werden.

import time
from openai import RateLimitError

class RobustEmbeddingClient:
    def __init__(self, max_retries: int = 5):
        self.client = OpenAI()
        self.max_retries = max_retries

    def create(
        self, input: str | list[str], model: str = "text-embedding-3-large",
        **kwargs
    ):
        """Aufruf mit exponentieller Wiederholung."""
        for attempt in range(self.max_retries):
            try:
                return self.client.embeddings.create(
                    input=input, model=model, **kwargs
                )
            except RateLimitError as e:
                if attempt == self.max_retries - 1:
                    raise
                wait = 2 ** attempt
                print(f"Ratenlimit erreicht, Pause für {wait}s...")
                time.sleep(wait)

Asynchrone Verarbeitung

Bei Ingestion-Pipelines ist die Flaschenhals nicht die Berechnung, sondern das Warten auf das Netzwerk: Ihr Programm verbringt den Großteil seiner Zeit damit, nichts zu tun. Asynchrone Verarbeitung ermöglicht es Ihnen, mehrere Anfragen gleichzeitig in der Luft zu halten. Der Semaphore begrenzt diese Parallelität absichtlich, da das Starten von hundert gleichzeitigen Aufrufen nur die Ratenlimits auslöst, die Sie gerade gelernt haben, zu handhaben – drei bis fünf gleichzeitige Anfragen sind eine gesunde Einstellung. Texte werden zunächst in Batches von hundert aufgeteilt, und der Startindex jedes Batches reist mit der Aufgabe mit, sodass as_completed, das Ergebnisse in beliebiger Reihenfolge zurückgibt, sie dennoch korrekt platzieren kann.

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def generate_embeddings_async(
    texts: list[str],
    model: str = "text-embedding-3-large",
    max_concurrent: int = 5
) -> list[list[float]]:
    """Generierung von Embeddings parallel mit einem Semaphore."""
    semaphore = asyncio.Semaphore(max_concurrent)
    results = [None] * len(texts)

    async def process_batch(batch_idx, batch):
        async with semaphore:
            response = await async_client.embeddings.create(
                input=batch, model=model
            )
            return batch_idx, [
                d.embedding
                for d in sorted(response.data, key=lambda x: x.index)
            ]

    # Aufteilung in Batches von 100
    tasks = []
    batch_size = 100
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        tasks.append(process_batch(i, batch))

    for coro in asyncio.as_completed(tasks):
        batch_idx, embeddings = await coro
        for j, emb in enumerate(embeddings):
            results[batch_idx + j] = emb

    return results

# Nutzung
embeddings = asyncio.run(
    generate_embeddings_async(my_texts, max_concurrent=3)
)

Optimierung der Speicherung

Dann gibt es noch die Infrastrukturrechnung. Zehn Tausend Vektoren mit 3.072 Dimensionen in float32 belegen etwa 122 MB; eine Million belegt zwölf Gigabyte, die im Arbeitsspeicher gehalten werden müssen, damit die Suche schnell bleibt. Die Quantisierung auf int8 ordnet jede Koordinate auf 256 Ganzzahlwerte zu und teilt den Fußabdruck um das Vierfache auf, auf etwa 30 MB im folgenden Beispiel. Der Verlust an Präzision besteht, bleibt aber marginal im Vergleich zu den Ähnlichkeitslücken, mit denen Sie arbeiten.

import numpy as np

def quantize_int8(embeddings: np.ndarray) -> tuple:
    """Quantisierung von Embeddings auf int8 (4x weniger Speicher)."""
    min_val = embeddings.min(axis=1, keepdims=True)
    max_val = embeddings.max(axis=1, keepdims=True)
    scale = (max_val - min_val) / 255.0

    quantized = ((embeddings - min_val) / scale).astype(np.int8)
    return quantized, min_val, scale

def dequantize_int8(quantized, min_val, scale):
    """Wiederherstellung von Embeddings aus int8."""
    return quantized.astype(np.float32) * scale + min_val

# Größenvergleich
embeddings = np.random.randn(10000, 3072).astype(np.float32)
print(f"float32 : {embeddings.nbytes / 1e6:.1f} MB")  # ~122 MB

quantized, min_val, scale = quantize_int8(embeddings)
print(f"int8 :    {quantized.nbytes / 1e6:.1f} MB")    # ~30 MB

Das einfachste Hebel bleibt jedoch der dimensions-Parameter der API, der kürzere Vektoren direkt erzeugt, ohne dass Sie nachträglich etwas tun müssen: 3.072 Dimensionen wiegen etwa 24 KB pro Vektor, 1.024 Dimensionen sinken auf 8 KB und 256 Dimensionen auf 2 KB.

# 3072 dim -> ~24 KB pro Vektor
# 1024 dim -> ~8 KB pro Vektor  (API-Reduktion)
# 256 dim  -> ~2 KB pro Vektor  (API-Reduktion)

response = client.embeddings.create(
    input="text",
    model="text-embedding-3-large",
    dimensions=1024  # 3x weniger Speicher
)

Produktionsarchitektur

Zusammengesetzt ergeben diese Teile eine schlankes Pipeline: Redis-Cache vorne, robuster Client für Aufrufe, die sowieso durchkommen. Die Ingestion nimmt den Batch-Weg, die Suche cache sogar die Abfrage-Embedding – und dieser letzte Punkt ist nicht trivial, weil Benutzerabfragen sich viel öfter wiederholen, als man denkt.

class ProductionPipeline:
    """RAG-Pipeline, optimiert für die Produktion."""

    def __init__(self):
        self.cache = EmbeddingCacheRedis()
        self.robust_client = RobustEmbeddingClient()

    def ingest(self, documents: list[dict]):
        """Ingestion mit Cache und Retry."""
        texts = [d["text"] for d in documents]
        embeddings = self.cache.get_embeddings_batch(texts)
        # Speichern in Vektordatenbank...

    def search(self, query: str, k: int = 5):
        """Suche mit Abfrage-Caching."""
        query_emb = self.cache.get_embedding(query)
        # Suche in Vektordatenbank...
        return query_emb

Produktions-Checkliste

Zwei Zeilen in dieser Tabelle bringen keine Leistungsgewinn, sind aber die wichtigsten. Die Überwachung der Cache-Trefferquote ist Ihr Indikator für die wirtschaftliche Gesundheit: Wenn sie zusammenbricht, verdreifacht sich Ihre Rechnung, bevor jemand merkt. Und das Sicherung von Embeddings rettet Sie am Tag eines Vorfalls davor, das volle Kodierung des Korpus erneut bezahlen und warten zu müssen.

Zusammenfassung

• Redis-Cache reduziert Kosten und Latenz um 90%+
• Exponentieller Retry behandelt OpenAI-Ratenlimits
• Asynchrone Verarbeitung beschleunigt die Ingestion großer Korpora
• int8-Quantisierung reduziert Speicher um das 4-fache
• Dimensionsreduktion an der API ist der einfachste Hebel

Testen Sie Ihr Wissen

Zwölf Lektionen über Embeddings und RAG: Das Wesentliche passt in fünf Fragen.

Der Faden des Kurses: Bedeutung wird messbar – Suche, Ranking, Empfehlung sind nur Anwendungen. Produktion (Skalierung, Caching) erledigt den Rest.