Embeddings erstellen
Auf dieser Seite
Chunks in Vektoren umwandeln
Ihre Dokumente sind nun in saubere Chunks unterteilt. Der nächste Schritt besteht darin, diese mithilfe der mistral-embed-API in Embeddings umzuwandeln und diese Vektoren in einem Vektorspeicher zu speichern, der als Ihre Suchmaschine fungieren wird. In diesem Moment hört Ihr Korpus auf, reiner Text zu sein, und wird zu abfragbarer Geometrie. Ein paar technische Vorsichtsmaßnahmen – Batch-Verarbeitung, Wiederholungsversuche und Persistenz – entscheiden zwischen einem fragilen Prototypen und einem zuverlässigen Index.
Bei einem kleinen Korpus, sagen wir weniger als hundert Chunks, reicht ein einzelner direkter Aufruf. Extrahieren Sie den Text aus Ihren Chunk-Dictionaries, übergeben Sie diesen unverändert an die API und rufen Sie die vollständige Matrix ab. Die Form (87, 1024), die am Ende der Ausführung angezeigt wird, ist Ihr erster Plausibilitätscheck: so viele Zeilen wie Chunks und 1024 Spalten, da dies die Dimension von mistral-embed ist. Wenn die Zeilenanzahl nicht übereinstimmt, ist ein leerer Chunk in den Batch geschlüpft.
import numpy as np
from mistralai.client import Mistral
client = Mistral(api_key=api_key)
# Angenommen, chunks ist die Liste unserer vorbereiteten Chunks
chunk_texts = [c["text"] for c in chunks]
# Kodierung in einem einzigen Aufruf
response = client.embeddings.create(
model="mistral-embed",
inputs=chunk_texts,
)
embeddings = np.array([d.embedding for d in response.data])
print(f"Embeddings: {embeddings.shape}")
# Beispiel: Embeddings: (87, 1024)Batch-Kodierung
Bei mehr als ein paar Dutzend Chunks stoßen Sie mit einem einzigen Aufruf an die API-Grenzen. Die Lösung besteht darin, die Liste in Batches aufzuteilen und die Ergebnisse zu verknüpfen. Die folgende Funktion verarbeitet Sätze von 25 und zeigt den Fortschritt an. Dies ist kein kosmetisches Detail: Bei einem Korpus mit mehreren tausend Chunks dauert die Kodierung mehrere Minuten, und Sie möchten wissen, wo der Prozess steht, anstatt auf einen stummen Terminalbildschirm zu starren. Die Reihenfolge bleibt erhalten, sodass jeder Vektor weiterhin mit dem entsprechenden Chunk in Ihrer ursprünglichen Liste übereinstimmt.
def encode_chunks(chunks, batch_size=25, model="mistral-embed"):
"""Kodiert eine Liste von Chunks in Batches."""
texts = [c["text"] for c in chunks]
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
print(f" Kodiere Batch {i // batch_size + 1}/{(len(texts) - 1) // batch_size + 1} "
f"({len(batch)} Chunks)...")
response = client.embeddings.create(
model=model,
inputs=batch,
)
all_embeddings.extend([d.embedding for d in response.data])
return np.array(all_embeddings)
# Alle Chunks kodieren
embeddings = encode_chunks(chunks, batch_size=25)
print(f"Total: {embeddings.shape[0]} Vektoren mit {embeddings.shape[1]} Dimensionen")Überleben von Netzwerkfehlern
In der Produktion wird ein API-Aufruf irgendwann fehlschlagen: Netzwerkausfall, Erreichung des Rate-Limits oder vorübergehende Ausfallzeit. Ohne Schutz bleibt Ihre Indexierung bei Batch 180 von 400 stecken, und Sie sind wieder beim Nullpunkt. Die robuste Version umgibt jeden Batch in einer Wiederholungs-Schleife mit exponentiellem Backoff – eine Sekunde, dann zwei, dann vier. Diese zunehmende Verzögerung ist entscheidend: Ein sofortiges Neustarten nach einer Quota-Abweisung verschlimmert die Situation nur. Nach drei aufeinanderfolgenden Fehlern wirft die Funktion eine explizite Ausnahme, die den fehlerhaften Batch benennt, anstatt eine abgeschnittene Matrix zurückzugeben, bei der Sie die Lücke erst während der Suche entdecken würden.
import time
def encode_chunks_robust(chunks, batch_size=25, max_retries=3):
"""Robuste Kodierung mit Fehlerbehandlung."""
texts = [c["text"] for c in chunks]
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
for attempt in range(max_retries):
try:
response = client.embeddings.create(
model="mistral-embed",
inputs=batch,
)
all_embeddings.extend([d.embedding for d in response.data])
break # Erfolg, Wiederholungsschleife verlassen
except Exception as e:
wait_time = 2 ** attempt # Exponentieller Backoff: 1s, 2s, 4s
print(f" Fehler in Batch {i // batch_size + 1}, "
f"Versuch {attempt + 1}/{max_retries}: {e}")
if attempt < max_retries - 1:
print(f" Warte {wait_time}s vor dem Neustart...")
time.sleep(wait_time)
else:
raise RuntimeError(
f"Nach {max_retries} Versuchen fehlgeschlagen für Batch {i // batch_size + 1}"
)
return np.array(all_embeddings)Index auf Festplatte speichern
Jede Kodierungslauf kostet Zeit und abgerechnete API-Aufrufe. Die Neuberechnung von Vektoren bei jedem Start Ihrer Anwendung ist reine Verschwendung, insbesondere da sich das Korpus selbst nicht geändert hat. Das Speichern erfolgt in zwei komplementären Dateien: Vektoren werden in NumPy gespeichert, ein kompaktes Binärformat, das sofort lädt, während Metadaten – Text, Quelle, Position – in lesbares JSON gespeichert werden, das Sie öffnen können, um einen verdächtigen Chunk zu inspizieren. Die beiden Dateien sind voneinander abhängig und durch die Zeilenreihenfolge verknüpft: Generieren Sie niemals eine ohne die andere neu, sonst zeigen Ihre Suchergebnisse auf die falschen Texte.
import json
def save_index(chunks, embeddings, filepath="rag_index"):
"""Speichert Chunks und Embeddings."""
# Speichert Embeddings (NumPy)
np.save(f"{filepath}_embeddings.npy", embeddings)
# Speichert Metadaten (JSON)
metadata = []
for c in chunks:
metadata.append({
"text": c["text"],
"source": c.get("source", ""),
"chunk_index": c.get("chunk_index", 0),
})
with open(f"{filepath}_metadata.json", "w", encoding="utf-8") as f:
json.dump(metadata, f, ensure_ascii=False, indent=2)
print(f"Index gespeichert: {len(chunks)} Chunks, {embeddings.shape}")
def load_index(filepath="rag_index"):
"""Lädt Chunks und Embeddings von der Festplatte."""
embeddings = np.load(f"{filepath}_embeddings.npy")
with open(f"{filepath}_metadata.json", "r", encoding="utf-8") as f:
metadata = json.load(f)
print(f"Index geladen: {len(metadata)} Chunks, {embeddings.shape}")
return metadata, embeddings
# Speichern
save_index(chunks, embeddings)
# Laden (beim nächsten Start)
chunks_loaded, embeddings_loaded = load_index()Verifizieren vor Weiterverarbeitung
Ein still korrupter Index zeigt sich spät, meist wenn ein Benutzer eine absurde Antwort meldet. Ein paar Offline-Checks dauern dreißig Sekunden und sparen Ihnen diese Entdeckung. Die Verifizierungsfunktion zeigt zunächst grundlegende Metriken an – Anzahl der Chunks, Dimension, durchschnittliche Norm – und misst dann die Ähnlichkeit zwischen benachbarten Chunks: Da sie aus demselben Dokument stammen, sollten sie deutlich höhere Werte aufweisen als zwei zufällige Passagen. Der letzte Test ist der einfachste und aufschlussreichste: Ein Chunk im Vergleich zu sich selbst muss genau 1,0 ergeben. Jeder andere Wert signalisiert ein Ausrichtungsproblem oder ein numerisches Konversionsproblem.
from sklearn.metrics.pairwise import cosine_similarity
def verify_embeddings(chunks, embeddings, n_tests=5):
"""Verifiziert Embedding-Qualität mit einfachen Tests."""
print("=== Verifiziere Embeddings ===")
print(f"Anzahl der Chunks: {len(chunks)}")
print(f"Dimensionen: {embeddings.shape[1]}")
print(f"Durchschnittliche Norm: {np.linalg.norm(embeddings, axis=1).mean():.4f}")
# Test: Chunks desselben Dokuments sollten näher beieinander liegen
for i in range(min(n_tests, len(chunks) - 1)):
sim = cosine_similarity([embeddings[i]], [embeddings[i + 1]])[0][0]
print(f" Ähnlichkeit Chunk {i} <-> {i+1}: {sim:.4f}")
# Test: Ein Chunk im Vergleich zu sich selbst = 1,0
self_sim = cosine_similarity([embeddings[0]], [embeddings[0]])[0][0]
print(f" Selbstähnlichkeit (muss 1,0 sein): {self_sim:.6f}")
verify_embeddings(chunks, embeddings)Zusammengenommen lässt sich der Arbeitsablauf auf vier nützliche Zeilen reduzieren. Sie bereiten Chunks mit dem Pipeline aus der vorherigen Lektion vor, kodieren im robusten Modus, verifizieren und speichern. Dies ist das Skript, das Sie jedes Mal neu ausführen, wenn sich das Korpus aktualisiert.
# 1. Chunks vorbereiten (vorherige Lektion)
chunks = prepare_documents(raw_docs)
# 2. Kodieren
embeddings = encode_chunks_robust(chunks, batch_size=25)
# 3. Verifizieren
verify_embeddings(chunks, embeddings)
# 4. Speichern
save_index(chunks, embeddings)
print("Index bereit für die Suche!")Kernaussagen
- Nutzen Sie Batch-Verarbeitung zur Kodierung großer Korpora (Batches von 25–50 Chunks)
- Fügen Sie Wiederholungslogik mit exponentiellem Backoff für Robustheit hinzu
- Speichern Sie Embeddings auf der Festplatte (NumPy), um Neuberechnungen zu vermeiden
- Halten Sie Metadaten (Text, Quelle, Position) in einer separaten JSON-Datei
- Verifizieren Sie die Embedding-Qualität, bevor Sie sie in der Produktion verwenden