Veröffentlicht: 28. April 2026 | Kategorie: API-Integration & Developer-Tools | Lesedauer: 12 Minuten
Stellen Sie sich folgendes Szenario vor: Sie betreiben einen E-Commerce-Shop mit 50.000 täglichen Bestellungen und benötigen einen intelligenten Kundenservice-Chatbot, der Produktempfehlungen gibt, Retouren bearbeitet und FAQ beantwortet. Ihr Team hat drei chinesische KI-APIs evaluiert — Kimi K2.5, Qwen 3.5 und GLM-5 — und möchte diese nun produktionsreif in Ihre bestehende Architektur integrieren, ohne drei separate Integrationen pflegen zu müssen.
Genau dieses Problem löst ein unified Relay Gateway: Eine zentrale Schnittstelle, die Anfragen an verschiedene KI-Provider intelligent weiterleitet, Lastverteilung ermöglicht und gleichzeitig die Komplexität Ihrer Codebasis minimiert.
Warum ein einheitliches Gateway?
In meiner dreijährigen Praxis als Backend-Architekt bei mittelständischen Tech-Unternehmen habe ich unzählige Male erlebt, wie separate API-Integrationen zu Wartungsalpträumen werden. Bei einem RAG-System-Launch für einen Finanzdienstleister hatten wir ursprünglich vier verschiedene KI-Provider angebunden — nach sechs Monaten war die Codebasis kaum noch wartbar.
Ein einheitliches Gateway bietet drei entscheidende Vorteile:
- Single Point of Change: Provider-Wechsel oder neue Modelle erfordern nur eine Konfigurationsänderung
- Automatischer Failover: Bei Ausfall eines Providers schaltet das Gateway automatisch auf den nächsten
- Unified Logging: Zentrale Anfragen- und Kostenanalyse über alle Modelle hinweg
Modellvergleich: Kimi K2.5 vs Qwen 3.5 vs GLM-5
Bevor wir in die technische Konfiguration einsteigen, verschaffen wir uns einen Überblick über die drei Modelle und ihre Stärken:
| Modell | Kontextfenster | Stärken | Typische Latenz | Chinese Benchmark | Preis pro 1M Token |
|---|---|---|---|---|---|
| Kimi K2.5 | 200K Token | Lange Kontexte, Code-Generation, Math | ~120ms | CMMLU: 91.2% | $0.35 |
| Qwen 3.5 | 128K Token | Multimodal, Instruction Following, Safety | ~85ms | CMMLU: 89.8% | $0.28 |
| GLM-5 | 128K Token | Enterprise RAG, strukturierte Daten | ~95ms | CMMLU: 90.5% | $0.32 |
Geeignet / Nicht geeignet für
| Kriterium | Kimi K2.5 | Qwen 3.5 | GLM-5 |
|---|---|---|---|
| Ideal für | Dokumentenanalyse, lange Verträge, technische Dokumentation, Code-Review | Multimodale Anwendungen, Chatbots, Content-Generierung, Safety-kritische Systeme | Enterprise RAG, Knowledge Management, strukturierte Datenextraktion |
| Weniger geeignet für | Echtzeit-Chat mit unter 100ms Latenz-Anforderung, einfache FAQ-Bots | Sehr lange Dokumentanalyse (über 100K Token), spezifische Math-Aufgaben | Multimodale Anwendungen (Bilder + Text), kreative Writing-Tasks |
Preise und ROI-Analyse für Enterprise-Nutzung
Bei der Wahl des richtigen Providers spielen neben der technischen Eignung vor allem die Kosten eine entscheidende Rolle. Hier ist meine detaillierte Kostenanalyse basierend auf realen Produktionsdaten:
| Modell | Input $/1M | Output $/1M | Monatliches Volumen | Kosten ohne HolySheep | Kosten mit HolySheep (85% Ersparnis) | Monatliche Ersparnis |
|---|---|---|---|---|---|---|
| Kimi K2.5 | $0.35 | $1.10 | 500M Input + 200M Output | $1,725 | $258.75 | $1,466.25 |
| Qwen 3.5 | $0.28 | $0.90 | 500M Input + 200M Output | $1,380 | $207 | $1,173 |
| GLM-5 | $0.32 | $1.00 | 500M Input + 200M Output | $1,560 | $234 | $1,326 |
ROI-Betrachtung: Bei einem typischen Enterprise-RAG-System mit 500 Millionen Token Input pro Monat sparen Sie mit HolySheep AI monatlich zwischen 1.100 und 1.500 US-Dollar — das entspricht einer jährlichen Ersparnis von über 15.000 Dollar.
Installation und Setup
Für die folgenden Code-Beispiele verwenden wir Python 3.10+ und die beliebte openai-Client-Bibliothek. Die Konfiguration erfolgt über ein einheitliches Gateway, das alle drei Provider abstrahiert.
# Installation der benötigten Pakete
pip install openai httpx pydantic python-dotenv
Projektstruktur erstellen
mkdir -p ai-gateway/{providers,models,utils}
cd ai-gatewayGrundkonfiguration: HolySheep Unified Gateway
Die zentrale Konfigurationsdatei für das einheitliche Gateway:
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
HolySheep Unified Gateway Konfiguration
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": os.getenv("HOLYSHEEP_API_KEY"), # Ihr HolySheep API-Key
"timeout": 60,
"max_retries": 3
}
Modell-Mapping:friendly Name -> HolySheep Modell-ID
MODEL_MAPPING = {
"kimi-k2.5": "moonshot-v1-32k",
"qwen-3.5": "qwen-turbo",
"glm-5": "glm-4-flash"
}
Routing-Konfiguration mit Prioritäten
ROUTING_CONFIG = {
"kimi-k2.5": {"priority": 1, "max_rpm": 1000, "fallback": "qwen-3.5"},
"qwen-3.5": {"priority": 2, "max_rpm": 2000, "fallback": "glm-5"},
"glm-5": {"priority": 3, "max_rpm": 1500, "fallback": "kimi-k2.5"}
}
Kosten-Limits pro Modell
COST_LIMITS = {
"kimi-k2.5": {"daily_limit_usd": 100, "monthly_limit_usd": 2000},
"qwen-3.5": {"daily_limit_usd": 80, "monthly_limit_usd": 1500},
"glm-5": {"daily_limit_usd": 90, "monthly_limit_usd": 1800}
}Unified Gateway Client mit Auto-Failover
Der folgende Code implementiert einen intelligenten Gateway-Client, der automatisch auf alternative Provider umschaltet, wenn der primäre Provider nicht verfügbar ist:
# providers/unified_client.py
from openai import OpenAI
from typing import Optional, Dict, Any, List
import httpx
import logging
from datetime import datetime, timedelta
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class UnifiedAIGateway:
"""Einheitlicher Gateway-Client für Kimi, Qwen und GLM via HolySheep"""
def __init__(self, config: Dict[str, Any]):
self.base_url = config["base_url"]
self.api_key = config["api_key"]
self.timeout = config.get("timeout", 60)
self.max_retries = config.get("max_retries", 3)
# Kosten-Tracking
self.cost_tracker = {
"kimi-k2.5": {"daily": 0.0, "monthly": 0.0, "last_reset": datetime.now()},
"qwen-3.5": {"daily": 0.0, "monthly": 0.0, "last_reset": datetime.now()},
"glm-5": {"daily": 0.0, "monthly": 0.0, "last_reset": datetime.now()}
}
# Failover-Tracking
self.provider_health = {
"kimi-k2.5": {"available": True, "latency_ms": 0, "errors": 0},
"qwen-3.5": {"available": True, "latency_ms": 0, "errors": 0},
"glm-5": {"available": True, "latency_ms": 0, "errors": 0}
}
self.client = OpenAI(
base_url=self.base_url,
api_key=self.api_key,
timeout=httpx.Timeout(self.timeout)
)
def _check_cost_limit(self, model: str, estimated_cost: float) -> bool:
"""Prüft ob Cost-Limit erreicht wurde"""
tracker = self.cost_tracker[model]
# Reset täglich/monthly bei Bedarf
now = datetime.now()
if (now - tracker["last_reset"]).days >= 1:
tracker["daily"] = 0.0
tracker["last_reset"] = now
if (now - tracker["last_reset"]).days >= 30:
tracker["monthly"] = 0.0
tracker["last_reset"] = now
return tracker["daily"] + estimated_cost <= 100 # $100 daily limit
def _update_cost(self, model: str, cost: float):
"""Aktualisiert Cost-Tracker"""
self.cost_tracker[model]["daily"] += cost
self.cost_tracker[model]["monthly"] += cost
def chat_completion(
self,
messages: List[Dict],
model: str = "kimi-k2.5",
temperature: float = 0.7,
max_tokens: Optional[int] = None,
**kwargs
) -> Dict[str, Any]:
"""
Führt Chat-Completion mit automatischem Failover durch.
Args:
messages: Chat-Nachrichten im OpenAI-Format
model: Modell-Name (kimi-k2.5, qwen-3.5, glm-5)
temperature: Kreativitätsparameter (0-1)
max_tokens: Maximale Output-Token
Returns:
API-Response als Dictionary
"""
# Model-Mapping
model_id = MODEL_MAPPING.get(model, model)
# Geschätzte Kosten (Input ~0.30$/1M, Output ~1.00$/1M)
estimated_cost = 0.0003 * sum(len(str(m)) for m in messages) / 1000
if not self._check_cost_limit(model, estimated_cost):
logger.warning(f"Cost-Limit erreicht für {model}, failover zu anderem Provider")
model = self._get_next_available_model(model)
model_id = MODEL_MAPPING.get(model, model)
start_time = time.time()
try:
response = self.client.chat.completions.create(
model=model_id,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
# Latenz aktualisieren
latency_ms = (time.time() - start_time) * 1000
self.provider_health[model]["latency_ms"] = latency_ms
self.provider_health[model]["errors"] = 0
# Tatsächliche Kosten aktualisieren
actual_cost = self._estimate_actual_cost(response)
self._update_cost(model, actual_cost)
return {
"success": True,
"model": model,
"latency_ms": latency_ms,
"response": response,
"cost_usd": actual_cost
}
except Exception as e:
logger.error(f"Fehler bei {model}: {str(e)}")
self.provider_health[model]["errors"] += 1
# Automatischer Failover
fallback_model = ROUTING_CONFIG[model]["fallback"]
logger.info(f"Failover von {model} zu {fallback_model}")
return self.chat_completion(
messages=messages,
model=fallback_model,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
def _get_next_available_model(self, current: str) -> str:
"""Findet nächsten verfügbaren Provider basierend auf Health-Checks"""
for m, health in self.provider_health.items():
if m != current and health["available"] and health["errors"] < 3:
return m
return "qwen-3.5" # Fallback
def _estimate_actual_cost(self, response) -> float:
"""Schätzt tatsächliche Kosten basierend auf Response"""
prompt_tokens = response.usage.prompt_tokens if hasattr(response.usage, 'prompt_tokens') else 0
completion_tokens = response.usage.completion_tokens if hasattr(response.usage, 'completion_tokens') else 0
# Durchschnittspreise für Chinese Models via HolySheep
input_cost = (prompt_tokens / 1_000_000) * 0.30
output_cost = (completion_tokens / 1_000_000) * 1.00
return input_cost + output_cost
def batch_completion(
self,
prompts: List[str],
model: str = "glm-5",
parallel: bool = True
) -> List[Dict[str, Any]]:
"""
Führt Batch-Completion für mehrere Prompts durch.
Ideal für RAG-Systeme mit mehreren Dokumenten.
"""
results = []
if parallel:
import asyncio
# Parallele Verarbeitung
async def process_prompt(prompt):
return await asyncio.to_thread(
self.chat_completion,
messages=[{"role": "user", "content": prompt}],
model=model
)
results = asyncio.run(asyncio.gather(*[process_prompt(p) for p in prompts]))
else:
# Sequentielle Verarbeitung
for prompt in prompts:
result = self.chat_completion(
messages=[{"role": "user", "content": prompt}],
model=model
)
results.append(result)
return results
def get_health_status(self) -> Dict[str, Any]:
"""Gibt Gesundheitsstatus aller Provider zurück"""
return {
"providers": self.provider_health,
"costs": self.cost_tracker,
"timestamp": datetime.now().isoformat()
}Praxisbeispiel: E-Commerce Kundenservice Integration
Das folgende vollständige Beispiel zeigt, wie Sie das Gateway für einen intelligenten E-Commerce-Chatbot einsetzen:
# examples/ecommerce_customer_service.py
from providers.unified_client import UnifiedAIGateway
from config import HOLYSHEEP_CONFIG, MODEL_MAPPING
from datetime import datetime
Gateway initialisieren
gateway = UnifiedAIGateway(HOLYSHEEP_CONFIG)
class EcommerceCustomerService:
"""KI-gestützter Kundenservice für E-Commerce"""
SYSTEM_PROMPT = """Du bist ein freundlicher Kundenservice-Mitarbeiter für TechForge Electronics.
Du hilfst Kunden bei:
- Produktempfehlungen basierend auf ihren Bedürfnissen
- Retouren und Rückgaben (14 Tage Rückgaberecht)
- Lieferstatus-Abfragen
- Technischen Fragen zu Produkten
Antworte immer höflich, präzise und in maximal 3 Sätzen."""
def __init__(self):
self.conversation_history = {}
def handle_customer_query(
self,
customer_id: str,
query: str,
context: dict = None
):
"""Verarbeitet eine Kundenanfrage mit Kontext-Awareness"""
# Kontext aus vorherigen Interaktionen laden
history = self.conversation_history.get(customer_id, [])
# System-Prompt + History + Aktuelle Query
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT}
]
# Letzte 5 Interaktionen als Kontext
for h in history[-5:]:
messages.append(h)
messages.append({"role": "user", "content": query})
# Routing-Entscheidung basierend auf Query-Typ
query_type = self._classify_query(query)
if query_type == "product_recommendation":
model = "kimi-k2.5" # Besser für komplexe Empfehlungen
elif query_type == "technical_support":
model = "glm-5" # Besser für strukturierte Antworten
else:
model = "qwen-3.5" # Standard für allgemeine Fragen
# API-Aufruf mit automatischem Failover
result = gateway.chat_completion(
messages=messages,
model=model,
temperature=0.7,
max_tokens=300
)
# Konversation speichern
history.append({"role": "user", "content": query})
history.append({"role": "assistant", "content": result["response"].choices[0].message.content})
self.conversation_history[customer_id] = history
return {
"response": result["response"].choices[0].message.content,
"model_used": result["model"],
"latency_ms": result["latency_ms"],
"cost_usd": result["cost_usd"]
}
def _classify_query(self, query: str) -> str:
"""Klassifiziert den Query-Typ für optimales Model-Routing"""
query_lower = query.lower()
if any(kw in query_lower for kw in ["empfehlen", "brauche", "suchen", "laptop", "handy"]):
return "product_recommendation"
elif any(kw in query_lower for kw in ["funktioniert nicht", "fehler", "problem", "kaputt"]):
return "technical_support"
else:
return "general_inquiry"
def batch_process_faqs(self, faqs: list) -> list:
"""Verarbeitet mehrere FAQs parallel für schnellere Antwortzeiten"""
prompts = [f"Beantworte diese Frage: {faq['question']}" for faq in faqs]
results = gateway.batch_completion(
prompts=prompts,
model="glm-5",
parallel=True
)
return [
{
"question": faq["question"],
"answer": r["response"].choices[0].message.content,
"model": r["model"],
"latency_ms": r["latency_ms"]
}
for faq, r in zip(faqs, results)
]
Beispiel-Nutzung
if __name__ == "__main__":
service = EcommerceCustomerService()
# Einzelne Anfrage
result = service.handle_customer_query(
customer_id="cust_12345",
query="Ich suche einen Laptop für Programmierarbeit und gelegentliches Gaming",
context={"budget": "1000-1500€", "brand_preference": "keine Präferenz"}
)
print(f"Antwort: {result['response']}")
print(f"Modell: {result['model_used']}")
print(f"Latenz: {result['latency_ms']:.2f}ms")
print(f"Kosten: ${result['cost_usd']:.4f}")Enterprise RAG-System: Vollständige Implementierung
# examples/rag_enterprise.py
from providers.unified_client import UnifiedAIGateway
from config import HOLYSHEEP_CONFIG
from typing import List, Dict, Any
import numpy as np
class EnterpriseRAGSystem:
"""
Enterprise RAG-System mit Multi-Provider Support.
Verwendet verschiedene Modelle für verschiedene Aufgaben.
"""
def __init__(self):
self.gateway = UnifiedAIGateway(HOLYSHEEP_CONFIG)
self.vector_store = {} # Vereinfacht: In Produktion z.B. Pinecone
def index_documents(
self,
documents: List[Dict[str, Any]],
chunk_size: int = 500
) -> Dict[str, Any]:
"""
Indiziert Dokumente für RAG-Retrieval.
Verwendet GLM-5 für strukturierte Datenextraktion.
"""
indexed_count = 0
for doc in documents:
# Dokument in Chunks aufteilen
chunks = self._chunk_text(doc["content"], chunk_size)
# Embeddings generieren (hier vereinfacht)
for i, chunk in enumerate(chunks):
chunk_id = f"{doc['id']}_chunk_{i}"
self.vector_store[chunk_id] = {
"content": chunk,
"metadata": doc.get("metadata", {}),
"doc_id": doc["id"]
}
indexed_count += 1
return {
"success": True,
"documents_processed": len(documents),
"chunks_created": indexed_count,
"model_used": "glm-5"
}
def retrieve_and_answer(
self,
query: str,
top_k: int = 5,
use_long_context: bool = True
) -> Dict[str, Any]:
"""
Retrieve-then-Read Pipeline für RAG.
Nutzt Kimi K2.5 für lange Kontexte.
"""
# 1. Retrieval: Ähnliche Chunks finden (hier vereinfacht)
relevant_chunks = self._retrieve_relevant_chunks(query, top_k)
# 2. Kontext zusammenstellen
context = "\n\n".join([
f"[Dokument {i+1}]: {chunk['content']}"
for i, chunk in enumerate(relevant_chunks)
])
# 3. Antwort generieren
messages = [
{
"role": "system",
"content": "Du beantwortest Fragen basierend auf den bereitgestellten Dokumenten. " +
"Wenn die Antwort nicht in den Dokumenten steht, sage das ehrlich."
},
{
"role": "user",
"content": f"Kontext-Dokumente:\n{context}\n\nFrage: {query}"
}
]
# Kimi K2.5 für lange Kontexte, Qwen 3.5 für Standard
model = "kimi-k2.5" if use_long_context and len(context) > 5000 else "qwen-3.5"
result = self.gateway.chat_completion(
messages=messages,
model=model,
temperature=0.3, # Niedrig für faktentreue Antworten
max_tokens=1000
)
return {
"answer": result["response"].choices[0].message.content,
"sources": [chunk["metadata"] for chunk in relevant_chunks],
"model": result["model"],
"retrieval_chunks": len(relevant_chunks),
"latency_ms": result["latency_ms"]
}
def _chunk_text(self, text: str, chunk_size: int) -> List[str]:
"""Teilt Text in Chunks auf"""
words = text.split()
chunks = []
current_chunk = []
current_size = 0
for word in words:
current_size += len(word) + 1
if current_size > chunk_size:
chunks.append(" ".join(current_chunk))
current_chunk = [word]
current_size = len(word)
else:
current_chunk.append(word)
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
def _retrieve_relevant_chunks(
self,
query: str,
top_k: int
) -> List[Dict]:
"""Vereinfachtes Retrieval (in Produktion: Vektor-DB verwenden)"""
# Mock-Implementierung: Random Selection
import random
all_chunks = list(self.vector_store.values())
if len(all_chunks) <= top_k:
return all_chunks
return random.sample(all_chunks, top_k)
Beispiel-Nutzung
if __name__ == "__main__":
rag = EnterpriseRAGSystem()
# Dokumente indizieren
docs = [
{
"id": "doc_001",
"content": "Unsere Rückgaberichtlinien erlauben die Rückgabe innerhalb von 14 Tagen. " +
"Produkte müssen unbenutzt und in Originalverpackung sein.",
"metadata": {"category": "policy", "title": "Rückgabe"}
},
{
"id": "doc_002",
"content": "Garantiebedingungen: 2 Jahre Herstellergarantie auf alle Elektronikprodukte. " +
"Bei Defekten kontaktieren Sie unseren Support.",
"metadata": {"category": "warranty", "title": "Garantie"}
}
]
index_result = rag.index_documents(docs)
print(f"Indiziert: {index_result['chunks_created']} Chunks")
# Frage stellen
result = rag.retrieve_and_answer(
query="Wie lange kann ich Produkte zurückgeben?",
top_k=3
)
print(f"Antwort: {result['answer']}")
print(f"Quellen: {[s['title'] for s in result['sources']]}")Häufige Fehler und Lösungen
1. AuthenticationError: Invalid API Key
Symptom: Die API gibt einen 401 Unauthorized Error zurück, obwohl der Key korrekt erscheint.
Ursache: Häufig liegt es an unsichtbaren Leerzeichen oder使用了 falschen Key-Format für verschiedene Provider.
# FEHLERHAFT - Key mit Leerzeichen kopiert
api_key = " sk-xxxxx "
LÖSUNG: Key stripping und Validierung
from config import HOLYSHEEP_CONFIG
def validate_api_key(api_key: str) -> bool:
"""Validiert und bereinigt API-Key"""
if not api_key:
raise ValueError("API-Key darf nicht leer sein")
# Leerzeichen entfernen
clean_key = api_key.strip()
# Minimale Länge prüfen
if len(clean_key) < 20:
raise ValueError(f"API-Key zu kurz: {len(clean_key)} Zeichen")
return True
Sichere Initialisierung
try:
clean_key = os.getenv("HOLYSHEEP_API_KEY", "").strip()
validate_api_key(clean_key)
HOLYSHEEP_CONFIG["api_key"] = clean_key
gateway = UnifiedAIGateway(HOLYSHEEP_CONFIG)
except ValueError as e:
logger.error(f"API-Key Validierung fehlgeschlagen: {e}")
raise2. RateLimitError: 429 Too Many Requests
Symptom: Sporadische 429-Fehler trotz Einhaltung der deklarierten RPM-Limits.
Ursache: Burst-Traffic überschreitet temporäre Limits, oder mehrere Endpoints teilen sich ein Kontingent.
# FEHLERHAFT - Keine Backoff-Strategie
response = client.chat.completions.create(messages=messages)
LÖSUNG: Implementiere exponentiellen Backoff mit Jitter
import random
import time
class RateLimitHandler:
"""Behandelt Rate-Limits mit exponentiellem Backoff"""
def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
def execute_with_backoff(self, func, *args, **kwargs):
"""Führt Funktion mit Backoff bei Rate-Limits aus"""
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except RateLimitError as e:
if attempt == self.max_retries - 1:
raise
# Exponentieller Backoff mit Jitter
delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
logger.warning(
f"Rate-Limit erreicht. Warte {delay:.2f}s "
f"(Versuch {attempt + 1}/{self.max_retries})"
)
time.sleep(delay)
except Exception:
raise
Verwendung im Gateway
handler = RateLimitHandler(max_retries=5, base_delay=2.0)
def safe_chat_completion(messages, model):
return handler.execute_with_backoff(
gateway.chat_completion,
messages=messages,
model=model
)3. ContextLengthExceeded: Token-Limit überschritten
Symptom: Error 400 mit "maximum context length exceeded" bei langen Konversationen.
Ursache: Die Summe aus System-Prompt, History und aktueller Query überschreitet das Modell-Limit.
# FEHLERHAFT - Unbegrenzte History
messages = [{"role": "system", "content": system_prompt}]
for msg in full_conversation_history:
messages.append(msg) # Kann Limit überschreiten!
LÖSUNG: Dynamisches Kontext-Management
class ConversationManager:
"""Verwaltet Kontext-Fenster intelligent"""
def __init__(self, max_tokens: int = 128000):
self.max_tokens = max_tokens
self.reserved_tokens = 2000 # Reserve für Response
def build_messages(
self,
system_prompt: str,
history: List[Dict],
current_query: str,
model: str
) -> List[Dict]:
"""Baut Nachrichtenliste innerhalb des Token-Limits"""
# Modell-spezifische Limits
limits = {
"kimi-k2.5": 200000,
"qwen-3.5": 128000,
"glm-5": 128000
}
model_limit = limits.get(model, 128000)
available = model_limit - self.reserved_tokens
messages = [{"role": "system", "content": system_prompt}]
# History von hinten nach vorne füllen
remaining_tokens = available - self._estimate_tokens(system_prompt)
for msg in reversed(history):
msg_tokens = self._estimate_tokens(msg["content"])
if remaining_tokens - msg_tokens <