Generieren Sie Embeddings für Ihre Daten
Auf dieser Seite
Lernziele
- Bereiten Sie Ihre Textdaten für die Erstellung von Embeddings vor
- Generieren Sie Embeddings effizient im großen Maßstab
- Speichern Sie Vektoren für die spätere Nutzung
Daten vorbereiten
Die Qualität Ihrer Embeddings hängt direkt von der Qualität Ihrer Daten ab. Text, der mit zufälligen Leerzeichen, Steuerzeichen aus einer falsch konfigurierten Exportdatei oder unregelmäßigen Zeilenumbrüchen behaftet ist, erzeugt einen leicht verzerrten Vektor. Diese Verzerrung summiert sich bei jeder Suche. Die Bereinigung ist kein kosmetischer Akt: Sie ist die erste und günstigste Optimierung der Relevanz.
import re
def nettoyer_texte(texte: str) -> str:
"""Bereitet Text für die Erstellung von Embeddings vor."""
# Entfernt mehrere Leerzeichen
texte = re.sub(r"\s+", " ", texte)
# Entfernt Steuerzeichen
texte = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", texte)
# Schneiden
texte = texte.strip()
return texte
# Beispiel
brut = " Texte avec des espaces\n\n\nen trop "
propre = nettoyer_texte(brut)
print(propre) # "Texte avec des espaces en trop"Kontext anreichern
Ein isolierter Ausschnitt aus einer Dokumentation verliert ohne seinen Titel einen Teil seiner Bedeutung. Nehmen Sie einen Absatz, der mit “Laden Sie das Paket von der offiziellen Website herunter” beginnt: Ohne Kontext deutet nichts darauf hin, dass es sich um PostgreSQL handelt, und der resultierende Vektor wird niemals mit einer Abfrage zu PostgreSQL übereinstimmen. Indem Sie den Inhalt mit seinem Titel und seiner Sektion voranstellen, injizieren Sie diese Informationen zurück in den Text, den das Modell sieht.
def enrichir_document(titre: str, section: str, contenu: str) -> str:
"""Erstellt angereicherten Text für bessere Embeddings."""
return f"Titre : {titre}\nSection : {section}\n\n{contenu}"
texte_enrichi = enrichir_document(
titre="Installation de PostgreSQL",
section="Getting Started Guide",
contenu="Download the package from the official site..."
)Embeddings in Batches generieren
Bei einem echten Korpus aus Hunderten oder Tausenden von Dokumenten muss die Verarbeitung in Blöcken erfolgen und gegen Netzwerkprobleme robust sein. Die folgende Funktion iteriert durch Texte in Slices, sortiert die Antwort systematisch nach index, um sicherzustellen, dass Vektor #7 dem Dokument #7 entspricht, und versucht es einmal erneut nach einer Pause, wenn ein Fehler auftritt.
from openai import OpenAI
import time
import json
client = OpenAI()
def generer_embeddings(
textes: list[str],
model: str = "text-embedding-3-large",
batch_size: int = 100
) -> list[list[float]]:
"""Generiert Embeddings in Batches mit Fehlerbehandlung."""
tous_les_embeddings = []
for i in range(0, len(textes), batch_size):
batch = textes[i:i + batch_size]
print(f"Verarbeite Batch {i // batch_size + 1}/"
f"{(len(textes) - 1) // batch_size + 1}")
try:
response = client.embeddings.create(
input=batch,
model=model
)
# Nach Index sortieren, um die Reihenfolge zu garantieren
batch_embeddings = sorted(
response.data, key=lambda x: x.index
)
tous_les_embeddings.extend(
[item.embedding for item in batch_embeddings]
)
except Exception as e:
print(f"Fehler bei Batch {i}: {e}")
# Erneuter Versuch nach einer Pause
time.sleep(5)
response = client.embeddings.create(
input=batch,
model=model
)
batch_embeddings = sorted(
response.data, key=lambda x: x.index
)
tous_les_embeddings.extend(
[item.embedding for item in batch_embeddings]
)
return tous_les_embeddingsDas Sortieren nach index ist eine Vorsichtsmaßnahme, die Sie niemals überspringen dürfen. Eine stillschweigende Verschiebung einer Zeile reicht aus, um jedes Dokument mit dem Vektor seines Nachbarn zu verknüpfen. Das Symptom, das Sie beobachten werden, ist “Suche liefert Unsinn”, ohne technischen Fehler, der Sie in die richtige Richtung weist.
Embeddings speichern
Für ein kleines Korpus aus wenigen Tausend Dokumenten funktioniert eine einfache JSON-Datei, die Texte, Metadaten und Vektoren enthält, perfekt. Sie hat den Vorteil, dass sie beim Debuggen eines seltsamen Ergebnisses für den Menschen lesbar ist.
import json
documents = [
{"id": 1, "texte": "Premier document...", "metadata": {"source": "faq"}},
{"id": 2, "texte": "Deuxième document...", "metadata": {"source": "docs"}},
]
# Embeddings generieren
textes = [d["texte"] for d in documents]
embeddings = generer_embeddings(textes)
# Vektoren zu Dokumenten hinzufügen
for doc, emb in zip(documents, embeddings):
doc["embedding"] = emb
# Speichern
with open("corpus_embeddings.json", "w") as f:
json.dump(documents, f)Jenseits dieser Größe wird JSON zum Flaschenhals: Jede Zahl wird vollständig ausgegeben, was die Datei aufbläht und bei jedem Laden teures Parsing auferlegt. NumPy speichert die gleichen Werte in Binärformat auf 4 Bytes und lädt sie sofort neu.
import numpy as np
# In NumPy-Matrix umwandeln
matrice = np.array(embeddings, dtype=np.float32)
print(f"Shape : {matrice.shape}") # (n_docs, 3072)
# Im Binärformat speichern
np.save("embeddings.npy", matrice)
# Laden
matrice_chargee = np.load("embeddings.npy")Komplette Pipeline: Von Textdateien zu Embeddings
Bringen wir diese Elemente in einer Funktion zusammen, die einen Ordner liest, jede Datei bereinigt, Vektoren generiert und die Speicherung in zwei Teile aufteilt: Metadaten in JSON (lesbar und bearbeitbar) und die Matrix in NumPy (kompakt und schnell).
from openai import OpenAI
from pathlib import Path
import json
import numpy as np
client = OpenAI()
def pipeline_embeddings(dossier: str, model: str = "text-embedding-3-large"):
"""Komplette Pipeline: Dateien → Embeddings → Speicherung."""
dossier_path = Path(dossier)
# 1. Dateien lesen
documents = []
for fichier in sorted(dossier_path.glob("*.txt")):
contenu = fichier.read_text(encoding="utf-8")
contenu = nettoyer_texte(contenu)
if contenu:
documents.append({
"id": fichier.stem,
"fichier": fichier.name,
"texte": contenu
})
print(f"{len(documents)} Dokumente gefunden")
# 2. Embeddings generieren
textes = [d["texte"] for d in documents]
embeddings = generer_embeddings(textes, model=model)
# 3. Speichern
# Metadaten in JSON
meta = [{"id": d["id"], "fichier": d["fichier"]} for d in documents]
with open("metadata.json", "w") as f:
json.dump(meta, f, ensure_ascii=False, indent=2)
# Vektoren in NumPy
np.save("embeddings.npy", np.array(embeddings, dtype=np.float32))
print(f"Embeddings gespeichert: {len(embeddings)} Vektoren "
f"mit {len(embeddings[0])} Dimensionen")
return documents, embeddings
# Nutzung
documents, embeddings = pipeline_embeddings("./mes_documents/")Beachten Sie das sorted() beim Glob: Ohne diese Funktion hängt die Dateireihenfolge vom Dateisystem ab, und Ihre beiden Ausgabedateien riskieren, nach einer einfachen Maschinenänderung aus der Synchronisation zu geraten.
Kosten schätzen
Bevor Sie eine massive Aufgabe starten, zählen Sie lokal die Token mit tiktoken. Die Operation dauert nur wenige Sekunden und spart Ihnen die Entdeckung, dass ein schlecht gefiltertes Korpus zehnmal mehr Volumen enthält als erwartet.
import tiktoken
encoder = tiktoken.get_encoding("cl100k_base")
def estimer_cout(textes: list[str], model: str = "text-embedding-3-large"):
"""Schätzt die Embedding-Kosten für eine Liste von Texten."""
total_tokens = sum(len(encoder.encode(t)) for t in textes)
# Näherungsweise Preise pro Million Token (aktuelle Raten prüfen)
prix = {
"text-embedding-3-large": 0.13,
"text-embedding-3-small": 0.02,
}
cout = total_tokens / 1_000_000 * prix.get(model, 0.13)
print(f"Geschätzte Token: {total_tokens:,}")
print(f"Geschätzte Kosten: ${cout:.4f}")
return total_tokens, coutDie Preise im Wörterbuch sind anzeigend und alternieren schnell: Prüfen Sie immer die aktuellen Raten, bevor Sie ein Angebot an einen Kunden unterbreiten.
Zusammenfassung
- Bereinigen und anreichern Sie Ihre Texte, bevor Sie Embeddings generieren
- Verwenden Sie Batch-Verarbeitung für große Korpora
- Speichern Sie Metadaten in JSON und Vektoren in NumPy
- Schätzen Sie Ihre Kosten, bevor Sie eine massive Aufgabe starten