Indexierung: Chunks einbetten und speichern
Sie betten jeden Chunk mithilfe der OpenAI Embeddings API ein und führen ein Upsert der resultierenden Vektoren mit Metadaten in einen Vektorspeicher durch, um einen durchsuchbaren Dokumentindex aufzubauen.
Indexierung: Chunks einbetten und speichern ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 3 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Die Indexierungsphase: Überblick
Nach dem Laden und Aufteilen Ihrer Dokumente erreichen Sie die Indexierungsphase: Sie wandeln Text-Chunks in Vektor-Embeddings um und speichern sie in einer durchsuchbaren Vektordatenbank. Dies ist der letzte Offline-Schritt, bevor Anfragen beantwortet werden können. Die Qualität Ihrer Embeddings sowie die Effizienz Ihrer Speicher- und Indexierungsstrategie bestimmen direkt, wie schnell und genau Ihr RAG-System zur Abfragezeit arbeitet.
Embeddings über die OpenAI API erzeugen
Der gängigste Ansatz besteht darin, die Embeddings API von OpenAI mit dem Text Ihres Chunks aufzurufen. Das Modell text-embedding-3-small erzeugt 1536-dimensionale Vektoren und kostet 0,02 $ pro einer Million Token – für die meisten Workloads äußerst günstig. Senden Sie mehrere Texte in einem einzigen API-Aufruf (bis zu 2048 Eingaben), um den Durchsatz zu maximieren. Die Antwort enthält für jeden Eingabetext einen Embedding-Vektor in derselben Reihenfolge.
from openai import OpenAI
client = OpenAI()
def embed_batch(texts, model='text-embedding-3-small'):
response = client.embeddings.create(
model=model,
input=texts # up to 2048 texts per call
)
return [item.embedding for item in response.data]
# Embed one batch of 100 chunk texts
texts = [chunk['text'] for chunk in chunks[:100]]
vectors = embed_batch(texts)
print(f'Embedding dimension: {len(vectors[0])}')
print(f'Embedded {len(vectors)} chunks')Batch-Verarbeitung für mehr Effizienz
Beim Indexieren von Tausenden von Chunks ist Effizienz entscheidend. Verarbeiten Sie Chunks in Stapeln von 100–500, um einen ausgewogenen Kompromiss zwischen Durchsatz und Speicherverbrauch zu erreichen. Speichern Sie Ihre Position, damit Sie nach einem Fehler fortsetzen können, ohne bereits verarbeitete Chunks erneut zu embedden. Protokollieren Sie regelmäßig den Fortschritt. Bei 100.000 Chunks und 500 Chunks pro Stapel führen Sie 200 API-Aufrufe aus – das ist typischerweise in wenigen Minuten abgeschlossen.
def embed_all_chunks(chunks, batch_size=200):
embedded = []
total = len(chunks)
for i in range(0, total, batch_size):
batch = chunks[i:i+batch_size]
texts = [c['text'] for c in batch]
vectors = embed_batch(texts)
for chunk, vector in zip(batch, vectors):
embedded.append({
**chunk,
'embedding': vector
})
if (i // batch_size) % 10 == 0:
print(f'Progress: {min(i+batch_size, total)}/{total}')
return embeddedUmgang mit Ratenbegrenzungen beim Indexieren
Die OpenAI Embeddings API hat Ratenbegrenzungen, die in Token pro Minute (TPM) gemessen werden. Große Indexierungsaufträge stoßen an diese Grenzen und erhalten den Fehler RateLimitError. Implementieren Sie exponentielles Backoff mit Jitter: Wenn ein Ratenbegrenzungsfehler auftritt, warten Sie vor dem erneuten Versuch ein kurzes zufälliges Intervall und verdoppeln Sie die Wartezeit bei jedem weiteren Fehler. Dadurch werden erneute Versuche zeitlich verteilt und es wird verhindert, dass alle parallelen Worker gleichzeitig die API überlasten.
import time
import random
from openai import RateLimitError
def embed_batch_with_retry(texts, max_retries=5):
for attempt in range(max_retries):
try:
return embed_batch(texts)
except RateLimitError:
if attempt == max_retries - 1:
raise
wait = (2 ** attempt) + random.uniform(0, 1)
print(f'Rate limited. Waiting {wait:.1f}s...')
time.sleep(wait)
return []Vektoren in Pinecone upserten
Nachdem Sie die Embeddings erzeugt haben, upserten Sie sie in den Vektorspeicher. Upserten bedeutet, neue Vektoren einzufügen oder vorhandene zu aktualisieren, wenn dieselbe ID bereits existiert – der Vorgang ist von vornherein idempotent. In Pinecone enthält jeder upsertete Datensatz die Vektor-ID, die Embedding-Werte und ein Metadatenverzeichnis mit Feldern, nach denen Sie später filtern oder die Sie anzeigen möchten. Upserten Sie für optimalen Durchsatz Stapel mit bis zu 100 Datensätzen pro Aufruf.
import pinecone
pc = pinecone.Pinecone(api_key='YOUR_KEY')
index = pc.Index('rag-index')
def upsert_to_pinecone(embedded_chunks, batch_size=100):
for i in range(0, len(embedded_chunks), batch_size):
batch = embedded_chunks[i:i+batch_size]
vectors = [
(
chunk['id'],
chunk['embedding'],
{
'text': chunk['text'],
'source': chunk['metadata']['source'],
'page': chunk['metadata'].get('page', 0)
}
)
for chunk in batch
]
index.upsert(vectors=vectors)
print(f'Upserted {min(i+batch_size, len(embedded_chunks))}/{len(embedded_chunks)}')Speichern in pgvector
Mit pgvector fügen Sie Embeddings direkt mithilfe von standardmäßigem SQL in eine PostgreSQL-Tabelle ein. Der Datentyp vector akzeptiert eine als Zeichenfolge serialisierte Python-Liste von Gleitkommazahlen. Erstellen Sie nach dem Einfügen aller Zeilen einen HNSW-Index für schnelle Abfragen der approximativen nächsten Nachbarn. Das Indexieren einer bestehenden Tabelle mit Millionen von Zeilen kann mehrere Minuten dauern. Erstellen Sie den Index daher nach dem Masseneinfügen und nicht vorher.
import psycopg2
from psycopg2.extras import execute_values
def upsert_to_pgvector(conn, embedded_chunks):
with conn.cursor() as cur:
records = [
(
chunk['id'],
chunk['text'],
chunk['metadata']['source'],
chunk['metadata'].get('page', 0),
chunk['embedding'] # list of floats
)
for chunk in embedded_chunks
]
execute_values(cur, '''
INSERT INTO document_chunks (id, text, source, page, embedding)
VALUES %s
ON CONFLICT (id) DO UPDATE
SET text = EXCLUDED.text, embedding = EXCLUDED.embedding
''', records)
conn.commit()Den HNSW-Index erstellen
HNSW (Hierarchical Navigable Small World) ist der Indextyp, der eine schnelle Suche nach approximativen nächsten Nachbarn ermöglicht. Anders als bei der Brute-Force-Suche, bei der der Abfragevektor mit jedem gespeicherten Vektor verglichen wird, erstellt HNSW eine mehrschichtige Graphstruktur, die den Suchraum verkleinert. Der Parameter m legt fest, wie viele Verbindungen jeder Knoten hat (höher = besserer Recall, aber mehr Speicherverbrauch), während ef_construction die Indexqualität während der Erstellung steuert.
-- Build HNSW index after bulk insertion
CREATE INDEX CONCURRENTLY ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Set ef_search at query time to trade recall vs speed
SET hnsw.ef_search = 100;
-- Verify index was created
SELECT indexname, indexdef
FROM pg_indexes
WHERE tablename = 'document_chunks';Metadatenschema entwerfen
Metadaten, die zusammen mit jedem Vektor gespeichert werden, ermöglichen eine leistungsfähige gefilterte Suche. Entwerfen Sie Ihr Metadatenschema vor dem Indexieren – das spätere Hinzufügen neuer Felder erfordert eine erneute Indexierung. Nehmen Sie Felder auf, nach denen Sie filtern werden (Abteilung, Dokumenttyp, Datumsbereich), Felder, die Sie in Quellenangaben anzeigen werden (Titel, Seite, Autor), sowie Felder, die beim Debugging nützlich sind (chunk_index, total_chunks, indexed_at). Halten Sie Metadatenwerte einfach: Zeichenfolgen, Zahlen und boolesche Werte lassen sich effizient indexieren und filtern, verschachtelte Objekte dagegen nicht.
# Well-designed metadata schema
METADATA_SCHEMA = {
# For filtering at retrieval time
'department': 'HR', # string
'doc_type': 'policy', # string
'year': 2025, # integer
'is_active': True, # boolean
# For display in citations
'title': 'Employee Handbook 2025',
'author': 'HR Team',
'page': 12,
'source': 's3://docs/handbook_2025.pdf',
# For debugging and updates
'chunk_index': 3,
'total_chunks': 24,
'indexed_at': '2025-09-01T10:00:00Z'
}Lange Indexierungsaufträge mit Checkpoints absichern
Das Indexieren eines großen Korpus kann Stunden dauern. Ein Absturz in der Mitte macht den gesamten Fortschritt zunichte. Implementieren Sie eine Checkpoint-Datei, in der festgehalten wird, welche Chunks erfolgreich indexiert wurden. Überspringen Sie beim Neustart bereits indexierte Chunks und setzen Sie die Verarbeitung an der unterbrochenen Stelle fort. Dadurch wird der Indexierungsauftrag idempotent und kann sicher fortgesetzt werden. Speichern Sie den Checkpoint als Menge der verarbeiteten Chunk-IDs in einer JSON-Datei oder einer Datenbanktabelle.
import json
from pathlib import Path
CHECKPOINT_FILE = '/tmp/index_checkpoint.json'
def load_checkpoint():
if Path(CHECKPOINT_FILE).exists():
return set(json.loads(Path(CHECKPOINT_FILE).read_text()))
return set()
def save_checkpoint(indexed_ids):
Path(CHECKPOINT_FILE).write_text(json.dumps(list(indexed_ids)))
def index_with_checkpoint(chunks, index):
done = load_checkpoint()
remaining = [c for c in chunks if c['id'] not in done]
print(f'Resuming: {len(done)} done, {len(remaining)} remaining')
for chunk in remaining:
upsert_to_pinecone([chunk], index)
done.add(chunk['id'])
save_checkpoint(done)Vollständigkeit des Index überprüfen
Überprüfen Sie nach dem Indexieren, ob alle Chunks im Vektorspeicher angekommen sind. Vergleichen Sie die Anzahl der von Ihrem Splitter erzeugten Chunks mit der vom Index gemeldeten Vektoranzahl. Fragen Sie den Index mit dem Text eines bekannten Dokuments ab und bestätigen Sie, dass das erwartete Ergebnis unter den Top 5 erscheint. Führen Sie einige bekannte Abfragen aus Ihrem Golden-Testdatensatz aus und prüfen Sie, ob die Präzision dem erwarteten Niveau entspricht. Gehen Sie niemals von einem vollständigen Index aus, ohne dies zu überprüfen.
def verify_index(index, chunks, sample_size=10):
index_stats = index.describe_index_stats()
total_vectors = index_stats.total_vector_count
expected = len(chunks)
print(f'Index vectors: {total_vectors}, Expected: {expected}')
if total_vectors != expected:
print('WARNING: mismatch — some chunks may not have been indexed')
# Spot-check retrieval
import random
sample = random.sample(chunks, sample_size)
for chunk in sample:
vec = embed_batch([chunk['text']])[0]
results = index.query(vector=vec, top_k=1, include_metadata=True)
top_id = results.matches[0].id if results.matches else None
if top_id != chunk['id']:
print(f'WARNING: expected {chunk["id"]}, got {top_id}')Lokale Alternativen für Embeddings
Verwenden Sie für datenschutzsensible Daten, die Ihre Infrastruktur nicht verlassen dürfen, lokal gehostete Embedding-Modelle. Die Bibliothek sentence-transformers stellt hochwertige Modelle wie all-MiniLM-L6-v2 (384 Dimensionen, 22 MB, sehr schnell) und bge-large-en-v1.5 (1024 Dimensionen, bessere Qualität) bereit. Führen Sie sie bei moderaten Workloads auf der CPU oder bei großen Indexierungsaufträgen auf der GPU aus. Lokale Modelle vermeiden API-Kosten und den Abfluss von Daten, erfordern jedoch die Verwaltung von Modelldateien und Rechenressourcen.
from sentence_transformers import SentenceTransformer
# Load once at startup
model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
def embed_locally(texts, batch_size=64):
# encode() handles batching internally
embeddings = model.encode(
texts,
batch_size=batch_size,
show_progress_bar=True,
convert_to_numpy=True
)
return embeddings.tolist() # convert numpy array to Python list
vectors = embed_locally([c['text'] for c in chunks])Kurztest
Testen Sie Ihr Verständnis der Konzepte aus dem Bereich AI Engineering, die in dieser Lektion behandelt wurden.
Zusammenfassung der Lektion
In dieser Lektion haben Sie Folgendes gelernt: Embeddings stapelweise mit der OpenAI API erzeugen und Ratenbegrenzungen mit exponentiellem Backoff behandeln, Vektoren in Pinecone und pgvector upserten und mit Metadaten versehen, HNSW-Indizes erstellen für eine schnelle Suche nach approximativen nächsten Nachbarn sowie Best Practices für den Produktivbetrieb, darunter die Fortsetzung anhand von Checkpoints, das Entwerfen von Metadatenschemata und die Überprüfung der Indexvollständigkeit. Als Nächstes erstellen wir die Abfragepipeline, die Chunks abruft und fundierte Antworten generiert.
Häufig gestellte Fragen
Ist die Lektion „Indexierung: Chunks einbetten und speichern“ kostenlos?
Ja — der vollständige Text von „Indexierung: Chunks einbetten und speichern“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Indexierung: Chunks einbetten und speichern“?
Sie betten jeden Chunk mithilfe der OpenAI Embeddings API ein und führen ein Upsert der resultierenden Vektoren mit Metadaten in einen Vektorspeicher durch, um einen durchsuchbaren Dokumentindex aufz… Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Engineering Academy zu starten?
Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 3 von 4.
Wie lange dauert die Lektion „Indexierung: Chunks einbetten und speichern“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?
Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- Dokumente laden und Text extrahieren
- Chunking-Strategien: fest, satzbasiert oder rekursiv
- Indexierung: Chunks einbetten und speichern
- Abfragen, abrufen und generieren