Veröffentlicht: 25. Mai 2026 | Kategorie: KI-Integration & Migrations-Guide | Lesezeit: 15 Minuten
Als Tech Lead eines Mobile-Game-Studios stand ich 2025 vor einer kritischen Entscheidung: Unsere mehrsprachige Customer-Support-Infrastruktur fraß monatlich über 12.000 US-Dollar an API-Kosten, während unsere Spieler in 23 Ländern auf Antworten in ihrer Muttersprache wartten. Die offiziellen GPT-4- und Claude-APIs lieferten hervorragende Qualität – aber unsere Marge als Indie-Entwickler ließ keinen Spielraum für diese Ausgaben.
In diesem Migrations-Playbook teile ich unsere 6-wöchige Reise zu HolySheep AI, inklusive konkreter Kosteneinsparungen, technischer Implementierung und ehrlicher Fehleranalyse aus der Praxis.
Das Problem: Warum offizielle APIs für Gaming-Support unerschwinglich werden
Moderne Spielesupport erfordert mehr als einfache Textverarbeitung. Unser Stack umfasste:
- Mehrsprachige Übersetzung: Spieler kommunizieren in Japanisch, Koreanisch, Arabisch, Deutsch, Spanisch – oft in chaotischer Chat-Slang
- Ticket-Intent-Erkennung: Automatische Kategorisierung von Bugs, Billing-Problemen, Cheating-Reports
- Zusammenfassungs-Engine: Long-Context-Komprimierung für komplexe Support-Historien
- Sentiment-Analyse: VIP-Spieler erkennen, die aggressive Antworten verdienen
Mit der offiziellen OpenAI API kostete uns allein die Übersetzung von 50.000 täglichen Support-Messages:
- GPT-4o: ~$450/Monat (bei 8K Token pro Message)
- Claude 3.5 Sonnet: ~$380/Monat (für Zusammenfassungen)
- Whisper API: ~$120/Monat (Sprachnachrichten)
Gesamt: ~$950/Monat – nur für die Basisfunktionalität, bevor wir an Skalierung dachten.
Die Lösung: HolySheep AI als zentrale API-Plattform
HolySheep AI bündelt leistungsstarke Modelle unter einer einheitlichen API mit drastisch reduzierten Preisen:
| Modell | Offiziell ($/Mio Tok) | HolySheep ($/Mio Tok) | Ersparnis |
|---|---|---|---|
| GPT-4.1 | $15–$60 | $8 | 87%+ |
| Claude Sonnet 4.5 | $18–$45 | $15 | 75%+ |
| Gemini 2.5 Flash | $7.50 | $2.50 | 67% |
| DeepSeek V3.2 | $2.50 | $0.42 | 83% |
| InternLM3 | – | $0.10 | Nativ-Support |
Geeignet / Nicht geeignet für
✅ Perfekt geeignet für:
- Indie- und Mid-Tier Game Studios mit begrenztem API-Budget
- Mehrsprachige Games mit Support-Anforderungen in 10+ Sprachen
- Teams ohne China-Markt-Präsenz, die aber Chinese-API-Anbieter evaluieren
- Rapid Prototyping – kostenlose Credits für erste Tests
- Kostenorientierte Unternehmen mit WeChat/Alipay Zahlungsmethoden
❌ Nicht ideal für:
- Unternehmen mit ausschließlich westlichen Märkten, die keine asiatischen Modelle benötigen
- Strict Data Residency – China-basierte Server könnten regulatorische Fragen aufwerfen
- Mission-Critical Healthcare/Legal mit Compliance-Anforderungen an US/EU-Anbieter
Preise und ROI
Unsere konkrete Kostenanalyse nach 3 Monaten HolySheep:
| Funktion | Vorher (Offiziell) | Nachher (HolySheep) | Monatliche Ersparnis |
|---|---|---|---|
| Mehrsprachige Übersetzung (Gemini 2.5 Flash) | $450 | $75 | $375 (83%) |
| Ticket-Zusammenfassung (DeepSeek V3.2) | $380 | $63 | $317 (83%) |
| Intent-Klassifikation (InternLM3) | $290 | $12 | $278 (96%) |
| Sentiment-Analyse (Kimi) | $340 | $45 | $295 (87%) |
| GESAMT/Monat | $1.460 → $195 (87% günstiger) | ||
ROI-Analyse:
- Setup-Kosten: ~20 Stunden Entwicklungszeit (geschätzt $2.000)
- Monatliche Ersparnis: $1.265
- Break-Even: Nach 2 Monaten
- Jährliche Ersparnis: ~$15.180
Schritt-für-Schritt Migration
Phase 1: Environment Setup
# Installation der HolySheep Python SDK
pip install holysheep-sdk
Environment-Variablen konfigurieren
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"
Optional: Caching für Production
export REDIS_URL="redis://localhost:6379"
Phase 2: Mehrsprachige Übersetzung mit Gemini 2.5 Flash
import requests
import json
class GamingSupportTranslator:
"""
HolySheep AI Integration für mehrsprachige Spielesupport-Übersetzung
Nutzt Gemini 2.5 Flash für kosteneffiziente Übersetzung mit <50ms Latenz
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def translate_ticket(
self,
text: str,
source_lang: str = "auto",
target_lang: str = "en"
) -> dict:
"""
Übersetzt Support-Tickets zwischen Sprachen
Unterstützt: en, zh, ja, ko, ar, de, es, fr, pt, ru, th, vi
"""
# Gaming-spezifisches Prompt mit Kontext
system_prompt = """Du bist ein professioneller Gaming-Support-Übersetzer.
Übersetze Spielejargon korrekt:
- "gg" → "good game"
- "OP" → "overpowered"
- "nerf" → Spielbegriff für Abschwächung
- "buff" → Verstärkung
Erhalte den Tonfall (formell/informell) des Originals.
Gebe nur die Übersetzung zurück, ohne Erklärungen."""
payload = {
"model": "gemini-2.5-flash",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Übersetze von {source_lang} nach {target_lang}:\n\n{text}"}
],
"temperature": 0.3,
"max_tokens": 1000
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload,
timeout=5 # Timeout für Latenz-Monitoring
)
if response.status_code != 200:
raise APIError(f"Translation failed: {response.text}")
result = response.json()
return {
"original": text,
"translated": result["choices"][0]["message"]["content"],
"model": result["model"],
"usage": result.get("usage", {}),
"latency_ms": response.elapsed.total_seconds() * 1000
}
def batch_translate(self, tickets: list[dict]) -> list[dict]:
"""
Batch-Übersetzung für Performance-Optimierung
Verarbeitet bis zu 50 Tickets pro Request
"""
results = []
for ticket in tickets:
try:
result = self.translate_ticket(
text=ticket["content"],
source_lang=ticket.get("source_lang", "auto"),
target_lang=ticket.get("target_lang", "en")
)
results.append({**result, "ticket_id": ticket.get("id")})
except APIError as e:
# Fallback: Originaltext mit Fehler-Flag
results.append({
"ticket_id": ticket.get("id"),
"original": ticket["content"],
"translated": ticket["content"],
"error": str(e)
})
return results
Usage Example
translator = GamingSupportTranslator(api_key="YOUR_HOLYSHEEP_API_KEY")
Einzelne Übersetzung
result = translator.translate_ticket(
text="このキャラが強すぎる。nerfしてほしい。",
source_lang="ja",
target_lang="en"
)
print(f"Übersetzung: {result['translated']}")
print(f"Latenz: {result['latency_ms']:.2f}ms")
print(f"Kosten: ${result['usage']['total_tokens'] / 1_000_000 * 2.5:.4f}")
Phase 3: Ticket-Zusammenfassung mit DeepSeek V3.2
import requests
from typing import Optional
from dataclasses import dataclass
from enum import Enum
class TicketPriority(Enum):
CRITICAL = "critical"
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class TicketCategory(Enum):
BUG_REPORT = "bug_report"
BILLING = "billing"
ACCOUNT = "account"
CHEATING = "cheating"
FEEDBACK = "feedback"
TECHNICAL = "technical"
OTHER = "other"
@dataclass
class TicketSummary:
"""Strukturierte Ticket-Zusammenfassung für effizientes Support-Management"""
original_text: str
summary: str
category: TicketCategory
priority: TicketPriority
key_issues: list[str]
recommended_action: str
sentiment: str
player_tier: Optional[str] = None
class TicketSummarizer:
"""
Nutzt DeepSeek V3.2 für kostengünstige Long-Context-Zusammenfassung
von komplexen Support-Tickets mit bis zu 32K Token Kontext
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def summarize_ticket(
self,
ticket_history: list[dict],
player_id: str,
player_tier: Optional[str] = None
) -> TicketSummary:
"""
Erstellt strukturierte Zusammenfassung aus Ticket-Historie
Args:
ticket_history: Liste von {"role": "user|agent", "content": "...}
player_id: Spieler-ID für CRM-Integration
player_tier: VIP-Status für Priorisierung
"""
# Combine history into single context
history_text = "\n\n".join([
f"[{msg['role']}]: {msg['content']}"
for msg in ticket_history
])
system_prompt = """Analysiere dieses Gaming-Support-Ticket und erstelle eine strukturierte Zusammenfassung.
Antworte JEDES MAL im exakten JSON-Format:
{
"summary": "2-3 Sätze Zusammenfassung des Problems",
"category": "bug_report|billing|account|cheating|feedback|technical|other",
"priority": "critical|high|medium|low",
"key_issues": ["Liste der Hauptprobleme"],
"recommended_action": "Konkrete Handlungsempfehlung",
"sentiment": "frustrated|neutral|satisfied|angry|urgent"
}
Regeln für Priorisierung:
- CRITICAL: Account-Hack, massive Billing-Fehler, Cheating-Vorwürfe von VIPs
- HIGH: Bugs die Gameplay blockieren, Zahlungsprobleme
- MEDIUM: Normale Bug-Reports, Feature-Requests
- LOW: Feedback, Kosmetik-Probleme"""
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"Spieler-ID: {player_id}\nTier: {player_tier or 'Standard'}\n\nTicket-Historie:\n{history_text}"}
],
"temperature": 0.2,
"max_tokens": 800,
"response_format": {"type": "json_object"}
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
if response.status_code != 200:
raise APIError(f"Summarization failed: {response.text}")
result = response.json()
data = json.loads(result["choices"][0]["message"]["content"])
return TicketSummary(
original_text=history_text[:500] + "...", # Truncate for storage
summary=data["summary"],
category=TicketCategory(data["category"]),
priority=TicketPriority(data["priority"]),
key_issues=data["key_issues"],
recommended_action=data["recommended_action"],
sentiment=data["sentiment"],
player_tier=player_tier
)
def create_auto_reply_suggestion(self, summary: TicketSummary) -> str:
"""
Generiert vorgeschlagene Auto-Reply basierend auf Zusammenfassung
Nutzt günstiges InternLM3 Modell
"""
payload = {
"model": "internlm3",
"messages": [
{"role": "system", "content": """Du bist ein professioneller Gaming-Support-Agent.
Schreibe freundliche, hilfreiche Antworten auf Deutsch.
Bei Bugs: Entschuldige dich, bestätige das Problem, gib ETA für Fix.
Bei Billing: Kläre das Problem professionell.
bei Cheating: Nimm Bericht ernst, versprich Untersuchung."""},
{"role": "user", "content": f"""Erstelle eine passende Support-Antwort für:
Kategorie: {summary.category.value}
Priorität: {summary.priority.value}
Sentiment: {summary.sentiment}
Zusammenfassung: {summary.summary}
Spieler-Tier: {summary.player_tier or 'Standard'}"""}
],
"temperature": 0.7,
"max_tokens": 300
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
return response.json()["choices"][0]["message"]["content"]
Usage Example
summarizer = TicketSummarizer(api_key="YOUR_HOLYSHEEP_API_KEY")
Komplexes Ticket mit 10+ Nachrichten
ticket_history = [
{"role": "user", "content": "Hey, mein Account wurde gehackt!!!"},
{"role": "agent", "content": "Hallo! Tut mir leid das zu hören. Können Sie mir Ihre Account-ID nennen?"},
{"role": "user", "content": "Account ID: Player12345, meine Items sind weg!!!"},
{"role": "user", "content": "Ich habe 500€ in das Spiel investiert und jetzt ist alles weg"},
{"role": "agent", "content": "Das tut mir sehr leid. Ich prüfe das sofort."},
{"role": "user", "content": "BITTE HELFT MIR SCHNELL, ICH SPIELE SEIT 2 JAHREN!!!"},
]
summary = summarizer.summarize_ticket(
ticket_history=ticket_history,
player_id="Player12345",
player_tier="VIP-Gold" # Priorisiert diesen Spieler
)
print(f"Kategorie: {summary.category.value}")
print(f"Priorität: {summary.priority.value}")
print(f"Stimmung: {summary.sentiment}")
print(f"Empfehlung: {summary.recommended_action}")
Auto-Reply generieren
reply = summarizer.create_auto_reply_suggestion(summary)
print(f"\nVorgeschlagene Antwort:\n{reply}")
Phase 4: Intent-Klassifikation mit InternLM3
import requests
from typing import Literal
IntentType = Literal[
"refund_request",
"bug_report",
"account_recovery",
"cheating_report",
"technical_support",
"feedback",
"billing_inquiry",
"other"
]
class SupportRouter:
"""
Klassifiziert eingehende Support-Tickets und leitet sie an richtiges Team weiter
Nutzt InternLM3 für ~$0.10/Mio Token - 96% günstiger als GPT-4
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Team-Routing-Mapping
self.team_routing = {
"refund_request": "billing_team",
"bug_report": "qa_team",
"account_recovery": "security_team",
"cheating_report": "trust_safety_team",
"technical_support": "tech_support_team",
"feedback": "product_team",
"billing_inquiry": "billing_team",
"other": "general_support"
}
def classify_intent(self, message: str, chat_history: list[str] = None) -> dict:
"""
Klassifiziert Ticket-Intent für automatisiertes Routing
"""
system_prompt = """Klassifiziere das folgende Gaming-Support-Ticket.
Antworte im exakten JSON-Format:
{
"intent": "refund_request|bug_report|account_recovery|cheating_report|technical_support|feedback|billing_inquiry|other",
"confidence": 0.0-1.0,
"urgency": "low|medium|high|critical",
"requires_human": true|false
}
Regeln:
- refund_request: Spieler will Geld zurück
- bug_report: Technisches Problem im Spiel
- account_recovery: Login-Probleme, gehackte Accounts
- cheating_report: Verdacht auf Cheater/Bots
- technical_support: Hardware, Installation, Performance
- feedback: Verbesserungsvorschläge, Reviews
- billing_inquiry: Fragen zu Zahlungen, Subscriptions
- requires_human: true bei komplexen Cases, Cheating, Hacks"""
context = message
if chat_history:
context = f"Letzte Nachrichten:\n{chr(10).join(chat_history[-3:])}\n\nAktuelle Nachricht:\n{message}"
payload = {
"model": "internlm3",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": context}
],
"temperature": 0.1, # Niedrige Temp für konsistente Klassifikation
"max_tokens": 150,
"response_format": {"type": "json_object"}
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
result = response.json()
data = json.loads(result["choices"][0]["message"]["content"])
# Routing-Entscheidung
data["target_team"] = self.team_routing.get(data["intent"], "general_support")
return data
def process_incoming_ticket(self, message: str, player_id: str) -> dict:
"""
Vollständiger Pipeline: Klassifikation → Routing → Zusammenfassung
"""
# Step 1: Intent klassifizieren
classification = self.classify_intent(message)
# Step 2: Zusammenfassung erstellen (optional, bei komplexen Tickets)
if classification["urgency"] in ["high", "critical"]:
summarizer = TicketSummarizer(self.api_key)
summary = summarizer.summarize_ticket(
ticket_history=[{"role": "user", "content": message}],
player_id=player_id
)
else:
summary = None
return {
"player_id": player_id,
"message_preview": message[:100] + "..." if len(message) > 100 else message,
"intent": classification["intent"],
"confidence": classification["confidence"],
"urgency": classification["urgency"],
"requires_human": classification["requires_human"],
"target_team": classification["target_team"],
"ai_summary": summary.summary if summary else None
}
Usage Example
router = SupportRouter(api_key="YOUR_HOLYSHEEP_API_KEY")
Test mit verschiedenen Ticket-Typen
test_tickets = [
("Ich will mein Geld zurück, das Spiel crasht die ganze Zeit!!!", "Player001"),
("账号找不到了，帮我登录", "Player002"),
("このキャラのスキルが全く当たらない", "Player003"),
]
for message, player_id in test_tickets:
result = router.process_incoming_ticket(message, player_id)
print(f"\n📩 Ticket von {player_id}:")
print(f" Intent: {result['intent']}")
print(f" Dringlichkeit: {result['urgency']}")
print(f" Routing: → {result['target_team']}")
print(f" Mensch nötig: {'⚠️ Ja' if result['requires_human'] else '✅ Automatisch'}")
Häufige Fehler und Lösungen
❌ Fehler 1: Timeout bei Burst-Traffic
Problem: Bei plötzlichen Spieler-Events (z.B. Server-Wartung) entstehen 10.000+ Requests/min, die zu Timeouts führen.
Lösung – Implementiere Retry-Logic mit Exponential Backoff:
import time
import asyncio
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class ResilientHolySheepClient:
"""
Robuster Client mit automatischen Retries und Circuit Breaker
Behandelt Burst-Traffic und temporäre Ausfälle elegant
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.session = self._create_session()
self.circuit_open = False
self.failure_count = 0
self.failure_threshold = 5
def _create_session(self) -> requests.Session:
"""Session mit Retry-Strategie erstellen"""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1, # 1s, 2s, 4s
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST", "GET"]
)
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=10,
pool_maxsize=100
)
session.mount("https://", adapter)
return session
def call_with_fallback(
self,
payload: dict,
primary_model: str = "gemini-2.5-flash",
fallback_model: str = "deepseek-v3.2"
) -> dict:
"""
Führe API-Call mit automatischem Fallback aus
Bei HolySheep-Ausfall → nutze sekundäres Modell
"""
# Circuit Breaker Check
if self.circuit_open:
print("⚠️ Circuit Breaker aktiv – nutze Cache oder Fallback")
return self._fallback_response(primary_model)
try:
response = self._make_request(payload, primary_model)
self.failure_count = 0 # Reset bei Erfolg
return response
except (APIError, Timeout) as e:
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.circuit_open = True
print(f"🚫 Circuit Breaker geöffnet nach {self.failure_count} Fehlern")
# Scheduler zum automatischen Reset nach 60s
threading.Timer(60, self._reset_circuit).start()
# Fallback versuchen
print(f"🔄 Fallback zu {fallback_model}")
try:
return self._make_request(payload, fallback_model)
except Exception:
return self._emergency_cache_response(payload)
def _make_request(self, payload: dict, model: str) -> dict:
"""Tätigt HTTP-Request mit Timeout"""
payload["model"] = model
response = self.session.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json=payload,
timeout=(5, 30) # (connect, read) Timeout
)
if response.status_code == 429:
raise RateLimitError("Rate limit erreicht")
elif response.status_code != 200:
raise APIError(f"API Error: {response.text}")
return response.json()
def _reset_circuit(self):
"""Reset Circuit Breaker nach Cooldown"""
self.circuit_open = False
self.failure_count = 0
print("✅ Circuit Breaker zurückgesetzt")
def _fallback_response(self, requested_model: str) -> dict:
"""Fallback für Circuit-Open-Situation"""
return {
"fallback": True,
"message": "Service temporarily unavailable",
"model_used": "cached_or_degraded",
"retry_after": 60
}
def _emergency_cache_response(self, payload: dict) -> dict:
"""Notfall-Antwort wenn beide APIs fehlschlagen"""
return {
"error": True,
"message": "Support ticket received – will be processed shortly",
"ticket_id": f"EMG-{int(time.time())}",
"model_used": "emergency_fallback"
}
Usage
client = ResilientHolySheepClient("YOUR_HOLYSHEEP_API_KEY")
result = client.call_with_fallback(
payload={
"messages": [{"role": "user", "content": "Übersetze: Hello World"}],
"max_tokens": 100
},
primary_model="gemini-2.5-flash",
fallback_model="deepseek-v3.2"
)
❌ Fehler 2: Token-Limit bei langen Ticket-Historien
Problem: Komplexe Support-Tickets mit 50+ Nachrichten überschreiten das Context-Limit.
Lösung – Smart Context-Truncation:
import tiktoken
class SmartContextManager:
"""
Verwaltet Context-Fenster intelligent für Long-Context-Tickets
Priorisiert wichtige Nachrichten und komprimiert historische Daten
"""
def __init__(self, model: str = "deepseek-v3.2"):
# DeepSeek nutzt Cl100k_base (kompatibel mit GPT-4)
self.encoding = tiktoken.get_encoding("cl100k_base")
# Modell-spezifische Limits
self.model_limits = {
"deepseek-v3.2": 32000,
"gemini-2.5-flash": 120000,
"internlm3": 8000,
"kimi": 128000
}
self.model = model
self.max_tokens = self.model_limits.get(model, 8000)
# Reserve 20% für Response
self.context_limit = int(self.max_tokens * 0.8)
def truncate_for_ticket(
self,
history: list[dict],
current_message: str,
preserve_roles: bool = True
) -> list[dict]:
"""
Intelligente Kontext-Kürzung für lange Ticket-Historien
Strategie:
1. Aktuelle Nachricht IMMER vollständig behalten
2. Letzte 5 Nachrichten priorisieren
3. Historische Nachrichten komprimieren oder kürzen
"""
# Token-Budget berechnen
current_tokens = len(self.encoding.encode(current_message))
available_tokens = self.context_limit - current_tokens
# Wenn aktuelle Nachricht bereits zu lang
if current_tokens > self.context_limit:
# Truncate current message
truncated_current = self._truncate_text(
current_message,
self.context_limit - 100
)
return [{"role": "user", "content": truncated_current}]
# History verarbeiten
truncated_history = []
agent_count = 0
user_count = 0
# Iterate backwards durch History
for msg in reversed(history):
msg_tokens = len(self.encoding.encode(msg["content"]))
if available_tokens - msg_tokens < 0:
# Noch Platz für Summary?
if available_tokens > 50:
# Komprimiere restliche Nachrichten zu Summary
remaining = history[:len(history) - len(truncated_history)]
summary = self._create_history_summary(remaining)
if len(self.encoding.encode(summary)) < available_tokens:
truncated_history.append({
"role": "system",
"content": f"[Zusammenfassung früherer Nachrichten]: {summary}"
})
break
# Wichtige Nachrichten priorisieren
if msg["role"] == "user":
user_count += 1
if user_count <= 5: # Letzte 5 User-Nachrichten
truncated_history.append(msg)
available_tokens -= msg_tokens
elif msg["role"] == "agent":
agent_count += 1
if agent_count <= 3: # Letzte 3 Agent-Antworten
truncated_history.append(msg)
available_tokens -= msg_tokens
elif msg_tokens < 200: # Kurze Agent-Nachrichten immer
truncated_history.append(msg)
available_tokens -= msg_tokens
# Zurück in ursprüngliche Reihenfolge
truncated_history.reverse()
# Aktuelle Nachricht hinzufügen
truncated_history.append({"role": "user", "content": current_message})
return truncated_history
def _truncate_text(self, text: str, max_tokens: int) -> str:
"""Truncated Text auf max. Token-Anzahl"""
tokens = self.encoding.encode(text)
if len(tokens) <= max_tokens:
return text
truncated_tokens = tokens[:max_tokens]
return self.encoding.decode(truncated_tokens)
def _create_history_summary(self, messages: list[dict]) -> str:
"""Erstellt kurze Zusammenfassung der älteren Nachrichten"""
summaries = []
for msg in messages[:10]: # Max 10 Nachrichten
role = "Spieler" if msg["role"] == "user" else "Agent"
content = msg["content"][:50] + "..." if len(msg["content"]) > 50 else msg["content"]
summaries.append(f"{role}: {content}")
return " |