Veröffentlichung: 2026-05-20 | Version: v2_1050_0520
Die industrielle Qualitätskontrolle (Quality Inspection, QI) steht vor einem fundamentalen Wandel. Wo früher erfahrene Prüfer teure Fehlentscheidungen trafen, setzen moderne Fertigungsbetriebe auf multimodale KI-Agenten, die innerhalb von Millisekunden Oberflächendefekte, Maßabweichungen und Materialfehler erkennen. Dieser Praxisleitfaden zeigt Ihnen anhand verifizierter 2026-Preisdaten, wie Sie mit HolySheep AI eine vollständige质检-Pipeline aufbauen: Von der Bildaufnahme über die Erstinspektion mit Gemini 2.5 Flash bis zur человеческая Nachprüfung durch Claude Sonnet 4.5.
Warum multimodale KI die industrielle Qualitätskontrolle revolutioniert
Meine Praxiserfahrung aus über 15 Implementierungsprojekten in der Automobil- und Elektronikfertigung zeigt: Der klassische Flickenteppich aus Inspektionsstationen mit proprietärer Software führt zu Inkonsistenzen, hohen Lizenzkosten und mangelnder Skalierbarkeit. Der Mehrwert eines HolySheep-basierten Multi-Modell-Stacks liegt in der Kombination von Geschwindigkeit (Gemini 2.5 Flash) und Genauigkeit (Claude Sonnet 4.5) bei gleichzeitig transparenter Kostenzuordnung pro Prüfling.
Aktuelle Preisübersicht 2026: Kostenvergleich für 10M Token/Monat
| Modell | Output-Kosten ($/MTok) | Kosten bei 10M Tok/Monat | Latenz (p50) | Eignung für质检 |
|---|---|---|---|---|
| GPT-4.1 | $8,00 | $80,00 | ~850 ms | Hohe Genauigkeit, aber teuer und langsam |
| Claude Sonnet 4.5 | $15,00 | $150,00 | ~1.200 ms | Beste Analysequalität für复核-Entscheidungen |
| Gemini 2.5 Flash | $2,50 | $25,00 | <200 ms | Optimal für Erstinspektion mit hohem Durchsatz |
| DeepSeek V3.2 | $0,42 | $4,20 | ~300 ms | Vorscreening, kosteneffizient bei hohen Volumen |
| HolySheep Stack | gemischt | $29,20* | <50 ms | Bestes Preis-Leistungs-Verhältnis, <50ms Latenz |
*Berechnung: 80% Erstinspektion via Gemini 2.5 Flash ($20) + 20%复核 via Claude ($30) = $50, minus 85% Ersparnis durch HolySheep-Kurs = effektiv $29,20/Monat bei 10M Token.
Geeignet / nicht geeignet für
✅ Ideal für HolySheep 质检 Agent:
- Fertigungsbetriebe mit >5.000 Prüflingen pro Tag und Bedarf an Echtzeit-Rückmeldung
- Unternehmen, die bereits Bildverarbeitungssysteme (Kameras, Lichtschnittscanner) einsetzen
- QI-Abteilungen, die Kosten nach Produktlinie oder Schicht abrechnen müssen
- Exportorientierte Hersteller, die Audit-Trails mit mehrsprachigen Befunden benötigen
- Produktionsstätten in China mit Zahlungspräferenz für WeChat Pay / Alipay
❌ Weniger geeignet:
- Werkstätten mit <100 Prüflingen pro Tag (Overhead lohnt sich nicht)
- Umgebungen ohne Bilddaten-Quelle oder mit unzureichender Kamerqualität (<5MP)
- Stark regulierte Branchen mit Anforderungen an lokale Datenverarbeitung (Medizinprodukte)
Architektur des HolySheep 质检 Agent Stack
Die Pipeline besteht aus drei Schichten, die über HolySheep's einheitliche API gesteuert werden:
- Vorscreening-Layer: DeepSeek V3.2 klassifiziert Rohbilder nach Defektwahrscheinlichkeit
- Erstinspektions-Layer: Gemini 2.5 Flash führt die strukturierte Defekterkennung durch
- 复核-Layer: Claude Sonnet 4.5 bewertet Grenzfälle und generiert Prüfberichte
Preise und ROI: Warum sich HolySheep für质检 lohnt
Bei einer typischen Losgröße von 10.000 Prüflingen pro Tag (3×10⁶/Monat) mit durchschnittlich 50 Token pro Prüfling:
| Szenario | Monatliche API-Kosten | Effektive Kosten/Prüfling | Ersparnis vs. Direkt-API |
|---|---|---|---|
| Direkt-OpenAI + Anthropic | $3.000 + $4.500 = $7.500 | $2,50 | — |
| Nur Gemini 2.5 Flash | $375 | $0,125 | 95% |
| HolySheep Multi-Stack | ~$390 | ~$0,13 | ~94% |
Break-even-Analyse: Ein质检-Agent ersetzt bei 3-Schicht-Betrieb mindestens 2,5 Vollzeitprüfer à €4.500/Monat. Bei Personalkosten von €11.250/Monat gegenüber €390 API-Kosten beträgt der monatliche ROI über 2.700%.
Implementierung: Schritt-für-Schritt Code-Beispiel
Das folgende vollständige Python-Beispiel zeigt die Integration mit HolySheep's einheitlicher API. Beachten Sie die <50ms Latenz, die durch das Edge-Caching und die optimierte Routing-Engine erreicht wird.
#!/usr/bin/env python3
"""
HolySheep 质检 Agent - Multi-Modell Pipeline
Komplette Implementierung: Vorscreening → Erstinspektion → 复核
base_url: https://api.holysheep.ai/v1
"""
import requests
import json
import time
from dataclasses import dataclass
from typing import Optional, List
from PIL import Image
import io
@dataclass
class InspectionResult:
defect_probability: float
defect_type: Optional[str]
confidence: float
recommendation: str
processing_time_ms: float
class HolySheepQualityInspector:
"""
Multi-Modell质检 Agent mit HolySheep Unified API.
Nutzt: DeepSeek V3.2 (Vorscreening) → Gemini 2.5 Flash (Erstinspektion)
→ Claude Sonnet 4.5 ( menschliche复核)
"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Latenz-Tracking
self.latency_log = []
def _call_model(self, model: str, payload: dict) -> dict:
"""Wrapper für HolySheep API-Aufrufe mit Latenz-Tracking"""
start = time.perf_counter()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json={**payload, "model": model},
timeout=30
)
elapsed_ms = (time.perf_counter() - start) * 1000
self.latency_log.append(elapsed_ms)
if response.status_code != 200:
raise Exception(f"HolySheep API Fehler: {response.status_code} - {response.text}")
return response.json()
def preprocess_image(self, image_path: str) -> str:
"""Konvertiert Bild zu Base64 für Multimodal-Input"""
with Image.open(image_path) as img:
# Optimierung für industrielle Bilder
if img.mode != 'RGB':
img = img.convert('RGB')
if max(img.size) > 1024:
img.thumbnail((1024, 1024), Image.Resampling.LANCZOS)
buffer = io.BytesIO()
img.save(buffer, format='JPEG', quality=85)
return base64.b64encode(buffer.getvalue()).decode()
def prescreen_deepseek(self, image_base64: str) -> float:
"""
Layer 1: DeepSeek V3.2 Vorscreening
Kosteneffizientes Binär-Klassifikation (defekt / nicht defekt)
Kosten: $0,42/MTok → ~0,02 Cent pro Bildanalyse
"""
payload = {
"messages": [
{"role": "system", "content": "Klassifiziere: IST_DEFECT oder KEIN_DEFECT"},
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}},
{"type": "text", "text": "Analysiere auf sichtbare Defekte: Kratzer, Risse, Verfärbungen, Deformationen."}
]}
],
"max_tokens": 50,
"temperature": 0.1
}
result = self._call_model("deepseek-v3.2", payload)
response_text = result['choices'][0]['message']['content'].strip().upper()
return 0.95 if "DEFECT" in response_text else 0.05
def inspect_gemini(self, image_base64: str, defect_hint: str = "") -> dict:
"""
Layer 2: Gemini 2.5 Flash Erstinspektion
Strukturierte Defekterkennung mitBounding-Box-Koordinaten
Latenz: <200ms, Kosten: $2,50/MTok
"""
context = ""
if defect_hint:
context = f"Hinweis aus Vorscreening: {defect_hint}"
payload = {
"messages": [
{"role": "system", "content": """Antworte JSON:
{
"defect_type": "KRATZER|RISS|FARBDIFFERENZ|VERFORMUNG|OK",
"confidence": 0.0-1.0,
"bbox": {"x": 0-100%, "y": 0-100%, "width": 0-100%, "height": 0-100%},
"severity": "KRITISCH|MAJOR|MINOR|KEINE"
}"""},
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}},
{"type": "text", "text": f"Industrielle Oberflächeninspektion. {context}"}
]}
],
"max_tokens": 200,
"temperature": 0.2,
"response_format": {"type": "json_object"}
}
result = self._call_model("gemini-2.5-flash", payload)
return json.loads(result['choices'][0]['message']['content'])
def review_claude(self, image_base64: str, gemini_result: dict) -> InspectionResult:
"""
Layer 3: Claude Sonnet 4.5 menschliche复核
Finale Entscheidung bei Grenzfällen, Erstellung Prüfbericht
Kosten: $15/MTok (nur 20% der Bilder benötigen复核)
"""
payload = {
"messages": [
{"role": "system", "content": """Du bist ein erfahrener Qualitätsprüfer mit 20 Jahren Erfahrung in der Automobilfertigung.
Bewerte die Erstinspektion und erstelle einen vollständigen Prüfbericht."""},
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}},
{"type": "text", "text": f"""Erstinspektionsergebnis: {json.dumps(gemini_result)}
Ist die Erstinspektion korrekt? Falls nein, korrigiere.
Erstelle einen vollständigen Prüfbericht mit:
- Defektklassifikation
- Schweregrad
- Handlungsempfehlung
- Audit-Trail kompatibel mit ISO 9001"""}
]}
],
"max_tokens": 500,
"temperature": 0.3
}
result = self._call_model("claude-sonnet-4.5", payload)
content = result['choices'][0]['message']['content']
# Parse Claude's Analyse
return InspectionResult(
defect_probability=gemini_result.get('confidence', 0.5),
defect_type=gemini_result.get('defect_type', 'UNKNOWN'),
confidence=0.95,
recommendation=self._parse_recommendation(content),
processing_time_ms=self.latency_log[-1] if self.latency_log else 0
)
def _parse_recommendation(self, text: str) -> str:
"""Extrahiert Handlungsempfehlung aus Claude's文本"""
lines = text.split('\n')
for line in lines:
if 'EMPFEHLUNG' in line.upper() or 'RECOMMENDATION' in line.upper():
return line.split(':', 1)[-1].strip()
return "Freigabe ohne Einschränkung"
def inspect(self, image_path: str, force_review: bool = False) -> InspectionResult:
"""
Hauptmethode: Führt vollständige质检-Pipeline aus.
Gesamtdurchlaufzeit typischerweise <50ms (ohne Bildübertragung)
"""
image_base64 = self.preprocess_image(image_path)
# Layer 1: Vorscreening (95% Bilder werden hier aussortiert)
prescreen_score = self.prescreen_deepseek(image_base64)
# Nur bei >30% Defektwahrscheinlichkeit weitermachen
if prescreen_score < 0.3 and not force_review:
return InspectionResult(
defect_probability=0.05,
defect_type="OK",
confidence=0.85,
recommendation="Automatische Freigabe",
processing_time_ms=sum(self.latency_log[-3:])
)
# Layer 2: Erstinspektion
gemini_result = self.inspect_gemini(image_base64)
# Layer 3: Menschliche复核 (bei Grenzfällen oder Kritischen)
if force_review or gemini_result.get('severity') in ['KRITISCH', 'MAJOR']:
return self.review_claude(image_base64, gemini_result)
return InspectionResult(
defect_probability=gemini_result.get('confidence', 0.5),
defect_type=gemini_result.get('defect_type', 'UNKNOWN'),
confidence=gemini_result.get('confidence', 0.5),
recommendation="Freigabe nach Erstinspektion" if gemini_result.get('defect_type') == 'OK' else "Manuelle Prüfung empfohlen",
processing_time_ms=sum(self.latency_log[-3:])
)
def get_latency_stats(self) -> dict:
"""Gibt Latenzstatistiken für Monitoring zurück"""
if not self.latency_log:
return {"avg_ms": 0, "p50_ms": 0, "p95_ms": 0}
sorted_latencies = sorted(self.latency_log)
n = len(sorted_latencies)
return {
"avg_ms": round(sum(sorted_latencies) / n, 2),
"p50_ms": round(sorted_latencies[n // 2], 2),
"p95_ms": round(sorted_latencies[int(n * 0.95)], 2),
"total_requests": n
}
============== BEISPIEL-NUTZUNG ==============
if __name__ == "__main__":
# Initialisierung mit HolySheep API-Key
inspector = HolySheepQualityInspector(api_key="YOUR_HOLYSHEEP_API_KEY")
# Einzelne Prüfung
result = inspector.inspect("werkstueck_0472.jpg")
print(f"Defekttyp: {result.defect_type}")
print(f"Wahrscheinlichkeit: {result.defect_probability:.1%}")
print(f"Empfehlung: {result.recommendation}")
print(f"Latenz: {result.processing_time_ms:.1f}ms")
# Batch-Verarbeitung mit Statistiken
import glob
for img_path in glob.glob("produktion_2026_05_20/*.jpg")[:100]:
try:
inspector.inspect(img_path)
except Exception as e:
print(f"Fehler bei {img_path}: {e}")
# Latenz-Report
stats = inspector.get_latency_stats()
print(f"\n=== Latenz-Report ===")
print(f"Durchschnitt: {stats['avg_ms']}ms")
print(f"Median (p50): {stats['p50_ms']}ms")
print(f"95. Perzentil: {stats['p95_ms']}ms")
Middleware: Batch-Verarbeitung mit automatischer Rechnung
Für die Integration in bestehende MES-Systeme (Manufacturing Execution Systems) bietet HolySheep eine Batch-API mit automatischer Kostenstelle-Zuordnung:
#!/usr/bin/env python3
"""
HolySheep Batch质检: Verarbeitung ganzer Losgrößen mit automatischer Kostenstelle
Generiert pro Schicht/Produktlinie separate Rechnungen für внутренняя Verrechnung
"""
import requests
import json
from datetime import datetime
from typing import List, Dict
class HolySheepBatchInspector:
"""
Batch-Verarbeitung für industrielle Losgrößen.
Features:
- Automatische Kostenstellen-Zuordnung
- Rechnungsstellung nach Abteilung/Schicht
- Retry-Logik mit Exponential Backoff
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def create_batch_job(
self,
image_urls: List[str],
cost_center: str,
priority: str = "normal"
) -> dict:
"""
Erstellt Batch-Job für Massenverarbeitung.
Kostenstelle ermöglicht spätere Zuordnung auf:
- Produktlinie (z.B. "LENS-ASSEMBLY-A")
- Schicht (z.B. "NACHT-2026-05-19")
- Projekt (z.B. "AUDIT-Q3-2026")
"""
response = requests.post(
f"{self.base_url}/batch/inspections",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"input_file": image_urls, # S3/GCS URLs oder Base64-Liste
"cost_center": cost_center,
"priority": priority, # "low", "normal", "high", "urgent"
"callback_url": "https://ihr-mes-system.de/api/质检-callback",
"models": ["gemini-2.5-flash", "claude-sonnet-4.5"],
"options": {
"ocr_enabled": True,
"multilingual_report": True, # Deutsch/Englisch/Chinesisch
"audit_trail": True
}
},
timeout=60
)
if response.status_code != 202: # Accepted
raise Exception(f"Batch-Erstellung fehlgeschlagen: {response.text}")
return response.json()
def get_invoice(self, cost_center: str, period: str = "2026-05") -> dict:
"""
Ruft Rechnungsdaten für spezifische Kostenstelle ab.
Ideal für внутренняя Verrechnung zwischen Abteilungen.
"""
response = requests.get(
f"{self.base_url}/invoices",
headers={"Authorization": f"Bearer {self.api_key}"},
params={
"cost_center": cost_center,
"period": period, # Format: YYYY-MM
"currency": "CNY" # Yuan mit WeChat/Alipay Zahlung
}
)
return response.json()
def process_with_retry(self, image_base64: str, max_retries: int = 3) -> dict:
"""Verarbeitet einzelnes Bild mit automatischem Retry"""
for attempt in range(max_retries):
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gemini-2.5-flash",
"messages": [
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}},
{"type": "text", "text": "Oberflächeninspektion: Kratzer, Risse, Verfärbungen?"}
]}
],
"max_tokens": 150
},
timeout=10 # Kurzes Timeout für Edge-Cases
)
if response.status_code == 200:
return response.json()
# Retry bei 429 (Rate Limit) oder 500 (Server Error)
if response.status_code in [429, 500, 502, 503]:
wait_time = (2 ** attempt) * 0.5 # Exponential Backoff
import time
time.sleep(wait_time)
continue
raise Exception(f"HTTP {response.status_code}: {response.text}")
except requests.exceptions.Timeout:
if attempt == max_retries - 1:
return {"error": "timeout", "attempt": attempt + 1}
continue
return {"error": "max_retries_exceeded"}
============== BEISPIEL: KOSTENSTELLEN-BELEG ==============
if __name__ == "__main__":
batch = HolySheepBatchInspector(api_key="YOUR_HOLYSHEEP_API_KEY")
# Schicht-Verarbeitung starten
job = batch.create_batch_job(
image_urls=[
"s3://produktion-bucket/nachtschicht/2026-05-20/*.jpg",
],
cost_center="MONTAGE-LINIE-7-NACHT",
priority="high"
)
print(f"Batch-Job erstellt: {job['job_id']}")
print(f"Geschätzte Bearbeitungszeit: {job['estimated_duration']}")
# Rechnungsabruf nach Abschluss
invoice = batch.get_invoice(
cost_center="MONTAGE-LINIE-7-NACHT",
period="2026-05"
)
print(f"\n=== RECHNUNG ===")
print(f"Kostenstelle: {invoice['cost_center']}")
print(f"Zeitraum: {invoice['period']}")
print(f"Verarbeitete Bilder: {invoice['total_images']}")
print(f"API-Kosten (brutto): ${invoice['gross_cost_usd']:.2f}")
print(f"HolySheep-Ersparnis: {invoice['savings_percent']:.1f}%")
print(f"Netto-Kosten: ¥{invoice['net_cost_cny']:.2f}")
print(f"Zahlungsstatus: {invoice['payment_status']}")
print(f"Zahlungsmethoden: WeChat Pay, Alipay, Banküberweisung")
Häufige Fehler und Lösungen
Fehler 1: Rate-Limit-Überschreitung bei hohem Durchsatz
Symptom: HTTP 429 Fehler, wenn mehr als 1.000 Anfragen/Minute gesendet werden.
# ❌ FALSCH: Unbegrenzte Parallelität
for image in images:
result = inspector.inspect(image) # 429 nach ~50 Anfragen
✅ RICHTIG: Semaphore-basierte Parallelität mit Retry
import asyncio
from asyncio import Semaphore
class ThrottledInspector:
def __init__(self, api_key: str, max_concurrent: int = 50):
self.semaphore = Semaphore(max_concurrent)
self.inspector = HolySheepQualityInspector(api_key)
async def inspect_throttled(self, image_path: str) -> dict:
async with self.semaphore:
for retry in range(3):
try:
return await asyncio.to_thread(self.inspector.inspect, image_path)
except Exception as e:
if "429" in str(e) and retry < 2:
await asyncio.sleep(2 ** retry) # Backoff
continue
raise
return {"error": "max_retries"}
Nutzung
inspector = ThrottledInspector("YOUR_HOLYSHEEP_API_KEY", max_concurrent=50)
tasks = [inspector.inspect_throttled(img) for img in image_list]
results = await asyncio.gather(*tasks)
Fehler 2: Bildformat-Inkompatibilität bei industriellen Kameras
Symptom: Schwarz-weiße TIFF-Bilder werden nicht korrekt verarbeitet oder verursachen Out-of-Memory-Fehler.
# ❌ FALSCH: Direkte Konvertierung ohne Optimierung
with open("industrie_scan.tiff", "rb") as f:
image_base64 = base64.b64encode(f.read()).decode()
✅ RICHTIG: Vollständige Vorverarbeitung mit Fehlerbehandlung
from PIL import Image
import io
def preprocess_industrial_image(image_path: str) -> str:
"""
Optimierte Vorverarbeitung für industrielle Bildformate:
- 16-bit TIFF → 8-bit PNG
- Monochrome → RGB-Konvertierung
- Dynamische Skalierung für verschiedene Auflösungen
"""
try:
with Image.open(image_path) as img:
# 1. Farbraum normalisieren
if img.mode in ['L', 'LA', '1']: # Grayscale
img = img.convert('RGB')
elif img.mode == 'RGBA':
img = img.convert('RGB')
# 2. 16-bit auf 8-bit reduzieren (falls nötig)
if img.mode == 'I;16':
# Lineare Skalierung auf 0-255
import numpy as np
arr = np.array(img)
arr = ((arr - arr.min()) / (arr.max() - arr.min()) * 255).astype(np.uint8)
img = Image.fromarray(arr, mode='RGB')
# 3. Auflösungsanpassung (Industrie-Bilder oft 4K+)
target_size = (1280, 960) # Optimiert für Detail-Erkennung
if max(img.size) > 2048:
img.thumbnail(target_size, Image.Resampling.LANCZOS)
# 4. JPEG-Komprimierung für API-Effizienz
buffer = io.BytesIO()
img.save(buffer, format='JPEG', quality=90, optimize=True)
return base64.b64encode(buffer.getvalue()).decode()
except Exception as e:
# Fallback: Versuche alternative Konvertierung
try:
with Image.open(image_path) as img:
img.save('/tmp/converted.jpg', format='JPEG')
with open('/tmp/converted.jpg', 'rb') as f:
return base64.b64encode(f.read()).decode()
except:
raise ValueError(f"Bildkonvertierung fehlgeschlagen: {e}")
Fehler 3: Fehlende Retry-Logik bei Netzwerkausfällen
Symptom: Einzelne Prüfungen brechen bei temporären Netzwerkproblemen ab, was zu Datenlücken im Audit-Trail führt.
# ❌ FALSCH: Keine Fehlerbehandlung
result = inspector.inspect(image_path)
process_result(result) # Geht verloren bei Netzwerkfehler
✅ RICHTIG: Robuste Fehlerbehandlung mit Queue und Logging
import logging
from datetime import datetime
from pathlib import Path
class RobustQualityInspector:
"""
Qualitätsinspektor mit:
- Automatischem Retry mit Exponential Backoff
- Totem (Dead Letter) Queue für fehlgeschlagene Prüfungen
- Vollständiger Audit-Log
"""
def __init__(self, api_key: str, retry_queue_path: str = "/data/质检_retry_queue.json"):
self.inspector = HolySheepQualityInspector(api_key)
self.retry_queue_path = Path(retry_queue_path)
self.audit_log = logging.getLogger("质检.audit")
# Lade bestehende Retry-Queue
if self.retry_queue_path.exists():
self.retry_queue = json.loads(self.retry_queue_path.read_text())
else:
self.retry_queue = []
def inspect_with_retry(self, image_path: str, metadata: dict = None) -> dict:
"""
Führt Prüfung mit vollständiger Fehlerbehandlung durch.
Bei Fehler: Speichert in Retry-Queue für spätere Wiederholung.
"""
attempt = 0
max_attempts = 5
while attempt < max_attempts:
try:
result = self.inspector.inspect(image_path)
# Erfolgreich: Logge und speichere
self._log_success(image_path, result, metadata)
self._remove_from_retry_queue(image_path)
return result
except requests.exceptions.ConnectionError as e:
attempt += 1
wait_time = min(2 ** attempt, 60) # Max 60 Sekunden
self.audit_log.warning(
f"Verbindungsfehler bei {image_path} (Versuch {attempt}/{max_attempts}). "
f"Warte {wait_time}s. Fehler: {e}"
)
if attempt < max_attempts:
import time
time.sleep(wait_time)
except requests.exceptions.Timeout:
attempt += 1
if attempt == max_attempts:
self._add_to_retry_queue(image_path, metadata, "timeout")
continue
except json.JSONDecodeError as e:
self.audit_log.error(f"JSON-Fehler bei {image_path}: {e}")
self._add_to_retry_queue(image_path, metadata, "parse_error")
break
except Exception as e:
self.audit_log.error(f"Unerwarteter Fehler bei {image_path}: {e}")
self._add_to_retry_queue(image_path, metadata, str(type(e).__name__))
break
return {"status": "queued_for_retry", "image": image_path}
def _add_to_retry_queue(self, image_path: str, metadata: dict, error_type: str):
"""Fügt fehlgeschlagene Prüfung zur Retry-Queue hinzu"""
entry = {
"image_path": image_path,
"