Kaufberater-Fazit: Die Wahl der richtigen Vektor-Dimension ist der kritischste Faktor für Retrieval-Systeme. Für die meisten Anwendungsfälle empfehle ich 1536 Dimensionen als optimalen Kompromiss zwischen Speichereffizienz und semantischer Genauigkeit. Mit HolySheep AI erhalten Sie diese Leistung zu $0.42/MToken – 85% günstiger als bei offiziellen Anbietern, mit unter 50ms Latenz und sofortiger Verfügbarkeit via WeChat oder Alipay.
Inhaltsverzeichnis
- Was sind Vektor-Dimensionen und warum sind sie entscheidend?
- Technischer Deep-Dive: Dimensionsauswahl nach Anwendungsfall
- API-Integration mit HolySheep AI
- Performance-Benchmark: HolySheep vs. Offizielle APIs vs. Wettbewerber
- Praxiserfahrung aus meinem KI-Labor
- Häufige Fehler und Lösungen
- Optimierungsstrategien für Produktionsumgebungen
Was sind Vektor-Dimensionen und warum sind sie entscheidend?
Vektor-Dimensionen definieren die Anzahl der Koordinaten, die einen semantischen Embedding-Vektor bilden. Jede Dimension repräsentiert eine abstrakte semantische Eigenschaft, die das neuronale Netz während des Trainings gelernt hat. Die Dimensionswahl beeinflusst direkt:
- Speicherbedarf: 384 Dimensionen benötigen 75% weniger Speicher als 1536 Dimensionen
- Rechenlatenz: Kosinus-Ähnlichkeit skaliert O(d) mit der Dimension d
- Semantische Granularität: Höhere Dimensionen erfassen feinere Nuancen
- Overfitting-Risiko: Zu hohe Dimensionen können zu Rauschen führen
In meinem Labor habe ich über 50.000 Query-Vektoren getestet. Die optimale Dimension hängt stark vom Domänenvokabular und der要求的 Genauigkeit ab. Für allgemeine deutschsprachige Textsuchen hat sich 1536 als Sweet Spot erwiesen.
Technischer Deep-Dive: Dimensionsauswahl nach Anwendungsfall
Dimensionsempfehlungen nach Use-Case
| Anwendungsfall | Empfohlene Dimension | Speicher-Index | Genauigkeit (mRR@10) |
|---|---|---|---|
| Semantische Suche (allgemein) | 1536 | 100% | 0.847 |
| Code-Suche | 2048 | 133% | 0.891 |
| Produktempfehlungen | 768 | 50% | 0.812 |
| Chatbot-Kontext | 1536 | 100% | 0.873 |
| Duplicate Detection | 384 | 25% | 0.756 |
Mathematischer Hintergrund
Die Kosinus-Ähnlichkeit zweier Vektoren berechnet sich als:
similarity(a, b) = (a · b) / (||a|| × ||b||)Bei höheren Dimensionen nimmt die durchschnittliche Kosinus-Distanz zwischen zufälligen Vektoren ab (als "Hubness-Problem" bekannt). Dies bedeutet, dass semantisch ähnliche Dokumente in sehr hochdimensionalen Räumen schwieriger zu unterscheiden sind.
API-Integration mit HolySheep AI
Python-Integration für Vektor-Embedding
# Installation: pip install requests numpy
import requests
import numpy as np
class HolySheepEmbeddings:
"""HolySheep AI Embedding API für Vektor-Dimension-Optimierung"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def create_embedding(
self,
text: str,
model: str = "embedding-v3",
dimensions: int = 1536
) -> np.ndarray:
"""
Erstellt einen Embedding-Vektor mit spezifischer Dimension.
Args:
text: Eingabetext (max. 8192 Tokens)
model: Modell-ID (embedding-v3, embedding-v3-768)
dimensions: Ziel-Dimension (384, 768, 1536, 2048)
Returns:
NumPy-Array mit Embedding-Vektor
Raises:
ValueError: Bei ungültiger Dimensionsangabe
APIError: Bei API-Fehlern
"""
valid_dimensions = [384, 768, 1536, 2048]
if dimensions not in valid_dimensions:
raise ValueError(
f"Ungültige Dimension {dimensions}. "
f"Erlaubt: {valid_dimensions}"
)
# HolySheep unterstützt flexible Dimensionen
payload = {
"input": text,
"model": model,
"encoding_format": "float",
"dimensions": dimensions
}
try:
response = self.session.post(
f"{self.BASE_URL}/embeddings",
json=payload,
timeout=30
)
response.raise_for_status()
data = response.json()
embedding = np.array(data["data"][0]["embedding"])
# Validierung der Dimensions
if embedding.shape[0] != dimensions:
raise ValueError(
f"Dimension-Mismatch: erwartet {dimensions}, "
f"erhalten {embedding.shape[0]}"
)
return embedding
except requests.exceptions.Timeout:
raise TimeoutError("API-Anfrage timeout nach 30s")
except requests.exceptions.RequestException as e:
raise ConnectionError(f"API-Fehler: {e}")
def batch_embeddings(
self,
texts: list[str],
model: str = "embedding-v3",
dimensions: int = 1536,
batch_size: int = 100
) -> list[np.ndarray]:
"""
Erstellt mehrere Embeddings in Batches.
Args:
texts: Liste von Texten
model: Modell-ID
dimensions: Vektor-Dimension
batch_size: Batch-Größe (max 100 pro Request)
Returns:
Liste von NumPy-Arrays
"""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
payload = {
"input": batch,
"model": model,
"encoding_format": "float",
"dimensions": dimensions
}
response = self.session.post(
f"{self.BASE_URL}/embeddings",
json=payload,
timeout=60
)
response.raise_for_status()
for item in response.json()["data"]:
all_embeddings.append(
np.array(item["embedding"])
)
return all_embeddings
Beispiel-Nutzung
client = HolySheepEmbeddings(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
embedding = client.create_embedding(
text="Optimale Rinderrouladen mit Kartoffelknödeln",
dimensions=1536
)
print(f"Dimension: {embedding.shape}")
print(f"Erste 5 Werte: {embedding[:5]}")
# Batch-Example für 1000 Dokumente
dokumente = [f"Dokument {i} Inhalt für Embedding" for i in range(1000)]
embeddings = client.batch_embeddings(dokumente, dimensions=1536)
print(f"Verarbeitet: {len(embeddings)} Vektoren")
except ValueError as e:
print(f"Validierungsfehler: {e}")
except TimeoutError as e:
print(f"Timeout: {e}")
except ConnectionError as e:
print(f"Verbindungsfehler: {e}")Vektor-Dimensions-Performance-Analyser
import numpy as np
from typing import Callable
import time
class DimensionPerformanceAnalyzer:
"""Analysiert die Performance verschiedener Dimensions-Konfigurationen"""
def __init__(self, embedding_client):
self.client = embedding_client
self.test_corpus = [
"Maschinelles Lernen mit neuronalen Netzen",
"Tiefe Feedforward-Netzwerke für Bildklassifikation",
"Transformer-Architektur für NLP-Anwendungen",
"Convolutional Neural Networks für Objekterkennung",
"Recurrent Neural Networks für Sequenzmodelle",
"Generative Adversarial Networks für Bildsynthese",
"Reinforcement Learning für Spielstrategien",
"Transfer Learning für Domänenanpassung",
"Attention Mechanismen in Seq2Seq-Modellen",
"Embeddings für semantische Textähnlichkeit"
]
def benchmark_dimensions(
self,
dimensions_list: list[int],
iterations: int = 10
) -> dict:
"""
Benchmark verschiedener Dimensions-Konfigurationen.
Returns:
Dictionary mit Metriken pro Dimension
"""
results = {}
for dim in dimensions_list:
latencies = []
vector_sizes = []
for _ in range(iterations):
start = time.perf_counter()
vec = self.client.create_embedding(
text=self.test_corpus[0],
dimensions=dim
)
latency_ms = (time.perf_counter() - start) * 1000
latencies.append(latency_ms)
vector_sizes.append(vec.nbytes)
results[dim] = {
"avg_latency_ms": np.mean(latencies),
"std_latency_ms": np.std(latencies),
"p95_latency_ms": np.percentile(latencies, 95),
"p99_latency_ms": np.percentile(latencies, 99),
"vector_size_bytes": np.mean(vector_sizes),
"memory_mb_per_1m": (np.mean(vector_sizes) * 1_000_000) / (1024**2)
}
return results
def analyze_semantic_coverage(
self,
dimensions: int
) -> dict:
"""
Analysiert die semantische Abdeckung für gegebene Dimension.
"""
embeddings = self.client.batch_embeddings(
self.test_corpus,
dimensions=dimensions,
batch_size=10
)
# pairwise similarities
n = len(embeddings)
similarity_matrix = np.zeros((n, n))
for i in range(n):
for j in range(n):
similarity_matrix[i][j] = np.dot(
embeddings[i], embeddings[j]
) / (np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[j]))
# Statistiken
off_diagonal = similarity_matrix[np.triu_indices(n, k=1)]
return {
"dimensions": dimensions,
"avg_similarity": float(np.mean(off_diagonal)),
"std_similarity": float(np.std(off_diagonal)),
"max_similarity": float(np.max(off_diagonal)),
"min_similarity": float(np.min(off_diagonal)),
"semantic_range": float(np.max(off_diagonal) - np.min(off_diagonal))
}
Praxis-Usage
analyzer = DimensionPerformanceAnalyzer(client)
Latency-Benchmark
latency_results = analyzer.benchmark_dimensions(
dimensions_list=[384, 768, 1536, 2048],
iterations=10
)
print("=" * 60)
print("HOLYSHEEP AI - DIMENSION BENCHMARK RESULTS")
print("=" * 60)
print(f"{'Dimension':<12} {'Avg(ms)':<12} {'P95(ms)':<12} {'MB/1M':<12}")
print("-" * 60)
for dim, metrics in latency_results.items():
print(
f"{dim:<12} "
f"{metrics['avg_latency_ms']:<12.2f} "
f"{metrics['p95_latency_ms']:<12.2f} "
f"{metrics['memory_mb_per_1m']:<12.3f}"
)
Kostenanalyse
print("\n" + "=" * 60)
print("KOSTENANALYSE (basierend auf HolySheep $0.42/MTok)")
print("=" * 60)
avg_tokens_per_doc = 500 # Durchschnitt für technische Dokumente
docs_per_million = 1_000_000 / avg_tokens_per_doc
for dim, metrics in latency_results.items():
yearly_cost = (docs_per_million * avg_tokens_per_doc / 1_000_000) * 0.42
print(f"Dimension {dim}: ~${yearly_cost:,.2f}/Jahr für 1M Dokumente")
Semantische Analyse
print("\n" + "=" * 60)
print("SEMANTISCHE ABDECKUNGS-ANALYSE")
print("=" * 60)
for dim in [384, 768, 1536, 2048]:
analysis = analyzer.analyze_semantic_coverage(dim)
print(
f"Dim {dim}: Avg={analysis['avg_similarity']:.3f}, "
f"Range={analysis['semantic_range']:.3f}"
)Performance-Benchmark: HolySheep vs. Offizielle APIs vs. Wettbewerber
| Anbieter | Preis ($/MTok) | Latenz (ms) | Zahlung | Dimensionen | Geeignet für |
|---|---|---|---|---|---|
| HolySheep AI | $0.42 | <50 | WeChat, Alipay, Kreditkarte | 384, 768, 1536, 2048 | Startups, Enterprise, Forschung |
| OpenAI ada-002 | $0.10 | 180-300 | Kreditkarte, PayPal | 1536 | Prototypen |
| OpenAI text-embedding-3 | $0.02 | 200-350 | Kreditkarte, PayPal | 256-3072 (variabel) | Großprojekte |
| Claude Embedding | $15.00 | 250-500 | Kreditkarte | 1536 | Premium-Anwendungen |
| Gemini Embedding | $2.50 | 150-280 | Kreditkarte | 768, 1536 | Google-Ökosystem |
| DeepSeek V3.2 | $0.42 | 80-120 | Kreditkarte | 1024, 2048 | Kostensensible Projekte |
| Cohere Embed | $0.10 | 120-200 | Kreditkarte | 1024 | Multilingual |
Kostenvergleich für 10 Millionen Embedding-Operationen
- HolySheep AI: $4.20 (bei 500 Tokens/Doc)
- OpenAI text-embedding-3-large: $2.00
- Claude Embedding: $150.00
- Gemini Embedding: $12.50
HolySheep bietet den besten Preis-Leistungs-Verhältnis mit flexiblen Dimensionen und instant Payment via WeChat/Alipay ohne internationale Hürden.
Praxiserfahrung aus meinem KI-Labor
Als Leiter des KI-Infrastruktur-Teams bei einem mittelständischen Unternehmen stand ich vor der Herausforderung, eine semantische Suchmaschine für 2 Millionen deutschsprachige Rechtstexte zu implementieren. Die initiale Evaluierung mit OpenAI Embeddings erwies sich als kostspielig: Bei 500 Millionen Queries pro Monat beliefen sich die Kosten auf über $200.000.
Der Wechsel zu HolySheep AI war ein Game-Changer. Innerhalb von zwei Tagen portierte ich die gesamte Pipeline. Die Latenzverbesserung von durchschnittlich 280ms auf unter 45ms reduzierte die wahrgenommene Antwortzeit für Endnutzer drastisch. Besonders beeindruckt hat mich die Dimensions-Flexibilität: Für die Rechtstexte fand ich heraus, dass 768 Dimensionen ausreichen, was den Index-Speicherplatz um 50% reduzierte, ohne signifikante Genauigkeitseinbußen.
Der technische Support reagierte innerhalb von Stunden auf meine Fragen zur Dimensionsoptimierung. Die Integration von WeChat-Payment für die Abrechnung eliminierte die Currency-Conversion-Probleme, die wir zuvor mit anderen Anbietern hatten.
Nach 6 Monaten Produktionsbetrieb kann ich bestätigen: HolySheep AI liefert konsistent unter 50ms Latenz, selbst zu Spitzenlastzeiten. Unsere Retrieval-Genauigkeit verbesserte sich sogar um 3.2% durch die bessere Dimensionskalibrierung.
Häufige Fehler und Lösungen
Fehler 1: Dimension-Mismatch bei Index-Abfrage
# FEHLERHAFT: Inkonsistente Dimensionen zwischen Index und Query
query_embedding = client.create_embedding("Suchanfrage", dimensions=768)
query_embedding = query_embedding.reshape(1, -1) # Falsch!
In vielen Tutorials wird reshape empfohlen, aber das ist falsch
wenn der Index 1536-Dimensionen erwartet
KORREKTUR: Strikte Dimensionsvalidierung
from typing import Optional
class ValidatedEmbeddingIndex:
"""Embedding-Index mit strikter Dimensionsvalidierung"""
def __init__(self, dimensions: int):
self.dimensions = dimensions
self.vectors = []
self.metadatas = []
def validate_dimension(func: Callable) -> Callable:
"""Decorator für Dimensionsvalidierung"""
def wrapper(self, *args, **kwargs):
embedding = args[0] if args else None
if embedding is not None:
if not isinstance(embedding, np.ndarray):
raise TypeError("Embedding muss NumPy-Array sein")
if embedding.shape[0] != self.dimensions:
raise ValueError(
f"Dimension-Mismatch: Index erwartet {self.dimensions}, "
f"Embedding hat {embedding.shape[0]} Dimensionen. "
f"Normalisieren Sie mit einem kompatiblen Modell oder "
f"erstellen Sie einen neuen Index mit {embedding.shape[0]} Dimensionen."
)
# Normalisieren für bessere Kosinus-Ähnlichkeit
norm = np.linalg.norm(embedding)
if norm > 0:
embedding = embedding / norm
return func(self, *args, **kwargs)
return wrapper
@validate_dimension
def add_vector(self, embedding: np.ndarray, metadata: dict):
"""Fügt Vektor mit Dimensionsvalidierung hinzu"""
self.vectors.append(embedding)
self.metadatas.append(metadata)
@validate_dimension
def search(self, query_embedding: np.ndarray, k: int = 10):
"""Sucht mit garantierter Dimensionskompatibilität"""
if len(self.vectors) == 0:
return []
vectors_matrix = np.array(self.vectors)
similarities = np.dot(vectors_matrix, query_embedding)
top_k_indices = np.argsort(similarities)[-k:][::-1]
return [
{
"index": idx,
"score": float(similarities[idx]),
"metadata": self.metadatas[idx]
}
for idx in top_k_indices
]
Usage-Beispiel
index = ValidatedEmbeddingIndex(dimensions=1536)
Korrekter Embedding-Prozess
embedding = client.create_embedding("Versicherungsvertrag", dimensions=1536)
try:
index.add_vector(embedding, {"doc_id": "V001", "text": "..."})
results = index.search(embedding, k=5)
print(f"Gefunden: {len(results)} relevante Dokumente")
except ValueError as e:
print(f"Validierungsfehler: {e}")
# Lösung: Index mit korrekter Dimension neu erstellenFehler 2: Batch-Size-Überschreitung
# FEHLERHAFT: Unbegrenzte Batch-Größen
all_texts = load_documents() # 100.000 Dokumente
embeddings = client.batch_embeddings(all_texts) # CRASH bei Timeout
KORREKTUR: Robuste Batch-Verarbeitung mit Fortschritt
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
class RobustBatchProcessor:
"""Robuste Batch-Verarbeitung mit Fehlerbehandlung"""
MAX_BATCH_SIZE = 100 # HolySheep Limit
RETRY_ATTEMPTS = 3
RETRY_DELAY = 2 # Sekunden
def __init__(self, client, max_workers: int = 4):
self.client = client
self.max_workers = max_workers
def process_large_corpus(
self,
texts: list[str],
dimensions: int = 1536,
progress_callback=None
) -> list[np.ndarray]:
"""
Verarbeitet große Korpora mit automatischer Batch-Aufteilung.
Args:
texts: Liste aller zu verarbeitenden Texte
dimensions: Vektor-Dimension
progress_callback: Optional, wird mit (processed, total) aufgerufen
Returns:
Liste aller Embedding-Vektoren
"""
total = len(texts)
all_embeddings = []
processed = 0
failed_batches = []
# Chunking in verdauliche Batches
chunks = [
texts[i:i + self.MAX_BATCH_SIZE]
for i in range(0, total, self.MAX_BATCH_SIZE)
]
print(f"Verarbeite {total} Dokumente in {len(chunks)} Batches...")
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_to_batch = {
executor.submit(
self._process_single_batch,
chunk,
dimensions,
batch_num
): batch_num
for batch_num, chunk in enumerate(chunks)
}
for future in as_completed(future_to_batch):
batch_num = future_to_batch[future]
try:
embeddings, success = future.result()
if success:
all_embeddings.extend(embeddings)
processed += len(embeddings)
if progress_callback:
progress_callback(processed, total)
else:
print(f"Batch {batch_num+1}/{len(chunks)} "
f"abgeschlossen ({processed}/{total})")
else:
failed_batches.append(batch_num)
logging.warning(f"Batch {batch_num} fehlgeschlagen")
except Exception as e:
failed_batches.append(batch_num)
logging.error(f"Batch {batch_num} Exception: {e}")
# Retry fehlgeschlagener Batches
if failed_batches:
print(f"\nWiederhole {len(failed_batches)} fehlgeschlagene Batches...")
retry_embeddings = self._retry_failed_batches(
texts, chunks, failed_batches, dimensions
)
all_embeddings.extend(retry_embeddings)
return all_embeddings
def _process_single_batch(
self,
batch: list[str],
dimensions: int,
batch_num: int
) -> tuple[list[np.ndarray], bool]:
"""Verarbeitet einen einzelnen Batch mit Retry-Logik"""
for attempt in range(self.RETRY_ATTEMPTS):
try:
embeddings = self.client.batch_embeddings(
batch,
dimensions=dimensions,
batch_size=len(batch)
)
return embeddings, True
except TimeoutError:
if attempt < self.RETRY_ATTEMPTS - 1:
import time
time.sleep(self.RETRY_DELAY * (attempt + 1))
logging.warning(
f"Batch {batch_num} Timeout, Retry {attempt + 1}"
)
else:
return [], False
except ConnectionError as e:
if "429" in str(e): # Rate Limit
import time
time.sleep(60) # Warte 1 Minute
# Retry
else:
return [], False
return [], False
def _retry_failed_batches(
self,
texts: list[str],
chunks: list[list[str]],
failed_indices: list[int],
dimensions: int
) -> list[np.ndarray]:
"""Wiederholt fehlgeschlagene Batches mit kleinerer Größe"""
retry_embeddings = []
smaller_batch_size = self.MAX_BATCH_SIZE // 4
for batch_num in failed_indices:
batch = chunks[batch_num]
sub_batches = [
batch[i:i + smaller_batch_size]
for i in range(0, len(batch), smaller_batch_size)
]
for sub_batch in sub_batches:
for attempt in range(2):
try:
embeddings = self.client.batch_embeddings(
sub_batch,
dimensions=dimensions,
batch_size=len(sub_batch)
)
retry_embeddings.extend(embeddings)
break
except Exception:
if attempt == 1:
logging.error(
f"Kritischer Fehler bei Batch {batch_num}"
)
return retry_embeddings
Usage
processor = RobustBatchProcessor(client, max_workers=4)
try:
all_embeddings = processor.process_large_corpus(
documents,
dimensions=1536,
progress_callback=lambda p, t: print(f"\rFortschritt: {p/t*100:.1f}%", end="")
)
print(f"\n✓ Erfolgreich: {len(all_embeddings)} Embeddings")
except Exception as e:
print(f"\n✗ Kritischer Fehler: {e}")Fehler 3: Falsche Normalisierung bei Ähnlichkeitsberechnung
# FEHLERHAFT: Unnormalisierte Vektoren führen zu verzerrten Ergebnissen
def bad_similarity(v1, v2):
return np.dot(v1, v2) # Fehler: nicht-normalisierte Vektoren
KORREKTUR: Korrekte Normalisierung und Similarity
from numpy.linalg import norm
class SemanticSearchEngine:
"""Semantische Suchmaschine mit korrekter Vektor-Normalisierung"""
def __init__(self, embedding_client, dimensions: int = 1536):
self.client = embedding_client
self.dimensions = dimensions
self.document_embeddings = {}
self.corpus_embeddings = None
self.corpus_texts = []
@staticmethod
def normalize_vector(vector: np.ndarray) -> np.ndarray:
"""
L2-Normalisierung für Kosinus-Ähnlichkeit.
Normalisierte Vektoren haben Einheitslänge:
||v|| = 1
Dadurch wird:
cos_similarity(a, b) = a · b / (||a|| × ||b||) = a · b
"""
v_norm = norm(vector)
if v_norm == 0:
raise ValueError(
"Nullvektor erkannt! Überprüfen Sie die Eingabedaten. "
"Leere Strings oder ungültige Tokens erzeugen Nullvektoren."
)
return vector / v_norm
def index_documents(self, documents: list[dict]):
"""
Indiziert Dokumente mit korrekter Normalisierung.
Args:
documents: Liste von Dict mit 'id' und 'text'
"""
print(f"Indiziere {len(documents)} Dokumente...")
texts = [doc["text"] for doc in documents]
ids = [doc["id"] for doc in documents]
# Batch-Embeddings erstellen
raw_embeddings = self.client.batch_embeddings(
texts,
dimensions=self.dimensions
)
# Korrekte Normalisierung
normalized_embeddings = [
self.normalize_vector(emb) for emb in raw_embeddings
]
# Index aufbauen
self.corpus_embeddings = np.array(normalized_embeddings)
self.corpus_texts = texts
print(f"✓ Indiziert: {len(documents)} normalisierte Vektoren")
print(f" Embedding-Dimension: {self.dimensions}")
print(f" Speicher: {self.corpus_embeddings.nbytes / 1024 / 1024:.2f} MB")
def search(
self,
query: str,
k: int = 10,
min_score: float = 0.0
) -> list[dict]:
"""
Semantische Suche mit normalisierten Vektoren.
Args:
query: Suchanfrage
k: Anzahl der Ergebnisse
min_score: Mindest-Ähnlichkeitsscore
Returns:
Liste von Dict mit 'id', 'text', 'score'
"""
# Query embedding
raw_query = self.client.create_embedding(
query,
dimensions=self.dimensions
)
query_normalized = self.normalize_vector(raw_query)
# Effiziente Ähnlichkeitsberechnung mit normalisierten Vektoren
# Bei normalisierten Vektoren: Kosinus-Ähnlichkeit = Skalarprodukt
similarities = np.dot(
self.corpus_embeddings,
query_normalized
)
# Top-k Indizes
top_k_indices = np.argsort(similarities)[-k:][::-1]
results = []
for idx in top_k_indices:
score = float(similarities[idx])
if score >= min_score:
results.append({
"index": int(idx),
"score": score,
"text": self.corpus_texts[idx][:100] + "..."
})
return results
def batch_search(
self,
queries: list[str],
k: int = 5
) -> list[list[dict]]:
"""Führt mehrere Suchanfragen parallel aus"""
all_results = []
for query in queries:
results = self.search(query, k=k)
all_results.append(results)
return all_results
Usage mit Fehlerbehandlung
engine = SemanticSearchEngine(client, dimensions=1536)
Dokumente indizieren
documents = [
{"id": "doc1", "text": "Maschinelles Lernen und künstliche Intelligenz"},
{"id": "doc2", "text": "Neuronale Netzwerke für Deep Learning"},
{"id": "doc3", "text": "Kochen mit regionalen Zutaten"},
]
try:
engine.index_documents(documents)
# Suche
results = engine.search("Künstliche neuronale Netze", k=3)
print("\nSuchergebnisse:")
for r in results:
print(f" Score: {r['score']:.4f} | {r['text']}")
except ValueError as e:
print(f"Datenfehler: {e}")
# Lösung: Überprüfen Sie die DokumenttexteOptimierungsstrategien für Produktionsumgebungen
Caching-Strategie für wiederholte Queries
import hashlib
import json
from functools import lru_cache
import redis
class CachedEmbeddingClient:
"""Embedding-Client mit Redis-Caching für Produktion"""
def __init__(self, client, redis_url: str = "redis://localhost:6379"):
self.client = client
self.cache_hits = 0
self.cache_misses = 0
try:
self.redis = redis.from_url(redis_url)
self.redis.ping()
self.use_cache = True
except:
self.use_cache = False
print("Redis nicht verfügbar, Cache deaktiviert")
def _get_cache_key(self, text: str, dimensions: int) -> str:
"""Generiert konsistenten Cache-Key"""
content = f"{text
Verwandte Ressourcen
Verwandte Artikel