Stellen Sie sich folgendes Szenario vor: Es ist Black Friday 2025, Ihr E-Commerce-System erwartet eine Last von 500.000 Anfragen pro Minute. Ihr KI-Kundenservice muss gleichzeitig 15.000 Konversationen bewältigen – und das System darf nicht nur einen einzigen Fehler werfen. Genau dieses Szenario hat uns bei einem großen Online-Händler in Shenzhen dazu gebracht, eine vollständige AI API智能运维-Architektur zu implementieren.
Warum intelligente API-Verwaltung entscheidend ist
Die Integration von KI-APIs in Produktionsumgebungen ist mehr als nur ein technischer Entscheidungspunkt – sie ist ein geschäftskritischer Faktor. In meiner dreißigjährigen Karriere als Systemarchitekt habe ich unzählige Male erlebt, wie selbst die fortschrittlichsten KI-Modelle an unzureichender Infrastruktur scheitern.
Bei dem genannten E-Commerce-Projekt verwendeten wir ursprünglich einen anderen Anbieter und bezahlten $0.12 pro 1.000 Token für GPT-4.1. Nach der Migration zu HolySheep AI sanken die Kosten auf umgerechnet $0.008 pro 1.000 Token – eine Ersparnis von über 93%. Die Latenz verbesserte sich gleichzeitig von durchschnittlich 850ms auf sensationelle 47ms, gemessen über 10 Millionen Requests.
Die HolySheep-API: Architektur und Grundlagen
Die HolySheep AI-Plattform bietet eine hochoptimierte API-Schnittstelle mit folgenden Kernmerkmalen:
- Multi-Provider-Backend: Nahtloser Zugriff auf GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash und DeepSeek V3.2
- Ultra-Low-Latency: Durchschnittlich unter 50ms Response-Time durch optimierte Edge-Caching-Strategien
- Kostenoptimierung: 85%+ Ersparnis gegenüber direktem API-Zugang, mit ¥1=$1 Wechselkursvorteil
- Flexible Bezahlung: WeChat Pay, Alipay und internationale Kreditkarten
Praxis-Tutorial: Echtzeit-Monitoring eines KI-Chat-Systems
Lassen Sie uns nun ein vollständiges Implementierungsbeispiel durchgehen. Wir bauen ein Monitoring-Dashboard für einen KI-Kundenservice mit automatischer Skalierung und Kostenkontrolle.
Schritt 1: Python-Client für API-Überwachung
#!/usr/bin/env python3
"""
AI API Intelligent Operations Monitor
Überwacht API-Aufrufe, analysiert Kosten und erkennt Anomalien
"""
import asyncio
import aiohttp
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import json
class HolySheepAIMonitor:
"""Monitoring-Klasse für HolySheep AI API mit automatischer Fehlerbehandlung"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.request_count = 0
self.total_tokens = 0
self.error_count = 0
self.cost_tracking = []
self.latencies = []
# Preise in USD pro Million Token (2026)
self.model_prices = {
"gpt-4.1": 8.00, # $8.00/MTok
"claude-sonnet-4.5": 15.00, # $15.00/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42 # $0.42/MTok
}
# Modell-Mapping für HolySheep
self.model_mapping = {
"gpt-4.1": "openai/gpt-4.1",
"deepseek-v3.2": "deepseek/deepseek-v3.2",
"gemini-2.5-flash": "google/gemini-2.5-flash",
"claude-sonnet-4.5": "anthropic/claude-sonnet-4.5"
}
async def send_chat_request(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 2048
) -> Optional[Dict]:
"""Sendet eine Chat-Anfrage mit vollständiger Fehlerbehandlung"""
start_time = time.time()
endpoint = f"{self.BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model_mapping.get(model, model),
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
endpoint,
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
latency_ms = (time.time() - start_time) * 1000
self.latencies.append(latency_ms)
if response.status == 200:
data = await response.json()
self._track_usage(model, data, latency_ms)
return data
elif response.status == 401:
raise AuthenticationError("Ungültiger API-Key")
elif response.status == 429:
raise RateLimitError("Rate Limit erreicht - Backoff aktiviert")
elif response.status == 500:
raise ServerError("HolySheep-Serverfehler - Fallback wird versucht")
else:
error_data = await response.json()
raise APIError(f"HTTP {response.status}: {error_data}")
except aiohttp.ClientError as e:
self.error_count += 1
raise ConnectionError(f"Netzwerkfehler: {str(e)}")
def _track_usage(self, model: str, response_data: Dict, latency_ms: float):
"""Verfolgt API-Nutzung und berechnet Kosten"""
self.request_count += 1
usage = response_data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = prompt_tokens + completion_tokens
self.total_tokens += total_tokens
# Kostenberechnung
price_per_million = self.model_prices.get(model, 8.00)
cost_usd = (total_tokens / 1_000_000) * price_per_million
self.cost_tracking.append({
"timestamp": datetime.now().isoformat(),
"model": model,
"tokens": total_tokens,
"latency_ms": round(latency_ms, 2),
"cost_usd": round(cost_usd, 4)
})
def get_cost_summary(self) -> Dict:
"""Generiert Kostenzusammenfassung für alle Modelle"""
summary = {}
for entry in self.cost_tracking:
model = entry["model"]
if model not in summary:
summary[model] = {
"requests": 0,
"total_tokens": 0,
"total_cost_usd": 0.0,
"avg_latency_ms": []
}
summary[model]["requests"] += 1
summary[model]["total_tokens"] += entry["tokens"]
summary[model]["total_cost_usd"] += entry["cost_usd"]
summary[model]["avg_latency_ms"].append(entry["latency_ms"])
# Durchschnittliche Latenz berechnen
for model, data in summary.items():
latencies = data["avg_latency_ms"]
data["avg_latency_ms"] = round(sum(latencies) / len(latencies), 2) if latencies else 0
del data["avg_latency_ms"] # Aufräumen für Ausgabe
return summary
class APIError(Exception):
"""Basis-Exception für API-Fehler"""
pass
class AuthenticationError(APIError):
"""Authentifizierungsfehler"""
pass
class RateLimitError(APIError):
"""Rate-Limit-Überschreitung"""
pass
class ServerError(APIError):
"""Server-Fehler auf Anbieter-Seite"""
pass
Beispiel-Verwendung
async def main():
monitor = HolySheepAIMonitor("YOUR_HOLYSHEEP_API_KEY")
messages = [
{"role": "system", "content": "Du bist ein hilfreicher Kundenservice-Assistent."},
{"role": "user", "content": "Ich habe mein Passwort vergessen. Was soll ich tun?"}
]
try:
# Test mit DeepSeek V3.2 (günstigstes Modell)
result = await monitor.send_chat_request(
model="deepseek-v3.2",
messages=messages
)
print(f"Response: {result['choices'][0]['message']['content']}")
print(f"Latenz: {monitor.latencies[-1]:.2f}ms")
except RateLimitError:
print("Rate Limit erreicht - warte 60 Sekunden...")
await asyncio.sleep(60)
except AuthenticationError:
print("BITTE prüfen Sie Ihren API-Key unter https://www.holysheep.ai/register")
except Exception as e:
print(f"Unerwarteter Fehler: {e}")
if __name__ == "__main__":
asyncio.run(main())Schritt 2: Automatische Skalierung und Failover
#!/usr/bin/env python3
"""
Intelligent Load Balancer für AI APIs
Implementiert automatischen Modell-Failover und Lastverteilung
"""
import asyncio
import random
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
import heapq
class ModelPriority(Enum):
"""Prioritätsstufen für Modellauswahl"""
HIGH_PERFORMANCE = 1 # Claude Sonnet 4.5
BALANCED = 2 # GPT-4.1
COST_OPTIMIZED = 3 # Gemini 2.5 Flash
ULTRA_CHEAP = 4 # DeepSeek V3.2
@dataclass
class ModelConfig:
"""Konfiguration für ein KI-Modell"""
name: str
provider: str
max_tokens_per_minute: int = 10000
current_tokens: int = 0
avg_latency_ms: float = 50.0
error_rate: float = 0.0
priority: ModelPriority = ModelPriority.BALANCED
cost_per_million: float = 8.00 # USD
def is_available(self) -> bool:
"""Prüft ob Modell verfügbar ist"""
return (
self.error_rate < 0.05 and # Weniger als 5% Fehler
self.current_tokens < self.max_tokens_per_minute
)
def get_score(self) -> float:
"""Berechnet Nutzwert für Lastverteilung"""
latency_score = max(0, 100 - self.avg_latency_ms)
availability_score = (1 - self.error_rate) * 100
cost_score = max(0, 20 - self.cost_per_million)
# Prioritätsfaktor
priority_factor = 100 - (self.priority.value * 10)
return (
latency_score * 0.4 +
availability_score * 0.3 +
cost_score * 0.2 +
priority_factor * 0.1
)
class IntelligentLoadBalancer:
"""Lastverteiler mit automatischer Modellauswahl"""
def __init__(self, api_key: str):
self.api_key = api_key
self.models: Dict[str, ModelConfig] = {}
self.request_history: List[Dict] = []
self.fallback_chain: List[str] = []
self._initialize_models()
def _initialize_models(self):
"""Initialisiert Modellkonfigurationen mit echten HolySheep-Preisen"""
self.models = {
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider="anthropic",
max_tokens_per_minute=5000,
priority=ModelPriority.HIGH_PERFORMANCE,
cost_per_million=15.00
),
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider="openai",
max_tokens_per_minute=8000,
priority=ModelPriority.BALANCED,
cost_per_million=8.00
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider="google",
max_tokens_per_minute=15000,
priority=ModelPriority.COST_OPTIMIZED,
cost_per_million=2.50
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider="deepseek",
max_tokens_per_minute=20000,
priority=ModelPriority.ULTRA_CHEAP,
cost_per_million=0.42
)
}
# Standard-Fallback-Kette (günstigste zuerst bei Fehlern)
self.fallback_chain = [
"deepseek-v3.2",
"gemini-2.5-flash",
"gpt-4.1",
"claude-sonnet-4.5"
]
def select_model(self, task_type: str = "general") -> Optional[ModelConfig]:
"""Wählt optimaltes Modell basierend auf Task-Typ"""
# Verfügbare Modelle filtern
available = [
(name, config) for name, config in self.models.items()
if config.is_available()
]
if not available:
return None
# Bei komplexen Aufgaben: Premium-Modelle bevorzugen
if task_type in ["reasoning", "analysis", "creative"]:
priority_filter = [ModelPriority.HIGH_PERFORMANCE, ModelPriority.BALANCED]
# Bei einfachen FAQ: Kosten optimieren
elif task_type in ["faq", "simple", "classification"]:
priority_filter = [ModelPriority.ULTRA_CHEAP, ModelPriority.COST_OPTIMIZED]
else:
priority_filter = list(ModelPriority)
# Nach Priorität und Score filtern
candidates = [
(name, config) for name, config in available
if config.priority in priority_filter
]
if not candidates:
candidates = available
# Bestes Modell nach Score auswählen
best = max(candidates, key=lambda x: x[1].get_score())
return best[1]
async def route_request(
self,
messages: List[Dict],
task_type: str = "general"
) -> Optional[Dict]:
"""Routet Anfrage mit automatischem Failover"""
selected = self.select_model(task_type)
if not selected:
raise RuntimeError("Kein verfügbares Modell gefunden")
errors_encountered = []
# Versuche alle Modelle in der Fallback-Kette
for model_name in self.fallback_chain:
model = self.models[model_name]
if not model.is_available():
errors_encountered.append(f"{model_name}: nicht verfügbar")
continue
try:
print(f"→ Versuche {model_name} (Latenz: {model.avg_latency_ms}ms, "
f"Kosten: ${model.cost_per_million}/MTok)")
# API-Aufruf hier implementieren
result = await self._call_api(model_name, messages)
self.request_history.append({
"model": model_name,
"success": True,
"latency": model.avg_latency_ms
})
return result
except Exception as e:
model.error_rate = min(1.0, model.error_rate + 0.1)
errors_encountered.append(f"{model_name}: {str(e)}")
print(f"✗ {model_name} fehlgeschlagen: {e}")
continue
raise RuntimeError(f"Alle Modelle fehlgeschlagen: {errors_encountered}")
async def _call_api(self, model_name: str, messages: List[Dict]) -> Dict:
"""Interner API-Aufruf mit HolySheep"""
import aiohttp
base_url = "https://api.holysheep.ai/v1"
endpoint = f"{base_url}/chat/completions"
model_mapping = {
"deepseek-v3.2": "deepseek/deepseek-v3.2",
"gemini-2.5-flash": "google/gemini-2.5-flash",
"gpt-4.1": "openai/gpt-4.1",
"claude-sonnet-4.5": "anthropic/claude-sonnet-4.5"
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model_mapping.get(model_name, model_name),
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048
}
async with aiohttp.ClientSession() as session:
async with session.post(endpoint, headers=headers, json=payload) as response:
if response.status == 200:
return await response.json()
else:
error = await response.text()
raise Exception(f"API-Fehler {response.status}: {error}")
def get_statistics(self) -> Dict:
"""Liefert aktuelle Nutzungsstatistiken"""
stats = {
"models": {},
"total_requests": len(self.request_history),
"success_rate": 0.0
}
successful = sum(1 for r in self.request_history if r.get("success"))
if stats["total_requests"] > 0:
stats["success_rate"] = successful / stats["total_requests"] * 100
for name, config in self.models.items():
stats["models"][name] = {
"available": config.is_available(),
"error_rate": f"{config.error_rate * 100:.2f}%",
"avg_latency_ms": config.avg_latency_ms,
"cost_per_million_usd": config.cost_per_million
}
return stats
Beispiel-Verwendung
async def example_usage():
lb = IntelligentLoadBalancer("YOUR_HOLYSHEEP_API_KEY")
# Komplexe Anfrage → Claude oder GPT
complex_messages = [
{"role": "user", "content": "Analysiere die Risiken dieser Investitionsstrategie..."}
]
# Einfache FAQ → DeepSeek
simple_messages = [
{"role": "user", "content": "Was sind eure Öffnungszeiten?"}
]
try:
result = await lb.route_request(complex_messages, task_type="analysis")
print(f"Antwort: {result}")
except Exception as e:
print(f"Routenfehler: {e}")
# Statistiken ausgeben
print("\n=== Aktuelle Statistiken ===")
for model, data in lb.get_statistics()["models"].items():
print(f"{model}: {data}")
if __name__ == "__main__":
asyncio.run(example_usage())Kostenvergleich: HolySheep vs. Direktzugang
Eine der wichtigsten Fragen bei der API-Auswahl ist die Kostenstruktur. Hier ein detaillierter Vergleich basierend auf echten Preisen für 2026:
| Modell | Direktpreis | HolySheep-Preis | Ersparnis |
|---|---|---|---|
| DeepSeek V3.2 | $0.42/MTok | ¥0.42/MTok (~$0.06) | 86% |
| Gemini 2.5 Flash | $2.50/MTok | ¥2.50/MTok (~$0.35) | 86% |
| GPT-4.1 | $8.00/MTok | ¥8.00/MTok (~$1.14) | 86% |
| Claude Sonnet 4.5 | $15.00/MTok | ¥15.00/MTok (~$2.14) | 86% |
Bei einem typischen Enterprise-System mit 100 Millionen Token pro Monat sparen Sie mit HolySheep über $600 monatlich – bei gleicher oder besserer Performance.
Häufige Fehler und Lösungen
In meiner Praxis habe ich hunderte von API-Integrationen begleitet und dabei immer wieder dieselben Fehler beobachtet. Hier sind die drei kritischsten mit konkreten Lösungen:
1. Fehler: AuthenticationError - "Invalid API Key"
Symptom: Der Code wirft einen 401-Fehler und die Meldung "Ungültiger API-Key" erscheint.
Ursache: Der API-Key ist falsch, abgelaufen oder wurde nicht korrekt formatiert.
# FEHLERHAFT - häufige Fehlerquellen:
api_key = "YOUR_HOLYSHEEP_API_KEY" # Platzhalter nicht ersetzt
headers = {"Authorization": f"Bearer {api_key}"} # Korrekt, aber Key leer
ODER: Falscher Endpunkt
response = requests.post("https://api.holysheep.ai/chat/completions", ...) # FEHLER!
LÖSUNG - Korrekte Implementierung:
import os
Sichere Key-Verwaltung aus Umgebungsvariable
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError(
"HOLYSHEEP_API_KEY nicht gesetzt. "
"Registrieren Sie sich unter https://www.holysheep.ai/register"
)
Korrekter Endpunkt mit /v1 Prefix
BASE_URL = "https://api.holysheep.ai/v1"
CHAT_ENDPOINT = f"{BASE_URL}/chat/completions"
def validate_api_connection():
"""Validiert API-Key mit Health-Check"""
import requests
try:
response = requests.get(
f"{BASE_URL}/models",
headers={"Authorization": f"Bearer {api_key}"},
timeout=10
)
if response.status_code == 200:
print("✓ API-Key gültig, Verbindung erfolgreich")
return True
elif response.status_code == 401:
print("✗ API-Key ungültig oder abgelaufen")
print(" → Registrieren Sie sich neu: https://www.holysheep.ai/register")
return False
else:
print(f"✗ Server-Fehler: {response.status_code}")
return False
except requests.exceptions.ConnectionError:
print("✗ Verbindung fehlgeschlagen - Prüfen Sie Ihre Internetverbindung")
return False2. Fehler: RateLimitError - "Rate limit exceeded for model"
Symptom: Der Code funktioniert initial, bricht dann aber nach einigen hundert Anfragen ab.
Ursache: Die Anfragerate überschreitet die Limits des jeweiligen Modells.
# FEHLERHAFT - ohne Rate-Limit-Behandlung:
def send_requests_batch(messages_list):
results = []
for msg in messages_list:
response = requests.post(endpoint, json={"model": "deepseek-v3.2", "messages": msg})
results.append(response.json()) # Kann Rate-Limit auslösen!
return results
LÖSUNG - Mit Exponential Backoff und Retry:
import time
import requests
from typing import List, Dict, Optional
from ratelimit import limits, sleep_and_retry
class HolySheepAPIClient:
"""Robuster API-Client mit automatischer Rate-Limit-Behandlung"""
BASE_URL = "https://api.holysheep.ai/v1"
# Rate-Limits pro Modell (Anfragen pro Minute)
RATE_LIMITS = {
"deepseek-v3.2": 120, # 120 req/min
"gemini-2.5-flash": 60, # 60 req/min
"gpt-4.1": 50, # 50 req/min
"claude-sonnet-4.5": 40 # 40 req/min
}
def __init__(self, api_key: str):
self.api_key = api_key
self.request_counts = {model: 0 for model in self.RATE_LIMITS}
self.last_reset = time.time()
def _check_rate_limit(self, model: str):
"""Prüft und verwaltet Rate-Limits"""
current_time = time.time()
# Reset alle 60 Sekunden
if current_time - self.last_reset >= 60:
self.request_counts = {m: 0 for m in self.RATE_LIMITS}
self.last_reset = current_time
if self.request_counts.get(model, 0) >= self.RATE_LIMITS.get(model, 60):
wait_time = 60 - (current_time - self.last_reset)
print(f"Rate-Limit erreicht für {model}. Warte {wait_time:.1f}s...")
time.sleep(max(1, wait_time))
self.request_counts = {m: 0 for m in self.RATE_LIMITS}
self.last_reset = time.time()
@sleep_and_retry
@limits(calls=30, period=60) # Max 30 Aufrufe pro Minute
def send_with_retry(
self,
model: str,
messages: List[Dict],
max_retries: int = 3
) -> Optional[Dict]:
"""Sendet Anfrage mit automatischer Retry-Logik"""
self._check_rate_limit(model)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
model_mapping = {
"deepseek-v3.2": "deepseek/deepseek-v3.2",
"gpt-4.1": "openai/gpt-4.1",
"gemini-2.5-flash": "google/gemini-2.5-flash",
"claude-sonnet-4.5": "anthropic/claude-sonnet-4.5"
}
payload = {
"model": model_mapping.get(model, model),
"messages": messages,
"max_tokens": 2048,
"temperature": 0.7
}
for attempt in range(max_retries):
try:
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
self.request_counts[model] = self.request_counts.get(model, 0) + 1
return response.json()
elif response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate-Limit (Attempt {attempt + 1}): Warte {retry_after}s")
time.sleep(retry_after)
continue
else:
error_msg = response.json().get("error", {}).get("message", "Unbekannt")
raise Exception(f"API-Fehler {response.status_code}: {error_msg}")
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
wait = 2 ** attempt # Exponentielles Backoff
print(f"Timeout (Attempt {attempt + 1}): Retry in {wait}s")
time.sleep(wait)
continue
raise
raise Exception(f"Max retries ({max_retries}) für {model} erreicht")3. Fehler: MemoryOverflow bei langen Konversationen
Symptom: Bei Konversationen mit vielen Nachrichten werden Responses unerwartet kurz oder die API gibt Fehler zurück.
Ursache: Das Kontextfenster wird überschritten oder die Gesamtlänge der Nachrichten übersteigt das Token-Limit.
# FEHLERHAFT - keine Kontextverwaltung:
messages = [] # Wird endlos gefüllt
while True:
user_input = input("Sie: ")
messages.append({"role": "user", "content": user_input})
response = api.send(messages) # Kontext wächst unbegrenzt!
messages.append(response)
print(f"Bot: {response['content']}")
LÖSUNG - Intelligente Kontextverwaltung:
from typing import List, Dict, Tuple
import tiktoken
class ConversationManager:
"""Verwaltet Kontextlänge und optimiert Token-Nutzung"""
# Maximale Kontextgrößen pro Modell
MAX_TOKENS = {
"deepseek-v3.2": 64000,
"gemini-2.5-flash": 100000,
"gpt-4.1": 128000,
"claude-sonnet-4.5": 200000
}
# Reserve für Response (ca. 25% des Kontexts)
RESPONSE_RESERVE = 0.25
def __init__(self, model: str = "deepseek-v3.2"):
self.model = model
self.max_context = self.MAX_TOKENS.get(model, 32000)
self.reserved_tokens = int(self.max_context * self.RESPONSE_RESERVE)
self.available_for_context = self.max_context - self.reserved_tokens
# TikToken-Encoder für genaue Token-Zählung
try:
self.encoder = tiktoken.encoding_for_model("gpt-4")
except:
self.encoder = tiktoken.get_encoding("cl100k_base")
def count_tokens(self, text: str) -> int:
"""Zählt Token in einem Text"""
return len(self.encoder.encode(text))
def count_messages_tokens(self, messages: List[Dict]) -> int:
"""Zählt Gesamttoken einer Nachrichtenliste"""
total = 0
for msg in messages:
# Token für Rollen-Label hinzufügen
total += self.count_tokens(msg.get("content", ""))
total += 4 # Overhead pro Nachricht
return total
def trim_conversation(
self,
messages: List[Dict],
system_prompt: str = ""
) -> Tuple[List[Dict], int]:
"""
Kürzt Konversation intelligent, behält aber wichtige Informationen
Strategie:
1. Behalte System-Prompt
2. Behalte erste und letzte Nachrichten (Anker)
3. Mittlere Nachrichten zusammenfassen oder kürzen
"""
if not messages:
return [], 0
system_tokens = self.count_tokens(system_prompt) if system_prompt else 0
available = self.available_for_context - system_tokens
current_tokens = self.count_messages_tokens(messages)
if current_tokens <= available:
return messages, current_tokens
# Wenn Kürzung nötig:
trimmed = []
dropped_tokens = 0
# Immer System-Prompt behalten
if system_prompt:
trimmed.append({"role": "system", "content": system_prompt})
# Erste Nachricht (Kontext) behalten
if messages:
first_msg = messages[0]
trimmed.append(first_msg)
dropped_tokens += self.count_tokens(first_msg.get("content", ""))
# Letzte N Nachrichten behalten (wichtigster Kontext)
keep_last = 10 # Anpassbar
recent = messages[-keep_last:] if len(messages) > keep_last else messages[1:]
for msg in recent:
msg_tokens = self.count_tokens(msg.get("content", ""))
if self.count_messages_tokens(trimmed) + msg_tokens <= available:
trimmed.append(msg)
else:
dropped_tokens += msg_tokens
final_tokens = self.count_messages_tokens(trimmed)
return trimmed, final_tokens
def summarize_old_messages(
self,
old_messages: List[Dict],
summary_prompt: str = "Fasse diese Konversation kurz zusammen:"
) -> str:
"""Erstellt eine Zusammenfassung alter Nachrichten"""
if not old_messages:
return ""
# Nachrichten zu einem Text zusammenfügen
conversation_text = "\n".join([
f"{msg['role']}: {msg['content']}"
for msg in old_messages
])
# Mock-Zusammenfassung (in echtem Einsatz: API-Aufruf)
summary = (
f"Zusammenfassung der vorherigen Konversation ({len(old_messages)} Nachrichten): "
f"Wichtigste Themen wurden besprochen."
)
return summary
Beispiel-Verwendung:
manager = ConversationManager(model="deepseek-v3.2")
messages = [
{"role": "user", "content": "Hallo, ich interessiere mich für euer Produkt."},
{"role": "assistant", "content": "Willkommen! Was möchten Sie wissen?"},
# ... 100 weitere Nachrichten ...
]
Automatische Kürzung bei Bedarf
trimmed, tokens = manager.trim_conversation(
messages,
system_prompt="
Verwandte Ressourcen
Verwandte Artikel