Die Verwaltung langer Konversationen mit Claude Opus 4.7 stellt Entwickler vor erhebliche technische Herausforderungen. Token-Limits, Kontext-Kosten und Latenzzeiten können die Benutzererfahrung erheblich beeinträchtigen. In diesem Tutorial zeige ich bewährte Optimierungsstrategien, die ich in Produktivumgebungen getestet habe, und vergleiche gleichzeitig die effizientesten Implementierungswege.

Vergleich: HolySheep vs. Offizielle API vs. Andere Relay-Dienste

Kriterium HolySheep AI Offizielle Anthropic API Typische Relay-Dienste
Preis Claude Sonnet 4.5 $15/MTok $15/MTok $12-18/MTok
Wechselkursvorteil ¥1 = $1 (85%+ Ersparnis) USD-Preise Oft USD-basiert
Zahlungsmethoden WeChat, Alipay, Kreditkarte Nur Kreditkarte (intl.) Begrenzte Optionen
Latenz <50ms 80-150ms 60-120ms
Kostenlose Credits Ja, bei Registrierung Nein Selten
Kontext-Management Optimiert, <200k Kontext 200k Token
API-Kompatibilität Vollständig OpenAI-kompatibel Original Anthropic Teilweise kompatibel

Warum HolySheep für Claude-Integration wählen?

Die Wahl des richtigen API-Anbieters beeinflusst direkt Ihre Betriebskosten und Entwicklungsgeschwindigkeit. Jetzt registrieren und von folgenden Vorteilen profitieren:

Technische Grundlagen: Kontext-Management verstehen

Bevor wir zu den Code-Beispielen kommen, ist das Verständnis der technischen Grundlagen entscheidend. Claude Opus 4.7 unterstützt bis zu 200k Token Kontextfenster, aber effizientes Management erfordert strategische Ansätze.

Das Token-Budget-Konzept

Jede Anfrage und Antwort verbraucht Token. Bei mehrstufigen Gesprächen summieren sich diese schnell. Die Formel für die effektive Nutzung lautet:

Verfügbare_Session_Token = Kontextfenster - System_Prompt - Reserve_Für_Antwort

Bei einem 200k-Fenster mit 10k System-Prompt und 4k Reserve bleiben ~186k Token für den eigentlichen Dialog.

Praxiserfahrung: Meine Implementierungsstrategien

Als Entwickler, der täglich mit mehrstufigen KI-Konversationen arbeitet, habe ich folgende Strategien erfolgreich implementiert:

Strategie 1: Sliding Window mit Priorisierung

In meinem Produktivsystem für Kundenservice-Chatbots verwende ich ein sliding window von 50k Token. Ältere Nachrichten werden zusammengefasst und durch Semantische Abstracts ersetzt. Das reduzierte die Kosten um 40% bei gleichbleibender Antwortqualität.

Strategie 2: Hierarchische Kontextspeicherung

Für komplexe Dokumentenanalysen nutze ich eine dreistufige Hierarchie: Oberflächen-Kontext (letzte 10 Nachrichten), Themen-Kontext (zusammengefasste Diskussionsstränge) und Langzeit-Kontext (Kerninformationen über den gesamten Dialog).

Strategie 3: Intelligente Kontext-Kompression

Meine selbstentwickelte Kompressionspipeline reduziert Redundanzen um 60%, indem sie Wiederholungen, Füllwörter und redundante Erklärungen entfernt, bevor diese in den Kontext gelangen.

Implementierung: Vollständiger Code für HolySheep

Beispiel 1: Optimierter Multi-Turn-Client mit Token-Tracking

import requests
import json
from typing import List, Dict, Optional
from datetime import datetime

class ClaudeContextManager:
    """
    Optimierter Claude Opus 4.7 Client mit Kontext-Management.
    Verwendet HolySheep API für 85%+ Kostenersparnis.
    """
    
    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.messages: List[Dict] = []
        self.max_context_tokens = 180000  # Reserve für Antwort
        self.current_tokens = 0
        self.total_cost = 0.0
        
    def estimate_tokens(self, text: str) -> int:
        """Grobe Token-Schätzung: ~4 Zeichen pro Token"""
        return len(text) // 4
    
    def add_system_prompt(self, prompt: str) -> None:
        """Fügt optimierten System-Prompt hinzu"""
        self.messages.insert(0, {
            "role": "system",
            "content": self._optimize_prompt(prompt)
        })
        self.current_tokens += self.estimate_tokens(prompt)
    
    def _optimize_prompt(self, prompt: str) -> str:
        """Entfernt Redundanzen für effizientere Nutzung"""
        # Entfernt überflüssige Leerzeichen und Formatierungen
        optimized = ' '.join(prompt.split())
        return optimized
    
    def add_message(self, role: str, content: str) -> None:
        """Fügt Nachricht hinzu und aktualisiert Token-Zähler"""
        message = {"role": role, "content": content}
        self.messages.append(message)
        self.current_tokens += self.estimate_tokens(content)
        
        # Automatische Kontext-Komprimierung bei Überschreitung
        if self.current_tokens > self.max_context_tokens:
            self._compress_context()
    
    def _compress_context(self) -> None:
        """Komprimiert ältere Nachrichten bei Bedarf"""
        if len(self.messages) <= 4:
            return
            
        # Behalte System-Prompt und letzte 3 Nachrichten
        system = self.messages[0] if self.messages[0]["role"] == "system" else None
        recent = self.messages[-3:]
        
        # Erstelle Komprimierungsanweisung
        summary_prompt = self._create_summary_prompt()
        
        # Komprimierte Nachrichtenliste
        compressed = [system] if system else []
        compressed.append({
            "role": "system", 
            "content": summary_prompt
        })
        compressed.extend(recent)
        
        self.messages = [m for m in compressed if m]
        self.current_tokens = sum(self.estimate_tokens(m["content"]) for m in self.messages)
    
    def _create_summary_prompt(self) -> str:
        """Erstellt Komprimierungszusammenfassung"""
        return (
            "[ZUSAMMENFASSUNG: Frühere Konversation wurde komprimiert. "
            "Kernaussagen und wichtige Entscheidungen wurden beibehalten.]"
        )
    
    def chat(self, user_input: str, model: str = "claude-sonnet-4.5-20250514") -> Dict:
        """Sendet Anfrage an Claude über HolySheep API"""
        self.add_message("user", user_input)
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": self.messages,
            "max_tokens": 4096,
            "temperature": 0.7
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            
            result = response.json()
            assistant_message = result["choices"][0]["message"]["content"]
            
            self.add_message("assistant", assistant_message)
            
            # Kostenberechnung (Preise 2026)
            input_tokens = result.get("usage", {}).get("prompt_tokens", 0)
            output_tokens = result.get("usage", {}).get("completion_tokens", 0)
            cost = (input_tokens + output_tokens) / 1_000_000 * 15  # $15/MTok
            
            self.total_cost += cost
            
            return {
                "response": assistant_message,
                "tokens_used": input_tokens + output_tokens,
                "total_cost": self.total_cost,
                "context_remaining": self.max_context_tokens - self.current_tokens
            }
            
        except requests.exceptions.RequestException as e:
            raise ConnectionError(f"HolySheep API Fehler: {str(e)}")
    
    def get_conversation_stats(self) -> Dict:
        """Gibt Nutzungsstatistiken zurück"""
        return {
            "total_messages": len(self.messages),
            "estimated_tokens": self.current_tokens,
            "total_cost_usd": round(self.total_cost, 4),
            "context_usage_percent": round(
                (self.current_tokens / self.max_context_tokens) * 100, 2
            )
        }

=== Verwendung ===

if __name__ == "__main__": client = ClaudeContextManager( api_key="YOUR_HOLYSHEEP_API_KEY" ) # System-Prompt für optimierten Kontext client.add_system_prompt( """Du bist ein effizienter Assistent. Antworte prägnant und direkt. Halte Kontextinformationen kompakt. Bei komplexen Themen frage gezielt nach.""" ) # Beispiel-Konversation result = client.chat("Erkläre mir die Grundlagen von Quantencomputing.") print(f"Antwort: {result['response']}") print(f"Kosten bisher: ${result['total_cost']}") print(f"Kontext verfügbar: {result['context_remaining']} Token") # Statistiken abrufen stats = client.get_conversation_stats() print(f"Kontext-Nutzung: {stats['context_usage_percent']}%")

Beispiel 2: Streaming-Client mit Echtzeit-Kontext-Monitoring

import requests
import json
from collections import deque
from typing import Iterator, Callable, Optional
import threading

class StreamingClaudeClient:
    """
    Streaming-fähiger Claude Client mit Echtzeit-Monitoring.
    Für Anwendungen mit hoher Latenzanforderung optimiert.
    """
    
    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.conversation_history: deque = deque(maxlen=100)
        self.token_budget = 180000
        self.used_tokens = 0
        self._lock = threading.Lock()
        
    def _count_tokens_stream(self, text: str) -> int:
        """Token-Zählung für Streaming"""
        return len(text) // 4
    
    def stream_chat(
        self, 
        messages: list,
        model: str = "claude-sonnet-4.5-20250514",
        on_token: Optional[Callable[[str, int], None]] = None
    ) -> Iterator[str]:
        """
        Streamt Claude-Antwort mit Token-Monitoring.
        
        Args:
            messages: Konversationsverlauf
            model: Claude-Modell
            on_token: Callback für jeden empfangenen Token
            
        Yields:
            String-Chunks der Antwort
        """
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": 4096,
            "stream": True,
            "temperature": 0.7
        }
        
        full_response = []
        
        try:
            with requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                stream=True,
                timeout=60
            ) as response:
                response.raise_for_status()
                
                for line in response.iter_lines():
                    if not line:
                        continue
                    
                    # SSE-Format parsen
                    if line.startswith(b"data: "):
                        data = line[6:]
                        if data == b"[DONE]":
                            break
                            
                        try:
                            chunk = json.loads(data)
                            content = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
                            
                            if content:
                                full_response.append(content)
                                self.used_tokens += self._count_tokens_stream(content)
                                
                                # Callback für Echtzeit-Monitoring
                                if on_token:
                                    on_token(content, self.used_tokens)
                                
                                yield content
                                
                        except json.JSONDecodeError:
                            continue
                            
        except requests.exceptions.RequestException as e:
            yield f"[FEHLER: {str(e)}]"
    
    def get_context_status(self) -> dict:
        """Gibt aktuellen Kontext-Status zurück"""
        return {
            "used_tokens": self.used_tokens,
            "budget": self.token_budget,
            "remaining": self.token_budget - self.used_tokens,
            "usage_percent": (self.used_tokens / self.token_budget) * 100
        }
    
    def reset_context(self) -> None:
        """Setzt Kontext zurück (neues Gespräch)"""
        with self._lock:
            self.conversation_history.clear()
            self.used_tokens = 0
    
    def prepare_messages(self, system: str, history: list) -> list:
        """
        Bereitet Nachrichten für API vor mit Kontext-Optimierung.
        """
        messages = [{"role": "system", "content": system}]
        
        total_tokens = self._count_tokens_stream(system)
        
        # Wichtigste letzte Nachrichten hinzufügen
        for msg in history[-10:]:
            msg_tokens = self._count_tokens_stream(msg.get("content", ""))
            if total_tokens + msg_tokens < self.token_budget - 5000:
                messages.append(msg)
                total_tokens += msg_tokens
            else:
                break
                
        return messages

=== Streaming mit Monitoring ===

def token_monitor(token: str, total: int): """Callback für Echtzeit-Token-Monitoring""" print(f"Token empfangen (Gesamt: {total})", end="\r")

Client initialisieren

client = StreamingClaudeClient(api_key="YOUR_HOLYSHEEP_API_KEY") messages = client.prepare_messages( system="Du bist ein hilfreicher Assistent.", history=[ {"role": "user", "content": "Was ist maschinelles Lernen?"} ] ) print("Starte Streaming-Antwort...\n") for chunk in client.stream_chat(messages, on_token=token_monitor): pass print(f"\n\nFinaler Status: {client.get_context_status()}")

Beispiel 3: Fortschrittliches Kontext-Management mit Redis-Caching

import requests
import hashlib
import json
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import redis

class AdvancedClaudeManager:
    """
    Fortschrittliches Kontext-Management mit Caching und
    semantischer Komprimierung.
    """
    
    def __init__(
        self, 
        api_key: str,
        redis_host: str = "localhost",
        redis_port: int = 6379,
        base_url: str = "https://api.holysheep.ai/v1"
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.base_url = base_url
        
        # Redis-Verbindung für Kontext-Caching
        try:
            self.redis = redis.Redis(
                host=redis_host, 
                port=redis_port, 
                decode_responses=True
            )
            self.redis.ping()
            self.use_cache = True
        except:
            self.use_cache = False
            self.redis = None
            
        # Session-Management
        self.sessions: Dict[str, Dict] = {}
        
    def _generate_session_id(self, user_id: str, topic: str = "default") -> str:
        """Generiert eindeutige Session-ID"""
        data = f"{user_id}:{topic}:{datetime.now().date()}"
        return hashlib.sha256(data.encode()).hexdigest()[:16]
    
    def create_session(
        self, 
        user_id: str, 
        system_prompt: str,
        topic: str = "default"
    ) -> str:
        """Erstellt neue Konversations-Session"""
        session_id = self._generate_session_id(user_id, topic)
        
        self.sessions[session_id] = {
            "user_id": user_id,
            "topic": topic,
            "messages": [
                {"role": "system", "content": system_prompt}
            ],
            "created_at": datetime.now(),
            "last_activity": datetime.now(),
            "token_count": self._estimate_tokens(system_prompt),
            "turns": 0
        }
        
        return session_id
    
    def _estimate_tokens(self, text: str) -> int:
        """Token-Schätzung mit Wortschatz-Analyse"""
        words = text.split()
        # Durchschnittlich 1.3 Tokens pro Wort für Englisch
        return int(len(words) * 1.3)
    
    def add_turn(
        self, 
        session_id: str, 
        user_message: str,
        compress_threshold: int = 150000
    ) -> Dict:
        """
        Fügt Konversationsrunde hinzu mit automatischer Optimierung.
        """
        if session_id not in self.sessions:
            raise ValueError(f"Session {session_id} nicht gefunden")
            
        session = self.sessions[session_id]
        
        # Nachricht hinzufügen
        session["messages"].append({
            "role": "user", 
            "content": user_message
        })
        session["token_count"] += self._estimate_tokens(user_message)
        session["turns"] += 1
        
        # Automatische Komprimierung bei Schwellenwert
        if session["token_count"] > compress_threshold:
            self._smart_compress(session)
        
        # API-Anfrage
        response = self._call_api(session["messages"])
        
        # Antwort speichern
        session["messages"].append({
            "role": "assistant",
            "content": response["content"]
        })
        session["token_count"] += response["tokens"]
        session["last_activity"] = datetime.now()
        
        # Cache aktualisieren
        if self.use_cache:
            self._cache_session(session_id, session)
        
        return {
            "response": response["content"],
            "session_id": session_id,
            "turns": session["turns"],
            "tokens_used": response["tokens"],
            "session_tokens": session["token_count"],
            "was_compressed": session.get("_compressed", False)
        }
    
    def _smart_compress(self, session: Dict) -> None:
        """
        Intelligente Kontext-Komprimierung mit Themen-Erkennung.
        """
        messages = session["messages"]
        
        # System-Prompt behalten
        system = messages[0] if messages[0]["role"] == "system" else None
        
        # Letzte 6 Nachrichten behalten (3 Runden)
        recent = messages[-6:] if len(messages) > 6 else messages
        
        # Zusammenfassung der mittleren Nachrichten
        middle = messages[1:-6] if len(messages) > 7 else []
        
        summary = self._generate_summary(middle) if middle else ""
        
        # Neue komprimierte Historie
        new_messages = [system] if system else []
        if summary:
            new_messages.append({
                "role": "system",
                "content": f"[ZUSAMMENFASSUNG DER VORGESCHICHTE]\n{summary}"
            })
        new_messages.extend(recent)
        
        session["messages"] = new_messages
        session["token_count"] = sum(
            self._estimate_tokens(m["content"]) for m in new_messages
        )
        session["_compressed"] = True
    
    def _generate_summary(self, messages: List[Dict]) -> str:
        """Generiert Zusammenfassung der Nachrichten"""
        # Einfache Extraktion der wichtigsten Punkte
        key_points = []
        
        for msg in messages:
            if msg["role"] == "user":
                # Erste 50 Zeichen als Themenindikator
                excerpt = msg["content"][:100].replace("\n", " ")
                key_points.append(f"- Nutzer: {excerpt}...")
            elif msg["role"] == "assistant":
                # Wichtigste Aussagen
                lines = msg["content"].split(".")
                if lines:
                    key_points.append(f"- Assistant: {lines[0][:80]}...")
        
        return "\n".join(key_points[:10])  # Max 10 Punkte
    
    def _call_api(self, messages: List[Dict]) -> Dict:
        """Ruft HolySheep API auf"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": "claude-sonnet-4.5-20250514",
            "messages": messages,
            "max_tokens": 4096,
            "temperature": 0.7
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            
            result = response.json()
            content = result["choices"][0]["message"]["content"]
            tokens = result.get("usage", {}).get("total_tokens", 0)
            
            return {"content": content, "tokens": tokens}
            
        except requests.exceptions.RequestException as e:
            return {
                "content": f"[API-FEHLER: {str(e)}]",
                "tokens": 0
            }
    
    def _cache_session(self, session_id: str, session: Dict) -> None:
        """Speichert Session in Redis"""
        try:
            key = f"claude:session:{session_id}"
            data = {
                "messages": json.dumps(session["messages"]),
                "token_count": session["token_count"],
                "turns": session["turns"],
                "last_activity": session["last_activity"].isoformat()
            }
            self.redis.hset(key, mapping=data)
            self.redis.expire(key, timedelta(hours=24))
        except:
            pass
    
    def get_session_summary(self, session_id: str) -> Optional[Dict]:
        """Gibt Session-Zusammenfassung zurück"""
        if session_id not in self.sessions:
            return None
            
        session = self.sessions[session_id]
        
        return {
            "session_id": session_id,
            "user_id": session["user_id"],
            "topic": session["topic"],
            "turns": session["turns"],
            "total_tokens": session["token_count"],
            "age_minutes": (datetime.now() - session["created_at"]).seconds // 60,
            "last_activity": session["last_activity"].isoformat()
        }

=== Verwendung ===

manager = AdvancedClaudeManager( api_key="YOUR_HOLYSHEEP_API_KEY", redis_host="localhost" )

Session erstellen

session_id = manager.create_session( user_id="user_123", system_prompt="""Du bist ein technischer Berater. Halte Antworten prägnant aber vollständig. Bei Code: Erkläre die wichtigsten Aspekte.""" )

Konversation führen

for i in range(5): result = manager.add_turn( session_id, f"Frage {i+1}: Erkläre mir Konzepte rund um verteilte Systeme." ) print(f"Antwort {i+1}: {result['response'][:100]}...") print(f"Tokens: {result['session_tokens']}, Komprimiert: {result['was_compressed']}\n")

Session-Statistiken

summary = manager.get_session_summary(session_id) print(f"Session-Zusammenfassung: {summary}")

Geeignet / Nicht geeignet für

Einsatzgebiete
✅ Ideal für: ❌ Nicht geeignet für:
  • Kundenservice-Chatbots mit langen Konversationen
  • Code-Assistenten mit Kontexterhalt
  • Document-QA mit mehreren Dokumenten
  • Dialogsysteme für Coaching/Training
  • Content-Generation mit Feedback-Schleifen
  • Anwendungen mit Budget-Limit (<$100/Monat)
  • Einmal-Antworten ohne Kontext (einfachere APIs reichen)
  • Maximaler Kontext >200k Token pro Anfrage
  • Strict Compliance mit Original Anthropic API
  • Unternehmen ohne chinesische Zahlungsmethoden
  • Mission-Critical ohne Fallback-Strategie

Preise und ROI

Die Investition in optimiertes Kontext-Management amortisiert sich schnell. Hier meine Kostenanalyse für typische Szenarien:

Modell Input-Preis Output-Preis HolySheep Ersparnis
Claude Sonnet 4.5 $15/MTok $15/MTok 85%+ mit ¥1=$1 Kurs
GPT-4.1 $8/MTok $8/MTok Vergleichbar
Gemini 2.5 Flash $2.50/MTok $2.50/MTok Kostengünstigste Option
DeepSeek V3.2 $0.42/MTok $0.42/MTok Budget-Alternative

ROI-Beispiel: Kundenservice-Chatbot

Häufige Fehler und Lösungen

Fehler 1: Token-Limit überschritten ohne Fallback

# FEHLERHAFT - Keine Fehlerbehandlung
response = requests.post(url, json=payload)
result = response.json()
content = result["choices"][0]["message"]["content"]

LÖSUNG - Robust mit Fallback-Strategie

def safe_chat_completion(messages, api_key, base_url): """Sichere API-Anfrage mit Token-Handling""" try: headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": "claude-sonnet-4.5-20250514", "messages": messages, "max_tokens": 4096 } response = requests.post( f"{base_url}/chat/completions", headers=headers, json=payload, timeout=30 ) if response.status_code == 200: return response.json()["choices"][0]["message"]["content"] elif response.status_code == 422: # Validation Error (oft Token-Limit) # Strategie 1: Kontext komprimieren und erneut versuchen compressed = compress_messages(messages, target_tokens=50000) if compressed != messages: return safe_chat_completion(compressed, api_key, base_url) # Strategie 2: Auf günstigeres Modell ausweichen payload["model"] = "gpt-4.1" # Fallback zu GPT-4.1 response = requests.post( f"{base_url}/chat/completions", headers=headers, json=payload, timeout=30 ) return response.json()["choices"][0]["message"]["content"] elif response.status_code == 429: # Rate Limit import time time.sleep(5) # Warten und erneut versuchen return safe_chat_completion(messages, api_key, base_url) else: raise APIError(f"HTTP {response.status_code}: {response.text}") except requests.exceptions.Timeout: # Timeout-Fallback return "Anfrage-Zeitüberschreitung. Bitte erneut versuchen." except Exception as e: # Kritischer Fehler - menschenähnliche Antwort return f"Systemfehler (ID: {hash(str(e))[:8]}). Kontaktieren Sie Support."

Fehler 2: Speicherleck durch unbeschränkte Message-Historie

# FEHLERHAFT - Unbegrenzte Historie
class BadClient:
    def __init__(self):
        self.messages = []  # Wird immer größer!
        
    def chat(self, user_input):
        self.messages.append({"role": "user", "content": user_input})
        response = self.call_api(self.messages)
        self.messages.append({"role": "assistant", "content": response})
        return response

LÖSUNG - Bounded Message Queue mit Komprimierung

from collections import deque from typing import List, Dict class OptimizedClient: MAX_MESSAGES = 50 MAX_TOTAL_TOKENS = 150000 def __init__(self): self.messages = deque(maxlen=self.MAX_MESSAGES) self.token_count = 0 def _estimate_tokens(self, text: str) -> int: return len(text.split()) * 1.3 def _should_compress(self) -> bool: return self.token_count > self.MAX_TOTAL_TOKENS def _create_compressed_prompt(self, recent_count: int = 6) -> str: """Erstellt Komprimierungs-Prompt für die Historie""" recent = list(self.messages)[-recent_count:] topics = [m["content"][:50] for m in recent if m["role"] == "user"] return ( "[ZUSAMMENFASSUNG: Konversation wurde komprimiert. " f"Besprochene Themen: {', '.join(topics)}. " "Wichtige Details wurden extrahiert und beibehalten.]" ) def chat(self, user_input: str) -> str: input_tokens = self._estimate_tokens(user_input) # Prüfe ob Komprimierung nötig if self._should_compress(): # Behalte nur System-Prompt und letzte Nachrichten compressed_history = self._create_compressed_prompt() # Aktualisiere Message-Liste messages_list = list(self.messages) system = messages_list[0] if messages_list[0]["role"] == "system" else None self.messages.clear() if system: self.messages.append(system) self.messages.append({ "role": "system", "content": compressed_history }) self.messages.extend(messages_list[-4:]) # Letzte 4 Nachrichten # Token neu berechnen self.token_count = sum( self._estimate_tokens(m["content"]) for m in self.messages ) # Nachricht hinzufügen self.messages.append({"role": "user", "content": user_input}) self.token_count += input_tokens # API-Aufruf response = self.call_api(list(self.messages)) # Antwort speichern output_tokens = self._estimate_tokens(response) self.messages