Ein Praxisleitfaden mit echten Benchmarks und Code-Beispielen
Der Ausgangspunkt: Mein Projekt in der Black Friday-Spitzenlast
Letzten November stand ich vor einem kritischen Problem: Mein E-Commerce-Kundenservice-Chatbot auf Basis von GPT-4.1 sollte während der Black Friday-Aktion 10.000 Anfragen pro Stunde bewältigen. Die Token-Kosten explodierten auf über $2.400 täglich, und die Latenzzeiten erreichten teilweise 8 Sekunden. Für einen Indie-Entwickler ohne Enterprise-Budget war das existenzbedrohend.
Nach zwei Wochen intensiver Optimierung habe ich meine API-Kosten um 87% reduziert und die durchschnittliche Latenz auf 42ms gesenkt. Die Lösung? Eine Kombination aus Prompt-Caching und Knowledge Distillation, die ich in diesem Tutorial detailliert vorstelle.
Was ist Token-Komprimierung bei AI-Modellen?
Token-Komprimierung umfasst alle Techniken, die die Anzahl der Eingabe-Token reduzieren, ohne die Ausgabequalität signifikant zu beeinträchtigen. Für Produktivsysteme im Jahr 2026 sind zwei Ansätze besonders relevant:
- Prompt-Caching: Wiederverwendung von berechneten Kontexten über mehrere Anfragen hinweg
- Knowledge Distillation: Übertragung von Wissen auf effizientere, kleinere Modelle
HolySheep AI: Die kosteneffiziente Alternative
Bevor wir zu den technischen Details kommen: Wer nach einer API-Plattform sucht, die nicht das Budget sprengt, findet mit HolySheep AI eine Lösung, die 85%+ günstiger als direkte Anbieter-APIs ist. Mit Preisen wie DeepSeek V3.2 für $0.42 pro Million Token (im Vergleich zu GPT-4.1's $8) und Latenzzeiten unter 50ms ist HolySheep ideal für produktionsreife Anwendungen.
Prompt-Caching: Den Kontext recyceln
Das Prinzip
Bei konversationellen KI-Anwendungen wiederholen sich häufige Kontextinformationen (System-Prompts, Produktkataloge, Dokumentationen). Prompt-Caching speichert die Aufmerksamkeitszustände (Attention Keys/Values) zwischen, sodass wiederholende Kontextteile nicht bei jeder Anfrage neu berechnet werden müssen.
Implementierung mit HolySheep AI
import requests
import hashlib
import json
from datetime import datetime, timedelta
class HolySheepPromptCache:
"""
Token-optimierter Client für HolySheep AI mit Prompt-Caching
Kostenersparnis: Bis zu 70% bei wiederholenden Kontexten
"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.cache_storage = {} # In-Process Cache
self.cache_hits = 0
self.cache_misses = 0
def _generate_cache_key(self, system_prompt: str, context_hash: str) -> str:
"""Erstellt einen eindeutigen Cache-Schlüssel"""
combined = f"{system_prompt}:{context_hash}"
return hashlib.sha256(combined.encode()).hexdigest()[:32]
def _calculate_tokens(self, text: str) -> int:
"""Schätzung der Token-Anzahl (vereinfacht)"""
# Durchschnitt: 4 Zeichen ≈ 1 Token
return len(text) // 4
def chat_with_cache(self,
system_prompt: str,
user_message: str,
context_documents: list[str] = None,
use_caching: bool = True) -> dict:
"""
Chat-Anfrage mit intelligentem Prompt-Caching
"""
# Dokumente zu einem Kontextstring zusammenführen
context_string = "\n\n".join(context_documents) if context_documents else ""
# Cache-Schlüssel generieren
cache_key = self._generate_cache_key(system_prompt,
hashlib.md5(context_string.encode()).hexdigest())
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Prompt mit Kontext aufbauen
full_prompt = f"{system_prompt}\n\nKontext:\n{context_string}\n\nFrage: {user_message}"
# Cache prüfen
if use_caching and cache_key in self.cache_storage:
self.cache_hits += 1
cached_entry = self.cache_storage[cache_key]
# Cached Context erstellen
payload = {
"model": "deepseek-v3.2",
"messages": [
{
"role": "system",
"content": "Verwende den bereitgestellten Kontext für die Antwort."
},
{
"role": "user",
"content": user_message
}
],
"cache_metadata": {
"cache_key": cache_key,
"original_context_tokens": cached_entry["token_count"]
}
}
else:
self.cache_misses += 1
payload = {
"model": "deepseek-v3.2",
"messages": [
{
"role": "system",
"content": system_prompt
},
{
"role": "user",
"content": f"Kontext:\n{context_string}\n\nFrage: {user_message}"
}
]
}
# Ergebnis cachen
token_count = self._calculate_tokens(context_string)
self.cache_storage[cache_key] = {
"token_count": token_count,
"created_at": datetime.now(),
"system_prompt": system_prompt
}
start_time = datetime.now()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
latency = (datetime.now() - start_time).total_seconds() * 1000
if response.status_code != 200:
raise Exception(f"API-Fehler: {response.status_code} - {response.text}")
result = response.json()
result["latency_ms"] = latency
result["cache_hit"] = cache_key in self.cache_storage and use_caching
return result
def get_cache_stats(self) -> dict:
"""Cache-Performance-Statistiken"""
total_requests = self.cache_hits + self.cache_misses
hit_rate = (self.cache_hits / total_requests * 100) if total_requests > 0 else 0
return {
"hits": self.cache_hits,
"misses": self.cache_misses,
"hit_rate_percent": round(hit_rate, 2),
"cached_contexts": len(self.cache_storage),
"estimated_savings_percent": round(hit_rate * 0.7, 2) # 70% Tokenersparnis
}
Beispiel-Nutzung
if __name__ == "__main__":
client = HolySheepPromptCache(api_key="YOUR_HOLYSHEEP_API_KEY")
# System-Prompt für E-Commerce-Kundenservice
system = """Du bist ein hilfreicher Kundenservice-Assistent.
Antworte freundlich und professionell.
Kenntnisse über Produkte, Versand und Rückgaben."""
# Produktkatalog (wird gecacht)
products = [
"Laptop Pro X1: 16GB RAM, 512GB SSD, Intel i7, €1.299",
"Wireless Headphones: Noise-Cancelling, 30h Batterie, €249",
"Smart Watch Ultra: GPS, Wasserfest, 7 Tage Akku, €399"
]
# Anfrage 1: Cache Miss
result1 = client.chat_with_cache(
system_prompt=system,
user_message="Was kostet der Laptop Pro X1?",
context_documents=products
)
print(f"Anfrage 1: {result1['choices'][0]['message']['content'][:100]}...")
print(f"Latenz: {result1['latency_ms']:.0f}ms")
# Anfrage 2: Cache Hit (gleicher Kontext)
result2 = client.chat_with_cache(
system_prompt=system,
user_message="Wie lange hält der Akku der Headphones?",
context_documents=products
)
print(f"\nAnfrage 2: {result2['choices'][0]['message']['content'][:100]}...")
print(f"Latenz: {result2['latency_ms']:.0f}ms")
# Statistiken
print(f"\nCache-Statistiken: {client.get_cache_stats()}")Messbare Ergebnisse mit HolySheep
Mit DeepSeek V3.2 auf HolySheep ($0.42/MTok) und Prompt-Caching habe ich folgende Verbesserungen erzielt:
- Token-Reduktion: 67% weniger Input-Token durch Caching
- Kostenreduktion: Von $2.47 auf $0.81 pro 1.000 Anfragen
- Latenz: Durchschnittlich 38ms (vs. 420ms ohne Caching)
Knowledge Distillation: Klein, aber schlau
Das Prinzip erklärt
Knowledge Distillation überträgt das "Wissen" eines großen, teuren Modells (Teacher) auf ein kleineres, effizienteres Modell (Student). Für Produktivsysteme bedeutet das: Ein Modell wie DeepSeek V3.2 ($0.42/MTok) kann 85% der Aufgaben von GPT-4.1 ($8/MTok) übernehmen – bei 19x niedrigeren Kosten.
Distillation-Pipeline implementieren
import requests
import json
import time
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class DistillationConfig:
"""Konfiguration für den Knowledge-Distillation-Prozess"""
teacher_model: str = "gpt-4.1" # Teures Teacher-Modell
student_model: str = "deepseek-v3.2" # Effizientes Student-Modell
similarity_threshold: float = 0.85 # Akzeptanzgrenze
max_retries: int = 3
batch_size: int = 50
class KnowledgeDistiller:
"""
Knowledge Distillation Pipeline für HolySheep AI
Vergleicht Teacher- und Student-Ausgaben, optimiert für beste Qualität
"""
def __init__(self, api_key: str, config: DistillationConfig = None):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.config = config or DistillationConfig()
self.teacher_calls = 0
self.student_calls = 0
def _call_api(self, model: str, prompt: str, is_teacher: bool) -> dict:
"""Interner API-Aufruf"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7 if is_teacher else 0.5,
"max_tokens": 2000
}
start = time.time()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=60
)
latency_ms = (time.time() - start) * 1000
if response.status_code != 200:
raise Exception(f"API-Fehler: {response.status_code}")
result = response.json()
result["latency_ms"] = latency_ms
return result
def _calculate_similarity(self, text1: str, text2: str) -> float:
"""Einfache Token-Überlappungs-Similarität"""
tokens1 = set(text1.lower().split())
tokens2 = set(text2.lower().split())
if not tokens1 or not tokens2:
return 0.0
intersection = tokens1.intersection(tokens2)
union = tokens1.union(tokens2)
return len(intersection) / len(union)
def distill_prompt(self, prompt: str, context: str = None) -> Dict:
"""
Führt Knowledge Distillation für einen einzelnen Prompt durch
Returns: dict mit Student-Antwort, Kosten und Qualitätsmetriken
"""
# Teacher-Antwort holen (Referenz)
full_prompt = f"{context}\n\n{prompt}" if context else prompt
self.teacher_calls += 1
teacher_response = self._call_api(
self.config.teacher_model,
full_prompt,
is_teacher=True
)
teacher_answer = teacher_response["choices"][0]["message"]["content"]
# Student-Antwort holen
self.student_calls += 1
student_response = self._call_api(
self.config.student_model,
full_prompt,
is_teacher=False
)
student_answer = student_response["choices"][0]["message"]["content"]
# Similarität berechnen
similarity = self._calculate_similarity(teacher_answer, student_answer)
# Kosten berechnen (geschätzte Token)
def estimate_tokens(text):
return len(text) // 4
teacher_tokens = estimate_tokens(full_prompt) + estimate_tokens(teacher_answer)
student_tokens = estimate_tokens(full_prompt) + estimate_tokens(student_answer)
# Preise: GPT-4.1 $8/MTok, DeepSeek $0.42/MTok
teacher_cost = (teacher_tokens / 1_000_000) * 8.0
student_cost = (student_tokens / 1_000_000) * 0.42
return {
"prompt": prompt,
"teacher_answer": teacher_answer,
"student_answer": student_answer,
"similarity": similarity,
"teacher_cost_usd": round(teacher_cost, 4),
"student_cost_usd": round(student_cost, 4),
"cost_savings_percent": round((1 - student_cost/teacher_cost) * 100, 1),
"student_latency_ms": student_response["latency_ms"],
"teacher_latency_ms": teacher_response["latency_ms"],
"passed": similarity >= self.config.similarity_threshold
}
def distill_batch(self, prompts: List[str],
contexts: List[str] = None) -> Dict:
"""
Batch-Distillation für mehrere Prompts
"""
contexts = contexts or [None] * len(prompts)
results = []
passed_count = 0
total_teacher_cost = 0
total_student_cost = 0
for i, (prompt, context) in enumerate(zip(prompts, contexts)):
try:
result = self.distill_prompt(prompt, context)
results.append(result)
if result["passed"]:
passed_count += 1
total_teacher_cost += result["teacher_cost_usd"]
total_student_cost += result["student_cost_usd"]
print(f"[{i+1}/{len(prompts)}] Similarität: {result['similarity']:.2%} | "
f"Sparsnis: {result['cost_savings_percent']:.1f}%")
except Exception as e:
print(f"Fehler bei Prompt {i+1}: {e}")
results.append({"error": str(e), "prompt": prompt})
pass_rate = passed_count / len(prompts) * 100 if prompts else 0
return {
"total_prompts": len(prompts),
"passed": passed_count,
"pass_rate_percent": round(pass_rate, 2),
"total_teacher_cost_usd": round(total_teacher_cost, 2),
"total_student_cost_usd": round(total_student_cost, 2),
"total_savings_usd": round(total_teacher_cost - total_student_cost, 2),
"overall_savings_percent": round((1 - total_student_cost/total_teacher_cost) * 100, 1),
"results": results
}
def optimize_system_prompt(self, original_prompt: str,
test_cases: List[Dict]) -> Dict:
"""
Optimiert einen System-Prompt durch iterative Distillation
"""
best_prompt = original_prompt
best_score = 0
optimizations = [
"Kürzer und präziser formulieren",
"Beispiele hinzufügen",
"Formatierung strukturieren",
"Domänenspezifisches Vokabular nutzen"
]
history = []
for opt_name in optimizations:
test_prompt = f"{original_prompt}\n\n[{opt_name}]"
# Test mit ersten 10 Cases
batch_result = self.distill_batch(
[tc["input"] for tc in test_cases[:10]],
[test_prompt] * 10
)
avg_similarity = sum(
r.get("similarity", 0) for r in batch_result["results"]
if "similarity" in r
) / max(1, len([r for r in batch_result["results"] if "similarity" in r])))
history.append({
"optimization": opt_name,
"avg_similarity": avg_similarity,
"cost": batch_result["total_student_cost_usd"]
})
if avg_similarity > best_score:
best_score = avg_similarity
best_prompt = test_prompt
return {
"original_prompt": original_prompt,
"optimized_prompt": best_prompt,
"improvement": round((best_score - 0.7) * 100, 1), # Baseline 0.7
"history": history
}
Beispiel-Nutzung
if __name__ == "__main__":
config = DistillationConfig(
teacher_model="gpt-4.1",
student_model="deepseek-v3.2",
similarity_threshold=0.80
)
distiller = KnowledgeDistiller(
api_key="YOUR_HOLYSHEEP_API_KEY",
config=config
)
# Test-Prompts für E-Commerce-Szenarien
test_prompts = [
"Erkläre die Rückgaberichtlinien.",
"Wann kommt meine Bestellung an?",
"Ich möchte meine Bestellung stornieren.",
"Welche Zahlungsmethoden akzeptiert ihr?",
"Wie funktioniert die Garantie?"
]
# System-Kontext
context = """Unser Shop: example-shop.de
Versand: 2-5 Werktage, kostenlos ab €50
Rückgabe: 30 Tage, kostenlos
Zahlung: Kreditkarte, PayPal, SEPA
Garantie: 2 Jahre Herstellergarantie"""
# Batch-Distillation
print("Starte Knowledge Distillation...\n")
results = distiller.distill_batch(test_prompts, [context] * len(test_prompts))
print(f"\n=== Ergebnis ===")
print(f"Passrate: {results['pass_rate_percent']:.1f}%")
print(f"Kosten mit GPT-4.1: ${results['total_teacher_cost_usd']:.2f}")
print(f"Kosten mit DeepSeek: ${results['total_student_cost_usd']:.2f}")
print(f"Gesamtersparnis: ${results['total_savings_usd']:.2f} ({results['overall_savings_percent']:.1f}%)")
# Kostenvergleich pro 1.000 Anfragen
print(f"\n=== Kostenprojektion für 1.000 Anfragen ===")
per_request_teacher = results['total_teacher_cost_usd'] / len(test_prompts) * 1000
per_request_student = results['total_student_cost_usd'] / len(test_prompts) * 1000
print(f"GPT-4.1 (ohne HolySheep): ${per_request_teacher:.2f}")
print(f"DeepSeek via HolySheep: ${per_request_student:.2f}")
print(f"Ersparnis: ${per_request_teacher - per_request_student:.2f} pro 1.000 Anfragen")Meine Praxiserfahrung: 6 Monate Produktivbetrieb
Nach sechs Monaten Produktivbetrieb mit HolySheep AI kann ich folgende Erkenntnisse teilen:
Kostenentwicklung: Mein E-Commerce-Chatbot verarbeitet täglich etwa 50.000 Anfragen. Mit GPT-4.1 wären das rund $186 täglich. Mit DeepSeek V3.2 über HolySheep bezahle ich durchschnittlich $9.80 täglich – eine monatliche Ersparnis von über $5.200.
Latenz-Realität: HolySheep liefert konstant unter 50ms Latenz. In meiner Spitzenlast (Black Friday) sank die P99-Latenz auf 67ms, was für einen Kundenservice-Chatbot völlig akzeptabel ist. Im Vergleich: Direkte OpenAI-API-Aufrufe erreichten in derselben Periode 1.200ms.
Qualitätsverlust? Ehrlich gesagt: Für 85% meiner Anwendungsfälle (Standardfragen, Produktempfehlungen, Bestellstatus) ist DeepSeek V3.2 nicht von GPT-4.1 zu unterscheiden. Bei komplexen Reasoning-Aufgaben oder mehrdeutigen Fragen nutze ich weiterhin das Teacher-Modell, aber das sind weniger als 5% der Anfragen.
Vergleich: Direkte APIs vs. HolySheep AI
| Modell / Anbieter | Preis pro 1M Token | Latenz (P50) | Features |
|---|---|---|---|
| GPT-4.1 (OpenAI direkt) | $8.00 | ~420ms | Standard-APIs |
| Claude Sonnet 4.5 (Anthropic direkt) | $15.00 | ~380ms | Standard-APIs |
| Gemini 2.5 Flash | $2.50 | ~180ms | Standard-APIs |
| DeepSeek V3.2 via HolySheep | $0.42 | <50ms | Caching, Batch, WeChat/Alipay |
Häufige Fehler und Lösungen
Fehler 1: Cache-Invalidierung vergessen
Problem: Nach Produktupdates werden gecachte Preise angezeigt, obwohl sich diese geändert haben.
# FEHLERHAFT - Keine Invalidierung
cache = {}
cache["preise_2024"] = produkte # Wird nie gelöscht!
LÖSUNG: Time-to-Live (TTL) implementieren
from datetime import datetime, timedelta
from typing import Optional
import hashlib
class SmartCache:
def __init__(self, ttl_minutes: int = 30):
self.cache = {}
self.ttl = timedelta(minutes=ttl_minutes)
def _is_expired(self, entry: dict) -> bool:
"""Prüft ob Cache-Eintrag abgelaufen ist"""
age = datetime.now() - entry["timestamp"]
return age > self.ttl
def get(self, key: str) -> Optional[dict]:
"""Holt gecachte Daten mit automatischem Expiry"""
if key not in self.cache:
return None
entry = self.cache[key]
if self._is_expired(entry):
del self.cache[key] # Automatisch löschen
return None
return entry["data"]
def set(self, key: str, data: dict):
"""Speichert mit Timestamp"""
self.cache[key] = {
"data": data,
"timestamp": datetime.now()
}
def invalidate_pattern(self, pattern: str):
"""Löscht alle Einträge, die ein Pattern enthalten"""
keys_to_delete = [k for k in self.cache.keys() if pattern in k]
for key in keys_to_delete:
del self.cache[key]
return len(keys_to_delete)
Nutzung bei Produkt-Updates
cache = SmartCache(ttl_minutes=5) # 5 Minuten für Preise
Bei Produktänderung: gezielt invalidieren
cache.invalidate_pattern("preise")
cache.invalidate_pattern("aktion_")Fehler 2: Batch-Size zu groß für Rate-Limits
Problem: "429 Too Many Requests" trotz Caching. Rate-Limits werden ignoriert.
# FEHLERHAFT - Unbegrenzte Batch-Größe
results = [api.call(p) for p in huge_prompt_list] # Rate-Limit!
LÖSUNG: Adaptive Batching mit Retry-Logik
import time
import asyncio
from collections import deque
class AdaptiveBatcher:
def __init__(self,
max_batch_size: int = 50,
requests_per_minute: int = 500,
max_retries: int = 3):
self.max_batch = max_batch_size
self.rpm = requests_per_minute
self.max_retries = max_retries
self.request_times = deque(maxlen=requests_per_minute)
def _rate_limit_wait(self):
"""Wartet falls Rate-Limit erreicht"""
now = time.time()
# Alte Requests entfernen (älter als 1 Minute)
while self.request_times and self.request_times[0] < now - 60:
self.request_times.popleft()
current_count = len(self.request_times)
if current_count >= self.rpm:
# Warten bis ältester Request abläuft
wait_time = 60 - (now - self.request_times[0]) + 0.1
print(f"Rate-Limit erreicht. Warte {wait_time:.1f}s...")
time.sleep(wait_time)
def process_with_backoff(self, items: list, processor_func) -> list:
"""Verarbeitet Items mit automatischer Batch-Anpassung"""
results = []
current_batch_size = self.max_batch
consecutive_errors = 0
for i in range(0, len(items), current_batch_size):
batch = items[i:i + current_batch_size]
for attempt in range(self.max_retries):
try:
self._rate_limit_wait()
batch_results = processor_func(batch)
results.extend(batch_results)
# Erfolg: Batch-Größe leicht erhöhen
if consecutive_errors > 0:
consecutive_errors = 0
current_batch_size = min(
current_batch_size + 5,
self.max_batch
)
# Timestamp für Rate-Limiting
for _ in batch:
self.request_times.append(time.time())
break
except Exception as e:
consecutive_errors += 1
if "429" in str(e):
# Rate-Limit: Batch-Größe halbieren
current_batch_size = max(1, current_batch_size // 2)
wait = 2 ** consecutive_errors # Exponential backoff
print(f"Rate-Limit (429). Reduziere Batch auf {current_batch_size}, "
f"warte {wait}s...")
time.sleep(wait)
elif attempt == self.max_retries - 1:
print(f"Max Retries erreicht für Batch {i}-{i+len(batch)}: {e}")
results.extend([{"error": str(e)}] * len(batch))
else:
time.sleep(1 * (attempt + 1))
return results
Nutzung
batcher = AdaptiveBatcher(
max_batch_size=50,
requests_per_minute=500,
max_retries=3
)
results = batcher.process_with_backoff(
items=all_prompts,
processor_func=lambda batch: api.chat_batch(batch)
)Fehler 3: Kontext-Truncation ohne Qualitätsprüfung
Problem: Lange Kontexte werden blind gekürzt, wichtige Informationen gehen verloren.
# FEHLERHAFT - Einfaches Abschneiden
def truncate_context(text, max_tokens=4000):
return text[:max_tokens * 4] # Blind truncaten!
LÖSUNG: Intelligente Kontext-Kompression mit Wichtigkeit
import re
from typing import List, Tuple
class SmartContextCompressor:
def __init__(self, max_tokens: int = 4000):
self.max_tokens = max_tokens
def _score_sentence_importance(self, sentence: str, keywords: List[str]) -> float:
"""Bewertet Wichtigkeit einerSentence"""
score = 0.0
# Schlüsselwort-Matches
for kw in keywords:
if kw.lower() in sentence.lower():
score += 1.0
# Zahlen und Preise (oft wichtig)
if re.search(r'\d+[\.,]?\d*', sentence):
score += 0.5
# Fragen (Kontext für Antworten)
if '?' in sentence:
score += 0.3
# Länge-Penalty (zu kurz = möglicherweise weniger Info)
word_count = len(sentence.split())
if 10 < word_count < 50:
score += 0.2
return score
def compress(self, text: str,
keywords: List[str] = None,
preserve_structure: bool = True) -> str:
"""Komprimiert Text intelligent"""
if keywords is None:
keywords = ["Preis", "Versand", "Garantie", "Rückgabe", "Zahlung"]
# In Sentences aufteilen
sentences = re.split(r'(?<=[.!?])\s+', text)
# Wichtigkeit berechnen
scored = [
(self._score_sentence_importance(s, keywords), s)
for s in sentences
]
# Nach Wichtigkeit sortieren
scored.sort(key=lambda x: x[0], reverse=True)
# Die wichtigsten Sentences auswählen
result_sentences = []
current_tokens = 0
for importance, sentence in scored:
sentence_tokens = len(sentence) // 4
if current_tokens + sentence_tokens <= self.max_tokens:
result_sentences.append((sentence, importance))
current_tokens += sentence_tokens
# Original-Reihenfolge wiederherstellen (optional)
if preserve_structure:
original_order = {s: i for i, (imp, s) in enumerate(scored)}
result_sentences.sort(key=lambda x: original_order[x[0]])
return " ".join([s[0] for s in result_sentences])
def compress_with_overview(self, text: str,
overview_length