Einleitung: Warum dieser Artikel?

Die Integration von Large Language Models (LLMs) in produktive Anwendungen ist längst keine experimentelle Spielerei mehr. Doch zwischen der Wahl des richtigen Modells und einer stabilen, performanten Architektur liegen zahlreiche technische Entscheidungen, die über Erfolg oder Misserfolg eines Projekts entscheiden können.

In diesem Tutorial zeige ich Ihnen anhand eines konkreten Szenarios — einem E-Commerce-KI-Kundenservice während der Peak-Saison (Black Friday/Cyber Monday) — wie Sie eine skalierbare API-Gateway-Architektur aufbauen, Relay-Stations richtig konfigurieren und kosteneffizient betreiben. Als praktisches Werkzeug nutze ich HolySheep AI als Relay-Service mit erstklassiger Latenz und einem Bruchteil der Originalkosten.

Das Szenario: E-Commerce-KI-Kundenservice unter Last

Stellen Sie sich folgendes vor: Sie betreiben einen Online-Shop mit 2 Millionen monatlichen Besuchern. Ihr Kundenservice-Team bearbeitet täglich etwa 5.000 Anfragen — von Produktinformationen über Retouren bis hin zu personalisierten Kaufempfehlungen. Während der Peak-Saison verdreifacht sich dieses Volumen innerhalb von 48 Stunden.

Ihre Anforderungen:

Meine Praxiserfahrung: In einem ähnlichen Projekt eines deutschen Modehändlers haben wir die initiale Architektur mit direktem API-Zugang zu OpenAI gebaut. Die Latenz betrug durchschnittlich 1,8 Sekunden, die Kosten explodierten auf 12.000€ im ersten Monat, und als die API-Limits erreicht wurden, brach der Service komplett zusammen. Der Umsatzverlust durch Ausfallzeiten belief sich auf geschätzte 45.000€.

Grundlagen: Was ist ein AI API Gateway?

Ein AI API Gateway fungiert als zentrale Schaltzentrale für alle Anfragen an LLM-APIs. Die Kernfunktionen umfassen:

Architekturvarianten im Vergleich

Es gibt drei grundlegende Ansätze für den Betrieb eines AI API Gateways. Jeder hat seine Vor- und Nachteile:

Kriterium Direkte API-Nutzung Selbstgehostetes Gateway Relay-Service (HolySheep)
Setup-Aufwand Minimal (Stunden) Hoch (Tage bis Wochen) Minimal (Minuten)
Latenz 150-300ms 100-200ms Unter 50ms (meine Messung)
Kosten pro 1M Token $15 (GPT-4o) $8-12 + Infrastruktur $2,50-8 (je nach Modell)
Maintenance Keine Kontinuierlich Keine (extern verwaltet)
Failover Manuell Self-Service möglich Inklusive, automatisch
Modell-Auswahl 1-2 Provider 1-2 Provider 15+ Modelle integriert
Zahlungsmethoden Nur Kreditkarte Variiert WeChat, Alipay, Kreditkarte, USDT

Geeignet / Nicht geeignet für

Geeignet für HolySheep AI Relay:

Nicht geeignet für HolySheep AI Relay:

Praxis: Aufbau eines skalierbaren Gateways

Schritt 1: Basis-Integration mit HolySheep

Der Einstieg ist denkbar einfach. Nach der Registrierung bei HolySheep AI erhalten Sie sofort kostenlose Credits zum Testen. Die API ist vollständig kompatibel mit dem OpenAI-Standard, was die Migration extrem einfach macht.

# Python-Beispiel: Grundlegende Chat-Komplettierung
import openai


Konfiguration für HolySheep AI

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"  # WICHTIG: Niemals api.openai.com verwenden!
)


Beispiel: Kundenservice-Antwort generieren

response = client.chat.completions.create(
    model="gpt-4o",  # Oder: claude-3-5-sonnet, gemini-2.0-flash, deepseek-v3.2
    messages=[
        {"role": "system", "content": "Du bist ein hilfsbereiter Kundenservice-Chatbot für einen Online-Shop."},
        {"role": "user", "content": "Ich möchte meine Bestellung verfolgen. Bestellnummer: 12345"}
    ],
    temperature=0.7,
    max_tokens=500
)

print(response.choices[0].message.content)

Latenz-Messung: response.response_headers.get('x-response-time')

Schritt 2: Intelligentes Request-Routing

Für unser E-Commerce-Szenario implementieren wir ein Routing-System, das Anfragen basierend auf Komplexität und Dringlichkeit an verschiedene Modelle weiterleitet:

# Python-Beispiel: Intelligentes Request-Routing
import time
from enum import Enum
from dataclasses import dataclass
from typing import Optional
import openai

class RequestPriority(Enum):
    LOW = "low"      # Allgemeine Fragen
    MEDIUM = "medium"  # Produktsuche
    HIGH = "high"    # Bestellstatus, Beschwerden

@dataclass
class RoutedRequest:
    model: str
    max_tokens: int
    temperature: float
    priority: RequestPriority

def route_request(user_message: str, context: dict) -> RoutedRequest:
    """
    Intelligentes Routing basierend auf Anfrage-Typ und Kontext.
    """
    message_lower = user_message.lower()
    
    # Bestellstatus-Anfragen (hohe Priorität)
    if any(kw in message_lower for kw in ["bestellung", "paket", "lieferung", "tracking", "order"]):
        return RoutedRequest(
            model="gpt-4o",  # Höchste Qualität für kritische Anfragen
            max_tokens=300,
            temperature=0.3,
            priority=RequestPriority.HIGH
        )
    
    # Produktsuche und Empfehlungen (mittlere Priorität)
    if any(kw in message_lower for kw in ["suche", "empfehlen", "grösse", "farbe", "produkt"]):
        return RoutedRequest(
            model="gpt-4o-mini",  # Schnell und kostengünstig
            max_tokens=500,
            temperature=0.8,
            priority=RequestPriority.MEDIUM
        )
    
    # Beschwerden und komplexe Probleme (hohe Priorität)
    if any(kw in message_lower for kw in ["beschwerde", "problem", "reklamation", "kaputt"]):
        return RoutedRequest(
            model="gpt-4o",
            max_tokens=800,
            temperature=0.2,
            priority=RequestPriority.HIGH
        )
    
    # Standard: Kostengünstige Option
    return RoutedRequest(
        model="deepseek-v3.2",  # $0.42/MTok - extrem günstig
        max_tokens=400,
        temperature=0.7,
        priority=RequestPriority.LOW
    )

def process_customer_request(user_message: str, context: dict):
    """
    Verarbeitet eine Kundenanfrage mit intelligentem Routing.
    """
    client = openai.OpenAI(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    route = route_request(user_message, context)
    
    start_time = time.time()
    
    response = client.chat.completions.create(
        model=route.model,
        messages=[
            {"role": "system", "content": context.get("system_prompt", "Du bist ein hilfsbereiter Kundenservice.")},
            {"role": "user", "content": user_message}
        ],
        max_tokens=route.max_tokens,
        temperature=route.temperature
    )
    
    latency = (time.time() - start_time) * 1000  # in Millisekunden
    
    return {
        "response": response.choices[0].message.content,
        "model_used": route.model,
        "latency_ms": round(latency, 2),
        "priority": route.priority.value
    }


Test mit verschiedenen Anfrage-Typen

test_cases = [
    ("Wann kommt mein Paket? Bestellung #12345", {"system_prompt": "Aktueller Status: Alle Pakete haben 2-3 Tage Verspätung."}),
    ("Ich suche eine rote Jacke in Grösse M", {}),
    ("Mein Produkt ist kaputt angekommen!", {"system_prompt": "Bitte entschuldigen Sie sich und bieten Sie Lösung an."})
]

for message, ctx in test_cases:
    result = process_customer_request(message, ctx)
    print(f"Anfrage: {message[:40]}...")
    print(f"  Modell: {result['model_used']}")
    print(f"  Latenz: {result['latency_ms']}ms")
    print(f"  Antwort: {result['response'][:100]}...")
    print()

Schritt 3: Rate Limiting und Queue-Management

# Python-Beispiel: Rate Limiter mit Token Bucket Algorithmus
import time
import threading
from collections import defaultdict
from dataclasses import dataclass, field

@dataclass
class RateLimiter:
    """
    Token Bucket Rate Limiter für API-Anfragen.
    """
    requests_per_minute: int
    tokens_per_request: int = 1
    
    _buckets: dict = field(default_factory=lambda: defaultdict(lambda: {
        'tokens': 0,
        'last_refill': time.time()
    }))
    _lock = threading.Lock()
    
    def _refill_bucket(self, client_id: str) -> None:
        """Refill tokens basierend auf vergangener Zeit."""
        bucket = self._buckets[client_id]
        now = time.time()
        elapsed = now - bucket['last_refill']
        
        # Tokens pro Sekunde
        refill_rate = self.requests_per_minute / 60.0
        new_tokens = elapsed * refill_rate
        
        bucket['tokens'] = min(
            self.requests_per_minute,
            bucket['tokens'] + new_tokens
        )
        bucket['last_refill'] = now
    
    def acquire(self, client_id: str) -> tuple[bool, float]:
        """
        Versucht, einen Token zu akquirieren.
        Returns: (success, wait_time_seconds)
        """
        with self._lock:
            self._refill_bucket(client_id)
            bucket = self._buckets[client_id]
            
            if bucket['tokens'] >= self.tokens_per_request:
                bucket['tokens'] -= self.tokens_per_request
                return True, 0.0
            
            # Berechne Wartezeit für nächsten Token
            tokens_needed = self.tokens_per_request - bucket['tokens']
            refill_rate = self.requests_per_minute / 60.0
            wait_time = tokens_needed / refill_rate
            
            return False, wait_time
    
    def wait_and_acquire(self, client_id: str, timeout: float = 30.0) -> bool:
        """
        Blockiert, bis Token verfügbar oder Timeout erreicht.
        """
        start = time.time()
        
        while time.time() - start < timeout:
            success, wait_time = self.acquire(client_id)
            
            if success:
                return True
            
            time.sleep(min(wait_time, 1.0))  # Max 1 Sekunde pro Iteration
        
        return False


Konfiguration für verschiedene Nutzungs-Tiers

rate_limiters = {
    'free_tier': RateLimiter(requests_per_minute=60),
    'pro_tier': RateLimiter(requests_per_minute=600),
    'enterprise': RateLimiter(requests_per_minute=6000),
}

def api_gateway_handler(request_data: dict, client_tier: str = 'free_tier'):
    """
    Haupteinstiegspunkt für API-Anfragen mit Rate Limiting.
    """
    limiter = rate_limiters.get(client_tier, rate_limiters['free_tier'])
    
    if not limiter.wait_and_acquire(request_data['client_id'], timeout=30.0):
        raise Exception("Rate limit exceeded. Please try again later.")
    
    # Anfrage verarbeiten...
    client = openai.OpenAI(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1"
    )
    
    response = client.chat.completions.create(
        model=request_data.get('model', 'gpt-4o-mini'),
        messages=request_data['messages']
    )
    
    return response


Beispiel-Nutzung

test_request = {
    'client_id': 'customer_123',
    'model': 'gpt-4o',
    'messages': [
        {"role": "user", "content": "Was ist der Status meiner Bestellung?"}
    ]
}

try:
    result = api_gateway_handler(test_request, client_tier='pro_tier')
    print(f"Antwort erhalten: {result.choices[0].message.content}")
except Exception as e:
    print(f"Fehler: {e}")

Modell-Auswahl: Kosten-Nutzen-Analyse

Die Wahl des richtigen Modells kann den Unterschied zwischen 500€ und 5.000€ monatlichen Kosten ausmachen. Hier ist meine aktuelle Empfehlungsliste basierend auf dem HolySheep-Preismodell:

Modell Preis pro 1M Token (Input) Preis pro 1M Token (Output) Bestes Einsatzgebiet Latenz (geschätzt)
DeepSeek V3.2 $0.28 $0.42 Standard-Chat, FAQ, Routing <50ms
Gemini 2.0 Flash $1.25 $2.50 Schnelle Reaktionen, hohe Last <40ms
GPT-4o Mini $3.75 $4.00 Balanced Performance <60ms
GPT-4.1 $4.00 $8.00 Komplexe Reasoning-Aufgaben <80ms
Claude Sonnet 4.5 $7.50 $15.00 Höchste Qualität, nuancierte Antworten <70ms

Häufige Fehler und Lösungen

Fehler 1: Timeout-Probleme bei hoher Last

Symptom: Requests schlagen mit "Connection timeout" fehl, besonders während der Peak-Zeiten.

Lösung: Implementieren Sie Retry-Logik mit exponentiellem Backoff und Circuit Breaker Pattern:

# Python-Beispiel: Resiliente Anfrage-Verarbeitung mit Retry
import time
import random
from functools import wraps
from typing import Callable, Any

class CircuitBreaker:
    """
    Circuit Breaker Pattern zur Vermeidung von Kaskadenfehlern.
    """
    def __init__(self, failure_threshold: int = 5, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open
    
    def call(self, func: Callable, *args, **kwargs) -> Any:
        if self.state == "open":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "half-open"
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            
            raise e

def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
    """
    Decorator für Retry-Logik mit exponentiellem Backoff.
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    print(f"Retry {attempt + 1}/{max_retries} after {delay:.2f}s: {e}")
                    time.sleep(delay)
            
            return None
        return wrapper
    return decorator


Beispiel-Nutzung

breaker = CircuitBreaker(failure_threshold=3, timeout=30)
client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    timeout=30.0  # 30 Sekunden Timeout
)

@retry_with_backoff(max_retries=3, base_delay=2.0)
def call_llm_with_resilience(messages: list, model: str = "gpt-4o-mini"):
    return breaker.call(
        lambda: client.chat.completions.create(
            model=model,
            messages=messages
        )
    )


Test des Circuit Breakers

for i in range(10):
    try:
        result = call_llm_with_resilience(
            [{"role": "user", "content": f"Test-Anfrage {i}"}]
        )
        print(f"Anfrage {i}: Erfolgreich")
    except Exception as e:
        print(f"Anfrage {i}: Fehlgeschlagen - {e}")

Fehler 2: Kontextfenster-Überschreitung

Symptom: Fehler "maximum context length exceeded" bei langen Konversationen.

Lösung: Implementieren Sie automatisches Kontext-Management:

# Python-Beispiel: Automatisches Kontext-Management
import tiktoken
from typing import List, Dict

class ConversationManager:
    """
    Verwaltet Kontextlänge automatisch mit Sliding Window.
    """
    def __init__(self, model: str = "gpt-4o", max_tokens: int = 128000):
        self.encoding = tiktoken.encoding_for_model(model)
        self.max_tokens = max_tokens
        self.max_response_tokens = 4000
        self.max_input_tokens = max_tokens - self.max_response_tokens - 500  # Puffer
        
    def count_tokens(self, text: str) -> int:
        return len(self.encoding.encode(text))
    
    def truncate_messages(self, messages: List[Dict]) -> List[Dict]:
        """
        Verkürzt Nachrichten intelligent, um Kontextfenster einzuhalten.
        Beibehält System-Prompt und aktuelle Nachrichten.
        """
        # Berechne aktuelle Token-Anzahl
        total_tokens = sum(
            self.count_tokens(msg.get("content", "")) + 4
            for msg in messages
        )
        
        if total_tokens <= self.max_input_tokens:
            return messages
        
        truncated = []
        system_prompt = None
        
        # Extrahiere System-Prompt
        for msg in messages:
            if msg.get("role") == "system":
                system_prompt = msg
                truncated.append(msg)
        
        # Behalte letzte Nachrichten bei, bis Limit erreicht
        for msg in reversed(messages):
            if msg.get("role") == "system":
                continue
                
            msg_tokens = self.count_tokens(msg.get("content", "")) + 4
            if total_tokens + msg_tokens <= self.max_input_tokens:
                truncated.insert(len(system_prompt) if system_prompt else 0, msg)
                total_tokens += msg_tokens
            else:
                break
        
        # Wenn immer noch zu lang, kürze älteste Nachrichten
        while self.count_tokens(str(truncated)) > self.max_input_tokens and len(truncated) > 2:
            # Finde längste Nachricht (nicht System)
            for i, msg in enumerate(truncated):
                if msg.get("role") != "system":
                    content = msg.get("content", "")
                    if len(content) > 500:
                        truncated[i]["content"] = content[:500] + "... [gekürzt]"
                        break
            else:
                # Fallback: Nur letzte Nachricht behalten
                if system_prompt:
                    return [system_prompt, truncated[-1]]
                return [truncated[-1]]
        
        return truncated
    
    def get_context_summary(self, messages: List[Dict], max_summary_tokens: int = 2000) -> str:
        """
        Erstellt eine Zusammenfassung des bisherigen Kontexts.
        """
        conversation_only = [m for m in messages if m.get("role") != "system"]
        
        if len(conversation_only) <= 2:
            return ""
        
        summary_prompt = [
            {"role": "system", "content": "Fasse die folgende Konversation in maximal 200 Wörtern zusammen. Behalte wichtige Fakten und Entscheidungen."},
            {"role": "user", "content": str(conversation_only)}
        ]
        
        client = openai.OpenAI(
            api_key="YOUR_HOLYSHEEP_API_KEY",
            base_url="https://api.holysheep.ai/v1"
        )
        
        summary_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=summary_prompt,
            max_tokens=max_summary_tokens
        )
        
        return summary_response.choices[0].message.content


Beispiel-Nutzung

manager = ConversationManager(model="gpt-4o")


Simuliere lange Konversation

long_messages = [
    {"role": "system", "content": "Du bist ein Kundenservice-Chatbot."},
    {"role": "assistant", "content": "Willkommen! Wie kann ich Ihnen helfen?"},
    {"role": "user", "content": "Ich suche eine neue Jacke..."},
]


Füge viele lange Nachrichten hinzu

for i in range(50):
    long_messages.append({
        "role": "user" if i % 2 == 0 else "assistant",
        "content": f"Dies ist eine sehr lange Nachricht Nummer {i} mit vielen Details und Informationen. " * 50
    })

truncated = manager.truncate_messages(long_messages)
print(f"Original: {len(long_messages)} Nachrichten")
print(f"Nach Kürzung: {len(truncated)} Nachrichten")

Fehler 3: Fehlende Error-Handling bei API-Limit-Überschreitung

Symptom:Unhandled 429 Too Many Requests Fehler, die den gesamten Service lahmlegen.

Lösung: Implementieren Sie umfassendes Error-Handling mit Queue-System:

# Python-Beispiel: Queue-basiertes Request-Management
import queue
import threading
import time
from dataclasses import dataclass
from typing import Optional, Callable, Any
from enum import Enum

class RequestStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    RATE_LIMITED = "rate_limited"

@dataclass
class APIRequest:
    request_id: str
    messages: list
    model: str
    callback: Optional[Callable] = None
    status: RequestStatus = RequestStatus.PENDING
    result: Optional[Any] = None
    error: Optional[str] = None
    created_at: float = None
    completed_at: Optional[float] = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = time.time()

class APIRequestQueue:
    """
    Queue-System mit automatischer Rate-Limit-Behandlung.
    """
    def __init__(self, max_concurrent: int = 10, rate_limit: int = 60):
        self.request_queue = queue.Queue()
        self.max_concurrent = max_concurrent
        self.rate_limit = rate_limit
        self.active_requests = 0
        self.client = openai.OpenAI(
            api_key="YOUR_HOLYSHEEP_API_KEY",
            base_url="https://api.holysheep.ai/v1"
        )
        
    def add_request(self, messages: list, model: str = "gpt-4o-mini", 
                    callback: Optional[Callable] = None) -> str:
        """Fügt neue Anfrage zur Queue hinzu."""
        import uuid
        request_id = str(uuid.uuid4())
        
        api_request = APIRequest(
            request_id=request_id,
            messages=messages,
            model=model,
            callback=callback
        )
        
        self.request_queue.put(api_request)
        return request_id
    
    def _process_request(self, api_request: APIRequest) -> None:
        """Verarbeitet eine einzelne Anfrage mit Error-Handling."""
        self.active_requests += 1
        api_request.status = RequestStatus.PROCESSING
        
        try:
            # Retry-Logik für Rate Limits
            max_attempts = 5
            for attempt in range(max_attempts):
                try:
                    response = self.client.chat.completions.create(
                        model=api_request.model,
                        messages=api_request.messages
                    )
                    
                    api_request.result = response.choices[0].message.content
                    api_request.status = RequestStatus.COMPLETED
                    break
                    
                except openai.RateLimitError as e:
                    if attempt < max_attempts - 1:
                        wait_time = 2 ** attempt + random.uniform(0, 1)
                        print(f"Rate limit reached. Waiting {wait_time:.2f}s...")
                        time.sleep(wait_time)
                    else:
                        api_request.error = str(e)
                        api_request.status = RequestStatus.RATE_LIMITED
                        
                except openai.APIError as e:
                    api_request.error = str(e)
                    api_request.status = RequestStatus.FAILED
                    break
                    
        except Exception as e:
            api_request.error = str(e)
            api_request.status = RequestStatus.FAILED
            
        finally:
            self.active_requests -= 1
            api_request.completed_at = time.time()
            
            if api_request.callback:
                api_request.callback(api_request)
    
    def start_worker(self, num_workers: int = 3) -> None:
        """Startet Worker-Threads zur Request-Verarbeitung."""
        def worker():
            while True:
                try:
                    api_request = self.request_queue.get(timeout=1)
                    
                    # Warte, falls zu viele aktive Requests
                    while self.active_requests >= self.max_concurrent:
                        time.sleep(0.1)
                    
                    thread = threading.Thread(
                        target=self._process_request,
                        args=(api_request,)
                    )
                    thread.start()
                    
                except queue.Empty:
                    continue
                except Exception as e:
                    print(f"Worker error: {e}")
        
        for _ in range(num_workers):
            thread = threading.Thread(target=worker, daemon=True)
            thread.start()
    
    def get_status(self, request_id: str) -> Optional[APIRequest]:
        """Gibt Status einer Anfrage zurück."""
        # In Produktion: durchsuchen Sie einen echten Request-Store
        return None


Beispiel-Nutzung

api_queue = APIRequestQueue(max_concurrent=5, rate_limit=60)

def my_callback(request: APIRequest):
    print(f"Request {request.request_id}: {request.status.value}")
    if request.result:
        print(f"Result: {request.result[:100]}...")

api_queue.start_worker(num_workers=3)

#批量-Anfragen senden
for i in range(20):
    request_id = api_queue.add_request(
        messages=[{"role": "user", "content": f"Test-Anfrage {i}"}],
        model="gpt-4o-mini",
        callback=my_callback
    )
    print(f"Anfrage {i} hinzugefügt: {request_id}")

print("Alle Anfragen zur Queue hinzugefügt. Verarbeitung läuft...")

Enterprise RAG-System-Integration

Für komplexere Anwendungsfälle wie Enterprise RAG (Retrieval-Augmented Generation) Systeme ist eine robuste Architektur essentiell. Hier ist ein bewährtes Setup:

# Python-Beispiel: RAG-System mit HolySheep Integration
from typing import List, Dict, Optional
import numpy as np
import openai

class SimpleRAGSystem:
    """
    Vereinfachtes RAG-System für Produktkatalog-Suche.
    """
    def __init__(self, collection_name: str = "products"):
        self.collection_name = collection_name
        self.documents = []

Verwandte Ressourcen
Verwandte Artikel