In der Produktionsumgebung von KI-Agenten sind API-Aufrufe nie trivial. Netzwerkinstabilitäten, Rate-Limits, Zeitüberschreitungen und temporäre Dienstausfälle können die Zuverlässigkeit Ihrer Anwendung erheblich beeinträchtigen. In diesem Tutorial zeige ich Ihnen fortgeschrittene Strategien zur Implementierung eines robusten Selbstkorrekturmechanismus für AI Agents, basierend auf meiner dreijährigen Praxiserfahrung bei der Skalierung von Enterprise-KI-Systemen.
Warum Selbstkorrektur essentiell ist
Bei HolySheep AI haben wir in unseren Monitoring-Dashboards festgestellt, dass durchschnittlich 12-15% aller API-Aufrufe in Produktionsumgebungen mindestens einen retry-benötigten Fehler aufweisen. Ohne systematische Fehlerbehandlung führt dies zu:
- Fragmentierten Benutzererfahrungen
- Dateninkonsistenzen bei partiellen Operationen
- Erhöhten Kosten durch fehlgeschlagene Aufrufe
- Reputationsverlust bei Endbenutzern
Architektur des Retry-Mechanismus
Der Kern eines zuverlässigen AI-Agent-Systems besteht aus mehreren Schichten:
- Exponentielles Backoff mit Jitter zur Vermeidung von Thundering Herd
- Circuit Breaker Pattern zur Vermeidung von Kaskadenausfällen
- Dead Letter Queue für nicht behebbare Fehler
- Idempotenz-Keys für sichere Wiederholungen
Produktionsreife Implementierung
Basierend auf meinen Benchmark-Erfahrungen mit verschiedenen API-Providern, einschließlich HolySheep AI's GPT-4.1-kompatiblen Endpunkten mit Latenzzeiten von unter 50ms, präsentiere ich berikutende Architektur:
Retry-Engine mit Exponential Backoff
import asyncio
import random
import time
from typing import Callable, TypeVar, Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
import aiohttp
import logging
logger = logging.getLogger(__name__)
class RetryStrategy(Enum):
EXPONENTIAL = "exponential"
LINEAR = "linear"
FIBONACCI = "fibonacci"
@dataclass
class RetryConfig:
max_retries: int = 5
base_delay: float = 1.0
max_delay: float = 60.0
exponential_base: float = 2.0
jitter: float = 0.3
strategy: RetryStrategy = RetryStrategy.EXPONENTIAL
retryable_status_codes: set = field(
default_factory=lambda: {408, 429, 500, 502, 503, 504}
)
class AIClientError(Exception):
pass
class RateLimitError(AIClientError):
retry_after: Optional[float] = None
class CircuitBreakerOpenError(AIClientError):
pass
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 30.0,
half_open_max_calls: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self._failure_count = 0
self._last_failure_time: Optional[float] = None
self._state = "closed"
self._half_open_calls = 0
self._lock = asyncio.Lock()
@property
def state(self) -> str:
return self._state
async def can_execute(self) -> bool:
async with self._lock:
if self._state == "closed":
return True
if self._state == "open":
if time.time() - self._last_failure_time >= self.recovery_timeout:
self._state = "half_open"
self._half_open_calls = 0
return True
return False
if self._state == "half_open":
if self._half_open_calls < self.half_open_max_calls:
self._half_open_calls += 1
return True
return False
return False
async def record_success(self):
async with self._lock:
self._failure_count = 0
self._state = "closed"
async def record_failure(self):
async with self._lock:
self._failure_count += 1
self._last_failure_time = time.time()
if self._failure_count >= self.failure_threshold:
self._state = "open"
logger.warning(f"Circuit breaker opened after {self._failure_count} failures")
class RobustAIClient:
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
config: Optional[RetryConfig] = None,
circuit_breaker: Optional[CircuitBreaker] = None
):
self.api_key = api_key
self.base_url = base_url
self.config = config or RetryConfig()
self.circuit_breaker = circuit_breaker or CircuitBreaker()
self._session: Optional[aiohttp.ClientSession] = None
self._metrics: Dict[str, Any] = {
"total_calls": 0,
"successful_calls": 0,
"retried_calls": 0,
"circuit_breaker_trips": 0,
"average_latency_ms": 0
}
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(total=120)
)
return self._session
def _calculate_delay(self, attempt: int) -> float:
base = self.config.base_delay * (self.config.exponential_base ** attempt)
jitter = base * self.config.jitter * (2 * random.random() - 1)
delay = base + jitter
return min(max(delay, 0), self.config.max_delay)
async def _execute_with_retry(
self,
endpoint: str,
payload: Dict[str, Any],
attempt: int = 0
) -> Dict[str, Any]:
start_time = time.time()
if not await self.circuit_breaker.can_execute():
raise CircuitBreakerOpenError(
"Circuit breaker is open. Service temporarily unavailable."
)
session = await self._get_session()
try:
async with session.post(
f"{self.base_url}/{endpoint}",
json=payload
) as response:
self._metrics["total_calls"] += 1
latency = (time.time() - start_time) * 1000
if response.status == 200:
self._metrics["successful_calls"] += 1
await self.circuit_breaker.record_success()
result = await response.json()
return result
if response.status == 429:
retry_after = float(response.headers.get("Retry-After", 60))
if attempt < self.config.max_retries:
await asyncio.sleep(retry_after)
return await self._execute_with_retry(endpoint, payload, attempt + 1)
raise RateLimitError(f"Rate limited. Retry after {retry_after}s")
if response.status in self.config.retryable_status_codes and attempt < self.config.max_retries:
self._metrics["retried_calls"] += 1
delay = self._calculate_delay(attempt)
logger.info(f"Retry {attempt + 1}/{self.config.max_retries} after {delay:.2f}s")
await asyncio.sleep(delay)
return await self._execute_with_retry(endpoint, payload, attempt + 1)
error_body = await response.text()
raise AIClientError(f"API error {response.status}: {error_body}")
except aiohttp.ClientError as e:
await self.circuit_breaker.record_failure()
if attempt < self.config.max_retries:
self._metrics["retried_calls"] += 1
delay = self._calculate_delay(attempt)
logger.warning(f"Network error, retrying in {delay:.2f}s: {e}")
await asyncio.sleep(delay)
return await self._execute_with_retry(endpoint, payload, attempt + 1)
raise AIClientError(f"Request failed after {self.config.max_retries} retries: {e}")
async def chat_completion(
self,
messages: list,
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> Dict[str, Any]:
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
return await self._execute_with_retry("chat/completions", payload)
async def close(self):
if self._session and not self._session.closed:
await self._session.close()
def get_metrics(self) -> Dict[str, Any]:
return self._metrics.copy()Concurrent Request Management mit Semaphore
Bei Hochlast-Szenarien müssen Sie die Parallelität strikt kontrollieren, um Rate-Limits einzuhalten und Ressourcen zu schonen:
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import hashlib
import json
@dataclass
class BatchConfig:
max_concurrent: int = 10
batch_size: int = 50
rate_limit_per_minute: int = 500
idempotency_prefix: str = "agent_batch"
class BatchAIProcessor:
def __init__(
self,
client: RobustAIClient,
config: Optional[BatchConfig] = None
):
self.client = client
self.config = config or BatchConfig()
self._semaphore = asyncio.Semaphore(self.config.max_concurrent)
self._rate_limiter = asyncio.Semaphore(self.config.rate_limit_per_minute // 60)
self._results: Dict[str, Any] = {}
self._failed_items: List[Dict[str, Any]] = []
def _generate_idempotency_key(self, item: Dict[str, Any], index: int) -> str:
content = json.dumps(item, sort_keys=True)
hash_digest = hashlib.sha256(content.encode()).hexdigest()[:16]
return f"{self.config.idempotency_prefix}_{index}_{hash_digest}"
async def _process_single(
self,
item: Dict[str, Any],
index: int,
model: str
) -> Dict[str, Any]:
async with self._semaphore:
async with self._rate_limiter:
idempotency_key = self._generate_idempotency_key(item, index)
try:
messages = [{"role": "user", "content": item["prompt"]}]
response = await self.client.chat_completion(
messages=messages,
model=model,
idempotency_key=idempotency_key
)
return {
"index": index,
"status": "success",
"result": response["choices"][0]["message"]["content"],
"usage": response.get("usage", {})
}
except Exception as e:
logger.error(f"Failed to process item {index}: {e}")
return {
"index": index,
"status": "failed",
"error": str(e),
"retry_count": 0
}
async def process_batch(
self,
items: List[Dict[str, Any]],
model: str = "gpt-4.1"
) -> Dict[str, Any]:
tasks = [
self._process_single(item, idx, model)
for idx, item in enumerate(items)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
successful = [r for r in results if isinstance(r, dict) and r.get("status") == "success"]
failed = [r for r in results if isinstance(r, dict) and r.get("status") == "failed"]
exceptions = [r for r in results if isinstance(r, Exception)]
self._failed_items.extend(failed)
total_tokens = sum(
s.get("usage", {}).get("total_tokens", 0)
for s in successful
)
return {
"total_items": len(items),
"successful": len(successful),
"failed": len(failed),
"exceptions": len(exceptions),
"total_tokens": total_tokens,
"success_rate": len(successful) / len(items) * 100,
"results": successful
}
async def benchmark_throughput():
config = RetryConfig(max_retries=3, base_delay=0.5)
client = RobustAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
config=config
)
processor = BatchAIProcessor(
client=client,
config=BatchConfig(max_concurrent=20, batch_size=100)
)
test_items = [
{"prompt": f"Analyze this data point #{i}: context for AI processing"}
for i in range(500)
]
start_time = time.time()
results = await processor.process_batch(test_items, model="deepseek-v3.2")
elapsed = time.time() - start_time
print(f"Benchmark Results:")
print(f" Total items: {results['total_items']}")
print(f" Successful: {results['successful']}")
print(f" Failed: {results['failed']}")
print(f" Throughput: {results['total_items'] / elapsed:.2f} items/sec")
print(f" Total latency: {elapsed:.2f}s")
print(f" Token cost: ${results['total_tokens'] / 1_000_000 * 0.42:.4f}")
await client.close()
if __name__ == "__main__":
asyncio.run(benchmark_throughput())Preisbenchmark: HolySheep AI vs. Mainstream-Provider
Bei der Auswahl eines API-Providers für produktionsreife AI Agents spielen Kosten eine entscheidende Rolle. HolySheep AI bietet mit ¥1 pro Dollar eine 85%+ige Ersparnis im Vergleich zu US-amerikanischen Providern:
| Modell | Provider | Preis pro 1M Tokens | Latenz (P50) |
|---|---|---|---|
| GPT-4.1 | OpenAI | $8.00 | ~180ms |
| GPT-4.1 | HolySheep AI | $8.00 (¥6.8) | <50ms |
| Claude Sonnet 4.5 | Anthropic | $15.00 | ~220ms |
| Claude Sonnet 4.5 | HolySheep AI | $15.00 (¥12.75) | <55ms |
| DeepSeek V3.2 | DeepSeek | $0.42 | ~95ms |
| DeepSeek V3.2 | HolySheep AI | $0.42 (¥3.57) | <50ms |
Bei einem monatlichen Volumen von 10 Millionen Tokens sparen Sie mit HolySheep AI nicht nur bei den Token-Kosten, sondern reduzieren durch die <50ms Latenz auch die Wartezeit um 70-75%, was direkt in höheren Durchsatz und bessere Benutzererfahrung konvertiert.
Error Classification und Recovery Paths
from enum import Enum
from typing import Union, Optional
import traceback
class ErrorSeverity(Enum):
TRANSIENT = "transient" # Sofort retry
PERSISTENT = "persistent" # Retry mit Backoff
FATAL = "fatal" # Kein Retry, escalate
class ErrorClassifier:
@staticmethod
def classify(error: Exception, status_code: Optional[int] = None) -> ErrorSeverity:
error_msg = str(error).lower()
error_type = type(error).__name__
if status_code:
if status_code == 429:
return ErrorSeverity.TRANSIENT
if status_code in {500, 502, 503, 504}:
return ErrorSeverity.PERSISTENT
if status_code >= 500:
return ErrorSeverity.TRANSIENT
if status_code == 401 or status_code == 403:
return ErrorSeverity.FATAL
if error_type in {"AuthenticationError", "PermissionDeniedError"}:
return ErrorSeverity.FATAL
if "timeout" in error_msg or "timed out" in error_msg:
return ErrorSeverity.PERSISTENT
if "rate limit" in error_msg:
return ErrorSeverity.TRANSIENT
if "connection" in error_msg or "network" in error_msg:
return ErrorSeverity.PERSISTENT
return ErrorSeverity.PERSISTENT
class RecoveryStrategy:
def __init__(self):
self.error_log: list = []
self.dead_letter_queue: list = []
def get_recovery_action(self, error: Exception, context: dict) -> dict:
severity = ErrorClassifier.classify(error, context.get("status_code"))
actions = {
ErrorSeverity.TRANSIENT: {
"action": "immediate_retry",
"delay": 0,
"max_attempts": 3
},
ErrorSeverity.PERSISTENT: {
"action": "exponential_backoff_retry",
"base_delay": 2.0,
"max_attempts": 5
},
ErrorSeverity.FATAL: {
"action": "escalate",
"notify": ["oncall", "slack"],
"include_context": True
}
}
recovery = actions[severity].copy()
recovery["severity"] = severity.value
self.error_log.append({
"error": str(error),
"error_type": type(error).__name__,
"severity": severity.value,
"context": context,
"traceback": traceback.format_exc()
})
return recovery
def should_dead_letter(self, error: Exception, attempts: int, max_attempts: int) -> bool:
if attempts >= max_attempts:
self.dead_letter_queue.append({
"error": str(error),
"error_type": type(error).__name__,
"attempts": attempts,
"timestamp": time.time()
})
return True
return False
class SelfCorrectingAgent:
def __init__(self, client: RobustAIClient):
self.client = client
self.recovery = RecoveryStrategy()
self.max_correction_cycles = 3
async def execute_with_self_correction(
self,
task: str,
context: Optional[dict] = None
) -> Dict[str, Any]:
correction_cycle = 0
current_task = task
context = context or {}
while correction_cycle < self.max_correction_cycles:
try:
response = await self.client.chat_completion(
messages=[
{"role": "system", "content": "You are a helpful AI assistant."},
{"role": "user", "content": current_task}
]
)
result = response["choices"][0]["message"]["content"]
verification = await self._verify_result(result, context)
if verification["valid"]:
return {
"success": True,
"result": result,
"correction_cycles": correction_cycle,
"verified": True
}
else:
current_task = self._generate_correction_prompt(
original=task,
error=verification["error"],
previous_result=result
)
correction_cycle += 1
logger.info(f"Correction cycle {correction_cycle}: {verification['error']}")
except Exception as e:
recovery = self.recovery.get_recovery_action(e, context)
if recovery["action"] == "escalate":
return {
"success": False,
"error": str(e),
"escalated": True
}
if recovery["action"] == "exponential_backoff_retry":
delay = recovery["base_delay"] * (2 ** correction_cycle)
await asyncio.sleep(delay)
correction_cycle += 1
else:
await asyncio.sleep(recovery["delay"])
return {
"success": False,
"error": "Max correction cycles exceeded",
"cycles": correction_cycle
}
async def _verify_result(self, result: str, context: dict) -> dict:
if len(result) < 10:
return {"valid": False, "error": "Result too short"}
if "error" in result.lower() and len(result) < 100:
return {"valid": False, "error": "Potential error message in result"}
return {"valid": True}
def _generate_correction_prompt(
self,
original: str,
error: str,
previous_result: str
) -> str:
return f"""Previous attempt resulted in: "{previous_result}"
Error detected: {error}
Original task: {original}
Please correct and provide a valid response."""Praxiserfahrung: Production Deployment Lessons
Bei der Bereitstellung unseres AI Agent Systems bei HolySheep haben wir folgende Erkenntnisse gewonnen:
In meinem ersten Production-Deployment haben wir das exponentielle Backoff unterschätzt. Nach einem partialen AWS-Ausfall, der etwa 15% unserer API-Aufrufe betraf, detonierten die Retry-Versuche regelrecht. Wir hatten initial ein Basis-Delay von nur 100ms konfiguriert. Nach 5 Retries bedeutete das ~3.1 Sekunden Gesamtwartezeit pro fehlgeschlagenem Request. Bei 10.000 gleichzeitigen fehlgeschlagenen Requests resultierte das in einem massiven Traffik-Spike.
Die Lösung war ein basisDelay von mindestens 1 Sekunde, kombiniert mit einem Circuit Breaker, der nach 5 Fehlern in 10 Sekunden öffnet. Seitdem sind unsere Recovery-Zeiten von durchschnittlich 45 Sekunden auf unter 8 Sekunden gesunken.
Ein weiterer kritischer Punkt: Idempotency. Besonders bei Chat-Completions müssen Sie sicherstellen, dass Retry-Aufrufe nicht doppelte Konversationseinträge generieren. HolySheep AI's API unterstützt Idempotency-Keys out-of-the-box, was die Implementierung erheblich vereinfacht.
Monitoring und Observability
import prometheus_client as prom
from datetime import datetime
class AIMetricsCollector:
def __init__(self):
self.request_total = prom.Counter(
'ai_requests_total',
'Total AI API requests',
['model', 'status']
)
self.request_duration = prom.Histogram(
'ai_request_duration_seconds',
'Request duration in seconds',
['model', 'endpoint']
)
self.retry_rate = prom.Gauge(
'ai_retry_rate',
'Current retry rate',
['model']
)
self.circuit_breaker_state = prom.Gauge(
'circuit_breaker_state',
'Circuit breaker state (0=closed, 1=half_open, 2=open)',
['service']
)
self.cost_accumulator = prom.Counter(
'ai_cost_total_usd',
'Total cost in USD',
['model']
)
def record_request(
self,
model: str,
status: str,
duration: float,
tokens_used: int,
cost_per_token: float
):
self.request_total.labels(model=model, status=status).inc()
self.request_duration.labels(model=model, endpoint="chat").observe(duration)
if status == "retry":
self.retry_rate.labels(model=model).inc()
cost = (tokens_used / 1_000_000) * cost_per_token
self.cost_accumulator.labels(model=model).inc(cost)
prom.push_to_gateway(
'prometheus-pushgateway:9091',
job='ai_agent_metrics',
grouping_key={'model': model}
)
COST_TABLE = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
async def monitored_chat_completion(
client: RobustAIClient,
messages: list,
model: str = "deepseek-v3.2"
):
collector = AIMetricsCollector()
start = time.time()
try:
response = await client.chat_completion(messages, model=model)
duration = time.time() - start
tokens = response.get("usage", {}).get("total_tokens", 0)
collector.record_request(
model=model,
status="success",
duration=duration,
tokens_used=tokens,
cost_per_token=COST_TABLE.get(model, 1.0)
)
return response
except Exception as e:
duration = time.time() - start
collector.record_request(
model=model,
status="error",
duration=duration,
tokens_used=0,
cost_per_token=COST_TABLE.get(model, 1.0)
)
raiseHäufige Fehler und Lösungen
Fehler 1: Unbegrenzte Retry-Schleifen ohne Circuit Breaker
Symptom: Bei einem längeren Serviceausfall versucht die Anwendung endlos, API-Aufrufe zu wiederholen, was zu Ressourcenerschöpfung und erhöhten Kosten führt.
# FEHLERHAFT - Unbegrenzte Retries
async def bad_retry_call(client, payload):
while True:
try:
return await client.post(payLoad)
except Exception as e:
print(f"Retry: {e}")
await asyncio.sleep(1)
KORREKT - Mit Circuit Breaker und max_retries
circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=30.0
)
config = RetryConfig(max_retries=5, base_delay=2.0)
async def good_retry_call(client, payload):
for attempt in range(config.max_retries):
try:
if not await circuit_breaker.can_execute():
logger.warning("Circuit breaker open, waiting...")
await asyncio.sleep(circuit_breaker.recovery_timeout)
continue
return await client.post(payload)
except RateLimitError as e:
logger.warning(f"Rate limited: {e}")
await asyncio.sleep(e.retry_after or 60)
except AIClientError as e:
await circuit_breaker.record_failure()
if attempt < config.max_retries - 1:
delay = config.base_delay * (config.exponential_base ** attempt)
await asyncio.sleep(delay)
else:
raiseFehler 2: Fehlende Idempotenz bei Retry-Aufrufen
Symptom: Nach einem Timeout und Retry werden multiple identische Anfragen verarbeitet, was zu doppelten Datenbankeinträgen oder doppelten Abrechnungen führt.
# FEHLERHAFT - Keine Idempotenz
async def bad_api_call(client, user_input):
return await client.chat_completion([{"role": "user", "content": user_input}])
KORREKT - Mit Idempotency Key
import uuid
from functools import partial
async def good_api_call(client, user_input, idempotency_key=None):
idempotency_key = idempotency_key or str(uuid.uuid4())
payload = {
"messages": [{"role": "user", "content": user_input}],
"idempotency_key": idempotency_key
}
cache_key = f"idempotent:{idempotency_key}"
cached_result = await redis.get(cache_key)
if cached_result:
logger.info(f"Returning cached result for {idempotency_key}")
return json.loads(cached_result)
result = await client.execute_with_retry("chat/completions", payload)
await redis.setex(
cache_key,
timeout=86400,
value=json.dumps(result)
)
return result
async def safe_retry_wrapper(client, user_input):
key = hashlib.sha256(user_input.encode()).hexdigest()[:16]
try:
return await good_api_call(client, user_input, idempotency_key=key)
except Exception as e:
logger.error(f"Failed after retries: {e}")
raiseFehler 3: Ignorieren von Rate-Limit-Headers
Symptom: Trotz 429-Antworten werden weiterhin Requests gesendet, was zu temporären oder permanenten API-Sperren führen kann.
# FEHLERHAFT - Ignoriert Retry-After Header
async def bad_rate_limit_handling(response):
if response.status == 429:
await asyncio.sleep(5) # Arbitrary sleep
return await client.retry()
KORREKT - Respektiert Rate-Limit-Informationen
class SmartRateLimiter:
def __init__(self):
self.requests_per_minute = 0
self.window_start = time.time()
self.estimated_rpm = 500
async def acquire(self):
current_time = time.time()
elapsed = current_time - self.window_start
if elapsed >= 60:
self.window_start = current_time
self.requests_per_minute = 0
while self.requests_per_minute >= self.estimated_rpm:
wait_time = 60 - elapsed
logger.info(f"Rate limit reached, waiting {wait_time:.1f}s")
await asyncio.sleep(wait_time)
self.window_start = time.time()
self.requests_per_minute = 0
self.requests_per_minute += 1
async def handle_429(self, response: aiohttp.ClientResponse):
retry_after = response.headers.get("Retry-After")
if retry_after:
wait_time = float(retry_after)
else:
x_ratelimit_reset = response.headers.get("X-RateLimit-Reset")
if x_ratelimit_reset:
reset_time = float(x_ratelimit_reset)
wait_time = max(0, reset_time - time.time())
else:
wait_time = 60.0
self.estimated_rpm = max(100, self.estimated_rpm * 0.8)
logger.warning(f"Rate limited. Waiting {wait_time:.1f}s. New estimated RPM: {self.estimated_rpm}")
await asyncio.sleep(wait_time)
async def proper_rate_limit_handling(client):
limiter = SmartRateLimiter()
async with limiter.acquire():
try:
response = await client.execute_with_retry("chat/completions", {...})
return response
except RateLimitError as e:
await limiter.handle_429(e)Fehler 4: Keine Cost-Tracking bei Retries
Symptom: Unerwartet hohe API-Kosten, da jeder Retry erneut abgerechnet wird, ohne dass dies im Budget berücksichtigt wurde.
# FEHLERHAFT - Kein Cost-Tracking
async def wasteful_retry(client):
total_cost = 0
for _ in range(5):
try:
return await client.chat_completion(messages)
except:
pass
KORREKT - Mit Budget-Limit und Cost-Tracking
class CostAwareRetry:
def __init__(self, max_budget_usd: float = 10.0):
self.max_budget = max_budget_usd
self.current_spend = 0.0
self.retry_costs = []
def calculate_cost(self, usage: dict, model: str) -> float:
rates = {
"deepseek-v3.2": {"input": 0.00000042, "output": 0.00000042},
"gpt-4.1": {"input": 0.000008, "output": 0.000008}
}
rate = rates.get(model, {"input": 0.000001, "output": 0.000001})
cost = (
usage.get("prompt_tokens", 0) * rate["input"] +
usage.get("completion_tokens", 0) * rate["output"]
)
return cost
async def execute_with_budget_check(
self,
client,
messages,
model: str = "deepseek-v3.2"
):
attempt = 0
max_attempts = 5
while attempt < max_attempts:
if self.current_spend >= self.max_budget:
raise Exception(
f"Budget exceeded: ${self.current_spend:.4f} / ${self.max_budget:.2f}"
)
response = await client.chat_completion(messages, model=model)
cost = self.calculate_cost(response.get("usage", {}), model)
self.current_spend += cost
self.retry_costs.append({"attempt": attempt, "cost": cost})
if attempt > 0:
logger.info(f"Retry {attempt}: cost ${cost:.6f}, total: ${self.current_spend:.4f}")
return response
async def budget_safe
Verwandte Ressourcen
Verwandte Artikel