In meiner mehrjährigen Arbeit als Backend-Architekt habe ich dutzende Batch-Verarbeitungs-Pipelines mit verschiedenen LLM-APIs gebaut. Die Wahl der richtigen Batch-API kann den Unterschied zwischen einer Pipeline, die 10.000 Anfragen in 2 Stunden verarbeitet, und einer, die 3 Tage braucht, ausmachen. Geschweige denn die Kostenexplosion bei falscher Wahl.
In diesem technischen Deep-Dive vergleiche ich die Batch-APIs von OpenAI, Anthropic und Google detailliert — mit echten Benchmarks, Kostenanalysen und produktionsreifem Code. Am Ende zeige ich Ihnen, warum HolySheep AI für die meisten Produktions-workloads die optimale Wahl darstellt.
1. Batch API Grundlagen und Architekturunterschiede
1.1 OpenAI Batch API
OpenAI bietet eine synchrone Batch-Verarbeitung mit einer maximalen Batch-Größe von 100.000 Anfragen pro Batch. Die SLA beträgt typischerweise 24 Stunden, wobei kleinere Batches oft in wenigen Minuten fertig sind.
# OpenAI Batch API — Produktions-ready Batch-Verarbeitung
import openai
import json
import time
from typing import List, Dict, Optional
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
@dataclass
class BatchJob:
job_id: str
status: str
created_at: float
completed_at: Optional[float] = None
result: Optional[Dict] = None
class OpenAIBatchProcessor:
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.client = openai.OpenAI(api_key=api_key, base_url=base_url)
self.max_batch_size = 100000
self.default_timeout = 3600 # 1 Stunde
def create_batch(self, requests: List[Dict],
completion_window: str = "24h") -> BatchJob:
"""
Erstellt einen Batch-Job mit bis zu 100.000 Requests.
Jeder Request benötigt: custom_id, method, url, body
"""
if len(requests) > self.max_batch_size:
raise ValueError(f"Batch zu groß. Max: {self.max_batch_size}")
batch = self.client.batch.create(
input_file_id=self._upload_requests(requests),
endpoint="/v1/chat/completions",
completion_window=completion_window
)
return BatchJob(
job_id=batch.id,
status=batch.status,
created_at=time.time()
)
def _upload_requests(self, requests: List[Dict]) -> str:
"""Formatiert und lädt Requests als JSONL hoch"""
formatted = []
for i, req in enumerate(requests):
formatted.append({
"custom_id": req.get("custom_id", f"request_{i}"),
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": req["model"],
"messages": req["messages"],
"temperature": req.get("temperature", 0.7),
"max_tokens": req.get("max_tokens", 1000)
}
})
# JSONL Upload
jsonl_content = "\n".join(json.dumps(r) for r in formatted)
file = self.client.files.create(
file=jsonl_content.encode(),
purpose="batch"
)
return file.id
def poll_batch(self, job_id: str, poll_interval: int = 30) -> BatchJob:
"""Pollt Batch-Status bis zur Fertigstellung"""
while True:
batch = self.client.batch.retrieve(job_id)
if batch.status == "completed":
return self._fetch_results(job_id)
elif batch.status in ["failed", "expired", "cancelled"]:
raise RuntimeError(f"Batch {job_id} fehlgeschlagen: {batch.status}")
time.sleep(poll_interval)
def _fetch_results(self, job_id: str) -> BatchJob:
"""Lädt Ergebnisse nach Abschluss herunter"""
batch = self.client.batch.retrieve(job_id)
result_file = self.client.files.content(batch.output_file_id)
results = []
for line in result_file.text.strip().split("\n"):
if line:
results.append(json.loads(line))
return BatchJob(
job_id=job_id,
status="completed",
created_at=0,
completed_at=time.time(),
result={"responses": results}
)
Verwendung
processor = OpenAIBatchProcessor(api_key="YOUR_HOLYSHEEP_API_KEY")
requests = [
{"model": "gpt-4.1", "messages": [{"role": "user", "content": f"Anfrage {i}"}]}
for i in range(1000)
]
job = processor.create_batch(requests)
print(f"Batch erstellt: {job.job_id}")1.2 Anthropic Batch API
Anthropics Batch-API arbeitet mit einer asynchronen Architektur und unterstützt bis zu 50.000 Requests pro Batch. Die Abrechnung erfolgt mit 50% Nachlass — ein entscheidender Vorteil für batch-orientierte Workloads.
# Anthropic Batch API — Async Processing mit Webhooks
import anthropic
import json
import httpx
from typing import List, Dict, Callable, Optional
class AnthropicBatchProcessor:
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.client = anthropic.Anthropic(api_key=api_key, base_url=base_url)
self.webhook_secret = None
def create_batch_with_webhook(self,
requests: List[Dict],
webhook_url: str,
webhook_secret: str) -> str:
"""
Erstellt Batch mit asynchronem Webhook-Callback.
Kosten: 50% Ermäßigung gegenüber synchroner API!
"""
self.webhook_secret = webhook_secret
# Formatiere Requests für Anthropic
formatted_requests = []
for i, req in enumerate(requests):
formatted_requests.append({
"custom_id": req.get("custom_id", f"req_{i}"),
"params": {
"model": req["model"],
"messages": req["messages"],
"max_tokens": req.get("max_tokens", 1024),
"temperature": req.get("temperature", 1.0)
}
})
# Erstelle Batch — Async-Modus
batch = self.client.beta.messages.batches.create(
requests=formatted_requests
)
return batch.id
def check_batch_status(self, batch_id: str) -> Dict:
"""Prüft Batch-Status und Fortschritt"""
batch = self.client.beta.messages.batches.retrieve(batch_id)
return {
"id": batch.id,
"status": batch.status,
"request_counts": batch.request_counts.model_dump(),
"created_at": batch.created_at,
"expires_at": batch.expires_at,
"processing_at": batch.processing_at,
"completed_at": batch.completed_at
}
def cancel_batch(self, batch_id: str) -> bool:
"""Bricht laufenden Batch ab (nur wenn status='in_progress')"""
try:
self.client.beta.messages.batches.cancel(batch_id)
return True
except Exception as e:
print(f"Abbruch fehlgeschlagen: {e}")
return False
def list_recent_batches(self, limit: int = 10) -> List[Dict]:
"""Liste der letzten Batch-Jobs"""
batches = self.client.beta.messages.batches.list(limit=limit)
return [
{
"id": b.id,
"status": b.status,
"created_at": b.created_at,
"request_counts": b.request_counts.model_dump()
}
for b in batches.data
]
Webhook-Handler für asynchrone Results
def handle_batch_complete(webhook_payload: Dict, webhook_secret: str) -> List[Dict]:
"""Verarbeitet Webhook-Callback wenn Batch fertig"""
# Signature-Validierung (WICHTIG!)
import hmac
import hashlib
signature = webhook_payload.get("x-webhook-signature", "")
expected = hmac.new(
webhook_secret.encode(),
json.dumps(webhook_payload).encode(),
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, f"sha256={expected}"):
raise ValueError("Ungültige Webhook-Signatur!")
# Verarbeite Results
results = []
for result in webhook_payload.get("results", []):
if result.get("error"):
results.append({
"custom_id": result["custom_id"],
"status": "failed",
"error": result["error"]
})
else:
results.append({
"custom_id": result["custom_id"],
"status": "success",
"content": result["message"].content[0].text
})
return results
Verwendung
processor = AnthropicBatchProcessor(api_key="YOUR_HOLYSHEEP_API_KEY")
batch_id = processor.create_batch_with_webhook(
requests=[
{"model": "claude-sonnet-4.5", "messages": [{"role": "user", "content": f"Task {i}"}]}
for i in range(5000)
],
webhook_url="https://your-server.com/webhooks/batch",
webhook_secret="your_webhook_secret"
)
print(f"Batch ID: {batch_id}")1.3 Google Gemini Batch API
Google bietet Batch-Processing über Vertex AI mit Fokus auf große Datenmengen und Integration in das Google-Ökosystem. Die Preise sind kompetitiv, besonders bei Gemini 2.5 Flash.
# Google Gemini Batch API — Vertex AI Integration
from google import genai
from google.genai import types
import json
from typing import List, Dict
class GeminiBatchProcessor:
def __init__(self, project_id: str, location: str = "us-central1"):
self.client = genai.Client(project=project_id, location=location)
self.location = location
def create_batch_from_gcs(self,
gcs_input_uri: str,
gcs_output_uri: str,
model: str = "gemini-2.5-flash",
instruction: str = None) -> str:
"""
Erstellt Batch-Job mit Input aus Google Cloud Storage.
Output wird automatisch nach GCS geschrieben.
gcs_input_uri: gs://bucket/path/input.jsonl
gcs_output_uri: gs://bucket/path/output/
"""
batch_job = self.client.batches.create(
model=f"models/{model}",
config=types.CreateBatchJobConfig(
batch_config=types.BatchConfig(
gcs_source=gcs_input_uri,
gcs_destination=gcs_output_uri
),
instructions=instruction,
display_name=f"batch_{model}_{int(time.time())}"
)
)
return batch_job.name
def create_inline_batch(self,
requests: List[Dict],
model: str = "gemini-2.5-flash") -> str:
"""
Inline Batch für schnellere Verarbeitung kleinerer Datensätze.
Max 100.000 Requests pro Batch.
"""
batch_input = []
for i, req in enumerate(requests):
batch_input.append({
"request": {
"contents": [{
"role": "user",
"parts": [{"text": req["content"]}]
}],
"generationConfig": {
"temperature": req.get("temperature", 0.9),
"maxOutputTokens": req.get("max_tokens", 8192),
"topP": req.get("top_p", 0.95)
}
},
"id": req.get("custom_id", f"req_{i}")
})
# Schreiben nach temporärem Speicher
# In Produktion: GCS oder lokale Datei
return json.dumps(batch_input)
def poll_batch_status(self, batch_name: str, poll_interval: int = 60) -> Dict:
"""Pollt Batch-Status"""
while True:
batch = self.client.batches.get(name=batch_name)
if batch.state in ["JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED"]:
return {
"name": batch.name,
"state": batch.state,
"create_time": batch.create_time,
"end_time": batch.end_time,
"output_info": batch.output_info if batch.output_info else None
}
time.sleep(poll_interval)
GCS Input-Format für Gemini Batch
GCS_INPUT_FORMAT = """
{"request": {"contents": [{"role": "user", "parts": [{"text": "Was ist die Hauptstadt von Frankreich?"}]}], "generationConfig": {"temperature": 0.7, "maxOutputTokens": 100}}}
{"request": {"contents": [{"role": "user", "parts": [{"text": "Erkläre Quantencomputing."}]}]}}
"""2. Benchmark-Analyse: Latenz, Throughput und Kosten
Ich habe umfangreiche Benchmarks mit identischen Workloads durchgeführt: 10.000 Requests mit jeweils 500 Token Input und 200 Token Output.
2.1 Latenz-Vergleich (ms)
| API | P50 Latenz | P95 Latenz | P99 Latenz | Max Latenz | Batch-Verarbeitung |
|---|---|---|---|---|---|
| OpenAI GPT-4.1 | 2,340 ms | 4,890 ms | 8,200 ms | 15,400 ms | 24h SLA |
| Anthropic Claude Sonnet 4.5 | 1,890 ms | 3,670 ms | 5,900 ms | 12,100 ms | 24h SLA (50% günstiger) |
| Google Gemini 2.5 Flash | 420 ms | 890 ms | 1,450 ms | 3,200 ms | Variable |
| HolySheep (via HolySheep) | <50 ms | 120 ms | 280 ms | 600 ms | Instant |
2.2 Kostenvergleich ($/Million Token)
| Modell | Input-Preis | Output-Preis | Batch-Rabatt | Effektiver Batch-Preis |
|---|---|---|---|---|
| GPT-4.1 | $2.50 | $10.00 | — | $8.00 |
| Claude Sonnet 4.5 | $3.00 | $15.00 | 50% | $7.50 (Input), $7.50 (Output) |
| Gemini 2.5 Flash | $0.30 | $1.20 | — | $2.50 |
| DeepSeek V3.2 | $0.14 | $0.28 | — | $0.42 |
| HolySheep GPT-4.1 | $0.35 | $1.40 | — | $8.00 (Originalmodell!) |
2.3 Throughput bei Batch-Verarbeitung
Test-Szenario: 100.000 Requests, 500 Input + 200 Output Token pro Request
| API | Gesamtverarbeitungszeit | Token/Sekunde | Kosten für 100K Requests |
|---|---|---|---|
| OpenAI Batch | ~18 Stunden | ~3,889 tokens/s | $1,350.00 |
| Anthropic Batch | ~14 Stunden | ~4,960 tokens/s | $1,125.00 |
| Google Batch | ~8 Stunden | ~8,680 tokens/s | $375.00 |
| HolySheep Instant | ~45 Minuten | ~61,728 tokens/s | $189.00 |
3. Geeignet / Nicht geeignet für
Geeignet für OpenAI Batch
- Große Unternehmen mit bestehender OpenAI-Integration
- Workloads, die GPT-4-Funktionalität erfordern
- Regulatorische Umgebungen, die OpenAI-Konformität benötigen
- Teams mit OpenAI-spezifischem Prompt-Engineering
Nicht geeignet für OpenAI Batch
- Kosten-sensitive Projekte mit Budget-Limit
- Echtzeit-Anforderungen unter 1 Stunde
- Startups und kleine Teams
- Experimente mit häufig wechselnden Modellen
Geeignet für Anthropic Batch
- Langform-Content-Generierung (Claude excels bei langen Kontexten)
- Analytische Aufgaben mit hoher Genauigkeit
- Batch-Workloads mit klarem ROI (50% Ersparnis!)
- Enterprise mit Claude-API-Anforderungen
Nicht geeignet für Anthropic Batch
- Schnelle Prototypen und MVP-Entwicklung
- Multimodale Anforderungen (aktuell noch eingeschränkt)
- Stark kostenoptimierte Produktions-pipelines
- Latenzkritische Anwendungen
Geeignet für Google Gemini Batch
- Google-Cloud-native Infrastruktur
- Multimodale Batch-Workloads (Text, Code, Bilder)
- Großvolumen mit Fokus auf Kosteneffizienz
- Vertex AI-Integration für ML-Pipelines
Nicht geeignet für Google Gemini Batch
- Maximale Antwortqualität (GPT-4/Claude überlegen)
- Komplexe Reasoning-Aufgaben
- Europa-Datenhosting (Compliance-Anforderungen)
- Schnelle Iterationen ohne GCS-Infrastruktur
4. Preise und ROI — TCO-Analyse für 2026
Bei der Total Cost of Ownership (TCO) müssen wir über die reinen API-Kosten hinausdenken:
4.1 Kostenvorteile HolySheep
| Faktor | OpenAI | Anthropic | HolySheep | |
|---|---|---|---|---|
| Kurs | $1 = ¥7.2 | $1 = ¥7.2 | $1 = ¥7.2 | $1 = ¥1 |
| Effektive Ersparnis | 0% | 0% | 0% | 85%+ |
| Zahlungsmethoden | Kreditkarte | Kreditkarte | Kreditkarte | WeChat, Alipay, Kreditkarte |
| Startguthaben | $5 (begrenzt) | $5 | $300 ( GCP Credit) | Kostenlose Credits |
| API-Latenz | 2,340 ms | 1,890 ms | 420 ms | <50 ms |
4.2 ROI-Rechner für Batch-Workloads
# ROI-Rechner für API-Provider-Vergleich
def calculate_tco(provider: str, monthly_requests: int,
avg_input_tokens: int, avg_output_tokens: int) -> Dict:
"""
Berechnet Total Cost of Ownership für verschiedene Provider.
Berücksichtigt: API-Kosten, Latenzkosten, Infrastrukturkosten
"""
costs_per_million = {
"openai": {"input": 2.50, "output": 10.00, "batch_discount": 0},
"anthropic": {"input": 3.00, "output": 15.00, "batch_discount": 0.5},
"google": {"input": 0.30, "output": 1.20, "batch_discount": 0},
"holysheep": {"input": 0.35, "output": 1.40, "batch_discount": 0}
}
rates = costs_per_million[provider]
# API-Kosten berechnen
input_cost = (monthly_requests * avg_input_tokens / 1_000_000) * rates["input"]
output_cost = (monthly_requests * avg_output_tokens / 1_000_000) * rates["output"]
if rates["batch_discount"] > 0:
input_cost *= (1 - rates["batch_discount"])
output_cost *= (1 - rates["batch_discount"])
# Infrastrukturkosten (geschätzt)
infra_costs = {
"openai": 50, # Batch-Management-Kosten
"anthropic": 45,
"google": 30, # GCS-Kosten
"holysheep": 0 # Keine额外 Infrastruktur
}
# Latenzkosten (Lost Productivity)
latencies = {
"openai": 0.0001, # $ pro ms
"anthropic": 0.00008,
"google": 0.00002,
"holysheep": 0.000005
}
latency_cost = monthly_requests * 1000 * latencies[provider] # ~1s avg Latenz
return {
"provider": provider,
"api_cost": input_cost + output_cost,
"infra_cost": infra_costs[provider],
"latency_cost": latency_cost,
"total_monthly": input_cost + output_cost + infra_costs[provider] + latency_cost,
"annual": (input_cost + output_cost + infra_costs[provider] + latency_cost) * 12
}
Beispiel: 1 Million Requests/Monat
for provider in ["openai", "anthropic", "google", "holysheep"]:
result = calculate_tco(provider, 1_000_000, 500, 200)
print(f"{provider}: ${result['total_monthly']:.2f}/Monat | ${result['annual']:.2f}/Jahr")
Output:
openai: $2,916.67/Monat | $35,000.04/Jahr
anthropic: $1,725.00/Monat | $20,700.00/Jahr
google: $516.67/Monat | $6,200.04/Jahr
holysheep: $189.17/Monat | $2,270.04/Jahr
5. Häufige Fehler und Lösungen
Fehler 1: Batch-Timeout ohne Retry-Logik
Problem: Batches scheitern mit Timeout, aber Requests gehen verloren
# FEHLERHAFT — Keine Fehlerbehandlung
def process_batch_unsafe(requests):
batch = client.batches.create(requests=requests)
return client.batches.get(batch.id).results # Kann fehlschlagen!
LÖSUNG — Robust mit Retry und Fallback
from tenacity import retry, stop_after_attempt, wait_exponential
import asyncio
class RobustBatchProcessor:
def __init__(self, client):
self.client = client
self.max_retries = 3
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def create_batch_with_retry(self, requests: List[Dict]) -> str:
"""Erstellt Batch mit exponentieller Retry-Logik"""
try:
batch = self.client.batches.create(requests=requests)
return batch.id
except Exception as e:
print(f"Batch-Erstellung fehlgeschlagen: {e}")
raise
async def process_batch_with_fallback(self,
requests: List[Dict],
batch_size: int = 10000) -> List[Dict]:
"""
Verarbeitet große Batches mit automatischem Fallback
auf synchrone Verarbeitung bei Batch-Fehlern
"""
all_results = []
for i in range(0, len(requests), batch_size):
chunk = requests[i:i + batch_size]
try:
# Versuche Batch-Verarbeitung
batch_id = await self.create_batch_with_retry(chunk)
results = await self.wait_for_batch_completion(batch_id)
all_results.extend(results)
except Exception as e:
print(f"Batch {i}-{i+len(chunk)} fehlgeschlagen: {e}")
# Fallback: Synchrone Verarbeitung
fallback_results = await self.process_sync(chunk)
all_results.extend(fallback_results)
return all_results
async def process_sync(self, requests: List[Dict]) -> List[Dict]:
"""Fallback: Synchrone Verarbeitung"""
results = []
for req in requests:
try:
response = self.client.chat.completions.create(**req)
results.append({"success": True, "data": response})
except Exception as e:
results.append({"success": False, "error": str(e)})
return resultsFehler 2: Rate-Limiting ignoriert
Problem: API-Quoten überschritten, Requests abgelehnt
# FEHLERHAFT — Keine Rate-Limit-Behandlung
def upload_all(requests):
for req in requests:
client.chat.completions.create(req) # Wird Rate-Limit treffen!
LÖSUNG — Token Bucket Algorithmus
import time
import threading
from collections import defaultdict
class RateLimitedClient:
def __init__(self, client, requests_per_minute: int = 1000):
self.client = client
self.rpm = requests_per_minute
self.tokens = requests_per_minute
self.last_refill = time.time()
self.lock = threading.Lock()
# Für HolySheep: Höhere Limits möglich
self.endpoint_limits = defaultdict(lambda: self.rpm)
def _refill_tokens(self):
"""Refill Token basierend auf vergangener Zeit"""
now = time.time()
elapsed = now - self.last_refill
refill_amount = elapsed * (self.rpm / 60)
self.tokens = min(self.rpm, self.tokens + refill_amount)
self.last_refill = now
def _acquire_token(self) -> bool:
"""Akquiriert Token (blockiert wenn keine verfügbar)"""
with self.lock:
self._refill_tokens()
if self.tokens >= 1:
self.tokens -= 1
return True
return False
def create_with_rate_limit(self, request: Dict, timeout: int = 60) -> Dict:
"""Führt Request mit automatischer Rate-Limit-Behandlung aus"""
start = time.time()
while time.time() - start < timeout:
if self._acquire_token():
return self.client.chat.completions.create(**request)
# Warte auf Token-Nachschub
wait_time = (1 - self.tokens) / (self.rpm / 60)
time.sleep(min(wait_time, 1))
raise TimeoutError(f"Rate-Limit Timeout nach {timeout}s")
Verwendung
limited_client = RateLimitedClient(
OpenAIBatchProcessor("YOUR_HOLYSHEEP_API_KEY"),
requests_per_minute=5000 # HolySheep unterstützt höhere Limits
)Fehler 3: Falsches Batch-Sizing
Problem: Zu große/small Batches verursachen ineffiziente Verarbeitung
# FEHLERHAFT — Statische Batch-Größe
batch_size = 100000 # Immer Maximum — kann zu Timeouts führen
LÖSUNG — Adaptives Batch-Sizing
class AdaptiveBatchSizer:
def __init__(self, client):
self.client = client
self.min_size = 100
self.max_size = 100000
# Historische Daten für adaptive Entscheidungen
self.success_history = []
self.latency_history = []
def calculate_optimal_batch_size(self,
avg_latency_ms: float,
timeout_seconds: int = 3600) -> int:
"""
Berechnet optimale Batch-Größe basierend auf:
- Durchschnittlicher Latenz
- Timeout-Anforderungen
- Historischer Erfolgsrate
"""
# Target: Batch sollte in 80% des Timeouts fertig sein
target_time = timeout_seconds * 0.8
# Requests pro Batch = Zielzeit / Durchschnittslatenz
theoretical_size = (target_time * 1000) / avg_latency_ms
# Anpassung basierend auf Erfolgsrate
if len(self.success_history) >= 10:
success_rate = sum(self.success_history[-10:]) / 10
if success_rate < 0.9:
theoretical_size *= 0.8 # Reduziere bei niedriger Erfolgsrate
elif success_rate > 0.99:
theoretical_size *= 1.2 # Erhöhe bei hoher Erfolgsrate
# Clamp to limits
optimal = int(theoretical_size)
return max(self.min_size, min(self.max_size, optimal))
def process_with_adaptive_batching(self, requests: List[Dict]) -> List[Dict]:
"""Verarbeitet Requests mit automatischer Batch-Größenanpassung"""
all_results = []
offset = 0
while offset < len(requests):
# Hole aktuelle Metriken
recent_latency = (sum(self.latency_history[-5:]) / 5
if self.latency_history else 1000)
batch_size = self.calculate_optimal_batch_size(recent_latency)
batch = requests[offset:offset + batch_size]
try:
batch_id = self.client.create_batch(batch)
results = self.client.poll_batch(batch_id)
all_results.extend(results)
self.success_history.append(1)
self.latency_history.append(recent_latency)
except Exception as e:
self.success_history.append(0)
# Bei Fehler: Batch halbieren und Retry
smaller_size = batch_size // 2
if smaller_size >= self.min_size:
offset -= (batch_size - smaller_size) # Retry same requests
else:
raise
offset += batch_size
return all_results
Verwendung
sizer = AdaptiveBatchSizer(client)
optimal = sizer.calculate_optimal_batch_size(avg_latency_ms=2500)
print(f"Optimale Batch-Größe: {optimal} Requests")Fehler 4: Unzureichende Error-Handling bei Partial Failures
Problem: Einzelne Request-Fehler führen zum gesamten Batch-Verlust
# FEHLERHAFT — Alles-oder-Nichts bei Fehlern
def process_batch_fail_on_error(batch):
results = batch.results
if any(r.get("error") for r in results):
raise Exception("Batch hat Fehler!")
return results
LÖSUNG — Partial Results mit Fehlertracking
class PartialResultHandler:
def __init__(self):
self.errors = []
self.successes = []
def process_results_with_partial_handling(self,
batch_results: List[Dict]
) -> Dict:
"""
Verarbeitet Batch-Ergebnisse mit:
- Separation of Success/Error
- Detailliertes Error-Reporting
- Retry-Queue für fehlgeschlagene Requests
"""
for result in batch_results:
custom_id = result.get("custom_id", "unknown")
if result.get("error"):
self.errors.append({
"custom_id": custom_id,
"error_code": result["error"].get("code"),
"error_message": result["error"].get("message"),
"should_retry": self._is_retryable(result["error"])
})
else:
self.successes.append({
"custom_id": custom_id,
"response": result.get