KI-Engineering-Bereitstellung

Sie haben ein Modell auf Ihrem Laptop ausgeführt, und es generierte schnell und präzise Antworten. Doch wenn Sie es auf einen Server stellen und hundert Benutzer es gleichzeitig aufrufen lassen, ändert sich die Situation:Einige Benutzer warten zehn Sekunden auf eine Antwort, einige Anfragen laufen in ein Timeout und melden Fehler, der GPU-Speicher ist im Nu voll, und die Rechnung steigt auf ein schmerzhaftes Niveau.

Das ist das Problem, das KI-Deployment lösen muss:Ein lauffähiges Modell in einen nutzbaren Dienst verwandeln。

Der Engpass bei traditionellen Webdiensten ist normalerweise die CPU und die Datenbank.

Der Engpass bei KI-Diensten ist hauptsächlich die GPU — ob der Speicher ausreicht, ob die Berechnung schnell ist und wie gleichzeitige Anfragen in eine Warteschlange gestellt werden.

Die zentralen Herausforderungen des KI-Deployments lassen sich in drei Worten zusammenfassen: Latenz, Durchsatz und Kosten.

Metrik	Bedeutung	Benutzererfahrung
Latenz	Zeit vom Senden einer Anfrage durch einen Benutzer bis zum Empfang des ersten Zeichens	Schnell oder nicht
Durchsatz	Wie viele Anfragen pro Sekunde verarbeitet werden können	Ob viele Personen gleichzeitig bedient werden können
Kosten	GPU-/Serverkosten für den Betrieb des Dienstes	Teuer oder nicht

Gutes KI-Deployment bedeutet, ein Gleichgewicht zwischen diesen drei Aspekten zu finden: niedrig genug Latenz, hoch genug Durchsatz und kontrollierbare Kosten.

Modell-Serving-Framework

Um ein trainiertes Modell als API-Dienst zu verpacken, benötigen Sie ein spezielles Framework. Dieser Abschnitt stellt drei gängige Optionen vor: vLLM, TGI und Ollama.

vLLM: Hochleistungsinferenz angetrieben von PagedAttention

vLLM ist eine Inferenz-Engine, die vom Team der UC Berkeley entwickelt wurde, und ihr größtes Merkmal ist Geschwindigkeit.

Die zentrale Innovation von vLLM istPagedAttention– eine Technik zur effizienten Verwaltung des GPU-Speichers, inspiriert vom virtuellen Speicher-Paging von Betriebssystemen.

In traditionellen Inferenz-Frameworks muss der KV-Cache (Key-Value-Cache) jeder Anfrage zusammenhängenden GPU-Speicherplatz belegen.

Wenn Anfragen unterschiedlich lang sind, wird der GPU-Speicher fragmentiert, was zu einer geringen Auslastung führt.

PagedAttention unterteilt den GPU-Speicher in Seiten fester Größe. Der KV-Cache jeder Anfrage kann über verschiedene Seiten verteilt sein, wobei eine Seitentabelle ihre Positionen aufzeichnet.

Dies verbessert die GPU-Speicherauslastung erheblich und ermöglicht die gleichzeitige Bedienung von mehr Anfragen.

Beispiel

# vLLM installieren
pip install vllm

# Einen OpenAI-kompatiblen API-Dienst mit vLLM starten
# --model gibt das Modell an, --host und --port geben die Listen-Adresse an
# --tensor-parallel-size: Anzahl der GPUs für die parallele Inferenz
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--host 0.0.0.0 \
--port 8000 \
--tensor-parallel-size 1 \
--gpu-memory-utilization 0.9 \
--max-model-len 8192

Nachdem der Dienst gestartet ist, können Sie ihn direkt mit dem OpenAI SDK aufrufen:

Beispiel

# Dateipfad: test_vllm_client.py
from openai import OpenAI

# Verbindung zum lokalen vLLM-Dienst herstellen
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="example-demo-key" # vLLM benötigt standardmäßig keinen echten API-Schlüssel
)

# Chat-Completion aufrufen
response = client.chat.completions.create(
model="Qwen/Qwen2.5-7B-Instruct",
messages=[
{"role": "system", "content": "Du bist ein hilfreicher KI-Assistent."},
{"role": "user", "content": "Stelle das Python-Tutorial in einem Satz vor."}
],
temperature=0.7,
max_tokens=500,
stream=True # Streaming-Ausgabe
)

print("KI-Antwort:", end="", flush=True)
for chunk in response:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

vLLM unterstützt außerdem das direkte Schreiben benutzerdefinierter Dienste mit Python-Code:

Beispiel

# Dateipfad: custom_vllm_server.py
from vllm import LLM, SamplingParams

# Modell initialisieren
llm = LLM(
model="Qwen/Qwen2.5-7B-Instruct",
gpu_memory_utilization=0.9, # Verhältnis der GPU-Speichernutzung
tensor_parallel_size=1, # Tensor-Parallelität (Anzahl der GPUs)
max_model_len=8192, # Maximale Kontextlänge
)

# Sampling-Parameter konfigurieren
sampling_params = SamplingParams(
temperature=0.7,
top_p=0.9,
max_tokens=500,
stop=["</s>"],
)

# Batch-Inferenz
prompts = [
"Python vorstellen",
"Was ist KI?",
"Wie lernt man Programmieren?",
]

# Antwort generieren
outputs = llm.generate(prompts, sampling_params)

# Ergebnisse ausgeben
for output in outputs:
prompt = output.prompt
generated_text = output.outputs[0].text
print(f"Prompt: {prompt}")
print(f"Generiert: {generated_text}")
print("-" * 50)

TGI: Hugging Faces Inferenz-Engine

TGI (Text Generation Inference) ist ein von Hugging Face eingeführtes Inferenzdienst-Framework.

Es bietet ein vollständiges Ökosystem, eine tiefe Integration in den Hugging Face Hub und unterstützt Optimierungen wie Flash Attention und dynamisches Batching.

Beispiel

# TGI mit Docker ausführen (der einfachste Weg)
# Hinweis: Sie müssen zuerst Docker installieren und GPU-Unterstützung konfigurieren
docker run -d \
--gpus all \
-p 8080:80 \
-v $PWD/data:/data \
--name tgi-server \
ghcr.io/huggingface/text-generation-inference:latest \
--model-id Qwen/Qwen2.5-7B-Instruct \
--max-input-length 4096 \
--max-total-tokens 8192 \
--max-batch-prefill-tokens 4096

# Oder direkt mit Python installieren (Entwicklungsumgebung)
# pip install text-generation

TGI-Dienst aufrufen:

Beispiel

# Dateipfad: test_tgi_client.py
from text_generation import Client

# Verbindung zum TGI-Dienst herstellen
client = Client("http://localhost:8080", timeout=60)

# Aufruf ohne Streaming
response = client.generate(
"Stelle das Python-Tutorial vor",
max_new_tokens=500,
temperature=0.7,
top_p=0.9,
)
print(response.generated_text)

print("-" * 50)

# Streaming-Aufruf
print("Streaming-Ausgabe:")
for chunk in client.generate_stream(
"Schreibe ein Hello World in Python",
max_new_tokens=200,
):
if not chunk.token.special:
print(chunk.token.text, end="", flush=True)
print()

Ollama: Von der lokalen Demo zur Produktionsbereitstellung

Ollama ist bekannt für "lokales Ausführen von Modellen mit einem Klick", und viele Leute nutzen es für Entwicklung und Demos.

Tatsächlich ist Ollama auch für den Produktionseinsatz in kleinem Maßstab geeignet – es ist einfach, stabil und der Ressourcenverbrauch ist kontrollierbar.

Beispiel

# Ollama installieren (Linux-Server)
curl -fsSL https://ollama.com/install.sh | sh

# Ollama so konfigurieren, dass es auf allen Netzwerkschnittstellen lauscht (in Produktion auf die Firewall achten)
# systemd-Dienstdatei bearbeiten
sudo systemctl edit ollama.service

# Folgenden Inhalt hinzufügen:
# [Service]
# Environment="OLLAMA_HOST=0.0.0.0:11434"
# Environment="OLLAMA_MODELS=/var/lib/ollama/models"

# Dienst neu starten
sudo systemctl restart ollama

# Modell herunterladen und ausführen
ollama pull qwen2.5:7b

Ollama über die API aufrufen:

Beispiel

# Dateipfad: test_ollama_client.py
import requests
import json

# Ollama-API-Adresse
base_url = "http://localhost:11434/api"

# Chat-Schnittstelle (Streaming)
def chat_stream(model, messages):
url = f"{base_url}/chat"
data = {
"model": model,
"messages": messages,
"stream": True,
"options": {
"temperature": 0.7,
"num_ctx": 8192,
}
}

response = requests.post(url, json=data, stream=True)
full_response = ""

for line in response.iter_lines():
if line:
chunk = json.loads(line)
if "message" in chunk:
content = chunk["message"].get("content", "")
full_response += content
print(content, end="", flush=True)
if chunk.get("done", False):
break

print()
return full_response

# Beispielaufruf
messages = [
{"role": "system", "content": "Du bist ein hilfreicher KI-Assistent."},
{"role": "user", "content": "Stelle das Python-Tutorial vor"}
]

print("KI-Antwort:", end="")
result = chat_stream("qwen2.5:7b", messages)

Vergleich der Framework-Auswahl

Die drei Frameworks haben jeweils ihre eigenen Eigenschaften. Wie wählt man aus? Es hängt von Ihren Anforderungen ab.

Framework	Vorteile	Nachteile	Anwendbare Szenarien
vLLM	Stärkste Leistung, höchster Durchsatz	Relativ komplexe Konfiguration	Produktionsumgebungen mit hoher Parallelität, Leistung im Fokus
TGI	Hugging-Face-Ökosystem, einfache Bereitstellung	Leistung etwas geringer als vLLM	Bereits die Hugging-Face-Toolchain verwenden
Ollama	Minimaler Bereitstellungsaufwand, geringe Wartungskosten	Nicht geeignet für extrem hohe Parallelität	Kleine und mittlere Dienste, schneller Start

Frühe Phase: mit Ollama beginnen, einfach und zuverlässig. Wenn der Datenverkehr wächst: zu vLLM migrieren, mit Fokus auf Leistung.

Techniken zur Inferenzoptimierung

Dasselbe Modell kann je nach Ausführungsart bei Geschwindigkeit und Kosten um ein Mehrfaches variieren.

Dieser Abschnitt stellt Inferenzoptimierungstechniken vor, die in Produktionsumgebungen üblicherweise verwendet werden.

Kontinuierliches Batching

Herkömmliches Batching ist statisch: genügend Anfragen sammeln, gemeinsam inferieren und erst nach Abschluss aller den nächsten Batch verarbeiten.

Das Problem: Manche Anfragen sind kurz (z. B. nur 10 Zeichen generieren), andere lang (500 Zeichen generieren).

Kurze Anfragen müssen auf lange Anfragen warten, bevor Ressourcen gemeinsam freigegeben werden können, daher kann die GPU-Auslastung nicht steigen.

Kontinuierliches BatchingEs ist dynamisch: Wenn eine Anfrage abgeschlossen ist, werden sofort neue Anfragen aus der Warteschlange hinzugefügt, ohne auf das Ende des gesamten Batches zu warten.

So ist die GPU fast immer voll ausgelastet, was den Durchsatz deutlich erhöht.

Sowohl vLLM als auch TGI haben integriertes kontinuierliches Batching. Sie müssen es nicht manuell implementieren; aktivieren Sie es einfach.

Spekulatives Decoding

Das Generieren eines Tokens mit einem großen Modell erfordert einen vollständigen Forward-Pass, was langsam ist.

Die Idee des spekulativen Decodings ist:Ein kleines Modell verwendet man, um die nächsten paar Tokens zu "erraten", dann lässt man das große Modell sie alle auf einmal überprüfen.。

Das kleine Modell rät schnell, und obwohl es nicht vollständig korrekt sein muss, kann der korrekt geratene Teil im Batch ausgegeben werden.

Die falschen Vermutungen werden verworfen und neu geraten.

Dies kann die Gesamtgeschwindigkeit um das 2- bis 3-Fache verbessern, während die Ausgabequalität unverändert bleibt.

Beispiel

# Spekulatives Decoding in vLLM aktivieren
# --speculative-model gibt das kleine Modell (Draft-Modell) an
# --num-speculative-tokens gibt an, wie viele Tokens jeweils geraten werden
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--speculative-model Qwen/Qwen2.5-0.5B-Instruct \
--num-speculative-tokens 5 \
--host 0.0.0.0 \
--port 8000

Quantisierte Inferenz: AWQ und GPTQ

Die Modellquantisierung komprimiert Modellgewichte von 16-Bit-Gleitkommazahlen (FP16) auf 8-Bit- oder 4-Bit-Ganzzahlen.

Der Speicherverbrauch wird halbiert oder sogar weniger, und die Inferenz kann schneller sein.

Die zwei am häufigsten verwendeten Quantisierungsmethoden sind AWQ und GPTQ.

Quantisierungsmethode	Merkmal	Genauigkeitsverlust	Inferenzgeschwindigkeit
AWQ	Aktivierungsbewusste Gewichtsquantisierung	Minimal	Schnell
GPTQ	Schichtweise optimierte Quantisierung	gering	Schnell
GGUF	Geeignet für lokale Bereitstellung	Akzeptabel	Schnell

Beispiel

# Ein AWQ-quantisiertes Modell direkt mit vLLM ausführen
# Viele quantisierte Modelle können direkt von Hugging Face heruntergeladen werden
python -m vllm.entrypoints.openai.api_server \
--model TheBloke/Qwen2.5-7B-Instruct-AWQ \
--quantization awq \
--host 0.0.0.0 \
--port 8000

# Oder GGUF-quantisierte Modelle mit Ollama ausführen
ollama run qwen2.5:7b-instruct-q4_0

Ein AWQ-quantisiertes Modell in Python laden:

Beispiel

# Dateipfad: awq_inference.py
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# AWQ-quantisiertes Modell laden
model_name = "TheBloke/Qwen2.5-7B-Instruct-AWQ"

# Hinweis: Die autoawq-Bibliothek muss installiert sein, um dies auszuführen
# pip install autoawq

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True,
)

# Inferenz
messages = [
{"role": "system", "content": "Du bist ein hilfreicher KI-Assistent."},
{"role": "user", "content": "Stelle Python vor"}
]

text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)

inputs = tokenizer([text], return_tensors="pt").to(model.device)

with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=500,
temperature=0.7,
)

response = tokenizer.decode(
outputs[0][inputs.input_ids.shape[1]:],
skip_special_tokens=True
)
print(response)

Modell-Destillation

Destillation verwendet ein großes Modell (Lehrer), um ein kleines Modell (Schüler) zu trainieren, wodurch das kleine Modell die Fähigkeiten des großen Modells erwerben kann.

Das kleine Modell ist schneller und günstiger, und obwohl seine Leistung etwas schlechter ist, reicht es in vielen Szenarien aus.

Wenn man beispielsweise Qwen2.5-72B als Lehrer verwendet, um einen Qwen2.5-7B-Schüler zu trainieren, können Ergebnisse erzielt werden, die nahe an das 72B-Modell heranreichen, jedoch mit Geschwindigkeit und Kosten auf dem Niveau des 7B-Modells.

Die Vorteile der Destillation sind typischerweise:Modellgröße auf 1/10 reduziert, Inferenzgeschwindigkeit um 5-10x verbessert, wobei über 90 % der Leistung erhalten bleiben.。

API-Service-Design

Das Verpacken des Modells ist nur der erste Schritt; die Produktions-API benötigt außerdem viele Funktionen: Authentifizierung, Ratenbegrenzung, Streaming-Antworten und Fehlerbehandlung.

FastAPI ist eine ausgezeichnete Wahl dafür.

LLM-Dienste mit FastAPI verpacken

Schreiben wir einen vollständigen API-Dienst mit Authentifizierung, Ratenbegrenzung und Streaming-Antworten.

Beispiel

# Dateipfad: llm_api_server.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import List, Optional, AsyncGenerator
from enum import Enum
import time
import asyncio
import uuid
from datetime import datetime, timedelta
from collections import defaultdict

# ============================================
# 1. Anwendung und Konfiguration initialisieren
# ============================================

app = FastAPI(title="EXAMPLE LLM API", version="1.0.0")

# Einfache API-Key-Authentifizierung (für die Produktion wird eine robustere Lösung empfohlen)
VALID_API_KEYS = {
"sk-example-123456": {"user": "demo_user", "quota": 1000},
"sk-example-789012": {"user": "pro_user", "quota": 10000},
}

# Einfacher Rate-Limiter (In-Memory-Implementierung; für die Produktion wird Redis empfohlen)
class RateLimiter:
def __init__(self):
self.requests = defaultdict(list) # Schlüssel -> [Zeitstempel, ...]

async def is_allowed(self, key: str, max_requests: int, window_seconds: int) -> bool:
"""Prüft, ob das Ratenlimit überschritten wurde"""
now = time.time()
# Abgelaufene Einträge bereinigen
self.requests[key] = [t for t in self.requests[key] if now - t < window_seconds]
# Prüfen, ob das Limit überschritten wurde
if len(self.requests[key]) >= max_requests:
return False
# Diese Anfrage erfassen
self.requests[key].append(now)
return True

rate_limiter = RateLimiter()

# ============================================
# 2. Datenmodelle
# ============================================

class Role(str, Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"

class Message(BaseModel):
role: Role = Field(..., description="Message role")
content: str = Field(..., description="Message content")

class ChatCompletionRequest(BaseModel):
model: str = Field(..., description="Model name")
messages: List[Message] = Field(..., description="Conversation message list")
temperature: Optional[float] = Field(default=0.7, ge=0, le=2, description="Sampling temperature")
top_p: Optional[float] = Field(default=0.9, ge=0, le=1, description="Top-P sampling")
max_tokens: Optional[int] = Field(default=500, ge=1, le=4096, description="Maximum number of generated tokens")
stream: Optional[bool] = Field(default=False, description="Whether to stream output")

class ChatCompletionResponse(BaseModel):
id: str
object: str = "chat.completion"
created: int
model: str
choices: List[dict]
usage: dict

# ============================================
⟦0▸# 3. Authentifizierungsabhängigkeit
# ============================================

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
⟦0▸"""API-Schlüssel validieren und Benutzerinformationen zurückgeben"""
api_key = credentials.credentials
if api_key not in VALID_API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API Key"
)
return VALID_API_KEYS[api_key]

# ============================================
⟦0▸# 4. LLM-Generierung simulieren (in echten Projekten durch echte Modellaufrufe ersetzen)
# ============================================

async def mock_llm_generate(messages: List[Message], max_tokens: int) -> str:
⟦0▸"""Den LLM-Generierungsprozess simulieren"""
⟦0▸# Dies ist nur eine Simulation; in einem echten Projekt das tatsächliche Modell aufrufen
⟦0▸response_text = (
⟦0▸"Hallo! Ich bin EXAMPLEs KI-Assistent.\n\n"
⟦0▸"EXAMPLE bietet eine Fülle von Programmier-Tutorials, einschließlich Python, Java, C++, Frontend-Entwicklung und mehr."\n"
⟦0▸"Besuchen Sie https://www.example.com, um mehr zu erfahren!"\n\n"
⟦0▸"Wie kann ich Ihnen helfen?"
)
⟦0▸# Generierungsverzögerung simulieren
await asyncio.sleep(0.5)
return response_text

async def mock_llm_generate_stream(messages: List[Message], max_tokens: int) -> AsyncGenerator[str, None]:
⟦0▸"""LLM-Streaming-Generierung simulieren"""
response_text = await mock_llm_generate(messages, max_tokens)
⟦0▸# Zeichen für Zeichen ausgeben, um einen Streaming-Effekt zu simulieren
for char in response_text:
yield char
await asyncio.sleep(0.03) # Delay 30ms per character

# ============================================
⟦0▸# 5. API-Endpunkte
# ============================================

@app.post("/v1/chat/completions")
async def chat_completions(
⟦0▸request: ChatCompletionRequest,
user: dict = Depends(get_current_user)
):
⟦0▸"""Chat-Completion-Endpunkt, kompatibel mit dem OpenAI-Format"""

# Ratenbegrenzungsprüfung: Jeder Benutzer kann höchstens 60 Anfragen pro Minute stellen
user_key = user["user"]
if not await rate_limiter.is_allowed(user_key, max_requests=60, window_seconds=60):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Ratenlimit überschritten"
)

# Request-ID generieren
request_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
created = int(datetime.now().timestamp())

if request.stream:
# Streaming-Antwort
async def stream_generator():
async for token in mock_llm_generate_stream(request.messages, request.max_tokens):
# SSE-formatierten Chunk erstellen
chunk = {
"id": request_id,
"object": "chat.completion.chunk",
"created": created,
"model": request.model,
"choices": [{
"index": 0,
"delta": {"content": token},
"finish_reason": None
}]
}
yield f"data: {__import__('json').dumps(chunk)}\n\n"
# Endmarker senden
yield "data: [DONE]\n\n"

return StreamingResponse(
stream_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"}
)

else:
# Nicht-Streaming-Antwort
response_text = await mock_llm_generate(request.messages, request.max_tokens)

return ChatCompletionResponse(
id=request_id,
created=created,
model=request.model,
choices=[{
"index": 0,
"message": {
"role": "assistant",
"content": response_text
},
"finish_reason": "stop"
}],
usage={
"prompt_tokens": sum(len(m.content) for m in request.messages),
"completion_tokens": len(response_text),
"total_tokens": sum(len(m.content) for m in request.messages) + len(response_text)
}
)

@app.get("/health")
async def health_check():
"""Health-Check-Endpunkt"""
return {"status": "ok", "timestamp": datetime.now().isoformat()}

@app.get("/v1/models")
async def list_models(user: dict = Depends(get_current_user)):
"""Verfügbare Modelle auflisten"""
return {
"object": "list",
"data": [
{
"id": "qwen2.5-7b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "example"
},
{
"id": "llama-3.1-8b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "example"
}
]
}

if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

Starten Sie den Dienst:

Beispiel

# Abhängigkeiten installieren
pip install fastapi uvicorn python-multipart pydantic

# Dienst starten
python llm_api_server.py

# In einem anderen Terminal testen
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-example-123456" \
-d '{
"model": "qwen2.5-7b-instruct",
"messages": [
{"role": "user", "content": "Hallo"}
],
"stream": false
}'

Aufruf mit Python-Client:

Beispiel

# Dateipfad: client_example.py
from openai import OpenAI

# Mit unserem eigenen API-Dienst verbinden
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="sk-example-123456"
)

# Nicht-Streaming testen
print("Nicht-Streaming-Aufruf:")
response = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "Python vorstellen"}],
stream=False
)
print(response.choices[0].message.content)

print("\n" + "="*50 + "\n")

# Streaming testen
print("Streaming-Aufruf:")
stream = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "Python vorstellen"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

Vorteile OpenAI-kompatibler Schnittstellen

Ihre API OpenAI-kompatibel zu machen, bietet mehrere klare Vorteile:

Benutzer müssen keinen Code ändern, nur base_url und api_key.

Reichhaltiges Ökosystem — jedes SDK, Tool oder jede Anwendung, die OpenAI unterstützt, kann Ihren Dienst direkt nutzen.

Sie müssen kein eigenes API-Format entwerfen; das Design von OpenAI ist bereits gut durchdacht.

Authentifizierung und Ratenbegrenzung in Produktionsumgebungen

Die vorherigen Beispiele verwendeten eine einfache In-Memory-Ratenbegrenzung; für die Produktion wird empfohlen, Redis für verteilte Ratenbegrenzung zu verwenden.

Beispiel

# Dateipfad: redis_rate_limiter.py
import redis.asyncio as redis
import time

class RedisRateLimiter:
"""Redis-basierter verteilter Ratenbegrenzer"""

def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)

async def is_allowed(
self,
key: str,
max_requests: int,
window_seconds: int
) -> bool:
"""
Algorithmus zur Ratenbegrenzung mit gleitendem Fenster
key: Ratenbegrenzungsschlüssel (z. B. user_id oder api_key)
max_requests: maximale Anzahl von Anfragen im Fenster
window_seconds: Fenstergröße (Sekunden)
"""
now = time.time()
window_start = now - window_seconds

# Redis-Schlüssel
redis_key = f"rate_limit:{key}"

# Atomare Pipeline-Operationen verwenden
async with self.redis.pipeline() as pipe:
# 1. Datensätze außerhalb des Fensters entfernen
pipe.zremrangebyscore(redis_key, 0, window_start)
# 2. Anfragen im aktuellen Fenster zählen
pipe.zcard(redis_key)
# 3. Aktuelle Anfrage hinzufügen
pipe.zadd(redis_key, {str(now): now})
# 4. Ablaufzeit festlegen
pipe.expire(redis_key, window_seconds)
# Ausführen
_, current_count, _, _ = await pipe.execute()

return current_count < max_requests

async def close(self):
await self.redis.close()

# Verwendungsbeispiel
async def main():
limiter = RedisRateLimiter("redis://localhost:6379")

# Test: Derselbe Benutzer höchstens 10 Anfragen pro Minute
user_id = "user_123"
for i in range(15):
allowed = await limiter.is_allowed(user_id, max_requests=10, window_seconds=60)
print(f"Request {i+1}: {'allowed' if allowed else 'denied'}")

await limiter.close()

if __name__ == "__main__":
import asyncio
asyncio.run(main())

Authentifizierung und Ratenbegrenzung sind die Infrastruktur einer API; besser etwas komplexer machen als später zu bereuen, überwältigt worden zu sein.

Lastverteilung und Skalierung

Wenn immer mehr Benutzer hinzukommen und ein Server nicht mehr ausreicht, müssen Sie Maschinen hinzufügen und den Datenverkehr auf mehrere Maschinen verteilen.

Das ist das Problem, das Lastverteilung und Auto-Scaling lösen müssen.

Design zustandsloser Dienste

Denken Sie zuerst daran:KI-Dienste sollten zustandslos entworfen werden。

Zustandslos bedeutet: Anfragen teilen keinen Speicher, und jeder Server kann jede Anfrage bearbeiten.

Das macht Skalierung einfach: einfach Maschinen hinzufügen, keine Sorge um Datensynchronisierung.

Sitzungszustand (z. B. Gesprächsverlauf) kann auf dem Client gespeichert werden (lassen Sie ihn vom Benutzer jedes Mal mitsenden) oder in Redis.

Nginx-Reverse-Proxy

Nginx ist der am häufigsten verwendete Load Balancer und verteilt den Datenverkehr auf mehrere Backend-Server.

Beispiel

# Dateipfad: nginx.conf
# Nginx-Konfiguration: Lastverteilung für mehrere LLM-Dienste

upstream llm_backends {
# Algorithmus der geringsten Verbindungen: Anfragen an den Server mit den wenigsten aktuellen Verbindungen senden
least_conn;

# Backend-Server-Liste
server 192.168.1.101:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.102:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.103:8000 weight=1 max_fails=3 fail_timeout=30s;

# Wenn Sie mehrere GPU-Modelle haben, können Sie unterschiedliche Verkehrsanteile nach Gewicht zuweisen
# weight=2 bedeutet, dass dieser Server doppelt so viel Verkehr erhält wie andere
}

server {
listen 80;
server_name api.example.com;

# API-Anfrageweiterleitung
location /v1/ {
# Weiterleiten an Backend
proxy_pass http://llm_backends;

# Timeout-Einstellungen (KI-Generierung kann langsam sein)
proxy_connect_timeout 10s;
proxy_send_timeout 600s;
proxy_read_timeout 600s;

# Die echte IP des Clients weitergeben
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

# Erforderliche Einstellungen für Streaming-Antworten
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection '';
proxy_http_version 1.1;
}

# Health-Check-Endpunkt
location /health {
proxy_pass http://llm_backends/health;
proxy_connect_timeout 5s;
proxy_read_timeout 5s;
}

# Ratenbegrenzung: Max. 10 Anfragen pro Sekunde pro IP
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
limit_req zone=api_limit burst=20 nodelay;
}

Kubernetes HPA Auto-Scaling

Wenn Sie mit Kubernetes bereitstellen, können Sie HPA (Horizontal Pod Autoscaler) verwenden, um basierend auf der Last automatisch hoch- und herunterzuskalieren.

Beispiel

# Dateipfad: llm-deployment.yaml
# Kubernetes-Deployment-Konfiguration für den LLM-Dienst

# 1. Deployment: LLM-Pod bereitstellen
apiVersion: apps/v1
kind: Deployment
metadata:
name: llm-server
namespace: example
spec:
replicas: 3 # Anfängliche Anzahl von Replikas
selector:
matchLabels:
app: llm-server
template:
metadata:
labels:
app: llm-server
spec:
containers:
- name: llm-server
image: example/llm-server:latest
ports:
- containerPort: 8000
Ressourcen:
requests:
cpu: "4"
memory: "16Gi"
nvidia.com/gpu: 1 # 1 GPU anfordern
limits:
cpu: "8"
memory: "32Gi"
nvidia.com/gpu: 1 # 1 GPU begrenzen
env:
- name: MODEL_NAME
value: "Qwen/Qwen2.5-7B-Instruct"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300 # Das Laden des Modells dauert Zeit
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300
periodSeconds: 10
tolerations:
- key: "nvidia.com/gpu"
operator: "Exists"
effect: "NoSchedule"

---
# 2. Service: Lastverteilungs-Service
apiVersion: v1
kind: Service
metadata:
name: llm-service
namespace: example
spec:
selector:
app: llm-server
ports:
- port: 80
targetPort: 8000
type: LoadBalancer

---
# 3. HPA: automatische Skalierung
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: llm-hpa
namespace: example
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: llm-server
minReplicas: 3 # Mindestens 3 Pods
maxReplicas: 20 # Maximal 20 Pods
metrics:
# Skalierung basierend auf CPU-Auslastung
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
# Skalierung basierend auf GPU-Auslastung (erfordert benutzerdefinierte Metriken)
- type: Pods
pods:
metric:
name: gpu_utilization
target:
type: AverageValue
averageValue: 80
# Skalierung basierend auf Warteschlangenlänge (benutzerdefinierte Metrik)
- type: Pods
pods:
metric:
name: pending_requests
target:
type: AverageValue
averageValue: 10
behavior:
scaleUp:
stabilizationWindowSeconds: 60 # 60 Sekunden beobachten, bevor hochskaliert wird
policies:
- type: Percent
value: 50
periodSeconds: 60
scaleDown:
stabilizationWindowSeconds: 300 # 5 Minuten beobachten, bevor herunterskaliert wird
policies:
- type: Percent
value: 20
periodSeconds: 120

GPU-Ressourcenpool-Verwaltung

GPUs sind teuer, daher sollte die Auslastung maximiert werden.

Eine gängige Strategie ist: nach Modelltyp gruppieren und Pods desselben Modells zusammen platzieren.

Zum Beispiel:

Ressourcenpool	GPU-Modell	Laufendes Modell	Zweck
small-pool	RTX 3090 / L4	7B-Modell	Täglicher Datenverkehr
medium-pool	A10 / L40	13B-70B-Modell	Komplexe Aufgaben
large-pool	A100 / H100	70B+-Modell	Hochwertige Nutzer
spot-pool	Verschiedene Maschinentypen	Offline-Batchverarbeitung	Günstige Spot-Instances verwenden

A/B-Tests für KI-Modelle

Sie haben eine neue Modellversion veröffentlicht. Woher wissen Sie, dass sie besser ist als die alte? Verwenden Sie A/B-Tests.

Der Kern von A/B-Tests ist:Teilen Sie den Traffic in Gruppen auf, verwenden Sie für verschiedene Gruppen unterschiedliche Modelle und vergleichen Sie die Metriken。

Strategie zur Traffic-Aufteilung

Es gibt mehrere gängige Methoden zur Traffic-Aufteilung:

Beispiel

# Dateipfad: ab_test_router.py
import hashlib
import random
from typing import Dict, List, Tuple
from dataclasses import dataclass

@dataclass
class Variant:
"""Eine Variante des A/B-Tests"""
name: str # Variantenname
model_name: str # Modellname
weight: int # Traffic-Gewicht
config: dict # Weitere Konfiguration

class ABTestRouter:
"""A/B-Test-Traffic-Router"""

def __init__(self, test_name: str, variants: List[Variante]):
self.test_name = test_name
self.variants = variants
# Gesamtgewicht berechnen
total_weight = sum(v.weight for v in variants)
# Segmentbereiche erstellen
self.ranges: List[Tuple[int, int, Variante]] = []
current = 0
for v in variants:
self.ranges.append((current, current + v.weight, v))
current += v.weight
self.total_weight = total_weight

def get_variant_by_user(self, user_id: str) -> Variant:
"""
Stabile Traffic-Aufteilung basierend auf der Benutzer-ID
Derselbe Nutzer wird immer derselben Gruppe zugewiesen, um eine inkonsistente Erfahrung zu vermeiden
"""
# Berechne den Score des Nutzers mithilfe eines Hashes
hash_input = f"{self.test_name}-{user_id}".encode()
hash_value = hashlib.md5(hash_input).hexdigest()
# In eine Zahl von 0-999 umwandeln
score = int(hash_value[:4], 16) % 1000
# Die entsprechende Variante finden
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

def get_variant_random(self) -> Variant:
"""Zufällige Aufteilung (geeignet für Szenarien, die keine Stickiness erfordern)"""
score = random.randint(0, self.total_weight - 1)
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

# Verwendungsbeispiel
def main():
# Drei Varianten definieren
variants = [
Variante(
name="control",
model_name="Qwen2.5-7B-Instruct",
weight=50, # 50% Traffic
config={"temperature": 0.7}
),
Variante(
name="treatment-v1",
model_name="Qwen2.5-14B-Instruct",
weight=30, # 30% Traffic
config={"temperature": 0.7}
),
Variante(
name="treatment-v2",
model_name="Llama-3.1-8B-Instruct",
weight=20, # 20% Traffic
config={"temperature": 0.8}
)
]

router = ABTestRouter("model-comparison-v1", variants)

# Aufteilung der Testnutzer
test_users = [f"user_{i}" for i in range(10)]
for user_id in test_users:
variant = router.get_variant_by_user(user_id)
print(f"Benutzer {user_id} → Gruppe {variant.name} → Modell {variant.model_name}")

# Verteilungsstatistik
count: Dict[str, int] = {}
for i in range(10000):
variant = router.get_variant_random()
count[variant.name] = count.get(variant.name, 0) + 1
print("\nZufällige Aufteilungsstatistik (10000 Mal):")
for name, cnt in count.items():
print(f" {name}: {cnt} Mal ({cnt/100}%)")

if __name__ == "__main__":
main()

Metrikdefinition und -erfassung

A/B-Tests dürfen nicht nur danach beurteilt werden, "welches Modell intelligenter antwortet"; es müssen quantifizierbare Metriken definiert werden.

Metriktypen	Spezifische Metriken	Beschreibung
Systemmetriken	Latenz, Durchsatz, Fehlerrate	Ob das Modell schnell und stabil ist
Nutzerverhalten	Nutzerzufriedenheit, Anzahl der Dialogrunden, Verlängerungsrate	Ob Nutzer es mögen
Geschäftsmetriken	Konversionsrate, Kundenbindungsrate, Umsatz	Ob es dem Geschäft hilft
Qualitätsmetriken	Menschliche Bewertungen, Schädlichkeitsrate, Faktenrichtigkeit	Ob die Ausgabequalität gut ist

Beispiel

# Dateipfad: metrics_collector.py
import time
import json
from dataclasses import dataclass, asdict
from typing import Optional
from datetime import datetime
import uuid

@dataclass
class ChatMetrics:
"""Vollständige Metriken für Chat-Anfragen"""
# Basisinformationen
request_id: str
user_id: str
variant_name: str # Name der A/B-Testgruppe
model_name: str

# Zeitmetriken
timestamp: str
ttft: float # Zeit bis zum ersten Token
total_time: float # Insgesamt verstrichene Zeit

# Generierungsmetriken
prompt_tokens: int
completion_tokens: int
total_tokens: int

# Geschäftsmetriken
user_rating: Optional[int] = None # Benutzerbewertung (1-5)
is_error: bool = False
error_message: Optional[str] = None

class MetricsCollector:
"""Metriksammler"""

def __init__(self):
# In einem echten Projekt sollte dies in Kafka oder eine Datenbank geschrieben werden
self.buffer = []

def collect(self, metrics: ChatMetrics):
"""Eine Metrik erfassen"""
self.buffer.append(asdict(metrics))
# Einfache Ausgabe; in einem echten Projekt in den Speicher schreiben
print(f"[Metriken] {json.dumps(asdict(metrics), ensure_ascii=False)}")

# Wenn der Puffer voll ist, in Stapeln schreiben
if len(self.buffer) >= 100:
self.flush()

def flush(self):
"""Stapelweises Schreiben in den Speicher"""
if self.buffer:
# In einem echten Projekt: in ClickHouse, BigQuery usw. schreiben
print(f"{len(self.buffer)} Metriken geschrieben")
self.buffer = []

# Verwendungsbeispiel
collector = MetricsCollector()

def process_chat_request(user_id: str, prompt: str, router):
"""Eine Chat-Anfrage verarbeiten und Metriken aufzeichnen"""
request_id = str(uuid.uuid4())

# Traffic aufteilen
variant = router.get_variant_by_user(user_id)

# Startzeit erfassen
start_time = time.time()
ttft = 0.0
prompt_tokens = len(prompt)
completion_tokens = 0

try:
# Dies durch einen echten Modellaufruf ersetzen
# response = call_model(variant.model_name, prompt)
response = "Dies ist die Antwort der KI"
completion_tokens = len(response)

# TTFT simulieren (annehmen, dass das erste Token nach 0.2 Sekunden empfangen wird)
ttft = 0.2

# Erfolgsmetriken aufzeichnen
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
is_error=False
)
collector.collect(metrics)
return response, request_id

except Exception as e:
# Fehlermetriken aufzeichnen
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=0,
total_tokens=prompt_tokens,
is_error=True,
error_message=str(e)
)
collector.collect(metrics)
raise

# Benutzerfeedback
def submit_user_feedback(request_id: str, rating: int):
"""Benutzer reicht Bewertungsfeedback ein"""
# In einem echten Projekt: den entsprechenden Datensatz in der Datenbank aktualisieren
print(f"Anfrage {request_id} erhielt Bewertung {rating}")

Bewertung der statistischen Signifikanz

Nachdem die Daten vorliegen, kann man nicht einfach 52% für Gruppe A und 55% für Gruppe B betrachten und daraus schließen, dass Gruppe B besser ist.

Ein statistischer Signifikanztest muss durchgeführt werden, um zu bestätigen, dass der Unterschied nicht durch Zufallsschwankungen verursacht wird.

Beispiel

# Dateipfad: ab_test_analysis.py
import math
from typing import Tuple

def calculate_z_test(
control_conv: float,
treatment_conv: float,
control_total: int,
treatment_total: int
) -> Tuple[float, float]:
"""
Berechne den p-Wert des Z-Tests
control_conv: Konversionsrate der Kontrollgruppe
treatment_conv: Konversionsrate der Testgruppe
control_total: Stichprobengröße der Kontrollgruppe
treatment_total: Stichprobengröße der Testgruppe
"""
# gepoolte Konversionsrate
p_pool = (control_conv * control_total + treatment_conv * treatment_total) / (control_total + treatment_total)
# Standardfehler
se_pool = math.sqrt(p_pool * (1 - p_pool) * (1 / control_total + 1 / treatment_total))
# Z-Statistik
z_score = (treatment_conv - control_conv) / se_pool
# p-Wert berechnen (zweiseitiger Test)
p_value = 2 * (1 - normal_cdf(abs(z_score)))
return z_score, p_value

def normal_cdf(x: float) -> float:
"""Kumulative Verteilungsfunktion der Standardnormalverteilung (Näherung)"""
return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0

def analyze_ab_test(
control_conversions: int,
control_total: int,
treatment_conversions: int,
treatment_total: int,
alpha: float = 0.05
):
"""A/B-Testergebnisse analysieren"""
control_conv = control_conversions / control_total
treatment_conv = treatment_conversions / treatment_total

z_score, p_value = calculate_z_test(
control_conv, treatment_conv,
control_total, treatment_total
)

relative_lift = (treatment_conv - control_conv) / control_conv

print("Analyse der A/B-Testergebnisse")
print("=" * 50)
print(f"Kontrollgruppe: {control_conversions}/{control_total} ({control_conv:.2%})")
print(f"Testgruppe: {treatment_conversions}/{treatment_total} ({treatment_conv:.2%})")
print(f"Relative Steigerung: {relative_lift:+.2%}")
print("-" * 50)
print(f"Z-Statistik: {z_score:.4f}")
print(f"p-Wert: {p_value:.4f}")
print("-" * 50)

if p_value < alpha:
print(f"✅ Ergebnis ist statistisch signifikant (p < {alpha})")
if relative_lift > 0:
print(" Testgruppe schneidet besser ab!")
else:
print(" Kontrollgruppe schneidet besser ab!")
else:
print(f"❌ Ergebnis ist nicht signifikant (p >= {alpha})")
print(" Es kann nicht bestimmt werden, welche Version besser ist; mehr Daten sind erforderlich")

# Beispiel: Annahme von A/B-Testergebnissen
if __name__ == "__main__":
# Szenario 1: signifikanter Unterschied
print("Szenario 1: signifikanter Unterschied")
analyze_ab_test(
control_conversions=450, # Kontrollgruppe: 450 zufrieden
control_total=1000, # Kontrollgruppe insgesamt: 1000 Personen
treatment_conversions=520, # Testgruppe: 520 zufrieden
treatment_total=1000 # Testgruppe insgesamt: 1000 Personen
)

print("\n" + "="*50 + "\n")

# Szenario 2: kein signifikanter Unterschied
print("Szenario 2: kein signifikanter Unterschied")
analyze_ab_test(
control_conversions=48,
control_total=100,
treatment_conversions=52,
treatment_total=100
)

Statistische Signifikanz ist wichtig. Eine ungetestete "Verbesserung" könnte bloß zufällige Schwankung sein.

MLflow-Experimentverfolgung

Sie haben Dutzende Modellversionen trainiert. Welche funktioniert am besten? Welche Version ist im Einsatz?

MLflow hilft Ihnen, Experimente zu verwalten, Modelle zu verfolgen und Metriken aufzuzeichnen.

Experimentverwaltung

Beispiel

# Dateipfad: mlflow_experiment.py
import mlflow
import mlflow.pyfunc
import mlflow.sklearn
from mlflow.tracking import MlflowClient
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score
import numpy as np
import time

# MLflow-Tracking-URI festlegen (lokaler oder Remote-Server)
mlflow.set_tracking_uri("http://localhost:5000")

# Experiment erstellen oder festlegen
experiment_name = "example-llm-finetuning"
try:
experiment_id = mlflow.create_experiment(experiment_name)
except mlflow.exceptions.MlflowException:
experiment_id = mlflow.get_experiment_by_name(experiment_name).experiment_id

mlflow.set_experiment(experiment_name)

# Ein LLM-Fine-Tuning-Experiment simulieren
def run_finetuning_experiment(
model_name: str,
learning_rate: float,
batch_size: int,
num_epochs: int,
lora_rank: int,
dataset_name: str = "example-demo"
):
"""Ein Fine-Tuning-Experiment ausführen und in MLflow protokollieren"""

with mlflow.start_run(run_name=f"{model_name}-lr{learning_rate}-bs{batch_size}") as run:
run_id = run.info.run_id

# Parameter protokollieren
mlflow.log_param("model_name", model_name)
mlflow.log_param("learning_rate", learning_rate)
mlflow.log_param("batch_size", batch_size)
mlflow.log_param("num_epochs", num_epochs)
mlflow.log_param("lora_rank", lora_rank)
mlflow.log_param("dataset_name", dataset_name)

# Trainingsprozess simulieren und Metriken aufzeichnen
print(f"Starte Trainingsexperiment {run_id}...")

for epoch in range(num_epochs):
# Trainingsverlust simulieren (sinkt mit den Epochen)
train_loss = 2.0 - 0.3 * epoch + np.random.normal(0, 0.1)
val_loss = 2.2 - 0.25 * epoch + np.random.normal(0, 0.15)

# Metriken aufzeichnen
mlflow.log_metric("train_loss", train_loss, step=epoch)
mlflow.log_metric("val_loss", val_loss, step=epoch)
mlflow.log_metric("epoch", epoch, step=epoch)

print(f"Epoche {epoch}: train_loss={train_loss:.4f}, val_loss={val_loss:.4f}")
time.sleep(0.5)

# Abschließende Evaluierungsmetriken
final_val_loss = 1.2 + np.random.normal(0, 0.1)
final_accuracy = 0.85 + np.random.normal(0, 0.03)
final_f1 = 0.83 + np.random.normal(0, 0.04)

mlflow.log_metric("final_val_loss", final_val_loss)
mlflow.log_metric("final_accuracy", final_accuracy)
mlflow.log_metric("final_f1", final_f1)

# Dateien protokollieren (z. B. Trainingsprotokolle, Beispielausgaben)
with open("training_log.txt", "w") as f:
f.write(f"Experiment-Lauf-ID: {run_id}\n")
f.write(f"Modell: {model_name}\n")
f.write(f"Endgültige Genauigkeit: {final_accuracy:.4f}\n")

mlflow.log_artifact("training_log.txt")

# Modell protokollieren (hier mit sklearn simuliert; in echten Projekten das tatsächliche LLM protokollieren)
# In echten Projekten können Sie mlflow.transformers.log_model verwenden
X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

mlflow.sklearn.log_model(
model,
"model",
registered_model_name=f"{model_name.replace('/', '-')}-demo"
)

print(f"Experiment abgeschlossen! run_id: {run_id}")
print(f"Endgültige Genauigkeit: {final_accuracy:.4f}")

return run_id, final_accuracy

# Mehrere Experimente ausführen
if __name__ == "__main__":
print("MLflow-UI starten (in einem anderen Terminal ausführen):")
print(" mlflow ui --port 5000")
print("\nExperimente werden gestartet...\n")

# Experiment 1: Basiskonfiguration
run_id1, acc1 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()

# Experiment 2: Größerer LoRA-Rang
run_id2, acc2 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=16
)

print()

# Experiment 3: Höhere Lernrate
run_id3, acc3 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=5e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()
print("Alle Experimente abgeschlossen!")
print(f"Experiment 1 (run_id={run_id1}): {acc1:.4f}")
print(f"Experiment 2 (run_id={run_id2}): {acc2:.4f}")
print(f"Experiment 3 (run_id={run_id3}): {acc3:.4f}")
print("\n"Ergebnisse anzeigen: http://localhost:5000")

Modell-Registry

Die MLflow-Modell-Registry verwaltet den Modelllebenszyklus: von Staging (Testen) bis Production.

Beispiel

# Dateipfad: mlflow_model_registry.py
import mlflow
from mlflow.tracking import MlflowClient

mlflow.set_tracking_uri("http://localhost:5000")

client = MlflowClient()

def promote_model_to_production(
model_name: str,
version: int,
description: str = ""
):
"""Das Modell in die Production-Phase überführen"""
# Versionsbeschreibung aktualisieren
client.update_model_version(
name=model_name,
version=version,
description=description
)

# Diese Version als Production festlegen
client.transition_model_version_stage(
name=model_name,
version=version,
stage="Production",
archive_existing_versions=True # Alte Versionen archivieren
)
print(f"Modell {model_name}, Version {version}, wurde in die Production-Phase überführt")

def get_production_model(model_name: str):
"""Das aktuelle Production-Modell abrufen"""
versions = client.get_latest_versions(model_name, stages=["Production"])
if versions:
return versions[0]
return None

def list_all_models():
"""Alle registrierten Modelle auflisten"""
models = client.search_registered_models()
print("Registrierte Modelle:")
for model in models:
print(f" - {model.name}")
# Alle Versionen dieses Modells auflisten
versions = client.search_model_versions(f"name='{model.name}'")
for v in versions:
print(f" Version {v.version}: {v.current_stage}")

# Verwendungsbeispiel
if __name__ == "__main__":
model_name = "Qwen2.5-7B-Instruct-demo"

print("Alle Modelle auflisten:")
list_all_models()

print("\n" + "="*50)
# Angenommen, wir möchten Version 1 in die Produktion überführen
# promote_model_to_production(
# model_name=model_name,
# version=1,
# description="Modell für Produktionsumgebung: 2024 optimierte Version"
# )

print("\nAktuelles Produktionsmodell abrufen: ")
prod_model = get_production_model(model_name)
if prod_model:
print(f" Modellname: {prod_model.name}")
print(f" Version: {prod_model.version}")
print(f" Phase: {prod_model.current_stage}")
print(f" Lauf-ID: {prod_model.run_id}")
else:
print(" Noch kein Produktionsmodell")

Monitoring und Alarmierung

Sobald der Dienst live ist, woher wissen Sie, ob er ordnungsgemäß funktioniert? Werden Benutzer auf Timeouts stoßen? Wird das Modell zunehmend ungenauer?

Sie benötigen ein vollständiges Monitoring-System.

Wichtige Metriken: TTFT, TPS, Fehlerrate

KI-Dienste haben mehrere zentrale Monitoring-Metriken:

Metrik	Vollständiger Name	Bedeutung	Beispiel für einen Alarm-Schwellenwert
TTFT	Zeit bis zum ersten Token	Zeit von der Benutzeranfrage bis zum Empfang des ersten Tokens	Alarm > 3 Sekunden
TPOT	Zeit pro Ausgabe-Token	Durchschnittliche Zeit pro Ausgabe-Token	Alarm > 100ms
TPS	Token pro Sekunde	Wie viele Token pro Sekunde generiert werden	Alarm < 100
Fehlerrate	Fehlerrate	Anteil fehlgeschlagener Anfragen	Alarm > 5%
GPU-Auslastung	GPU-Auslastung	GPU-Nutzungsrate	Alarm < 30% oder > 95%
VRAM-Nutzung	GPU-Speicher	GPU-VRAM-Belegungsrate	Alarm > 90%

Prometheus + Grafana

Prometheus sammelt Metriken, Grafana zeigt Dashboards an – die goldene Kombination für das Monitoring.

Beispiel

# Dateipfad: metrics_prometheus.py
from prometheus_client import (
Counter, Histogram, Gauge,
start_http_server, generate_latest
)
import time
import random

# ============================================
# Metriken definieren
# ============================================

# Counter: ein Zähler, der nur steigt
REQUEST_COUNT = Counter(
"llm_requests_total",
"Gesamtzahl der LLM-Anfragen",
["model", "variant", "status"] # Beschriftungen
)

# Histogramm: statistische Verteilung (z. B. Latenz)
REQUEST_LATENCY = Histogram(
"llm_request_latency_seconds",
"LLM-Anfragelatenz",
["model", "variant"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60] # Buckets
)

TTFT_HISTOGRAM = Histogram(
"llm_ttft_seconds",
"Zeit bis zum ersten Token",
["model", "variant"],
buckets=[0.05, 0.1, 0.2, 0.5, 1, 2, 5]
)

# Gauge: ein Messwert, der steigen oder fallen kann
ACTIVE_CONNECTIONS = Gauge(
"llm_active_connections",
"Anzahl aktiver Verbindungen",
["model"]
)

GPU_UTILIZATION = Gauge(
"llm_gpu_utilization_percent",
"GPU-Auslastung",
["gpu_id", "model"]
)

GPU_MEMORY_USED = Gauge(
"llm_gpu_memory_used_bytes",
"Verwendeter GPU-Speicher",
["gpu_id", "model"]
)

TOKENS_PER_SECOND = Gauge(
"llm_tokens_per_second",
"Generierte Tokens pro Sekunde",
["model"]
)

# ============================================
# Service simulieren
# ============================================

def process_request(model: str, variant: str):
"""Simuliert die Verarbeitung einer Anfrage und zeichnet Metriken auf"""
start_time = time.time()

# Aktive Verbindungen +1
ACTIVE_CONNECTIONS.labels(model=model).inc()

try:
# TTFT simulieren (Zeit bis zum ersten Token)
ttft = random.uniform(0.05, 0.3)
TTFT_HISTOGRAM.labels(model=model, variant=variant).observe(ttft)

# Verarbeitungszeit simulieren
processing_time = random.uniform(0.5, 3.0)
time.sleep(processing_time)

# Erfolg simulieren
REQUEST_COUNT.labels(
model=model, variant=variant, status="success"
).inc()

# Latenz aufzeichnen
latency = time.time() - start_time
REQUEST_LATENCY.labels(model=model, variant=variant).observe(latency)

# TPS simulieren
tps = random.uniform(50, 200)
TOKENS_PER_SECOND.labels(model=model).set(tps)

except Exception:
# Fehler simulieren
REQUEST_COUNT.labels(
model=model, variant=variant, status="error"
).inc()
raise
finally:
# Aktive Verbindungen -1
ACTIVE_CONNECTIONS.labels(model=model).dec()

def update_gpu_metrics():
"""Simuliert die Aktualisierung von GPU-Metriken"""
# Metriken für GPU 0 simulieren
gpu0_util = random.uniform(40, 90)
gpu0_mem = random.uniform(10, 20) * 1024**3 # 10-20 GB
GPU_UTILIZATION.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_util)
GPU_MEMORY_USED.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_mem)

# ============================================
# Service starten
# ============================================

if __name__ == "__main__":
# Prometheus-Metriken-Endpunkt starten
start_http_server(8081)
print("Prometheus-Metriken-Server gestartet auf Port 8081")
print("Metriken verfügbar unter http://localhost:8081/metrics")

# Datenverkehr simulieren
models = ["Qwen2.5-7B-Instruct", "Llama-3.1-8B-Instruct"]
variants = ["control", "treatment-v1"]

while True:
model = random.choice(models)
variant = random.choice(variants)
process_request(model, variant)
update_gpu_metrics()
time.sleep(random.uniform(0.1, 0.5))

Prometheus-Konfigurationsdatei:

Beispiel

# Dateipfad: prometheus.yml
global:
scrape_interval: 15s

scrape_configs:
# LLM-Service-Metriken scrapen
- job_name: "llm-service"
static_configs:
- targets: ["localhost:8081"]
labels:
service: "llm-api"
environment: "production"

# Node-Metriken scrapen (GPU, CPU, Speicher)
- job_name: "node"
static_configs:
- targets: ["localhost:9100"]

alerting:
alertmanagers:
- static_configs:
- targets: ["localhost:9093"]

# Alarmregeln
rule_files:
- "alerts.yml"

Alarmregeln:

Beispiel

# Dateipfad: alerts.yml
groups:
- name: llm_alerts
interval: 30s
rules:
# Alarm bei hoher Fehlerrate
- alert: HighErrorRate
expr: |
sum(rate(llm_requests_total{status="error"}[5m]))
/
sum(rate(llm_requests_total[5m])) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "Alarm bei hoher Fehlerrate"
description: "Fehlerrate überschreitet 5%, aktueller Wert: {{ $value | humanizePercentage }}"

# Warnung: TTFT zu hoch
- alert: SlowTTFT
expr: |
histogram_quantile(0.95, sum(rate(llm_ttft_seconds_bucket[5m])) by (le, model)) > 2
for: 5m
labels:
severity: warning
annotations:
summary: "TTFT zu hoch"
description: "Das 95. Perzentil der TTFT von Modell {{ $labels.model }} überschreitet 2 Sekunden"

# GPU-Auslastung zu niedrig (Ressourcenverschwendung)
- alert: LowGPUUtilization
expr: llm_gpu_utilization_percent < 30
for: 30m
labels:
severity: info
annotations:
summary: "GPU-Auslastung zu niedrig"
description: "GPU {{ $labels.gpu_id }} Auslastung {{ $value }}%, mögliche Ressourcenverschwendung"

# GPU-Auslastung zu hoch (Engpassrisiko)
- alert: HighGPUUtilization
expr: llm_gpu_utilization_percent > 95
for: 5m
labels:
severity: warning
annotations:
summary: "GPU-Auslastung zu hoch"
description: "GPU {{ $labels.gpu_id }} Auslastung {{ $value }}%, könnte zum Engpass werden"

KI-spezifische Überwachung: Drift-Erkennung

KI-Modelle "brechen" nicht nur, sie "driften" auch: Mit der Zeit ändert sich die Datenverteilung und die Modellleistung verschlechtert sich.

Beispiel

# Dateipfad: drift_detection.py
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import List, Tuple
import json

@dataclass
class DistributionStats:
"""Verteilungsstatistiken"""
mean: float
std: float
min: float
max: float
percentiles: dict

class DataDriftDetector:
"""Detektor für Daten-Drift"""

def __init__(self, baseline_data: np.ndarray):
"""Mit Baseline-Daten initialisieren"""
self.baseline_stats = self._compute_stats(baseline_data)
self.baseline_data = baseline_data

def _compute_stats(self, data: np.ndarray) -> DistributionStats:
"""Statistiken für die Daten berechnen"""
return DistributionStats(
mean=float(np.mean(data)),
std=float(np.std(data)),
min=float(np.min(data)),
max=float(np.max(data)),
percentiles={
"p10": float(np.percentile(data, 10)),
"p25": float(np.percentile(data, 25)),
"p50": float(np.percentile(data, 50)),
"p75": float(np.percentile(data, 75)),
"p90": float(np.percentile(data, 90))
}
)

def detect(self, new_data: np.ndarray, threshold: float = 0.05) -> dict:
"""
Drift erkennen
Gibt zurück: ob Drift vorliegt, p-Wert, Statistiken
"""
# KS-Test: vergleicht, ob zwei Verteilungen identisch sind
ks_statistic, ks_pvalue = stats.ks_2samp(self.baseline_data, new_data)

# Statistiken für neue Daten berechnen
new_stats = self._compute_stats(new_data)

# Mittlere Differenz berechnen (prozentual)
mean_diff_pct = abs((new_stats.mean - self.baseline_stats.mean) / self.baseline_stats.mean * 100)

# Feststellen, ob Drift aufgetreten ist
is_drift = ks_pvalue < threshold

return {
"is_drift": is_drift,
"ks_statistic": float(ks_statistic),
"ks_pvalue": float(ks_pvalue),
"mean_diff_percent": float(mean_diff_pct),
"baseline_stats": self.baseline_stats,
"new_stats": new_stats,
"threshold": threshold
}

# Simulation: Drift-Erkennung für die Länge der Benutzereingabe
def simulate_input_length_drift():
# Baseline-Daten: durchschnittliche Länge der Benutzereingabe 50
np.random.seed(42)
baseline = np.random.normal(loc=50, scale=15, size=1000)
baseline = np.clip(baseline, 5, 200)

detector = DataDriftDetector(baseline)

print("Beispiel zur Data-Drift-Erkennung")
print("="*50)
print(fBasislinie: Mittelwert={detector.baseline_stats.mean:.2f}, Standardabweichung={detector.baseline_stats.std:.2f})
print()

# Szenario 1: Normale Daten (keine Drift)
print(Szenario 1: Normale Daten)
normal_data = np.random.normal(loc=50, scale=15, size=500)
normal_data = np.clip(normal_data, 5, 200)
result = detector.detect(normal_data)
print(fMittelwert der neuen Daten: {result['new_stats'].mean:.2f})
print(f" KS-p-Wert: {result['ks_pvalue']:.4f}")
print(fDrift? {'Ja' if result['is_drift'] else 'Nein'})
print()

# Szenario 2: Drift (Eingabe wurde länger)
print(Szenario 2: Eingabe wurde länger)
drift_data = np.random.normal(loc=100, scale=20, size=500)
drift_data = np.clip(drift_data, 5, 300)
result = detector.detect(drift_data)
print(fMittelwert der neuen Daten: {result['new_stats'].mean:.2f})
print(f" KS-p-Wert: {result['ks_pvalue']:.4f}")
print(fDrift? {'Ja' if result['is_drift'] else 'Nein'})
print(fMittelwertdifferenz: {result['mean_diff_percent']:.1f}%)

if __name__ == "__main__":
simulate_input_length_drift()

Strategien zur Kostenoptimierung

GPUs sind teuer. Kosten zu optimieren ist nicht „billig“ – es ist „Engineering-Fähigkeit“.

Request-Caching: Semantisches Caching

Traditionelles Caching ist exakte Übereinstimmung: Bei genau derselben Frage wird direkt die vorherige Antwort zurückgegeben.

Semantisches Caching geht weiter: Ähnliche Fragen können ebenfalls vorherige Antworten oder Zwischenergebnisse wiederverwenden.

Beispiel

# Dateipfad: semantic_cache.py
import numpy as np
from typing import Optional, Tuple
import time

class SemanticCache:
"""Einfache Implementierung eines semantischen Caches"""

def __init__(self, threshold: float = 0.9):
self.threshold = threshold
self.queries = [] # Vektoren der Fragen speichern
self.answers = [] # Zugehörige Antworten speichern
self.timestamps = [] # Zeitstempel speichern

def _simulate_embedding(self, text: str) -> np.ndarray:
"""
Simuliertes Embedding; in realen Projekten OpenAI Embeddings, Sentence-Transformers usw. verwenden.
"""
# Einfache Hash-Simulation; in realen Projekten ein echtes Embedding-Modell verwenden
np.random.seed(hash(text) % (2**32))
return np.random.randn(128) # 128-dimensionaler Vektor

def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
"""Kosinusähnlichkeit berechnen"""
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def get(self, query: str) -> Optional[str]:
"""Abfrage-Cache: Wenn eine ähnliche Frage gefunden wird, die Antwort zurückgeben"""
query_vec = self._simulate_embedding(query)

for i, (q_vec, answer, ts) in enumerate(zip(self.queries, self.answers, self.timestamps)):
# Bereinigung abgelaufener Einträge (mehr als 1 Stunde)
if time.time() - ts > 3600:
continue

similarity = self._cosine_similarity(query_vec, q_vec)
if similarity >= self.threshold:
return answer

return None

def put(self, query: str, answer: str):
"""Im Cache speichern"""
query_vec = self._simulate_embedding(query)
self.queries.append(query_vec)
self.answers.append(answer)
self.timestamps.append(time.time())

# Einfache Kapazitätskontrolle: höchstens 1000 Einträge speichern
if len(self.queries) > 1000:
self.queries = self.queries[-1000:]
self.answers = self.answers[-1000:]
self.timestamps = self.timestamps[-1000:]

# Verwendungsbeispiel
def main():
cache = SemanticCache(threshold=0.9)

# Erste Anfrage
query1 = Wie liest man eine CSV-Datei in Python?
print(fAbfrage 1: {query1})
answer1 = cache.get(query1)
if answer1:
print(fCache-Treffer: {answer1})
else:
print(Cache-Miss, Modell wird aufgerufen...)
answer1 = Sie können pandas.read_csv('file.csv') verwenden, um CSV-Dateien zu lesen.
cache.put(query1, answer1)
print(fIm Cache speichern: {answer1})

print()

# Zweites Mal: ähnliche Frage
query2 = Wie liest man CSV in Python?
print(fAbfrage 2: {query2})
answer2 = cache.get(query2)
if answer2:
print(fCache-Treffer: {answer2})
else:
print(Cache-Miss, Modell wird aufgerufen...)

print()

# Drittes Mal: völlig andere Frage
query3 = Was ist maschinelles Lernen?
print(fAbfrage 3: {query3})
answer3 = cache.get(query3)
if answer3:
print(fCache-Treffer: {answer3})
else:
print(Cache-Miss, Modell wird aufgerufen...)

if __name__ == "__main__":
main()

Modell-Routing: Kleines Modell als Fallback

Nicht alle Anfragen benötigen ein großes Modell. Verwenden Sie ein kleines Modell für einfache Fragen und ein großes Modell für komplexe Fragen.

Beispiel

# Dateipfad: model_router.py
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class ModelConfig:
name: str
cost_per_1k_tokens: float
max_tokens: int
capabilities: List[str] # Liste der Fähigkeiten

class SmartModelRouter:
"""Intelligenter Modell-Router"""

def __init__(self):
self.models = {
"small": ModelConfig(
name="Qwen2.5-0.5B-Instruct",
cost_per_1k_tokens=0.001,
max_tokens=4096,
capabilities=["chat", "simple_qa", "summarization"]
),
"medium": ModelConfig(
name="Qwen2.5-7B-Instruct",
cost_per_1k_tokens=0.01,
max_tokens=8192,
capabilities=["chat", "qa", "coding", "reasoning"]
),
"large": ModelConfig(
name="Qwen2.5-72B-Instruct",
cost_per_1k_tokens=0.1,
max_tokens=32768,
capabilities=["complex_qa", "advanced_coding", "deep_reasoning"]
)
}

def _classify_query(self, query: str) -> str:
"""
Einfache Fragenklassifizierung (in echten Projekten können Sie einen Klassifikator verwenden)
Rückgabe: small/medium/large
"""
query_lower = query.lower()

# Schlüsselwörter für einfache Fragen
simple_keywords = [
"Hallo", "hallo", "Danke", "hi", "Auf Wiedersehen",
"Wetter heute", "Wie spät", "Datum",
"Hilf beim Übersetzen", "übersetzen",
"Kurze Einführung", "Was ist"
]

# Schlüsselwörter für komplexe Fragen
complex_keywords = [
"Code", "Schreibe", "Implementiere", "Code", "python",
"Analysiere", "Erkläre warum", "Leite ab", "Beweise",
"Vergleiche", "Stelle gegenüber", "Unterschied",
"Detailliert", "Tiefgehend", "Komplex"
]

# Prüfen, ob es eine einfache Frage ist
for keyword in simple_keywords:
if keyword in query_lower:
return "small"

# Prüfen, ob es eine komplexe Frage ist
for keyword in complex_keywords:
if keyword in query_lower:
return "large"

# Standardmäßig mittleres Modell verwenden
return "medium"

def route(self, query: str, user_tier: str = "free") -> str:
"""
Routing-Entscheidung
"""
# Zuerst die Frage klassifizieren
category = self._classify_query(query)

# Bezahlte Benutzer können ein besseres Modell verwenden
if user_tier == "pro":
upgrade_map = {
"small": "small",
"medium": "medium",
"large": "large"
}
category = upgrade_map[category]
elif user_tier == "enterprise":
upgrade_map = {
"small": "medium",
"medium": "large",
"large": "large"
}
category = upgrade_map.get(category, category)

# Modellnamen zurückgeben
return self.models[category].name

def get_cost_estimate(self, model_name: str, input_tokens: int, output_tokens: int) -> float:
"""Kostenschätzung"""
for model in self.models.values():
if model.name == model_name:
return (input_tokens + output_tokens) * model.cost_per_1k_tokens / 1000
return 0.0

# Verwendungsbeispiel
def main():
router = SmartModelRouter()

test_queries = [
"Hallo, ich würde gerne mehr erfahren",
"Was ist Python?",
"Hilf mir, einen Python-Quicksort zu schreiben",
Erstelle eine detaillierte Analyse der Transformer-Architektur.
]

print(Beispiel für intelligentes Routing)
print("="*50)

for query in test_queries:
model = router.route(query, user_tier="free")
cost = router.get_cost_estimate(model, 100, 200)
print(fAbfrage: {query})
print(fRoute zu: {model})
print(fGeschätzte Kosten: ${cost:.4f})
print()

# Verschiedene Benutzerstufen
print(Vergleich verschiedener Benutzerstufen)
print("-"*50)
query = Hilf mir, einen Python-Quicksort zu schreiben
for tier in ["free", "pro", "enterprise"]:
model = router.route(query, user_tier=tier)
print(f{tier} Benutzer: {model})

if __name__ == "__main__":
main()

Spot-Instanzen und Batch-Verarbeitung

Verwenden Sie On-Demand-Instanzen für Echtzeitanfragen (stabil, aber teuer) und Spot-Instanzen für die Offline-Batch-Verarbeitung (günstig, aber können unterbrochen werden).

Zusammenfassung mehrerer Kostenoptimierungsmethoden:

Strategie	Szenario	Kosteneinsparungen
Semantischer Cache	Viele wiederholte Abfragen	30-70%
Modell-Routing	Hoher Anteil einfacher Abfragen	50-80%
Modellquantisierung	Alle Szenarien	30-50%
Spot-Instanzen	Offline-Batchverarbeitung	60-90%
Auto-Skalierung	Große Traffic-Schwankungen	30-60%

Fehlerbehandlung und Degradation

In Produktionsumgebungen gibt es so etwas wie 'keine Fehler' nicht — nur 'wie man Fehler elegant behandelt'.

Degradationsstrategie

Beispiel

# Dateipfad: fallback_handler.py
import time
from typing import Optional
from dataclasses import dataclass
from enum import Enum

class ServiceStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
FAILED = "failed"

@dataclass
class FallbackResponse:
content: str
is_fallback: bool
fallback_reason: Optional[str] = None

class FallbackHandler:
"""Fallback-Handler"""

def __init__(self):
self.error_count = 0
self.last_error_time = 0
self.circuit_open = False
self.circuit_open_time = 0

# Konfiguration des Circuit Breakers
self.error_threshold = 5 # Auslösen nach 5 aufeinanderfolgenden Fehlern
self.circuit_timeout = 30 # Wiederherstellung nach 30 Sekunden offenem Circuit versuchen

def _call_primary_service(self, query: str) -> str:
"""Simuliert den Aufruf des Primärdienstes"""
# Dies durch einen echten Modellaufruf ersetzen
return "Dies ist die Antwort des Primärdienstes"

def _call_fallback_service(self, query: str) -> str:
"""Simuliert den Aufruf des Fallback-Dienstes (schneller, aber von geringerer Qualität)"""
# Könnte ein kleineres Modell, eine zwischengespeicherte Antwort oder sogar eine vordefinierte Vorlage sein
return "Entschuldigung, der Dienst ist vorübergehend ausgelastet. Dies ist eine vereinfachte Antwort."

def _call_static_fallback(self, query: str) -> str:
"""Statische Fallback-Antwort (letzte Verteidigungslinie)"""
return (
"Wir bitten vielmals um Entschuldigung, unser Dienst hat derzeit einige Probleme.\n"
"Bitte versuchen Sie es später erneut. Bei dringenden Problemen kontaktieren Sie bitte [email protected]."
)

def process_request(self, query: str) -> FallbackResponse:
Anfragen mit vollständiger Fallback-Logik bearbeiten

# 1. Prüfen, ob der Circuit Breaker offen ist
if self.circuit_open:
if time.time() - self.circuit_open_time > self.circuit_timeout:
# Timeout abgelaufen, Half-Open-Zustand versuchen, eine kleine Anzahl von Anfragen durchlassen
print("Circuit Breaker halb offen, Wiederherstellung wird versucht...")
self.circuit_open = False
else:
# Direkt auf Fallback zurückgreifen
print("Circuit Breaker ist offen, direkt auf Fallback zurückgreifen")
return FallbackResponse(
content=self._call_fallback_service(query),
is_fallback=True,
fallback_reason="circuit_open"
)

# 2. Primärdienst versuchen
try:
result = self._call_primary_service(query)
# Erfolg: Fehlerzähler zurücksetzen
self.error_count = 0
return FallbackResponse(
content=result,
is_fallback=False
)

except Exception as e:
print(f"Aufruf des Primärdienstes fehlgeschlagen: {e}")
self.error_count += 1
self.last_error_time = time.time()

# 3. Prüfen, ob Circuit Breaking erforderlich ist
if self.error_count >= self.error_threshold:
print("Zu viele Fehler, Circuit Breaker wird geöffnet")
self.circuit_open = True
self.circuit_open_time = time.time()

# 4. Fallback-Dienst versuchen
try:
print("Versuche, Fallback-Dienst aufzurufen")
result = self._call_fallback_service(query)
return FallbackResponse(
content=result,
is_fallback=True,
fallback_reason="primary_failed"
)
except Exception as e2:
print(f"Fallback-Dienst ebenfalls fehlgeschlagen: {e2}")

# 5. Letzte statische Antwort
return FallbackResponse(
content=self._call_static_fallback(query),
is_fallback=True,
fallback_reason="all_failed"
)

# Verwendungsbeispiel
def main():
handler = FallbackHandler()

print("Beispiel für Fallback-Behandlung")
print("="*50)

# Normale Anfrage
print("Normale Anfrage:")
resp = handler.process_request("Hallo")
print(f" Antwort: {resp.content}")
print(f" Ist Fallback? {resp.is_fallback}")
print()

# Fehler simulieren (in echten Projekten werden tatsächlich Ausnahmen ausgelöst)
# Hier demonstrieren wir einfach die Logik
print("Zusammenfassung der Fallback-Strategie:")
print(" 1. Primärdienst bevorzugen")
print(" 2. Primärdienst schlägt fehl → kleines Modell/Cache verwenden")
print(" 3. Aufeinanderfolgende Fehler → Circuit Breaking, um einen Avalanche-Effekt zu vermeiden")
print(" 4. Während Circuit Breaking → schnell Fallback-Antwort zurückgeben")
print(" 5. Letzte Verteidigungslinie → statische Antwort")

Canary-Release und Rollback

Rolle die neue Version nicht auf einmal für alle Benutzer aus. Gib sie zuerst 1% des Traffics, und wenn es keine Probleme gibt, dann 5%, 10%, 100%.

Wenn Probleme auftreten, musst du schnell ein Rollback durchführen können. Warte nicht, bis sich Benutzer lange beschwert haben, bevor du hektisch die Konfiguration änderst.

Weitere Erweiterungen