Módulo 4: Generación de Imágenes

8. Proyecto: Image Generator con Fallback

Siguiente →

Descripción

En las cápsulas anteriores aprendiste a generar imágenes con DALL-E 3 y Stable Diffusion, optimizar prompts, editar imágenes y construir pipelines. Ahora vas a integrar todo en un sistema de producción: un Image Generator con fallback multi-proveedor que intenta DALL-E 3 primero, usa Stable Diffusion como respaldo, valida la calidad con Vision API, trackea costos y optimiza prompts automáticamente.

Este no es un script de demo. Es la arquitectura que usarías en un servicio real: un endpoint de API que recibe "genera una imagen de X" y retorna una imagen verificada, con metadata de qué proveedor la generó, cuánto costó y cuánto tardó.

Lo que vas a construir:

Input (prompt + config)
    │
    ▼
┌──────────────────┐
│ Prompt Optimizer  │ ← Adapta al provider
└──────────┬───────┘
           │
    ┌──────▼──────┐
    │  DALL-E 3   │── Éxito ──→ Quality Check ──→ Resultado
    └──────┬──────┘                   │
           │ Fallo                    │ Fallo
    ┌──────▼──────┐                   │
    │  Stable     │── Éxito ──→ Quality Check ──→ Resultado
    │  Diffusion  │
    └──────┬──────┘
           │ Fallo
    ┌──────▼──────┐
    │    Error     │
    └─────────────┘

Paso 1: Configuración y Dataclasses

from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime


class Provider(Enum):
    DALLE = "dalle"
    STABLE_DIFFUSION = "sd"


class ImageSize(Enum):
    SQUARE = "1024x1024"
    LANDSCAPE = "1792x1024"
    PORTRAIT = "1024x1792"


class Quality(Enum):
    STANDARD = "standard"
    HD = "hd"


@dataclass
class GenerationConfig:
    provider_order: list[Provider] = field(
        default_factory=lambda: [Provider.DALLE, Provider.STABLE_DIFFUSION]
    )
    size: ImageSize = ImageSize.SQUARE
    quality: Quality = Quality.STANDARD
    max_retries_per_provider: int = 2
    enable_quality_check: bool = True
    quality_threshold: float = 0.7
    enable_prompt_optimization: bool = True


@dataclass
class CostEntry:
    step: str
    provider: str
    cost_usd: float
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())


@dataclass
class GenerationResult:
    success: bool
    image_b64: str = ""
    image_url: str = ""
    provider_used: str = ""
    prompt_original: str = ""
    prompt_optimized: str = ""
    prompt_revised: str = ""
    quality_score: float = 0.0
    quality_details: dict = field(default_factory=dict)
    costs: list[CostEntry] = field(default_factory=list)
    total_cost_usd: float = 0.0
    latency_seconds: float = 0.0
    attempts: list[dict] = field(default_factory=list)
    error: str = ""

    def add_cost(self, step: str, provider: str, cost: float):
        self.costs.append(CostEntry(step=step, provider=provider, cost_usd=cost))
        self.total_cost_usd = sum(c.cost_usd for c in self.costs)

    def cost_summary(self) -> str:
        lines = [f"  {c.step} ({c.provider}): ${c.cost_usd:.4f}" for c in self.costs]
        lines.append(f"  TOTAL: ${self.total_cost_usd:.4f}")
        return "\n".join(lines)

Paso 2: Tabla de Costos

COST_TABLE = {
    "dall-e-3": {
        "standard": {"1024x1024": 0.040, "1024x1792": 0.080, "1792x1024": 0.080},
        "hd": {"1024x1024": 0.080, "1024x1792": 0.120, "1792x1024": 0.120},
    },
    "sdxl": {"base": 0.004},
    "gpt-4o": {"vision": 0.005},
    "gpt-4o-mini": {"per_1k": 0.000150},
}


def get_dalle_cost(quality: Quality, size: ImageSize) -> float:
    return COST_TABLE["dall-e-3"].get(quality.value, {}).get(size.value, 0.040)

def get_sd_cost() -> float:
    return COST_TABLE["sdxl"]["base"]

def get_vision_cost() -> float:
    return COST_TABLE["gpt-4o"]["vision"]

Paso 3: Prompt Optimizer

from openai import OpenAI

client = OpenAI()

OPTIMIZATION_PROMPTS = {
    Provider.DALLE: (
        "Optimiza para DALL-E 3. Inglés, 2-3 oraciones naturales con sujeto, estilo, "
        "composición, iluminación, mood. Sin formato de tags. Solo el prompt."
    ),
    Provider.STABLE_DIFFUSION: (
        "Optimiza para Stable Diffusion XL. Tags en inglés separados por comas. "
        "Incluye sujeto, estilo, calidad (masterpiece, best quality). Máximo 40 tags. Solo el prompt."
    ),
}

NEGATIVE_PROMPT_SD = (
    "blurry, low quality, worst quality, distorted, ugly, watermark, text, "
    "signature, cropped, out of frame, bad anatomy, extra fingers, jpeg artifacts"
)


def optimize_prompt(prompt: str, provider: Provider) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": OPTIMIZATION_PROMPTS.get(provider, OPTIMIZATION_PROMPTS[Provider.DALLE])},
            {"role": "user", "content": prompt},
        ],
        max_tokens=250,
        temperature=0.3,
    )
    return response.choices[0].message.content.strip()

Paso 4: DALL-E Caller

import time
import logging

logger = logging.getLogger("image_generator")


class DalleError(Exception):
    def __init__(self, message: str, is_content_policy: bool = False, is_rate_limit: bool = False):
        super().__init__(message)
        self.is_content_policy = is_content_policy
        self.is_rate_limit = is_rate_limit


def generate_dalle(
    prompt: str,
    size: ImageSize = ImageSize.SQUARE,
    quality: Quality = Quality.STANDARD,
) -> dict:
    start = time.time()
    try:
        response = client.images.generate(
            model="dall-e-3", prompt=prompt, size=size.value,
            quality=quality.value, response_format="b64_json", n=1,
        )
        return {
            "success": True,
            "b64_json": response.data[0].b64_json,
            "revised_prompt": response.data[0].revised_prompt,
            "latency_seconds": round(time.time() - start, 2),
            "cost": get_dalle_cost(quality, size),
        }
    except Exception as e:
        error_msg = str(e).lower()
        raise DalleError(
            str(e),
            is_content_policy="content_policy" in error_msg or "safety" in error_msg,
            is_rate_limit="rate" in error_msg or "429" in error_msg,
        )

Paso 5: Stable Diffusion Caller

import replicate
import time

class SDError(Exception):
    def __init__(self, message: str, is_timeout: bool = False):
        super().__init__(message)
        self.is_timeout = is_timeout


def generate_sd(
    prompt: str,
    negative_prompt: str = "",
    width: int = 1024,
    height: int = 1024,
    guidance_scale: float = 7.5,
    seed: int = None,
) -> dict:
    start = time.time()
    if not negative_prompt:
        negative_prompt = NEGATIVE_PROMPT_SD

    input_params = {
        "prompt": prompt, "negative_prompt": negative_prompt,
        "width": width, "height": height, "guidance_scale": guidance_scale,
    }
    if seed is not None:
        input_params["seed"] = seed

    try:
        output = replicate.run(
            "stability-ai/sdxl:39a52a2a03a4faf0651640ac8a542059c52f2d04fc26c8b83e22b0a957ffedd3",
            input=input_params,
        )
        url = output[0] if isinstance(output, list) else str(output)
        return {
            "success": True, "url": url,
            "latency_seconds": round(time.time() - start, 2),
            "cost": get_sd_cost(),
        }
    except Exception as e:
        raise SDError(str(e), is_timeout="timeout" in str(e).lower())

Paso 6: Quality Checker con Vision API

import json
import base64
import urllib.request


def check_quality(image_b64: str, original_prompt: str, threshold: float = 0.7) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        f"Evalúa esta imagen para: '{original_prompt}'\n"
                        "Criterios (0-1): relevance, quality, coherence, artifacts (0=muchos, 1=ninguno)\n"
                        'JSON: {{"relevance": 0-1, "quality": 0-1, "coherence": 0-1, '
                        '"artifacts": 0-1, "overall": 0-1, "issues": [...], "pass": true/false}}'
                    ),
                },
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_b64}"}},
            ],
        }],
        max_tokens=300,
    )
    try:
        result = json.loads(response.choices[0].message.content)
    except json.JSONDecodeError:
        result = {"overall": 0.5, "issues": ["Parse error"], "pass": False}

    result["pass"] = result.get("overall", 0) >= threshold
    return result


def url_to_b64(url: str) -> str:
    return base64.b64encode(urllib.request.urlopen(url).read()).decode("utf-8")

Paso 7: Fallback Router

import time
import logging

logger = logging.getLogger("image_generator")


def generate_with_fallback(prompt: str, config: GenerationConfig = None) -> GenerationResult:
    if config is None:
        config = GenerationConfig()

    result = GenerationResult(success=False, prompt_original=prompt)
    start = time.time()
    image_b64 = None

    for provider in config.provider_order:
        optimized = prompt
        if config.enable_prompt_optimization:
            try:
                optimized = optimize_prompt(prompt, provider)
                result.prompt_optimized = optimized
                result.add_cost("prompt_optimization", "gpt-4o-mini", COST_TABLE["gpt-4o-mini"]["per_1k"] * 0.5)
            except Exception:
                optimized = prompt

        for attempt in range(config.max_retries_per_provider):
            attempt_info = {"provider": provider.value, "attempt": attempt + 1}

            try:
                if provider == Provider.DALLE:
                    gen = generate_dalle(optimized, config.size, config.quality)
                    image_b64 = gen["b64_json"]
                    result.prompt_revised = gen.get("revised_prompt", "")
                    result.add_cost("generation", "dall-e-3", gen["cost"])
                elif provider == Provider.STABLE_DIFFUSION:
                    size_parts = config.size.value.split("x")
                    gen = generate_sd(optimized, width=int(size_parts[0]), height=int(size_parts[1]))
                    image_b64 = url_to_b64(gen["url"])
                    result.image_url = gen["url"]
                    result.add_cost("generation", "sdxl", gen["cost"])

                attempt_info["success"] = True
                result.attempts.append(attempt_info)
                result.provider_used = provider.value
                result.image_b64 = image_b64

                if config.enable_quality_check and image_b64:
                    qc = check_quality(image_b64, prompt, config.quality_threshold)
                    result.add_cost("quality_check", "gpt-4o", get_vision_cost())
                    result.quality_score = qc.get("overall", 0)
                    result.quality_details = qc

                    if qc.get("pass", False):
                        result.success = True
                        result.latency_seconds = round(time.time() - start, 2)
                        return result
                    continue
                else:
                    result.success = True
                    result.latency_seconds = round(time.time() - start, 2)
                    return result

            except (DalleError, SDError) as e:
                attempt_info["success"] = False
                attempt_info["error"] = str(e)
                result.attempts.append(attempt_info)

                if isinstance(e, DalleError) and e.is_rate_limit:
                    time.sleep(15 * (attempt + 1))
                elif isinstance(e, DalleError) and e.is_content_policy:
                    break
            except Exception as e:
                attempt_info["success"] = False
                attempt_info["error"] = str(e)
                result.attempts.append(attempt_info)

    if image_b64:
        result.success = True
        result.image_b64 = image_b64
        result.latency_seconds = round(time.time() - start, 2)
        return result

    result.error = "Todos los proveedores fallaron"
    result.latency_seconds = round(time.time() - start, 2)
    return result

Paso 8: Función Principal

import logging


def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
        datefmt="%H:%M:%S",
    )


def main():
    setup_logging()

    print("=" * 60)
    print("IMAGE GENERATOR CON FALLBACK")
    print("=" * 60)

    print("\n--- Test 1: Generación estándar ---")
    r1 = generate_with_fallback(
        "Logo minimalista para startup de IA, colores azul y blanco",
        GenerationConfig(quality=Quality.STANDARD, quality_threshold=0.7),
    )
    print(f"Éxito: {r1.success}, Provider: {r1.provider_used}, Score: {r1.quality_score}")
    print(f"Costos:\n{r1.cost_summary()}")

    print("\n--- Test 2: Solo Stable Diffusion ---")
    r2 = generate_with_fallback(
        "Paisaje de montaña al atardecer con lago en primer plano",
        GenerationConfig(
            provider_order=[Provider.STABLE_DIFFUSION],
            quality_threshold=0.6,
        ),
    )
    print(f"Éxito: {r2.success}, Provider: {r2.provider_used}, Score: {r2.quality_score}")
    print(f"Costos:\n{r2.cost_summary()}")

    print("\n--- Test 3: Sin quality check ---")
    r3 = generate_with_fallback(
        "Interfaz de usuario moderna para app de finanzas",
        GenerationConfig(enable_quality_check=False),
    )
    print(f"Éxito: {r3.success}, Provider: {r3.provider_used}")
    print(f"Costos:\n{r3.cost_summary()}")

    print("\n" + "=" * 60)
    all_results = [r1, r2, r3]
    total_cost = sum(r.total_cost_usd for r in all_results)
    success_rate = sum(1 for r in all_results if r.success) / len(all_results)
    print(f"Éxito: {success_rate:.0%}, Costo total: ${total_cost:.4f}")


if __name__ == "__main__":
    main()

Extensión 1: Generación en Batch

import time
from dataclasses import dataclass, field


@dataclass
class BatchResult:
    total: int = 0
    successful: int = 0
    failed: int = 0
    total_cost: float = 0.0
    total_latency: float = 0.0
    results: list[GenerationResult] = field(default_factory=list)

    @property
    def success_rate(self) -> float:
        return self.successful / self.total if self.total > 0 else 0

    def summary(self) -> str:
        avg = self.total_cost / self.successful if self.successful > 0 else 0
        return (
            f"Batch: {self.successful}/{self.total} ({self.success_rate:.0%})\n"
            f"Costo: ${self.total_cost:.4f} (prom: ${avg:.4f})\n"
            f"Tiempo: {self.total_latency:.1f}s"
        )


def generate_batch(
    prompts: list[str],
    config: GenerationConfig = None,
    delay_seconds: float = 10.0,
) -> BatchResult:
    if config is None:
        config = GenerationConfig()

    batch = BatchResult(total=len(prompts))
    start = time.time()

    for i, prompt in enumerate(prompts):
        print(f"[{i + 1}/{len(prompts)}] {prompt[:50]}...")
        result = generate_with_fallback(prompt, config)
        batch.results.append(result)

        if result.success:
            batch.successful += 1
            batch.total_cost += result.total_cost_usd
        else:
            batch.failed += 1

        if i < len(prompts) - 1:
            time.sleep(delay_seconds)

    batch.total_latency = round(time.time() - start, 2)
    return batch


product_prompts = [
    "Audífonos inalámbricos premium en negro mate sobre fondo blanco",
    "Smartwatch deportivo con pantalla circular y correa azul",
    "Teclado mecánico compacto con iluminación RGB sobre madera",
]

batch = generate_batch(
    product_prompts,
    GenerationConfig(quality=Quality.STANDARD, quality_threshold=0.65),
    delay_seconds=12.0,
)
print(f"\n{batch.summary()}")

Extensión 2: Verificación Avanzada con Auto-mejora

import json
from dataclasses import dataclass, field


@dataclass
class DetailedQualityReport:
    overall_score: float = 0.0
    scores: dict = field(default_factory=dict)
    issues: list[str] = field(default_factory=list)
    suggestions: list[str] = field(default_factory=list)
    improved_prompt: str = ""
    approved: bool = False


def detailed_quality_check(image_b64: str, original_prompt: str, threshold: float = 0.7) -> DetailedQualityReport:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        f"Evalúa para: '{original_prompt}'\n"
                        "Criterios (0-1): subject_accuracy, style_match, composition, "
                        "technical_quality, no_artifacts, commercial_ready\n"
                        "Si overall < 0.7, sugiere prompt mejorado.\n"
                        'JSON: {{"scores": {{...}}, "overall": 0-1, "issues": [...], '
                        '"suggestions": [...], "improved_prompt": "...o vacío"}}'
                    ),
                },
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_b64}"}},
            ],
        }],
        max_tokens=400,
    )

    try:
        data = json.loads(response.choices[0].message.content)
    except json.JSONDecodeError:
        data = {"scores": {}, "overall": 0.5, "issues": ["Parse error"], "suggestions": [], "improved_prompt": ""}

    return DetailedQualityReport(
        overall_score=data.get("overall", 0),
        scores=data.get("scores", {}),
        issues=data.get("issues", []),
        suggestions=data.get("suggestions", []),
        improved_prompt=data.get("improved_prompt", ""),
        approved=data.get("overall", 0) >= threshold,
    )


def generate_with_auto_improvement(
    prompt: str,
    config: GenerationConfig = None,
    max_rounds: int = 2,
) -> GenerationResult:
    if config is None:
        config = GenerationConfig()

    current_prompt = prompt
    best_result = None

    for round_num in range(max_rounds + 1):
        result = generate_with_fallback(current_prompt, config)
        if not result.success:
            return best_result or result

        if best_result is None or result.quality_score > best_result.quality_score:
            best_result = result

        if result.quality_score >= config.quality_threshold:
            return result

        qc = detailed_quality_check(result.image_b64, prompt, config.quality_threshold)
        if qc.approved or not qc.improved_prompt:
            return best_result

        current_prompt = qc.improved_prompt

    return best_result


improved = generate_with_auto_improvement(
    "Infografía profesional sobre el ciclo del agua, estilo educativo moderno",
    GenerationConfig(quality=Quality.HD, quality_threshold=0.75),
    max_rounds=2,
)
print(f"Score: {improved.quality_score}, Provider: {improved.provider_used}")
print(f"Costos:\n{improved.cost_summary()}")

Troubleshooting

Todos los proveedores fallan

Causa: API keys inválidas o no configuradas.

import os


def validate_api_keys() -> dict:
    keys = {
        "OPENAI_API_KEY": os.getenv("OPENAI_API_KEY"),
        "REPLICATE_API_TOKEN": os.getenv("REPLICATE_API_TOKEN"),
    }
    status = {}
    for name, value in keys.items():
        if not value:
            status[name] = "MISSING"
        elif len(value) < 10:
            status[name] = "TOO_SHORT"
        elif name == "OPENAI_API_KEY" and not value.startswith("sk-"):
            status[name] = "INVALID_FORMAT"
        else:
            status[name] = "OK"
    return status


for key, st in validate_api_keys().items():
    print(f"  {key}: {st}")

SD es muy lento

Causa: Replicate tiene cold starts cuando un modelo no se ha usado recientemente.

Solución: Reducir parámetros para el primer intento:

FAST_SD_CONFIG = {"num_inference_steps": 20, "guidance_scale": 7.0, "width": 768, "height": 768}

Quality check rechaza imágenes buenas

Solución: Bajar quality_threshold a 0.6. Si persiste, usar criterios más específicos en el prompt de evaluación.

Content policy bloquea prompts legítimos

Causa: Palabras que activan filtros por contexto ("blood orange", "shooting star").

TRIGGER_WORDS = {
    "blood": "deep red", "weapon": "tool", "gun": "device",
    "shot": "capture", "naked": "bare", "war": "conflict",
}


def sanitize_for_dalle(prompt: str) -> str:
    result = prompt
    for trigger, safe in TRIGGER_WORDS.items():
        result = result.replace(trigger, safe)
    return result

Costos más altos de lo esperado

Solución: Audita con cost_summary():

def audit_batch(batch: BatchResult):
    by_step: dict[str, float] = {}
    for r in batch.results:
        for c in r.costs:
            by_step[c.step] = by_step.get(c.step, 0) + c.cost_usd

    print("Costos por paso:")
    for step, cost in sorted(by_step.items(), key=lambda x: -x[1]):
        pct = cost / batch.total_cost * 100 if batch.total_cost else 0
        print(f"  {step}: ${cost:.4f} ({pct:.0f}%)")

Checklist

  • GenerationConfig con todos los parámetros configurables
  • GenerationResult con tracking completo de costos y attempts
  • COST_TABLE centralizada con precios por modelo
  • optimize_prompt() adapta al formato de cada provider
  • generate_dalle() con manejo de content policy y rate limit
  • generate_sd() con negative prompts y parámetros configurables
  • check_quality() con Vision API y threshold
  • generate_with_fallback() orquesta providers con reintentos
  • main() demuestra 3 escenarios (estándar, SD only, sin QC)
  • generate_batch() con rate limiting y progreso
  • detailed_quality_check() con sugerencias y prompt mejorado
  • generate_with_auto_improvement() regenera con prompts mejorados
  • Logging configurado con niveles apropiados
  • Validación de API keys
  • Troubleshooting para los 5 problemas más comunes

Ejercicios

Ejercicio 1: Agregar proveedor adicional

Extiende el sistema para soportar Stability AI directamente (no vía Replicate). Agrega el caller y actualiza el fallback router sin modificar la lógica existente.

Ver solución 
import requests
import os
import time


class StabilityError(Exception):
    def __init__(self, message: str, status_code: int = 0):
        super().__init__(message)
        self.status_code = status_code


def generate_stability(
    prompt: str,
    negative_prompt: str = "",
    width: int = 1024,
    height: int = 1024,
    cfg_scale: float = 7.0,
) -> dict:
    start = time.time()
    api_key = os.getenv("STABILITY_API_KEY")
    if not api_key:
        raise StabilityError("STABILITY_API_KEY not set")

    if not negative_prompt:
        negative_prompt = NEGATIVE_PROMPT_SD

    response = requests.post(
        "https://api.stability.ai/v1/generation/stable-diffusion-xl-1024-v1-0/text-to-image",
        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Accept": "application/json"},
        json={
            "text_prompts": [
                {"text": prompt, "weight": 1.0},
                {"text": negative_prompt, "weight": -1.0},
            ],
            "cfg_scale": cfg_scale, "width": width, "height": height, "steps": 30, "samples": 1,
        },
    )

    if response.status_code != 200:
        raise StabilityError(f"API error: {response.text}", response.status_code)

    return {
        "success": True, "provider": "stability-ai",
        "b64_json": response.json()["artifacts"][0]["base64"],
        "latency_seconds": round(time.time() - start, 2),
        "cost": 0.006,
    }


def generate_with_three_providers(prompt: str, provider_order: list[str] = None) -> GenerationResult:
    if provider_order is None:
        provider_order = ["dalle", "stability", "sd"]

    result = GenerationResult(success=False, prompt_original=prompt)
    start = time.time()

    callers = {
        "dalle": lambda p: generate_dalle(p),
        "sd": lambda p: generate_sd(p),
        "stability": lambda p: generate_stability(p),
    }

    for name in provider_order:
        caller = callers.get(name)
        if not caller:
            continue
        try:
            optimized = optimize_prompt(
                prompt, Provider.DALLE if name == "dalle" else Provider.STABLE_DIFFUSION
            )
            gen = caller(optimized)
            b64 = gen.get("b64_json", "")
            if not b64 and gen.get("url"):
                b64 = url_to_b64(gen["url"])

            result.image_b64 = b64
            result.provider_used = name
            result.add_cost("generation", name, gen["cost"])
            result.success = True
            result.latency_seconds = round(time.time() - start, 2)
            return result
        except Exception as e:
            result.attempts.append({"provider": name, "error": str(e)})

    result.error = "Todos los proveedores fallaron"
    result.latency_seconds = round(time.time() - start, 2)
    return result

Ejercicio 2: Dashboard de métricas del generador

Crea un sistema que acumule métricas de todas las generaciones (éxito/fallo por provider, costos, latencias, scores) y genere un reporte para identificar el provider más confiable y económico.

Ver solución 
from dataclasses import dataclass, field


@dataclass
class MetricsCollector:
    entries: list[dict] = field(default_factory=list)

    def record(self, result: GenerationResult):
        self.entries.append({
            "success": result.success, "provider": result.provider_used,
            "quality": result.quality_score, "cost": result.total_cost_usd,
            "latency": result.latency_seconds, "attempts": len(result.attempts),
        })

    def report(self) -> dict:
        if not self.entries:
            return {"message": "No data"}

        successful = [e for e in self.entries if e["success"]]
        by_provider: dict[str, list] = {}
        for e in self.entries:
            by_provider.setdefault(e["provider"] or "none", []).append(e)

        stats = {}
        for prov, entries in by_provider.items():
            ok = [e for e in entries if e["success"]]
            costs = [e["cost"] for e in ok]
            latencies = [e["latency"] for e in ok]
            scores = [e["quality"] for e in ok if e["quality"] > 0]
            stats[prov] = {
                "total": len(entries),
                "success_rate": len(ok) / len(entries) if entries else 0,
                "avg_cost": sum(costs) / len(costs) if costs else 0,
                "avg_latency": sum(latencies) / len(latencies) if latencies else 0,
                "avg_quality": sum(scores) / len(scores) if scores else 0,
            }

        return {
            "total": len(self.entries),
            "success_rate": len(successful) / len(self.entries),
            "total_cost": sum(e["cost"] for e in self.entries),
            "provider_stats": stats,
            "most_reliable": max(stats, key=lambda k: stats[k]["success_rate"]) if stats else "none",
            "most_economical": min(
                (k for k in stats if stats[k]["avg_cost"] > 0),
                key=lambda k: stats[k]["avg_cost"], default="none",
            ),
        }

    def print_report(self):
        r = self.report()
        print(f"\nTotal: {r['total']}, Éxito: {r['success_rate']:.0%}, Costo: ${r['total_cost']:.4f}")
        for prov, s in r.get("provider_stats", {}).items():
            print(f"  {prov}: {s['success_rate']:.0%} éxito, ${s['avg_cost']:.4f}/img, {s['avg_latency']:.1f}s")
        print(f"Más confiable: {r['most_reliable']}, Más económico: {r['most_economical']}")


metrics = MetricsCollector()
for prompt in ["Modern office", "Tropical sunset", "Abstract geometric"]:
    metrics.record(generate_with_fallback(prompt))
metrics.print_report()

Resumen

Este proyecto integra todas las técnicas del módulo:

  • Dataclasses tipadas para configuración, resultados y costos
  • Fallback multi-proveedor (DALL-E → SD) con manejo específico de errores
  • Prompt optimization adaptada al formato de cada proveedor
  • Quality checking con Vision API y threshold configurable
  • Cost tracking granular por paso y proveedor
  • Batch generation con rate limiting y progreso
  • Auto-improvement que regenera con prompts mejorados
  • Extensibilidad para agregar proveedores sin modificar el router

La arquitectura sigue Chain of Responsibility: cada proveedor es un eslabón. Si falla, pasa al siguiente. El quality check actúa como gate keeper. El cost tracker observa cada operación sin interferir.

Próximo módulo: Módulo 5 — Procesamiento de Audio. Ya generas imágenes; ahora dominarás la otra modalidad de output: transcripción con Whisper y síntesis de voz con TTS.

Recursos Adicionales

  1. OpenAI Images API — Referencia completa
  2. Replicate Python Client — Cliente oficial
  3. OpenAI Pricing — Precios de DALL-E y GPT-4o
  4. Stability AI API — API directa
  5. Python Dataclasses — Documentación oficial
← Anterior7. Integración en PipelinesSiguiente →1. Introducción: Audio en IA