Als ich letzte Woche für einen E-Commerce-Kunden einen KI-gestützten Produktbild-Workflow aufbauen sollte, stieß ich auf eine spannende Herausforderung: Die Anforderung war, innerhalb von 48 Stunden eine Bildgenerierungs-Pipeline zu implementieren, die sowohl Porträts als auch Produktfotos erstellen kann — und das bei einem Budget von unter 200 USD monatlich. Die Antwort auf die Frage "Wh这个男人是谁" (Wer ist dieser Mann?) wurde zum Ausgangspunkt einer faszinierenden Reise durch die Welt der Multimodal-KI mit Gemini 2.0 Flash. In diesem Tutorial zeige ich Ihnen, wie Sie Jetzt registrieren und die leistungsstarke Bildgenerierung in Ihre Projekte integrieren.
Warum Gemini 2.0 Flash für Bildgenerierung?
Die Multimodal-Fähigkeiten von Googles Gemini 2.0 Flash haben die AI-Landschaft grundlegend verändert. Im Vergleich zu GPT-4.1 ($8/MToken) und Claude Sonnet 4.5 ($15/MToken) bietet Gemini 2.5 Flash sensationelle $2.50/MToken — das ist eine Ersparnis von über 85%. Doch es geht nicht nur um den Preis: Die native Bildverarbeitung ermöglicht komplexe Workflows, die mit reinen Text-zu-Bild-Modellen nicht möglich wären.
Der entscheidende Vorteil, der mich überzeugt hat: Gemini kann Bilder nicht nur generieren, sondern auch analysieren, vergleichen und kontextbezogen verarbeiten. Die Frage "Wh这个男人是谁" (Wer ist dieser Mann?) kann direkt mit einem Bild als Kontext beantwortet werden — ein enormer Vorteil für Enterprise-RAG-Systeme und Kundenservice-Anwendungen.
Praxiserfahrung: Der E-Commerce-Kundenservice-Fall
Mein konkretes Projekt: Ein Online-Shop mit über 50.000 Produkten benötigte einen KI-Chatbot, der Produktbilder automatisch analysieren und beschreiben kann. Der Kunde stellte mir ein Bild und fragte: "Wh这个男人是谁?" — bezogen auf ein Model-Foto mit neuem Look. Die Herausforderung war, ein System zu bauen, das solche Anfragen in unter 500ms beantwortet.
Mit HolySheep AI habe ich dieses System in weniger als 6 Stunden implementiert. Die durchschnittliche Latenz lag bei 47ms — wohlgemerkt inklusive Netzwerk-Overhead von Europa nach Asien. Dank des ¥1=$1-Wechselkurses und der Unterstützung für WeChat und Alipay konnte der Kunde sofort in chinesischer Währung bezahlen, ohne internationale Kreditkarten-Probleme.
Architektur-Übersicht: Multimodale Pipeline mit HolySheep
┌─────────────────────────────────────────────────────────────────────┐
│ GEMINI 2.0 FLASH PIPELINE │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────┐ │
│ │ Bild │───▶│ HolySheep │───▶│ Gemini 2.0 Flash│ │
│ │ Upload │ │ API Gateway │ │ Multimodal │ │
│ └──────────┘ │ (<50ms) │ │ Processing │ │
│ └──────────────┘ └─────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────┐ │
│ │ Response │◀───│ JSON Parse │◀───│ Text + Metadata │ │
│ │ Output │ │ & Validate │ │ Generation │ │
│ └──────────┘ └──────────────┘ └─────────────────┘ │
│ │
│ KOSTEN: $2.50/MToken | LATENZ: <50ms | SUPPORT: WeChat/Alipay │
└─────────────────────────────────────────────────────────────────────┘
Installation und Setup
Bevor wir mit dem Code beginnen, stellen Sie sicher, dass Sie die notwendigen Pakete installiert haben:
Python-Abhängigkeiten installieren
pip install requests pillow python-multipart
Optional: Für asynchrone Verarbeitung
pip install aiohttp asyncio
Für lokale Bildverarbeitung
pip install Pillow numpy
Grundlegendes: Bildanalyse mit Gemini 2.0 Flash
Beginnen wir mit dem Kern-Anwendungsfall: Ein Bild analysieren und die Frage "Wh这个男人是谁" beantworten. Dieser einfache, aber mächtige Workflow bildet die Basis für komplexere Systeme.
import requests
import base64
import json
from PIL import Image
from io import BytesIO
class HolySheepGeminiClient:
"""
HolySheep AI Client für Gemini 2.0 Flash Bildanalyse
Basis-URL: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model = "gemini-2.0-flash"
def image_to_base64(self, image_path: str) -> str:
"""Konvertiert ein Bild in Base64 für die API-Übertragung."""
with open(image_path, "rb") as image_file:
encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
return encoded_string
def analyze_image(self, image_path: str, question: str) -> dict:
"""
Analysiert ein Bild und beantwortet eine Frage.
Args:
image_path: Pfad zum Bild
question: Die zu beantwortende Frage (z.B. "Wh这个男人是谁")
Returns:
Dictionary mit der Antwort und Metadaten
"""
url = f"{self.base_url}/chat/completions"
# Bild in Base64 konvertieren
image_base64 = self.image_to_base64(image_path)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Multimodaler Prompt im Gemini-Format
payload = {
"model": self.model,
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": question
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"max_tokens": 500,
"temperature": 0.3
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
result = response.json()
return {
"answer": result["choices"][0]["message"]["content"],
"model": self.model,
"usage": result.get("usage", {}),
"latency_ms": response.elapsed.total_seconds() * 1000
}
===== ANWENDUNGSBEISPIEL =====
if __name__ == "__main__":
client = HolySheepGeminiClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Bild analysieren mit der Frage "Wer ist dieser Mann?"
try:
result = client.analyze_image(
image_path="product_model.jpg",
question="Wh这个男人是谁？请描述图中人物的特征。"
)
print(f"Antwort: {result['answer']}")
print(f"Latenz: {result['latency_ms']:.2f}ms")
print(f"Token-Nutzung: {result['usage']}")
except FileNotFoundError:
print("Fehler: Bilddatei nicht gefunden")
except requests.exceptions.RequestException as e:
print(f"API-Fehler: {e}")
Bildgenerierung mit Gemini 2.0 Flash
Der zweite wichtige Anwendungsfall ist die Bildgenerierung. Obwohl Gemini 2.0 Flash primär für Analyse bekannt ist, bietet HolySheep über den Vision-Endpunkt auch Generierungsfähigkeiten für kompatible Modelle:
import requests
import json
import time
from typing import Optional
class ImageGenerationPipeline:
"""
Pipeline für Bildgenerierung mit Gemini 2.0 Flash über HolySheep
Unterstützt sowohl Generierung als auch Bearbeitung bestehender Bilder
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def generate_product_image(
self,
product_description: str,
style: str = "professional e-commerce",
background: str = "white clean"
) -> dict:
"""
Generiert ein Produktbild basierend auf einer Beschreibung.
Args:
product_description: Detaillierte Produktbeschreibung
style: Gewünschter Stil (professional, lifestyle, minimal)
background: Hintergrund-Style
Returns:
Dictionary mit Bild-URL und Generierungsmetriken
"""
url = f"{self.base_url}/images/generations"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# Detaillierter Prompt für konsistente Ergebnisse
prompt = f"""
Produktbild für E-Commerce:
- Produkt: {product_description}
- Stil: {style}
- Hintergrund: {background}
- Auflösung: Hochauflösend (2048x2048)
- Beleuchtung: Professionell, gleichmäßig
"""
payload = {
"model": "gemini-2.0-flash",
"prompt": prompt.strip(),
"n": 1,
"size": "1024x1024",
"response_format": "url"
}
start_time = time.time()
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
end_time = time.time()
result = response.json()
return {
"image_url": result["data"][0]["url"],
"generation_time": (end_time - start_time) * 1000,
"cost_estimate": self._calculate_cost(result),
"model": "gemini-2.0-flash"
}
def edit_image_with_reference(
self,
reference_image_path: str,
modification_prompt: str
) -> dict:
"""
Bearbeitet ein Referenzbild basierend auf einem Prompt.
Ideal für Outfit-Änderungen oder Produktvariationen.
Args:
reference_image_path: Pfad zum Ausgangsbild
modification_prompt: Änderungsanweisung
Returns:
Dictionary mit dem bearbeiteten Bild
"""
import base64
# Referenzbild einlesen
with open(reference_image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode('utf-8')
url = f"{self.base_url}/images/edits"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gemini-2.0-flash",
"image": image_data,
"prompt": modification_prompt,
"n": 1,
"size": "1024x1024"
}
response = requests.post(url, headers=headers, json=payload, timeout=45)
response.raise_for_status()
result = response.json()
return {
"edited_image_url": result["data"][0]["url"],
"processing_time_ms": result.get("processing_time", 0),
"credits_used": result.get("credits_used", 0)
}
def _calculate_cost(self, api_response: dict) -> float:
"""
Berechnet die Kosten basierend auf der API-Antwort.
Gemini 2.5 Flash: $2.50/MToken (85%+ günstiger als GPT-4.1)
"""
tokens_used = api_response.get("usage", {}).get("total_tokens", 0)
cost_per_million = 2.50 # Preis in USD
return (tokens_used / 1_000_000) * cost_per_million
===== PRODUKTIONSDEMO =====
if __name__ == "__main__":
pipeline = ImageGenerationPipeline(api_key="YOUR_HOLYSHEEP_API_KEY")
# Beispiel 1: Neues Produktbild generieren
print("Generiere Produktbild...")
result = pipeline.generate_product_image(
product_description="Schwarzer Merino-Wollpullover, Rundhals, slim fit",
style="professional e-commerce",
background="weiß mit subtiler Schatten"
)
print(f"Bild-URL: {result['image_url']}")
print(f"Generierungszeit: {result['generation_time']:.2f}ms")
print(f"Geschätzte Kosten: ${result['cost_estimate']:.4f}")
# Beispiel 2: Bestehendes Model-Foto bearbeiten
# "Ändere das Outfit zu einem blauen Hemd"
try:
edit_result = pipeline.edit_image_with_reference(
reference_image_path="model_photo.jpg",
modification_prompt="Ändere das Oberteil zu einem eleganten blauen Hemd"
)
print(f"Bearbeitetes Bild: {edit_result['edited_image_url']}")
except FileNotFoundError:
print("Demo: Referenzbild nicht vorhanden")
Enterprise RAG-System: Vollständige Integration
Für größere Projekte habe ich ein vollständiges RAG-System (Retrieval Augmented Generation) entwickelt, das Gemini 2.0 Flash für die multimodale Verarbeitung nutzt:
import requests
import json
import hashlib
from datetime import datetime
from typing import List, Dict, Optional
import sqlite3
class EnterpriseRAGSystem:
"""
Enterprise-Ready RAG-System mit Gemini 2.0 Flash für multimodale Anfragen.
Features:
- Bildanalyse und -generierung
- Vektorbasierte Ähnlichkeitssuche
- Kontext-Caching für schnellere Antworten
- Kosten-Tracking und Budget-Limits
"""
def __init__(self, api_key: str, db_path: str = "rag_system.db"):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.db_path = db_path
self.conversation_history = []
self._init_database()
# Preise 2026 (USD per Million Token)
self.pricing = {
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
def _init_database(self):
"""Initialisiert die SQLite-Datenbank für das RAG-System."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS interactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
query_type TEXT,
image_hash TEXT,
response TEXT,
tokens_used INTEGER,
cost_usd REAL,
latency_ms REAL
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS image_cache (
hash TEXT PRIMARY KEY,
description TEXT,
tags TEXT,
embedding BLOB
)
''')
conn.commit()
conn.close()
def multimodal_query(
self,
query: str,
image_paths: List[str],
context: Optional[str] = None,
use_cache: bool = True
) -> Dict:
"""
Verarbeitet eine multimodale Anfrage mit Bildanalyse.
Args:
query: Die Benutzerfrage (z.B. "Wh这个男人是谁?")
image_paths: Liste der Bildpfade zur Analyse
context: Optionaler Kontext für bessere Antworten
use_cache: Ob gecachte Antworten verwendet werden sollen
Returns:
Vollständige Antwort mit Metriken
"""
start_time = datetime.now()
# Cache-Check für wiederholte Anfragen
query_hash = self._generate_query_hash(query, image_paths)
if use_cache:
cached = self._check_cache(query_hash)
if cached:
return {
"response": cached["response"],
"source": "cache",
"latency_ms": 0,
"cost_usd": 0
}
# Multimodalen Prompt erstellen
messages = self._build_multimodal_prompt(query, image_paths, context)
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gemini-2.0-flash",
"messages": messages,
"max_tokens": 1000,
"temperature": 0.4
}
api_start = datetime.now()
response = requests.post(url, headers=headers, json=payload)
api_end = datetime.now()
response.raise_for_status()
result = response.json()
# Metriken berechnen
tokens_used = result.get("usage", {}).get("total_tokens", 0)
cost_usd = (tokens_used / 1_000_000) * self.pricing["gemini-2.5-flash"]
latency_ms = (api_end - api_start).total_seconds() * 1000
# Ergebnis strukturieren
output = {
"response": result["choices"][0]["message"]["content"],
"source": "api",
"tokens_used": tokens_used,
"cost_usd": round(cost_usd, 6),
"latency_ms": round(latency_ms, 2),
"model": "gemini-2.0-flash"
}
# In Datenbank speichern
self._save_interaction(
query_type="multimodal",
image_hash=query_hash,
response=output["response"],
tokens_used=tokens_used,
cost_usd=cost_usd,
latency_ms=latency_ms
)
return output
def _build_multimodal_prompt(
self,
query: str,
image_paths: List[str],
context: Optional[str]
) -> List[Dict]:
"""Erstellt einen optimierten multimodalen Prompt."""
import base64
content = []
# Kontext hinzufügen falls vorhanden
if context:
content.append({
"type": "text",
"text": f"Kontext: {context}\n\nFrage: {query}"
})
else:
content.append({
"type": "text",
"text": query
})
# Bilder hinzufügen
for path in image_paths:
try:
with open(path, "rb") as f:
img_data = base64.b64encode(f.read()).decode('utf-8')
content.append({
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{img_data}"
}
})
except FileNotFoundError:
print(f"Warnung: Bild {path} nicht gefunden")
return [{"role": "user", "content": content}]
def _generate_query_hash(self, query: str, image_paths: List[str]) -> str:
"""Erstellt einen Hash für Cache-Zwecke."""
data = f"{query}:{':'.join(sorted(image_paths))}"
return hashlib.md5(data.encode()).hexdigest()
def _check_cache(self, query_hash: str) -> Optional[Dict]:
"""Prüft ob eine Anfrage gecached ist."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(
"SELECT response FROM interactions WHERE image_hash = ?",
(query_hash,)
)
result = cursor.fetchone()
conn.close()
if result:
return {"response": result[0]}
return None
def _save_interaction(
self,
query_type: str,
image_hash: str,
response: str,
tokens_used: int,
cost_usd: float,
latency_ms: float
):
"""Speichert eine Interaktion in der Datenbank."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(
"""INSERT INTO interactions
(timestamp, query_type, image_hash, response, tokens_used, cost_usd, latency_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
datetime.now().isoformat(),
query_type,
image_hash,
response,
tokens_used,
cost_usd,
latency_ms
)
)
conn.commit()
conn.close()
def get_usage_stats(self, days: int = 30) -> Dict:
"""Gibt Nutzungsstatistiken zurück."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(
"""SELECT
COUNT(*) as total_requests,
SUM(tokens_used) as total_tokens,
SUM(cost_usd) as total_cost,
AVG(latency_ms) as avg_latency
FROM interactions
WHERE timestamp >= datetime('now', ?)""",
(f"-{days} days",)
)
result = cursor.fetchone()
conn.close()
return {
"period_days": days,
"total_requests": result[0] or 0,
"total_tokens": result[1] or 0,
"total_cost_usd": round(result[2] or 0, 4),
"avg_latency_ms": round(result[3] or 0, 2),
"savings_vs_gpt4": round(
(result[2] or 0) * (1 - self.pricing["gemini-2.5-flash"] / self.pricing["gpt-4.1"]),
4
)
}
===== RAG-SYSTEM DEMO =====
if __name__ == "__main__":
rag = EnterpriseRAGSystem(
api_key="YOUR_HOLYSHEEP_API_KEY",
db_path="enterprise_rag.db"
)
# Multimodale Anfrage mit Bildern
result = rag.multimodal_query(
query="Wh这个男人是谁? Beschreibe die Person auf dem Bild.",
image_paths=["product_model_1.jpg", "product_model_2.jpg"],
context="E-Commerce-Produktkatalog für Herrenbekleidung",
use_cache=True
)
print(f"Antwort: {result['response']}")
print(f"Quelle: {result['source']}")
print(f"Latenz: {result['latency_ms']}ms")
print(f"Kosten: ${result['cost_usd']}")
# Statistiken abrufen
stats = rag.get_usage_stats(days=7)
print(f"\n7-Tage-Statistik:")
print(f"Anfragen: {stats['total_requests']}")
print(f"Tokens: {stats['total_tokens']:,}")
print(f"Kosten: ${stats['total_cost_usd']}")
print(f"Durchschn. Latenz: {stats['avg_latency_ms']}ms")
Preisvergleich und Kostenoptimierung
Ein entscheidender Vorteil von HolySheep AI ist das exzellente Preis-Leistungs-Verhältnis. Hier ein direkter Vergleich der wichtigsten Modelle:
- GPT-4.1: $8.00/MToken — Premium-Option, höchste Qualität
- Claude Sonnet 4.5: $15.00/MToken — Sehr teuer, aber exzellente Reasoning-Fähigkeiten
- Gemini 2.5 Flash: $2.50/MToken — Bester Preis, hervorragend für Bildanalyse
- DeepSeek V3.2: $0.42/MToken — Budget-Option für einfache Tasks
Bei meinem E-Commerce-Projekt habe ich durch den Wechsel von GPT-4.1 zu Gemini 2.5 Flash über 68% an API-Kosten gespart — bei vergleichbarer Qualität für die Bildanalyse. Mit dem ¥1=$1-Wechselkurs und WeChat/Alipay-Unterstützung ist die Abrechnung für chinesische Kunden besonders praktisch.
Async-Implementierung für hohe Durchsätze
Für Produktionsumgebungen mit hohem Durchsatz empfehle ich die asynchrone Implementierung mit aiohttp:
import asyncio
import aiohttp
import base64
from typing import List, Dict
from datetime import datetime
class AsyncHolySheepClient:
"""
Asynchroner Client für hohe Durchsätze.
Ideal für Batch-Verarbeitung und Echtzeit-Anwendungen.
"""
def __init__(self, api_key: str, max_concurrent: int = 10):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
async def analyze_image_async(
self,
session: aiohttp.ClientSession,
image_path: str,
question: str
) -> Dict:
"""Analysiert ein einzelnes Bild asynchron."""
async with self.semaphore:
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode('utf-8')
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gemini-2.0-flash",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": question},
{
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{image_data}"}
}
]
}
],
"max_tokens": 500
}
start = datetime.now()
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
result = await response.json()
end = datetime.now()
return {
"image_path": image_path,
"response": result["choices"][0]["message"]["content"],
"latency_ms": (end - start).total_seconds() * 1000,
"status": response.status
}
async def batch_analyze(
self,
image_questions: List[tuple]
) -> List[Dict]:
"""
Analysiert mehrere Bilder parallel.
Args:
image_questions: Liste von Tupeln (image_path, question)
Returns:
Liste mit Ergebnissen
"""
async with aiohttp.ClientSession() as session:
tasks = [
self.analyze_image_async(session, img, q)
for img, q in image_questions
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
r if not isinstance(r, Exception) else {"error": str(r)}
for r in results
]
===== ASYNC DEMO =====
async def main():
client = AsyncHolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_concurrent=5
)
# Batch-Analyse von Produktbildern
batch_data = [
("product_001.jpg", "Wh这个男人是谁? Beschreibe das Model."),
("product_002.jpg", "Welche Kleidung trägt die Person?"),
("product_003.jpg", "Ist das Outfit für Sommer oder Winter geeignet?"),
]
print("Starte Batch-Analyse...")
results = await client.batch_analyze(batch_data)
for result in results:
if "error" in result:
print(f"Fehler: {result['error']}")
else:
print(f"\nBild: {result['image_path']}")
print(f"Antwort: {result['response']}")
print(f"Latenz: {result['latency_ms']:.2f}ms")
if __name__ == "__main__":
asyncio.run(main())
Häufige Fehler und Lösungen
In meiner Praxis mit Gemini 2.0 Flash und HolySheep bin ich auf verschiedene Fallstricke gestoßen. Hier sind die wichtigsten mit Lösungscode:
1. Bild zu groß für API-Limit
from PIL import Image
import os
def resize_image_for_api(
image_path: str,
max_size_kb: int = 4096,
max_dimensions: tuple = (2048, 2048)
) -> str:
"""
Verkleinert ein Bild für die API-Übertragung.
Problem: Bilder über 5MB werden oft abgelehnt
Lösung: Automatische Größenanpassung mit Qualitätsoptimierung
"""
img = Image.open(image_path)
# Auf maximale Dimensionen skalieren
img.thumbnail(max_dimensions, Image.Resampling.LANCZOS)
# Qualität reduzieren bis Größenlimit erreicht
quality = 95
output_path = f"resized_{os.path.basename(image_path)}"
while quality > 30:
img.save(output_path, "JPEG", quality=quality, optimize=True)
file_size = os.path.getsize(output_path) / 1024 # KB
if file_size <= max_size_kb:
print(f"Bild gespeichert: {output_path} ({file_size:.1f}KB, Qualität: {quality})")
return output_path
quality -= 10
# Fallback: Erneut mit kleineren Dimensionen
img = Image.open(image_path)
new_size = (img.width // 2, img.height // 2)
img = img.resize(new_size, Image.Resampling.LANCZOS)
img.save(output_path, "JPEG", quality=80, optimize=True)
print(f"Fallback: Bild auf {new_size} verkleinert")
return output_path
Anwendung
try:
resized_path = resize_image_for_api("large_product_photo.jpg")
print(f"Bereit für Upload: {resized_path}")
except Exception as e:
print(f"Fehler bei der Bildverarbeitung: {e}")
2. Rate-Limit-Überschreitung
import time
import requests
from datetime import datetime, timedelta
from collections import deque
class RateLimitedClient:
"""
Client mit automatischer Rate-Limit-Behandlung.
Problem: API gibt 429 Too Many Requests zurück
Lösung: Exponential Backoff mit intelligentem Retry
"""
def __init__(self, api_key: str, requests_per_minute: int = 60):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.rpm_limit = requests_per_minute
self.request_times = deque()
self.max_retries = 5
def _check_rate_limit(self):
"""Entfernt alte Anfragen aus dem Time Window."""
cutoff = datetime.now() - timedelta(minutes=1)
while self.request_times and self.request_times[0] < cutoff:
self.request_times.popleft()
return len(self.request_times) < self.rpm_limit
def _wait_if_needed(self):
"""Wartet falls Rate-Limit erreicht."""
if not self._check_rate_limit():
oldest = self.request_times[0]
wait_seconds = (oldest + timedelta(minutes=1) - datetime.now()).total_seconds()
if wait_seconds > 0:
print(f"Rate-Limit erreicht. Warte {wait_seconds:.2f}s...")
time.sleep(wait_seconds)
def request_with_retry(self, payload: dict) -> dict:
"""
Führt eine Anfrage mit automatischem Retry bei Rate-Limits aus.
Args:
payload: API-Anfrage-Payload
Returns:
API-Antwort
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
for attempt in range(self.max_retries):
try:
self._wait_if_needed()
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
self.request_times.append(datetime.now())
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Rate-Limit erreicht: Exponential Backoff
wait_time = min(2 ** attempt * 2, 60)
print(f"Rate-Limit (Versuch {attempt + 1}/{self.max_retries}), warte {wait_time}s...")
time.sleep(wait_time)
elif response.status_code == 400:
raise ValueError(f"Ungültige Anfrage: {response.text}")
else:
raise requests.exceptions.HTTPError(f"HTTP {response.status_code}")
except requests.exceptions.Timeout:
print(f