Vektordatenbank
Im Zeitalter der KI hat sich die Form der Daten, die wir verarbeiten, grundlegend verändert.
Traditionelle Datenbanken speichern strukturierte Daten: Zahlen, Zeichenfolgen, Datumsangaben.
KI-Anwendungen verarbeiten Semantik: die Bedeutung eines Textstücks, den Inhalt eines Bildes, die Bedeutung eines Sprachclips. Diese semantischen Informationen werden als Vektoren dargestellt – eine Sequenz von Gleitkommazahlen, z. B. [0.123, -0.456, 0.789, ...].
Eine Vektordatenbank ist eine Datenbank, die speziell für das Speichern, Indizieren und Abfragen dieser Vektoren entwickelt wurde. Ohne sie kann RAG (Retrieval-Augmented Generation) nicht funktionieren, die semantische Suche kann nicht implementiert werden und Empfehlungssysteme können nicht effizient ausgeführt werden.
Kernwert von Vektordatenbanken:Finden Sie innerhalb von Millisekunden aus Millionen oder sogar Milliarden von Vektoren die Top-K-Ergebnisse, die einem Abfragevektor am ähnlichsten sind.。
Herausforderungen der Vektor-Ähnlichkeitssuche
Nur wenn man die Natur des Problems versteht, kann man nachvollziehen, warum spezielle Algorithmen erforderlich sind.
Komplexitätsproblem der Brute-Force-Suche
Der einfachste Ansatz ist die Brute-Force-Suche: Berechnen Sie die Ähnlichkeit zwischen dem Abfragevektor und jedem Vektor in der Datenbank, sortieren Sie sie und nehmen Sie die Top K.
Beispiel
# Brute-force Search Demo: Simple but Inefficient
# ============================================
import numpy as np
from typing import List, Tuple
def l2_distance(v1: np.ndarray, v2: np.ndarray) -> float:
"""Compute L2 Euclidean distance: the smaller, the more similar."""
return np.sqrt(np.sum((v1 - v2) ** 2))
def brute_force_search(
query: np.ndarray,
vectors: List[np.ndarray],
top_k: int = 5
) -> List[Tuple[int, float]]:
"""
Brute-force search: compute distances to all vectors, sort, and return Top K
Advantages: simple and accurate, 100% recall
Disadvantages: extremely slow with large data volumes
"""
# Compute distance to each vector
distances = []
for idx, vec in enumerate(vectors):
dist = l2_distance(query, vec)
distances.append((idx, dist))
# Sort by distance, take top K
distances.sort(key=lambda x: x[1])
return distances[:top_k]
# Generate 10,000 random 128-dimensional vectors (simulating a vector database)
np.random.seed(42)
vector_count = 10000
dimension = 128
vectors = [np.random.randn(dimension) for _ in range(vector_count)]
# Generate a query vector
query_vector = np.random.randn(dimension)
# Execute brute-force search
print(f"Searching Top 5 among {vector_count} vectors...")
results = brute_force_search(query_vector, vectors, top_k=5)
print("Search results:")
for idx, dist in results:
print(f" Vector {idx:4d}, distance = {dist:.4f}")
# Output looks like:
# Searching Top 5 among 10,000 vectors...
# Search results:
# Vector 8236, distance = 13.3986
# Vector 8946, distance = 13.5604
# Vector 9891, distance = 13.5688
# Vector 5172, distance = 13.5987
# Vector 5223, distance = 13.6081
Das Problem bei der Brute-Force-Suche liegt auf der Hand: Bei jeder Abfrage müssen alle Vektoren durchlaufen werden.
Wenn es 1 Million Vektoren gibt, erfordert jede Abfrage die Berechnung von 1 Million Distanzen.
Wenn es 1 Milliarde Vektoren gibt, erfordert jede Abfrage 1 Milliarde Berechnungen – das ist in der Produktion völlig inakzeptabel.
Der „Fluch der Dimensionalität“ im hocheingeschränkten Raum
Um die Sache noch schlimmer zu machen, sind Vektoren in der Regel hochdimensional.
text-embedding-3-small von OpenAI hat 1536 Dimensionen und large hat 3072 Dimensionen.
In hochdimensionalen Räumen kommt es bei traditionellen räumlichen Indizierungsmethoden (wie KD-Tree) zu einem drastischen Effizienzeinbruch, der manchmal sogar langsamer ist als die Brute-Force-Suche.
| Datenmenge | Suchzeit für Brute-Force-Suche (geschätzt) | Akzeptabel? |
|---|---|---|
| 10 Tausend | ~1 Millisekunde | Akzeptabel |
| 1 Million | ~100 Millisekunden | Etwas langsam |
| 10 Millionen | ~1 Sekunde | Zu langsam |
| 100 Millionen | ~10 Sekunden | Völlig inakzeptabel |
| 1 Milliarde | ~100 Sekunden | Unbrauchbar |
Aus diesem Grund benötigen wirAlgorithmen für die ungefähre nächste Nachbarsuche (ANN): Opferung von 100% Genauigkeit für eine 100-fache oder sogar 1000-fache Beschleunigung.
Algorithmen für die ungefähre nächste Nachbarsuche (ANN)
Die Kernidee von ANN-Algorithmen:Erstellung einer Indexstruktur, sodass Abfragen nicht alle Vektoren durchlaufen müssen。
Es gibt drei gängige Hauptalgorithmen: HNSW, IVF und PQ.
HNSW: Hierarchical Navigable Small World Graph (Hierarchischer navigierbarer Small-World-Graph)
HNSW (Hierarchical Navigable Small World) ist heute einer der beliebtesten Algorithmen. Er ist inspiriert von der Theorie der sechs Grade der Trennung – jeder Mensch auf der Welt kann über höchstens 6 Zwischenstationen verbunden werden.
HNSW verwendet einen mehrschichtigen, spärlichen Small-World-Graphen. Es lokalisiert schnell und grob ausgehend von den spärlichen Knoten der obersten Schicht und steigt dann Schicht für Schicht zur untersten Volladatenschicht für eine feingranulare Suche ab. Durch die Verwendung hierarchischer Sprünge wird die Anzahl der Distanzberechnungen stark reduziert, wodurch eine schnelle Suche nach ungefähren nächsten Nachbarn mit hoher Wiederfindungsrate (Recall) erreicht wird.
Hierarchische Struktur:Die unterste Schicht, Schicht 0, speichert alle Vektoren; je höher die Schicht, desto weniger Knoten fungieren als Überholspur. Derselbe Vektor ist vertraulich über Schichten hinweg mit gestrichelten Linien verknüpft.
Suchprozess (Pfad der blauen Linie):
- Die grobe Suche beginnt am Einstiegspunkt der obersten Schicht und findet den Punkt, der dem Abfragevektor in dieser Schicht am nächsten liegt.
- Steigen Sie vertikal in die mittlere Schicht ab und lokalisieren Sie erneut grob.
- Fallen Sie in Schicht 0, der Volladatenschicht, für einen feingranularen Durchlauf und geben Sie den nächsten Nachbarn aus.
Kernprinzip:Hohe Schichten schränken den Suchbereich schnell ein, während die unterste Schicht eine präzise Übereinstimmung liefert. Es optimiert die Brute-Force-Suche auf logarithmische Komplexität, bietet hohe Geschwindigkeit und hohe Wiederfindungsrate, was es zu einem Mainstream-Vektorsuchalgorithmus macht.
Beispiel
# Vereinfachte Demo der HNSW-Prinzipien (keine Produktionsimplementierung)
# ============================================
import numpy as np
import random
from typing import List, Dict, Set, Tuple
class SimpleHNSW:
"""Vereinfachtes HNSW, nur zur Demonstration der Prinzipien"""
def __init__(self, dim: int, max_level: int = 3, ef_construction: int = 5):
self.dim = dim
self.max_level = max_level # Maximale Anzahl an Schichten
self.ef_construction = ef_construction # Anzahl der Kandidaten, die beim Erstellen des Graphen berücksichtigt werden
self.levels: List[Dict[int, np.ndarray]] = [{} for _ in range(max_level)]
self.edges: List[Dict[int, Set[int]]] = [{} for _ in range(max_level)]
self.next_id = 0
self.enter_point = None # Einstiegspunkt (ein Punkt in der obersten Schicht)
def _random_level(self) -> int:
"""Zufällige Auswahl einer Schicht: je höher die Schicht, desto geringer die Wahrscheinlichkeit"""
level = 0
while random.random() < 0.5 and level < self.max_level - 1:
level += 1
return level
def _distance(self, v1: np.ndarray, v2: np.ndarray) -> float:
"""L2-Distanz"""
return np.sqrt(np.sum((v1 - v2) ** 2))
def _search_layer(
self,
query: np.ndarray,
level: int,
entry: int,
ef: int
) -> List[Tuple[float, int]]:
"""Suche in einer Schicht: Beginnend beim Einstiegspunkt gierig ef nächste Punkte finden"""
if entry not in self.levels[level]:
return []
visited = set([entry])
candidates = [(self._distance(query, self.levels[level][entry]), entry)]
result = list(candidates)
while candidates:
# Den aktuellen nächsten Kandidaten herausnehmen
candidates.sort()
dist, current = candidates.pop(0)
# Nachbarn durchlaufen
if current not in self.edges[level]:
continue
for neighbor in self.edges[level][current]:
if neighbor in visited:
continue
visited.add(neighbor)
neighbor_dist = self._distance(query, self.levels[level][neighbor])
# Zu Kandidaten und Ergebnismenge hinzufügen
candidates.append((neighbor_dist, neighbor))
result.append((neighbor_dist, neighbor))
# Die Größe der Ergebnismenge nicht größer als ef halten
result.sort()
result = result[:ef]
return result
def add(self, vector: np.ndarray) -> int:
"""Einen Vektor hinzufügen"""
vec_id = self.next_id
self.next_id += 1
# Zufällig bestimmen, in welchen Schichten dieser Vektor erscheint
max_level_for_vec = self._random_level()
# Den Vektor in allen Schichten speichern (in tatsächlichen Implementierungen wird der vollständige Vektor normalerweise nur in der untersten Schicht gespeichert)
for level in range(max_level_for_vec + 1):
self.levels[level][vec_id] = vector
self.edges[level][vec_id] = set()
# Wenn es der erste Punkt ist, als Einstiegspunkt festlegen
if self.enter_point is None:
self.enter_point = vec_id
return vec_id
# Von der obersten Schicht nach unten suchen, um die Einfügeposition in jeder Schicht zu finden
current_entry = self.enter_point
# Zuerst in Schichten oberhalb von max_level_for_vec suchen und den Einstiegspunkt aktualisieren
for level in range(self.max_level - 1, max_level_for_vec, -1):
if current_entry in self.levels[level]:
layer_result = self._search_layer(vector, level, current_entry, 1)
if layer_result:
current_entry = layer_result[0][1]
# In Schichten bis max_level_for_vec Verbindungen herstellen
for level in range(max_level_for_vec, -1, -1):
# Nach den nächstgelegenen ef_construction-Punkten in dieser Schicht suchen
layer_result = self._search_layer(vector, level, current_entry, self.ef_construction)
# Mit diesen Punkten verbinden (vereinfacht: bidirektionale Verbindung)
for dist, neighbor_id in layer_result:
self.edges[level][vec_id].add(neighbor_id)
self.edges[level][neighbor_id].add(vec_id)
# Den Eintrag für die nächste Schicht aktualisieren
if layer_result:
current_entry = layer_result[0][1]
# Den Eintrag der obersten Schicht aktualisieren
if max_level_for_vec > self._random_level(): # Vereinfachte Prüfung
self.enter_point = vec_id
return vec_id
def search(self, query: np.ndarray, top_k: int = 5, ef: int = 10) -> List[Tuple[int, float]]:
"""Suche nach den Top K Nächstgelegenen"""
if self.enter_point is None:
return []
current_entry = self.enter_point
# Von der obersten Schicht bis zur ersten Schicht
for level in range(self.max_level - 1, 0, -1):
if current_entry in self.levels[level]:
layer_result = self._search_layer(query, level, current_entry, 1)
if layer_result:
current_entry = layer_result[0][1]
# Ef-Punkte in der untersten Schicht suchen und dann Top K zurückgeben
layer_result = self._search_layer(query, 0, current_entry, ef)
layer_result.sort()
return [(vec_id, dist) for dist, vec_id in layer_result[:top_k]]
# Test
np.random.seed(42)
dim = 128
hnsw = SimpleHNSW(dim=dim, max_level=3, ef_construction=5)
# 100 Vektoren hinzufügen
print("HNSW-Index wird erstellt (100 Vektoren werden hinzugefügt)...")
for i in range(100):
vec = np.random.randn(dim)
hnsw.add(vec)
print(f"Index-Erstellung abgeschlossen, Einstiegspunkt = {hnsw.enter_point}")
# Abfragevektor generieren
query = np.random.randn(dim)
# Suchen
print("\nHNSW-Suche wird ausgeführt...")
results = hnsw.search(query, top_k=5, ef=10)
print("Search results:")
for vec_id, dist in results:
print(f" Vektor {vec_id:3d}, Abstand = {dist:.4f}")
# Ausgabe ähnlich wie:
# HNSW-Index wird erstellt (100 Vektoren werden hinzugefügt)...
# Index-Erstellung abgeschlossen, Einstiegspunkt = 54
#
# HNSW-Suche wird ausgeführt...
# Search results:
# Vektor 86, Abstand = 13.7515
# Vektor 30, Abstand = 13.8651
# Vektor 28, Abstand = 14.0754
# Vektor 56, Abstand = 14.1106
# Vektor 34, Abstand = 14.2131
Vorteile von HNSW:
1. Schnelle Abfrage: Normalerweise müssen nur wenige Knoten besucht werden, um den nächsten Nachbarn zu finden.
-
2. Schnelle Erstellung: Vektoren können inkrementell hinzugefügt werden, ohne den gesamten Index neu zu erstellen.
-
3. Einfache Parameter: Hauptsächlich Anpassung von ef_construction (Graphenqualität) und ef (Suchqualität).
IVF: Inverted File Index
Die Idee von IVF (Inverted File) ist „Teile und herrsche“.
Schritte:
-
1. Clustering: Den Vektorraum in K „Buckets“ unterteilen (mittels K-Means-Clustering).
-
2. Indizierung: Jeder Vektor wird seinem nächstgelegenen Bucket zugewiesen.
-
3. Abfrage: Zuerst die N Buckets finden, die dem Abfragevektor am nächsten liegen, und dann nur innerhalb dieser N Buckets suchen.
Beispiel
# Vereinfachte Demonstration des IVF-Prinzips
# ============================================
import numpy as np
from typing import List, Tuple
class SimpleIVF:
"""Vereinfachtes IVF, dient nur der Demonstration des Prinzips"""
def __init__(self, dim: int, nlist: int = 10):
self.dim = dim
self.nlist = nlist # Anzahl der Clusterzentren (Anzahl der Buckets)
self.centroids: np.ndarray = None # Clusterzentren
self.inverted_lists: List[List[Tuple[int, np.ndarray]]] = [[] for _ in range(nlist)]
self.next_id = 0
def _l2_distance(self, v1: np.ndarray, v2: np.ndarray) -> float:
"""L2-Distanz"""
return np.sqrt(np.sum((v1 - v2) ** 2))
def _find_nearest_centroid(self, vector: np.ndarray) -> int:
"""Das nächstgelegene Clusterzentrum finden"""
distances = [self._l2_distance(vector, c) for c in self.centroids]
return int(np.argmin(distances))
def fit(self, vectors: List[np.ndarray]):
"""Clusterzentren mit K-Means initialisieren (vereinfacht: zufällige Auswahl)"""
# Vereinfacht: nlist Vektoren zufällig als initiale Zentren auswählen
# In einer realen Implementierung sollte eine K-Means-Iteration verwendet werden
indices = np.random.choice(len(vectors), self.nlist, replace=False)
self.centroids = np.array([vectors[i] for i in indices])
def add(self, vector: np.ndarray) -> int:
"""Vektor hinzufügen"""
vec_id = self.next_id
self.next_id += 1
# Das nächstgelegene Bucket finden
centroid_idx = self._find_nearest_centroid(vector)
# In dieses Bucket legen
self.inverted_lists[centroid_idx].append((vec_id, vector))
return vec_id
def search(self, query: np.ndarray, top_k: int = 5, nprobe: int = 3) -> List[Tuple[int, float]]:
"""
Top K suchen
nprobe: Anzahl der zu durchsuchenden Buckets (mehr ist genauer, aber langsamer)
"""
# Die nprobe nächstgelegenen Buckets finden
centroid_distances = [
(self._l2_distance(query, c), i)
for i, c in enumerate(self.centroids)
]
centroid_distances.sort()
nearest_centroids = [i for dist, i in centroid_distances[:nprobe]]
# Nur in diesen nprobe Buckets suchen
candidates = []
for centroid_idx in nearest_centroids:
for vec_id, vec in self.inverted_lists[centroid_idx]:
dist = self._l2_distance(query, vec)
candidates.append((dist, vec_id))
# Sortieren und Top K zurückgeben
candidates.sort()
return [(vec_id, dist) for dist, vec_id in candidates[:top_k]]
# Test
np.random.seed(42)
dim = 128
# 1000 Vektoren generieren
vectors = [np.random.randn(dim) for _ in range(1000)]
# IVF-Index erstellen
ivf = SimpleIVF(dim=dim, nlist=20)
ivf.fit(vectors)
# Vektoren hinzufügen
for vec in vectors:
ivf.add(vec)
print(f"IVF index construction complete, bucket count = {ivf.nlist}")
for i in range(ivf.nlist):
print(f" Bucket {i:2d}: {len(ivf.inverted_lists[i])} vectors")
# Abfrage
query = np.random.randn(dim)
print("\nExecuting IVF search (nprobe=3)...")
results = ivf.search(query, top_k=5, nprobe=3)
print("Search results:")
for vec_id, dist in results:
print(f" Vektor {vec_id:3d}, Abstand = {dist:.4f}")
# Ausgabe ähnlich wie:
# IVF index construction complete, bucket count = 20
# Bucket 0: 63 vectors
# Bucket 1: 56 vectors
# ...
#
# Executing IVF search (nprobe=3)...
# Search results:
# Vector 863, distance = 13.5409
# Vector 189, distance = 13.7083
# Vector 491, distance = 13.8628
# Vector 264, distance = 13.9812
# Vector 781, distance = 14.0021
The advantage of IVF is its small memory footprint, and you can dynamically adjust the nprobe parameter at query time to balance speed and accuracy.
PQ: Produktquantisierung
PQ (Product Quantization) ist ein Komprimierungsalgorithmus.
Kernidee:
-
1. Aufteilung: Den hochdimensionalen Vektor in mehrere kleinere Segmente (Subvektoren) aufteilen.
-
2. Quantisierung: Jedes kleine Segment unabhängig clustern und den Originalvektor durch die ID des Clusterzentrums ersetzen.
-
3. Nachschlag: Zum Abfragezeitpunkt eine Nachschlagetabelle verwenden, um Näherungsdistanzen schnell zu berechnen.
Ein 128-dimensionaler Vektor wird beispielsweise in 8 Segmente zu je 16 Dimensionen aufgeteilt, und jedes Segment wird mit 256 Zentren quantisiert. Dadurch benötigt jeder Vektor nur 8 Byte Speicherplatz (ursprünglich 128 * 4 = 512 Byte).
IVF-PQ-Kombination
In Produktionsumgebungen werden IVF und PQ normalerweise kombiniert:
IVF ist für das „Bucketing“ verantwortlich, um den Suchbereich zu verkleinern, und PQ ist für die „Komprimierung“ zuständig, um Speicher und Rechenaufwand zu reduzieren.
| Algorithmus | Geschwindigkeit | Genauigkeit | Gedächtnis | Anwendbare Szenarien |
|---|---|---|---|---|
| Exakte Suche (Brute-Force) | Langsam | 100% | Hoch | Kleine Datenmenge, 100% Genauigkeit erforderlich |
| HNSW | Extrem schnell | Hoch | Mittel | Abfragelastig, geringe Latenz erforderlich |
| IVF | Schnell | Mittel | Niedrig | Großflächige Daten, geringer Genauigkeitsverlust akzeptabel |
| IVF-PQ | Sehr schnell | Mittel | Extrem niedrig | Ultra-große Datenmengen, speicherbeschränkt |
Ähnlichkeitsmetriken
Es gibt viele Möglichkeiten, „Ähnlichkeit“ zu berechnen, und die Wahl der richtigen Metrik ist wichtig.
Kosinus-Ähnlichkeit
Die Kosinus-Ähnlichkeit misst den Winkel zwischen den Richtungen zweier Vektoren mit einem Bereich von [-1, 1].
Je näher der Wert bei 1 liegt, desto ähnlicher; je näher bei -1, desto entgegengesetzter.
Beispiel
# Berechnung der Kosinus-Ähnlichkeit
# ============================================
import numpy as np
def cosine_similarity(v1: np.ndarray, v2: np.ndarray) -> float:
"""
Kosinus-Ähnlichkeit = (v1 · v2) / (|v1| * |v2|)
Bereich [-1, 1], näher an 1 bedeutet ähnlicher
"""
dot_product = np.dot(v1, v2)
norm1 = np.linalg.norm(v1)
norm2 = np.linalg.norm(v2)
return dot_product / (norm1 * norm2)
# Test
np.random.seed(42)
v1 = np.array([1, 2, 3, 4, 5])
v2 = np.array([2, 4, 6, 8, 10]) # Gleiche Richtung
v3 = np.array([-1, -2, -3, -4, -5]) # Entgegengesetzte Richtung
v4 = np.random.randn(5) # Zufällig
print(f"v1 = {v1}")
print(f"v2 = {v2}")
print(f"v3 = {v3}")
print(f"v4 = {v4}")
print()
print(f"cosine(v1, v2) = {cosine_similarity(v1, v2):.4f} (same direction, should be 1.0)")
print(f"cosine(v1, v3) = {cosine_similarity(v1, v3):.4f} (opposite direction, should be -1.0)")
print(f"cosine(v1, v4) = {cosine_similarity(v1, v4):.4f}")
# Output:
# v1 = [1 2 3 4 5]
# v2 = [ 2 4 6 8 10]
# v3 = [-1 -2 -3 -4 -5]
# v4 = [ 0.4967 -0.1383 0.6477 1.5230 -0.2342]
#
# cosine(v1, v2) = 1.0000 (gleiche Richtung, sollte 1.0 sein)
# cosine(v1, v3) = -1.0000 (entgegengesetzte Richtung, sollte -1.0 sein)
# cosine(v1, v4) = 0.2143
Eigenschaft der Kosinus-Ähnlichkeit: Sie kümmert sich nicht um die Länge des Vektors, sondern nur um die Richtung. Dies ist bei Texteinbettungen üblich, da die „Semantik“ von Text unabhängig von der Länge ist.
L2-Euklidische Distanz
Die L2-Distanz ist die Luftlinienentfernung zwischen zwei Punkten im Raum; je kleiner, desto ähnlicher.
Skalarprodukt (Punktprodukt)
Skalarprodukt = v1 · v2. Wenn die Vektoren normalisiert sind, entspricht das Skalarprodukt der Kosinus-Ähnlichkeit.
Auswahlleitfaden
| Metrik | Werbereich | Wann zu wählen | Typische Anwendungsfälle |
|---|---|---|---|
| Kosinus-Ähnlichkeit | [-1, 1] | Kümmert sich nur um die Richtung, nicht um die Länge | Texteinbettungen, semantische Suche |
| L2-Distanz | [0, ∞) | Räumliche Position ist wichtig | Bildmerkmale, Empfehlungssysteme |
| Skalarprodukt | (-∞, ∞) | Vektoren sind normalisiert oder die Länge ist aussagekräftig | Bestimmte spezifische Embedding-Modelle |
Die meisten Embedding-Modelle (z. B. die text-embedding-Serie von OpenAI) geben normalisierte Vektoren aus; in diesem Fall istKosinus-Ähnlichkeit = inneres Produkt, sodass es keine Rolle spielt, welches Sie verwenden.
Vergleich gängiger Vektordatenbanken
Vektordatenbanken lassen sich in drei Kategorien unterteilen: lokale Bibliotheken (FAISS), leichtgewichtige Open-Source-Bibliotheken (Chroma, Qdrant) und unternehmensweite Lösungen (Milvus, Pinecone, Weaviate).
FAISS: Eine lokale Bibliothek von Facebook
FAISS ist keine vollständige Datenbank, sondern eine C++-Bibliothek (mit Python-Bindings).
Beispiel
# Demonstration der grundlegenden FAISS-Nutzung
# ============================================
import numpy as np
# Hinweis: Sie müssen FAISS zuerst installieren, um es auszuführen
# pip install faiss-cpu (CPU-Version)
# oder
# pip install faiss-gpu (GPU-Version)
# Unten ist ein typisches FAISS-Nutzungsbeispiel (Codebeispiel, wird nicht tatsächlich ausgeführt)
def faiss_example():
"""FAISS-Nutzungsbeispiel"""
# 1. Daten vorbereiten
dimension = 128
nb = 10000 # Anzahl der Datenbankvektoren
nq = 10 # Anzahl der Abfragevektoren
np.random.seed(42)
xb = np.random.random((nb, dimension)).astype('float32') # Datenbank
xq = np.random.random((nq, dimension)).astype('float32') # Abfrage
# 2. Index erstellen (IVF-Beispiel)
import faiss
nlist = 100 # Anzahl der Clusterzentren
quantizer = faiss.IndexFlatL2(dimension) # Quantisierer
index = faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_L2)
# 3. Index trainieren
index.train(xb)
# 4. Vektoren hinzufügen
index.add(xb)
# 5. nprobe für Suche festlegen
index.nprobe = 10
# 6. Top 5 suchen
k = 5
distances, indices = index.search(xq, k)
print("Abfrageergebnisse (Top 5 für die ersten 2 Abfragen):")
for i in range(min(2, nq)):
print(f"Abfrage {i}:")
for j in range(k):
print(f" Index {indices[i][j]}, Distanz {distances[i][j]:.4f}")
return index
# HNSW-Index-Beispiel
def faiss_hnsw_example():
"""FAISS HNSW-Index"""
import faiss
dimension = 128
index = faiss.IndexHNSWFlat(dimension, 32) # 32 = M-Parameter (Anzahl der Verbindungen pro Knoten)
index.hnsw.efConstruction = 40 # ef während der Graphenkonstruktion
index.hnsw.efSearch = 16 # ef während der Suche
# Daten hinzufügen...
# index.add(xb)
return index
# Wenn FAISS installiert ist, können Sie die Zeilen unten auskommentieren, um sie auszuführen
# faiss_example()
print("FAISS-Beispielcode: Dies ist eine typische Verwendung von FAISS.")
print("Wichtigste Schritte: Index erstellen → trainieren → Vektoren hinzufügen → suchen.")
print()
print("Gängige FAISS-Indextypen:")
print(" - IndexFlatL2: Brute-Force-Suche (genau, aber langsam)")
print(" - IndexIVFFlat: IVF-Bucket-Suche")
print(" - IndexIVFPQ: IVF + PQ-Kompression")
print(" - IndexHNSWFlat: HNSW-Graphenindex")
FAISS-Funktionen:
-
1. Extrem schnell: C++-Implementierung, bis zum Äußersten optimiert.
-
2. Voll funktionsfähig: Unterstützt fast alle gängigen ANN-Algorithmen.
-
3. Lokal: Kein Netzwerk, keine Persistenz (Sie müssen die Indexdatei selbst speichern).
Anwendbare Szenarien: Offline-Batch-Verarbeitung, kein Wunsch nach Einrichtung eines Dienstes, relativ geringes Datenvolumen.
Chroma: Leichtgewicht und Local-First
Chroma ist eine einfache und benutzerfreundliche Vektordatenbank, die entwickelt wurde, um „die Entwicklung von KI-Anwendungen zu beschleunigen“.
Beispiel
# Demonstration der grundlegenden Chroma-Nutzung
# ============================================
# Installation: pip install chromadb
def chroma_example():
"""Chroma-Anwendungsbeispiel"""
import chromadb
from chromadb.utils import embedding_functions
# 1. Client initialisieren (lokaler Modus, Daten werden auf der Festplatte gespeichert)
client = chromadb.PersistentClient(path="./example-chroma-db")
# 2. Eine Collection erstellen oder abrufen
collection = client.get_or_create_collection(
name="example_docs",
metadata={"description": "Beispielhafte Tutorial-Dokumenten-Collection"}
)
# 3. Dokumente hinzufügen (Chroma übernimmt das Embedding automatisch)
# Sie können OpenAI, Cohere etc. verwenden oder Ihre eigenen Vektoren bereitstellen
documents = [
"Python ist eine interpretierte, höhere Programmiersprachen für allgemeine Zwecke.",
"JavaScript ist eine leichtgewichtige, interpretierte Programmiersprache.",
"Vektordatenbanken werden zum Speichern und Abrufen von Vektoreinbettungen verwendet.",
"RAG oder Retrieval-Augmented Generation ist eine Technik, die Abruf mit Generierung kombiniert.",
"HNSW ist ein approximativer Nearest-Neighbor-Algorithmus, der auf einem hierarchischen Small-World-Graphen basiert."
]
metadatas = [
{"category": "programming", "language": "Python"},
{"category": "programming", "language": "JavaScript"},
{"category": "database", "topic": "vector"},
{"category": "ai", "topic": "rag"},
{"category": "algorithm", "topic": "hnsw"}
]
ids = ["doc1", "doc2", "doc3", "doc4", "doc5"]
# Hinzufügen (hier werden Dummy-Vektoren verwendet; in der Praxis ein Embedding-Modell nutzen)
import numpy as np
np.random.seed(42)
embeddings = [np.random.randn(128).tolist() for _ in range(5)]
collection.add(
documents=documents,
metadatas=metadatas,
ids=ids,
embeddings=embeddings # Falls nicht angegeben, verwendet Chroma die Standard-Embed-Funktion
)
print(f"Es befinden sich {collection.count()} Dokumente in der Collection")
# 4. Abfrage
query_embedding = np.random.randn(128).tolist()
results = collection.query(
query_embeddings=[query_embedding],
n_results=3,
# Metadaten-Filterung kann hinzugefügt werden
# where={"category": "ai"}
)
print("\nAbfrageergebnisse:")
for i in range(len(results['ids'][0])):
print(f"ID: {results['ids'][0][i]}")
print(f"Dokument: {results['documents'])
print(f"Distanz: {results['distances'])
print("---")
return collection
# Wenn Chroma installiert ist, können Sie die Zeilen unten zum Ausführen auskommentieren
# chroma_example()
print("Chroma-Beispielcode: einfaches Design, funktioniert sofort.")
print()
print("Chroma-Funktionen:")
print(" - Einfach: benutzerfreundliche API, in Minuten startklar")
print(" - Lokal-zuerst: kann auf der Festplatte persistieren oder im Arbeitsspeicher laufen")
print(" - Integrierte Embeddings: unterstützt OpenAI, Cohere, Sentence-Transformers")
print(" - Unterstützt Metadaten-Filterung: kann bei Abfragen nach Metadaten filtern")
Qdrant: Produktionstaugliches Open Source
Qdrant ist eine in Rust geschriebene Vektordatenbank mit guter Leistung und vollem Funktionsumfang.
Beispiel
# Qdrant-Grundnutzungsdemo
# ============================================
# Installieren: pip install qdrant-client
# Dienst starten: docker run -p 6333:6333 qdrant/qdrant
def qdrant_example():
"""Qdrant-Nutzungsbeispiel"""
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
# 1. Client verbinden
client = QdrantClient(host="localhost", port=6333)
# 2. Collection erstellen
client.recreate_collection(
collection_name="example_collection",
vectors_config=VectorParams(
size=128,
distance=Distance.COSINE # oder L2, DOT
),
# Optional: HNSW-Index-Parameter konfigurieren
# hnsw_config=HnswConfig(m=16, ef_construct=100)
)
# 3. Punkte hinzufügen
import numpy as np
np.random.seed(42)
points = [
PointStruct(
id=i,
vector=np.random.randn(128).tolist(),
payload={
"text": f"Inhalt von Dokument {i}",
"category": "tech" if i % 2 == 0 else "life",
"author": f"author_{i % 3}"
}
)
for i in range(100)
]
client.upsert(
collection_name="example_collection",
points=points
)
# 4. Abfrage
query_vector = np.random.randn(128).tolist()
results = client.search(
collection_name="example_collection",
query_vector=query_vector,
limit=5,
# Filterung kann hinzugefügt werden
# query_filter=Filter(
# must=[
# FieldCondition(key="category", match=MatchValue(value="tech"))
# ]
# )
)
print(Abfrageergebnisse:")
for result in results:
print(f"ID: {result.id}, Score: {result.score:.4f}")
print(f"Nutzdaten (Payload): {result.payload}")
print("---")
return client
# Wenn Qdrant läuft, können Sie die Zeilen unten zum Ausführen auskommentieren
# qdrant_example()
print("Qdrant-Beispielcode: Rust-Implementierung, umfassende Funktionen.")
print()
print("Qdrant-Funktionen:")
print(" - Gute Leistung: Rust + HNSW")
print(" - Voller Funktionsumfang: unterstützt Filtern, Gruppieren, Aggregieren, Paginierung")
print(" - Einfache Bereitstellung: Start mit einem Klick über Docker")
print(" - Unterstützt Sharding: kann horizontal skaliert werden")
Vergleichstabelle gängiger Vektordatenbanken
| Produkt | Type | Bereitstellung | Funktionen | Anwendungsfälle |
|---|---|---|---|---|
| FAISS | Lokale Bibliothek | Keine | Extrem schnell, funktionsreich, erfordert Eigenwartung | Offline-Verarbeitung, kein Einrichten eines Dienstes gewünscht |
| Chroma | Leichtgewicht | Lokal/Server | Einfach und benutzerfreundlich, Python-zentriert | Prototyping, kleine Projekte |
| Qdrant | Open Source | Docker/K8s | Gute Leistung, vollständige Funktionen, Rust | Produktionsumgebung, mittlerer Maßstab |
| Milvus | Open Source | Verteilt | Unternehmensreif, vollständigster Funktionsumfang | Groß angelegte Produktionsumgebung |
| Weaviate | Open Source | Docker/K8s | Modular, gutes Ökosystem | Flexible Zusammensetzung von Funktionen erforderlich |
| Pinecone | Cloud-Dienst | SaaS | Vollständig verwaltet, Skalierung bei Bedarf | Kein Betrieb gewünscht, schneller Start |
Auswahl des Embedding-Modells
Vektordatenbanken sind nur Infrastruktur; was die Ergebnisse wirklich bestimmt, ist das Embedding-Modell.
Vergleich gängiger Embedding-Modelle
| Modell | Dimensionen | Funktionen | Preis |
|---|---|---|---|
| OpenAI text-embedding-3-small | 1536 | Gutes Preis-Leistungs-Verhältnis, gute Gesamtqualität | $0.00002 / 1K tokens |
| OpenAI text-embedding-3-large | 3072 | Höchste Qualität, hohe Dimensionen | $0.00013 / 1K tokens |
| Cohere Embed v3 | 1024 | Gute mehrsprachige Unterstützung | $0.0001 / 1K tokens |
| bge-large-zh-v1.5 | 1024 | Gut für Chinesisch, Open Source und kostenlos | Kostenlos |
| bge-m3 | 1024 | Mehrsprachig, multifunktional | Kostenlos |
| gte-large | 1024 | Ausgewogene Wahl, Open Source | Kostenlos |
Abwägung zwischen Dimensionen und Leistung
Je höher die Dimensionen, desto besser sind normalerweise die Ergebnisse, aber:
-
1. Speicher ist teurer: 3072 Dimensionen beanspruchen doppelt so viel Speicherplatz wie 1536 Dimensionen.
-
2. Berechnung ist langsamer: Die Kosten für die Distanzberechnung sind proportional zur Dimensionalität.
-
3. Größerer Index: HNSW-/IVF-Indizes werden ebenfalls größer.
Empfehlung:Testen Sie zuerst mit einem niedrigdimensionalen Modell; wenn das nicht ausreicht, gehen Sie zu einer höheren Dimension über.. In den meisten Szenarien sind 1536 Dimensionen bereits gut genug.
Hybridsuche
Reine Vektorsuche ist manchmal nicht gut genug, weil sie nur nach "Semantik" und nicht nach "Schlüsselwörtern" sucht.
Hybride Suche = Vektorsuche + Stichwortsuche, gefolgt von der Zusammenführung der Ergebnisse.
BM25 Sparse Retrieval
BM25 ist ein klassischer Algorithmus in der traditionellen Suche, der den Grad der Schlüsselwortübereinstimmung misst.
Beispiel
# BM25-Stichwortsuche-Demo
# ============================================
import math
from typing import List, Dict, Set
class SimpleBM25:
"""Vereinfachte BM25-Implementierung"""
def __init__(self, k1: float = 1.5, b: float = 0.75):
self.k1 = k1
self.b = b
self.documents: List[List[str]] = []
self.doc_lengths: List[int] = []
self.avg_doc_length: float = 0
self.term_freqs: List[Dict[str, int]] = []
self.doc_freq: Dict[str, int] = {}
self.num_docs: int = 0
def _tokenize(self, text: str) -> List[str]:
"""Einfache Tokenisierung (nach Leerzeichen und Satzzeichen)"""
# Vereinfacht: in Kleinbuchstaben umwandeln, nach Nicht-Alphanumerischen Zeichen aufteilen
import re
return [t.lower() for t in re.findall(r'\w+', text)]
def fit(self, documents: List[str]):
"""Index erstellen"""
self.documents = []
self.doc_lengths = []
self.term_freqs = []
self.doc_freq = {}
self.num_docs = len(documents)
for doc in documents:
tokens = self._tokenize(doc)
self.documents.append(tokens)
self.doc_lengths.append(len(tokens))
# Termfrequenz zählen
tf: Dict[str, int] = {}
for token in tokens:
tf[token] = tf.get(token, 0) + 1
self.term_freqs.append(tf)
# Dokumentfrequenz aktualisieren
for token in tf.keys():
self.doc_freq[token] = self.doc_freq.get(token, 0) + 1
self.avg_doc_length = sum(self.doc_lengths) / self.num_docs
def score(self, query: str) -> List[float]:
"""BM25-Score für jedes Dokument berechnen"""
query_tokens = self._tokenize(query)
scores = [0.0] * self.num_docs
for token in query_tokens:
if token not in self.doc_freq:
continue
# IDF-Berechnung
df = self.doc_freq[token]
idf = math.log(1 + (self.num_docs - df + 0.5) / (df + 0.5))
# Punktzahl für jedes Dokument berechnen
for doc_idx in range(self.num_docs):
doc_len = self.doc_lengths[doc_idx]
tf = self.term_freqs[doc_idx].get(token, 0)
# BM25-Kernformel
numerator = tf * (self.k1 + 1)
denominator = tf + self.k1 * (1 - self.b + self.b * doc_len / self.avg_doc_length)
scores[doc_idx] += idf * numerator / denominator
return scores
def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""Suche Top K"""
scores = self.score(query)
# Sortieren
results = [
{"doc_idx": i, "score": s}
for i, s in enumerate(scores)
]
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]
# Test
documents = [
"Python ist eine Programmiersprache, die leicht zu erlernen ist.",
"JavaScript wird für die Webentwicklung verwendet und ist sowohl im Frontend als auch im Backend einsetzbar.",
"Vektordatenbanken werden zum Speichern und Abrufen von Vektoreinbettungen verwendet.",
"Es gibt viele Arten von Datenbanken, wie zum Beispiel relationale Datenbanken und NoSQL-Datenbanken.",
"HNSW ist ein Suchalgorithmus für die ungefähren nächsten Nachbarn (Approximate Nearest Neighbor)."
]
bm25 = SimpleBM25()
bm25.fit(documents)
# Suchen
query = "database"
results = bm25.search(query, top_k=5)
print(f"Query:\"{query}\"")
print("BM25 search results:")
for r in results:
print(f" Document {r['doc_idx']}: {documents[r['doc_idx']]}")
print(f" Score = {r['score']:.4f}")
# Ausgabe ähnlich wie:
# Query: "database"
# BM25 search results:
# Document 2: Vector databases are used to store and retrieve vector embeddings.
# Score = 0.7134
# Document 3: There are many types of databases, such as relational databases and NoSQL databases.
# Score = 0.6148
# ...
RRF: Reciprocal Rank Fusion
RRF (Reciprocal Rank Fusion) ist eine einfache und effektive Methode zum Zusammenführen mehrerer Suchergebnisse.
Formula: score = 1 / (k + rank), where k is usually 60.
Beispiel
# RRF fusion demo
# ============================================
from typing import List, Dict, Any
def rrf_fuse(
result_lists: List[List[Any]],
k: int = 60
) -> Dict[Any, float]:
"""
RRF fusion: for each result list, add scores using 1/(k + rank)
result_lists: multiple sorted result lists
"""
scores: Dict[Any, float] = {}
for results in result_lists:
for rank, item in enumerate(results):
score = 1.0 / (k + rank)
scores[item] = scores.get(item, 0) + score
return scores
# Example: vector search results and BM25 search results
vector_results = ["doc3", "doc1", "doc4", "doc2", "doc5"] # Vector search considers doc3 most relevant
bm25_results = ["doc2", "doc4", "doc3", "doc5", "doc1"] # BM25 considers doc2 most relevant
# RRF fusion
fused = rrf_fuse([vector_results, bm25_results])
# Sortieren
final_rank = sorted(fused.items(), key=lambda x: x[1], reverse=True)
print("Vector search results:", vector_results)
print("BM25 search results:", bm25_results)
print()
print("Final ranking after RRF fusion:")
for doc, score in final_rank:
print(f" {doc}: {score:.6f}")
# Output:
# Vector search results: ['doc3', 'doc1', 'doc4', 'doc2', 'doc5']
# BM25 search results: ['doc2', 'doc4', 'doc3', 'doc5', 'doc1']
#
# Final ranking after RRF fusion:
# doc3: 0.032787
# doc2: 0.032787
# doc4: 0.032520
# doc1: 0.016393
# doc5: 0.016393
Fusion Weight Tuning
Neben RRF kann auch die gewichtete lineare Fusion verwendet werden:
final_score = α * vector_score + (1 - α) * bm25_score
α ist das Gewicht für die Vektorsuche, das basierend auf tatsächlichen Daten angepasst werden muss.
Re-Ranking
Die Hybridsuche kann weiter verbessert werden: Zuerst wird eine Charge von Kandidaten abgerufen und diese werden dann mit einem leistungsstärkeren Modell neu gerankt.
Cross-Encoder Principles
Bi-Encoder (für den Abruf): Kodiert Anfrage und Dokument separat und vergleicht sie anhand der Ähnlichkeit.
Cross-Encoder (für das Re-Ranking): Verknüpft Anfrage und Dokument als Eingabe für das Modell und gibt direkt einen Relevanzwert aus.
Cross-Encoder sind genauer, aber langsamer, weshalb sie sich nur für das Re-Ranking der Top 50-100 Kandidaten eignen.
Beispiel
# Re-ranking process demo
# ============================================
def reranking_pipeline_example():
"""
Zweistufiger Abrufprozess:
1. Grobabruf: Verwenden Sie die Vektorsuche, um schnell die Top 100 zu finden
2. Fein-Ranking: Verwenden Sie den Cross-Encoder, um die Top 100 neu zu ranken
"""
print("Two-stage retrieval process:")
print()
print("Stage 1: Coarse retrieval (vector search)")
print(" - Goal: fast, low latency")
print(" - Method: HNSW/IVF vector search")
print(" - Output: Top 100 candidates")
print()
print("Stage 2: Fine ranking (Cross-Encoder)")
print(" - Goal: accurate, high quality")
print(" - Method: Cross-Encoder pairwise scoring")
print(" - Output: final Top 10")
print()
print("Typical configuration:")
print(" - Retrieval stage: HNSW efSearch = 128, recall 100")
print(" - Reranking stage: use bge-reranker or Cohere Rerank")
print()
print("Why not directly use Cross-Encoder to search everything?")
print(" - Cross-Encoder is too slow; searching 1 million documents is unrealistic")
print(" - Vector search over 1 million is fast, and Cross-Encoder reranking over 100 is also fast")
# Cohere Rerank API example (pseudocode)
def cohere_rerank_example():
"""
Beispiel für die Verwendung der Cohere Rerank API
"""
print("Typical usage of the Cohere Rerank API:")
print()
print('''
import cohere
co = cohere.Client(api_key="your-api-key")
query = "What is a vector database?"
documents = [
"Inhalt von Dokument 1...",
"Inhalt von Dokument 2...",
... # Top-100-Kandidaten
]
results = co.rerank(
query=query,
documents=documents,
top_n=10,
model="rerank-english-v3.0"
)
for r in results:
print(r.index, r.relevance_score, documents[r.index])
''')
reranking_pipeline_example()
Leistung vs. Latenz-Abwägung
| Ansatz | Qualität | Geschwindigkeit | Kosten | Anwendungsszenario |
|---|---|---|---|---|
| Reine Vektorsuche | Mittel | Extrem schnell | Niedrig | Für extrem hohe Geschwindigkeitsanforderungen |
| Hybride Suche | Mittel-hoch | Schnell | Mittel | Ausgewogene Wahl |
| Vektor + Reranking | Hoch | Mittel | Mittel-hoch | Streben nach Qualität |
| Hybrid + Reranking | Höchste | Mittel-langsam | Hoch | Qualität zuerst |
Optimierung der Produktionsumgebung
Vom Demo- zum Produktionseinsatz gibt es viele Details zu beachten.
Index-Sharding-Strategien
Wenn das Datenvolumen die Kapazität einer einzelnen Maschine übersteigt, ist Sharding erforderlich:
-
1. Sharding nach ID: Einfach, aber Abfragen müssen alle Shards durchsuchen.
-
2. Sharding durch Vektor-Clustering: Bei Abfragen werden nur wenige relevante Shards durchsucht.
Metadaten-Filterung
In Produktionsumgebungen wird fast immer eine Metadaten-Filterung kombiniert.
Zwei Strategien:
-
1. Vorfilterung: Zuerst nach Metadaten filtern, dann Vektorsuche durchführen.
-
2. Nachfilterung: Zuerst Vektorsuche durchführen, dann nach Metadaten filtern.
Auswahlprinzip: Wenn nach dem Filtern viele Vektoren übrig bleiben, verwende Pre-Filtering; wenn nur wenige übrig bleiben, verwende Post-Filtering.
Inkrementelle Aktualisierungen
Die meisten Vektordatenbanken unterstützen inkrementelles Hinzufügen, aber beachte:
-
1. Stapelweises Hinzufügen: In Stapeln hinzufügen, nicht einzeln.
-
2. Index-Neuaufbau: Einige Indizes (z. B. IVF) müssen regelmäßig neu trainiert werden.
-
3. Versionsverwaltung: Behalte alte Indexversionen, damit du bei Aktualisierungen zurückrollen kannst.
Monitoring-Metriken
Produktionsumgebungen sollten überwachen:
-
1. Abfragelatenz：P50、P95、P99.
-
2. Recall-Rate: Überprüfe regelmäßig die Qualität der Top-K-Ergebnisse.
-
3. Indexgröße: Speicher-/Festplattennutzung.
-
4. QPSAbfragen pro Sekunde.