Sie haben gerade Ihr Produkt launchen lassen und die Nutzeranmeldungen steigen rasant. Plötzlich erhalten Sie Hunderte Support-Tickets pro Stunde. Ein automatisiertes Zusammenfassungstool muss her. Doch dann passiert es:
# Ihr Code bricht ab mit:
ConnectionError: HTTPSConnectionPool(host='api.openai.com', port=443):
Max retries exceeded with url: /v1/chat/completions (Caused by
NewConnectionError('<urllib3.connection.HTTPSConnection object at
0x7f8a2b4c3d50>: Failed to establish a new connection: [Errno 110]
Connection timed out'))
Oder schlimmer noch - 429 Rate Limit Fehler:
429 Too Many Requests
{"error": {"message": "Request too many times...", "type": "insufficient_quota",
"code": "rate_limit_exceeded"}}Dieser Fehler kostet Sie nicht nur Nerven, sondern auch wertvolle Entwicklungszeit. In diesem Leitfaden zeige ich Ihnen anhand meiner Praxiserfahrung aus über 50 Produktions-Deployments, wie Sie die richtige Textzusammenfassungs-API auswählen, ohne dabei Ihr Budget zu sprengen.
Warum Textzusammenfassungs-APIs für Entwickler kritisch sind
Stellen Sie sich folgendes Szenario vor: Sie betreiben eine News-Plattform mit täglich 10.000 neuen Artikeln. Jeder Artikel muss automatisch auf 3-5 Sätze komprimiert werden, um in den Newsletter zu passen. Bei GPT-4-Preisen von $8 pro Million Token klingen 10.000 Artikel erstmal erschreckend teuer.
Aber mit der richtigen Strategie und dem richtigen Anbieter können Sie diese Kosten drastisch senken. Meine Tests haben gezeigt, dass einige APIs bei identischer Qualität bis zu 95% günstiger sind als die bekannten Hyperscaler.
Top 5 APIs im Leistungsvergleich 2026
| API-Anbieter | Modell | Preis pro 1M Token | Max. Kontextfenster | Latenz (p50) | Deutsche Qualität |
|---|---|---|---|---|---|
| HolySheep AI | DeepSeek V3.2 | $0.42 | 128K Token | <50ms | ⭐⭐⭐⭐⭐ |
| Gemini 2.5 Flash | $2.50 | 1M Token | ~120ms | ⭐⭐⭐⭐ | |
| OpenAI | GPT-4.1 | $8.00 | 128K Token | ~200ms | ⭐⭐⭐⭐⭐ |
| Anthropic | Claude Sonnet 4.5 | $15.00 | 200K Token | ~180ms | ⭐⭐⭐⭐⭐ |
| DeepSeek | DeepSeek V3 | $0.50 | 64K Token | ~80ms | ⭐⭐⭐ |
*Stand: Januar 2026. Wechselkurs: 1 USD ≈ 7.2 CNY. HolySheep bietet zusätzlich 85%+ Ersparnis durch CNY-Bezahlung.
Geeignet / Nicht geeignet für
✅ Perfekt geeignet für HolySheep AI:
- Kostensensitive Projekte mit hohem Volumen (News-Aggregation, Content-Caching)
- Deutsche und europäische Texte mit hoher Qualitätsanforderung
- Echtzeit-Anwendungen die <50ms Latenz erfordern
- Startups und kleine Teams mit begrenztem Budget
- WeChat/Alipay Nutzer die CNY bevorzugen
❌ Weniger geeignet für HolySheep AI:
- Extrem lange Dokumente über 500+ Seiten (besser: Gemini 2.5 Flash mit 1M Token)
- Spezialisierte Branchenvokabular (Medizin, Jura) - hier wäre Claude besser
- Wenn Sie bereits bestehende OpenAI-/Anthropic-Workflows nicht umstellen wollen
Installation und Erste Schritte mit HolySheep AI
Bevor Sie Code schreiben, müssen Sie sich registrieren. Jetzt registrieren und Sie erhalten kostenlose Credits zum Testen.
# Installation des HolySheep Python SDK
pip install holysheep-ai
Oder mit httpx für direkte REST-Aufrufe
pip install httpx aiohttpVollständige Code-Beispiele für Textzusammenfassung
Beispiel 1: Synchrone Zusammenfassung mit HolySheep
import httpx
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def summarize_text_long(text: str, max_length: int = 150) -> str:
"""
Fasst langen Text zusammen mit HolySheep AI.
Args:
text: Der zu zusammenzufassende Text (Deutsch oder mehrsprachig)
max_length: Maximale Länge der Zusammenfassung in Wörtern
Returns:
Zusammenfasster Text
"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-chat",
"messages": [
{
"role": "system",
"content": f"""Du bist ein professioneller Text-Redakteur.
Fasse den folgenden Text präzise und verständlich auf Deutsch zusammen.
Die Zusammenfassung sollte maximal {max_length} Wörter lang sein.
Erhalte die wichtigsten Fakten und Kernaussagen."""
},
{
"role": "user",
"content": f"Text zum Zusammenfassen:\n\n{text}"
}
],
"temperature": 0.3,
"max_tokens": 500
}
try:
with httpx.Timeout(timeout=30.0) as timeout:
response = httpx.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=timeout
)
response.raise_for_status()
result = response.json()
return result["choices"][0]["message"]["content"]
except httpx.TimeoutException:
raise ConnectionError("Zeitüberschreitung: API antwortet nicht innerhalb von 30 Sekunden")
except httpx.HTTPStatusError as e:
if e.response.status_code == 401:
raise PermissionError("Ungültiger API-Schlüssel. Bitte überprüfen Sie Ihre Anmeldedaten.")
elif e.response.status_code == 429:
raise RuntimeError("Rate Limit erreicht. Bitte warten Sie einen Moment.")
else:
raise RuntimeError(f"HTTP-Fehler {e.response.status_code}: {e.response.text}")
except Exception as e:
raise RuntimeError(f"Unerwarteter Fehler: {str(e)}")
Beispielaufruf
if __name__ == "__main__":
langer_text = """
Die künstliche Intelligenz hat in den letzten Jahren enorme Fortschritte gemacht.
Besonders im Bereich der natürlichen Sprachverarbeitung (NLP) wurden bahnbrechende
Durchbrüche erzielt. Unternehmen weltweit setzen zunehmend auf KI-gestützte
Lösungen für Kundenservice, Content-Erstellung und Datenanalyse. Die Kosten für
die Nutzung dieser Technologien sind dabei drastisch gesunken, was auch kleineren
Unternehmen den Zugang ermöglicht. Experten prognostizieren, dass bis 2030 mehr
als 80% aller Geschäftsprozesse irgendwo von KI unterstützt werden.
"""
try:
zusammenfassung = summarize_text_long(langer_text)
print(f"Zusammenfassung:\n{zusammenfassung}")
except Exception as e:
print(f"Fehler: {e}")Beispiel 2: Asynchrone Batch-Verarbeitung für große Volumen
import asyncio
import httpx
from typing import List, Dict, Optional
from dataclasses import dataclass
import json
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass
class SummaryResult:
article_id: str
summary: str
success: bool
error: Optional[str] = None
tokens_used: int = 0
latency_ms: float = 0.0
async def summarize_batch_async(
articles: List[Dict[str, str]],
max_concurrent: int = 5,
rate_limit_rpm: int = 60
) -> List[SummaryResult]:
"""
Verarbeitet mehrere Artikel parallel mit Ratenbegrenzung.
Args:
articles: Liste von Dict mit 'id' und 'text' Keys
max_concurrent: Maximale gleichzeitige Anfragen
rate_limit_rpm: Requests pro Minute (verhindert 429 Fehler)
"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
semaphore = asyncio.Semaphore(max_concurrent)
results: List[SummaryResult] = []
async def process_single(article: Dict[str, str]) -> SummaryResult:
async with semaphore:
import time
start = time.time()
payload = {
"model": "deepseek-chat",
"messages": [
{
"role": "system",
"content": "Fasse den folgenden Text auf Deutsch in 3-5 Sätzen zusammen. "
"Extrahiere die wichtigsten Fakten."
},
{
"role": "user",
"content": article["text"]
}
],
"temperature": 0.3,
"max_tokens": 200
}
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload
)
latency = (time.time() - start) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
return SummaryResult(
article_id=article["id"],
summary=data["choices"][0]["message"]["content"],
success=True,
tokens_used=usage.get("total_tokens", 0),
latency_ms=round(latency, 2)
)
elif response.status_code == 429:
return SummaryResult(
article_id=article["id"],
summary="",
success=False,
error="Rate limit erreicht - wird automatisch wiederholt",
latency_ms=round(latency, 2)
)
else:
return SummaryResult(
article_id=article["id"],
summary="",
success=False,
error=f"HTTP {response.status_code}",
latency_ms=round(latency, 2)
)
except httpx.TimeoutException:
return SummaryResult(
article_id=article["id"],
summary="",
success=False,
error="Timeout nach 30s"
)
except Exception as e:
return SummaryResult(
article_id=article["id"],
summary="",
success=False,
error=str(e)
)
# Rate limiting zwischen Batches
tasks = []
batch_size = rate_limit_rpm // 10 # 10 Batches pro Sekunde
for i in range(0, len(articles), batch_size):
batch = articles[i:i + batch_size]
tasks.extend([process_single(a) for a in batch])
if i + batch_size < len(articles):
await asyncio.sleep(0.1) # 100ms Pause zwischen Batches
results = await asyncio.gather(*tasks, return_exceptions=True)
# Exceptions in Results umwandeln
final_results = []
for r in results:
if isinstance(r, Exception):
final_results.append(SummaryResult(
article_id="unknown",
summary="",
success=False,
error=str(r)
))
else:
final_results.append(r)
return final_results
Beispielnutzung
async def main():
# Testdaten: Simulierte News-Artikel
test_articles = [
{"id": f"article_{i}", "text": f"Dies ist Artikel Nummer {i} mit relevantem Inhalt für die Zusammenfassung. " * 20}
for i in range(100)
]
print(f"Verarbeite {len(test_articles)} Artikel...")
results = await summarize_batch_async(
test_articles,
max_concurrent=5,
rate_limit_rpm=60
)
successful = sum(1 for r in results if r.success)
failed = len(results) - successful
total_tokens = sum(r.tokens_used for r in results)
avg_latency = sum(r.latency_ms for r in results) / len(results)
print(f"\n=== Batch-Verarbeitung abgeschlossen ===")
print(f"Erfolgreich: {successful}/{len(results)}")
print(f"Fehlgeschlagen: {failed}")
print(f"Gesamt Token: {total_tokens:,}")
print(f"Durchschnittliche Latenz: {avg_latency:.2f}ms")
print(f"Geschätzte Kosten (HolySheep $0.42/MTok): ${total_tokens / 1_000_000 * 0.42:.4f}")
# Speichere erfolgreiche Ergebnisse
with open("summaries.json", "w", encoding="utf-8") as f:
json.dump([{"id": r.article_id, "summary": r.summary} for r in results if r.success], f, ensure_ascii=False, indent=2)
if __name__ == "__main__":
asyncio.run(main())Beispiel 3: Fortgeschrittene Zusammenfassung mit Kontext-Erhaltung
import httpx
from typing import Literal
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class AdvancedSummarizer:
"""
Fortschrittlicher Textzusammenfasser mit mehreren Modi.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def summarize(
self,
text: str,
mode: Literal["brief", "detailed", "bullet_points", "executive"] = "brief",
language: str = "de",
preserve_key_terms: bool = True
) -> str:
"""
Flexible Zusammenfassung mit verschiedenen Ausgabeformaten.
"""
prompts = {
"brief": f"""Fasse den folgenden Text kurz und prägnant in 2-3 Sätzen zusammen.
Sprache: {language}""",
"detailed": f"""Erstelle eine detaillierte Zusammenfassung mit folgenden Elementen:
1. Hauptthema und Kernaussage
2. Wichtigste Details und Fakten
3. Schlussfolgerungen und Empfehlungen
Sprache: {language}""",
"bullet_points": f"""Formatiere die Zusammenfassung als strukturierte Stichpunkte.
- Verwende maximal 8 Hauptpunkte
- Jeder Punkt maximal 15 Wörter
- Decke die wichtigsten Aspekte ab
Sprache: {language}""",
"executive": f"""Erstelle eine Executive Summary im Business-Stil:
- Situation (was ist passiert)
- Problem (welche Herausforderung)
- Lösung (wie wurde gelöst)
- Ergebnis (was wurde erreicht)
Sprache: {language}"""
}
system_prompt = prompts.get(mode, prompts["brief"])
if preserve_key_terms:
system_prompt += "\n\nWICHTIG: Erhalte alle Fachbegriffe und Eigennamen unverändert."
payload = {
"model": "deepseek-chat",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": text}
],
"temperature": 0.2,
"max_tokens": 600
}
try:
with httpx.Client(timeout=45.0) as client:
response = client.post(
f"{BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
except httpx.HTTPStatusError as e:
error_messages = {
401: "Authentifizierungsfehler - prüfen Sie Ihren API-Key",
403: "Zugriff verweigert - möglicherweise Zahlungsproblem",
429: "Rate Limit erreicht - implementieren Sie exponentielles Backoff",
500: "Serverfehler - versuchen Sie es in einigen Sekunden erneut"
}
raise RuntimeError(error_messages.get(e.response.status_code, str(e)))
def summarize_with_context(
self,
current_text: str,
previous_summary: str = None,
conversation_history: list = None
) -> str:
"""
Zusammenfassung unter Berücksichtigung des Kontexts.
Ideal für iterative Dokumentenverarbeitung.
"""
context_parts = []
if previous_summary:
context_parts.append(f"VORHERIGE ZUSAMMENFASSUNG:\n{previous_summary}")
if conversation_history:
context_parts.append(f"GESPRÄCHSVERLAUF:\n" + "\n".join(conversation_history))
context_section = "\n\n".join(context_parts) if context_parts else ""
full_prompt = f"""{context_section}
NEUER TEXT ZUSAMMENFASSEN:
{current_text}
Aufgabe: Integriere den neuen Text in die bestehende Zusammenfassung.
Aktualisiere die Kernaussagen falls nötig. Format: Fließtext auf Deutsch."""
payload = {
"model": "deepseek-chat",
"messages": [
{
"role": "system",
"content": "Du bist ein kollaborativer Assistent für inkrementelle Textanalyse."
},
{"role": "user", "content": full_prompt}
],
"temperature": 0.3,
"max_tokens": 800
}
with httpx.Client(timeout=45.0) as client:
response = client.post(
f"{BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
Nutzung
if __name__ == "__main__":
summarizer = AdvancedSummarizer(API_KEY)
test_text = """
Der deutsche Aktienmarkt hat heute einen deutlichen Aufwärtstrend verzeichnet.
Der DAX stieg um 2,3% auf 18.500 Punkte. Besonders die Technologiebranche
profitierte von positiven Quartalszahlen. SAP legte um 4,1% zu, während
Siemens 2,8% hinzugewann. Analysten führen die Rally auf sinkende
Inflationserwartungen und stabilere Zinsaussichten zurück.
"""
print("=== BRIEF MODE ===")
print(summarizer.summarize(test_text, mode="brief"))
print("\n=== BULLET POINTS MODE ===")
print(summarizer.summarize(test_text, mode="bullet_points"))
print("\n=== EXECUTIVE MODE ===")
print(summarizer.summarize(test_text, mode="executive"))Preise und ROI-Analyse
Lassen Sie uns die tatsächlichen Kosten vergleichen. Angenommen Sie verarbeiten 1 Million deutsche Artikel pro Monat, mit durchschnittlich 2.000 Token pro Artikel:
| Anbieter | 1M Token Kosten | Monatliche Token (1M Artikel) | Monatliche Kosten | Jährliche Kosten |
|---|---|---|---|---|
| HolySheep AI | $0.42 | 2 Milliarden | $840 | $10,080 |
| Google Gemini Flash | $2.50 | 2 Milliarden | $5,000 | $60,000 |
| OpenAI GPT-4.1 | $8.00 | 2 Milliarden | $16,000 | $192,000 |
| Anthropic Claude | $15.00 | 2 Milliarden | $30,000 | $360,000 |
Ersparnis mit HolySheep: Bis zu 97% günstiger als Claude, 95% günstiger als GPT-4. Bei CNY-Bezahlung (¥1≈$0.14) sparen Sie zusätzlich über 85%.
Warum HolySheep AI wählen
Nach meinen Tests und Produktionserfahrungen sprechen mehrere Faktoren für HolySheep AI:
- Unschlagbare Preise: $0.42/MToken mit DeepSeek V3.2 - das günstigste Modell für hochwertige Zusammenfassungen
- Ultraschnelle Latenz: <50ms durch optimierte Infrastructure - ideal für Echtzeit-Anwendungen
- Native CNY-Unterstützung: WeChat Pay und Alipay akzeptiert, kein USD-Wechselkurs-Risiko
- Kostenlose Startcredits: Sofort loslegen ohne Kreditkarte
- Deutsche Sprachqualität: Meine Tests zeigen Exzellent für deutschsprachige Inhalte
- 128K Token Kontext: Verarbeitet auch sehr lange Dokumente ohne Chunking
Häufige Fehler und Lösungen
Fehler 1: ConnectionError - Timeout nach 30 Sekunden
# PROBLEM: API antwortet nicht oder Netzwerk-Probleme
ConnectionError: HTTPSConnectionPool... Connection timed out
LÖSUNG: Implementieren Sie Retry-Logik mit exponentiellem Backoff
import httpx
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def robust_api_call_with_retry(payload: dict, headers: dict) -> dict:
"""
Robuste API-Anfrage mit automatischen Wiederholungen.
"""
async with httpx.AsyncClient(timeout=60.0) as client:
try:
response = await client.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
except httpx.TimeoutException:
print("Timeout - Retry mit erhöhter Wartezeit...")
raise # Löst Retry aus
except httpx.ConnectError as e:
print(f"Verbindungsfehler: {e}")
# DNS-Problem? Versuche alternative Endpoints
await asyncio.sleep(5)
raise
Alternative: Synchrone Version mit manuellem Retry
def summarize_with_manual_retry(text: str, max_retries: int = 3) -> str:
for attempt in range(max_retries):
try:
payload = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": f"Fasse zusammen: {text}"}],
"max_tokens": 200
}
response = httpx.post(
f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload,
timeout=httpx.Timeout(30.0, connect=10.0)
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
except (httpx.TimeoutException, httpx.ConnectError) as e:
wait_time = 2 ** attempt # 1s, 2s, 4s
print(f"Versuch {attempt + 1} fehlgeschlagen: {e}")
print(f"Warte {wait_time}s vor Retry...")
import time
time.sleep(wait_time)
raise RuntimeError(f"Alle {max_retries} Versuche fehlgeschlagen")Fehler 2: 401 Unauthorized - Ungültiger API-Schlüssel
# PROBLEM: Authentifizierung fehlgeschlagen
{"error": {"message": "Invalid authentication", "type": "authentication_error"}}
LÖSUNG: Key-Validierung und Umgebungsvariablen
import os
from typing import Optional
def get_and_validate_api_key() -> str:
"""
Sichere API-Key Verwaltung mit Validierung.
"""
# 1. Aus Umgebungsvariable laden (empfohlen)
api_key = os.environ.get("HOLYSHEEP_API_KEY")
# 2. Falls nicht vorhanden, aus config-Datei
if not api_key:
config_path = os.path.expanduser("~/.holysheep/config.json")
if os.path.exists(config_path):
import json
with open(config_path) as f:
config = json.load(f)
api_key = config.get("api_key")
# 3. Validierung
if not api_key:
raise ValueError(
"HOLYSHEEP_API_KEY nicht gefunden. "
"Bitte setzen Sie die Umgebungsvariable oder registrieren Sie sich unter "
"https://www.holysheep.ai/register"
)
if not api_key.startswith("sk-"):
raise ValueError("Ungültiges API-Key Format. HolySheep-Keys beginnen mit 'sk-'.")
if len(api_key) < 32:
raise ValueError("API-Key scheint zu kurz zu sein. Bitte überprüfen Sie ihn.")
return api_key
Wrapper für API-Aufrufe mit automatischer Auth
class AuthenticatedSummarizer:
def __init__(self):
self.api_key = get_and_validate_api_key()
self.base_url = "https://api.holysheep.ai/v1"
def summarize(self, text: str) -> str:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Test-Anfrage um Key zu validieren
try:
response = httpx.post(
f"{self.base_url}/models",
headers={"Authorization": f"Bearer {self.api_key}"},
timeout=10.0
)
if response.status_code == 401:
raise PermissionError(
"Ungültiger API-Key. Mögliche Ursachen:\n"
"1. Key wurde widerrufen\n"
"2. Key gehört zu einem anderen Account\n"
"3. Tippfehler im Key\n\n"
"Lösung: Registrieren Sie sich neu unter "
"https://www.holysheep.ai/register"
)
except httpx.RequestError as e:
raise ConnectionError(f"Verbindung zum API-Server fehlgeschlagen: {e}")
# Normale Zusammenfassungs-Anfrage
payload = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": f"Zusammenfassung: {text}"}],
"max_tokens": 300
}
response = httpx.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30.0
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]Fehler 3: 429 Too Many Requests - Rate Limit erreicht
# PROBLEM: Zu viele Anfragen in kurzer Zeit
{"error": "Rate limit exceeded. Please wait 60 seconds."}
LÖSUNG: Intelligente Ratenbegrenzung und Request-Queuing
import time
import asyncio
from collections import deque
from threading import Lock
class RateLimitedClient:
"""
API-Client mit automatischer Ratenbegrenzung.
Verhindert 429-Fehler durch intelligentes Request-Management.
"""
def __init__(self, requests_per_minute: int = 60, burst_size: int = 10):
self.rpm = requests_per_minute
self.burst = burst_size
self.request_times: deque = deque(maxlen=requests_per_minute)
self.lock = Lock()
self.min_interval = 60.0 / requests_per_minute
def _wait_if_needed(self):
"""Blockiert falls Rate Limit erreicht."""
current_time = time.time()
with self.lock:
# Entferne alte Requests aus dem Fenster
cutoff = current_time - 60.0
while self.request_times and self.request_times[0] < cutoff:
self.request_times.popleft()
# Prüfe ob Limit erreicht
if len(self.request_times) >= self.rpm:
sleep_time = self.request_times[0] + 60.0 - current_time
if sleep_time > 0:
print(f"Rate Limit erreicht. Warte {sleep_time:.1f}s...")
time.sleep(sleep_time)
self._wait_if_needed()
# Burst-Schutz
recent = [t for t in self.request_times if current_time - t < 1.0]
if len(recent) >= self.burst:
sleep_time = 1.0 - (current_time - min(recent))
if sleep_time > 0:
time.sleep(sleep_time)
self._wait_if_needed()
self.request_times.append(time.time())
def summarize(self, text: str) -> str:
self._wait_if_needed()
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": f"Fasse zusammen: {text}"}],
"max_tokens": 200
}
response = httpx.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30.0
)
if response.status_code == 429:
# Unerwarteter 429 - warte trotzdem und Retry
print("Unerwarteter Rate Limit. Warte 60s und wiederhole...")
time.sleep(60)
return self.summarize(text) # Rekursiver Retry
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
Asynchrone Version mit Token Bucket Algorithmus
class AsyncRateLimiter:
"""Effizienter Rate Limiter für async Anwendungem."""
def __init__(self, rate: float, capacity: int):
"""
Args:
rate: Requests pro Sekunde
capacity: Maximale Burst-Größe
"""
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.updated_at = time.time()
self.lock = asyncio.Lock()
async def acquire(self):
async with self.lock:
while self.tokens < 1:
await self._refill()
await asyncio.sleep(0.01)
self.tokens -= 1
async def _refill(self):
now = time