Module 4: Image Generation

8. Project: Image Generator with Fallback

Next →

Description

In the previous capsules you learned to generate images with DALL-E 3 and Stable Diffusion, optimize prompts, edit images and build pipelines. Now you'll integrate everything into a production system: an Image Generator with multi-provider fallback that tries DALL-E 3 first, uses Stable Diffusion as a backup, validates quality with the Vision API, tracks costs and optimizes prompts automatically.

This isn't a demo script. It's the architecture you'd use in a real service: an API endpoint that receives "generate an image of X" and returns a verified image, with metadata about which provider generated it, how much it cost and how long it took.

What you're going to build:

Input (prompt + config)
    │
    ▼
┌──────────────────┐
│ Prompt Optimizer  │ ← Adapts to the provider
└──────────┬───────┘
           │
    ┌──────▼──────┐
    │  DALL-E 3   │── Success ──→ Quality Check ──→ Result
    └──────┬──────┘                   │
           │ Failure                  │ Failure
    ┌──────▼──────┐                   │
    │  Stable     │── Success ──→ Quality Check ──→ Result
    │  Diffusion  │
    └──────┬──────┘
           │ Failure
    ┌──────▼──────┐
    │    Error     │
    └─────────────┘

Step 1: Configuration and 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)

Step 2: Cost Table

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"]

Step 3: Prompt Optimizer

from openai import OpenAI

client = OpenAI()

OPTIMIZATION_PROMPTS = {
    Provider.DALLE: (
        "Optimize for DALL-E 3. English, 2-3 natural sentences with subject, style, "
        "composition, lighting, mood. No tag format. Only the prompt."
    ),
    Provider.STABLE_DIFFUSION: (
        "Optimize for Stable Diffusion XL. Comma-separated tags in English. "
        "Include subject, style, quality (masterpiece, best quality). Maximum 40 tags. Only the 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()

Step 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,
        )

Step 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())

Step 6: Quality Checker with 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"Evaluate this image for: '{original_prompt}'\n"
                        "Criteria (0-1): relevance, quality, coherence, artifacts (0=many, 1=none)\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")

Step 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 = "All providers failed"
    result.latency_seconds = round(time.time() - start, 2)
    return result

Step 8: Main Function

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 WITH FALLBACK")
    print("=" * 60)

    print("\n--- Test 1: Standard generation ---")
    r1 = generate_with_fallback(
        "Minimalist logo for an AI startup, blue and white colors",
        GenerationConfig(quality=Quality.STANDARD, quality_threshold=0.7),
    )
    print(f"Success: {r1.success}, Provider: {r1.provider_used}, Score: {r1.quality_score}")
    print(f"Costs:\n{r1.cost_summary()}")

    print("\n--- Test 2: Stable Diffusion only ---")
    r2 = generate_with_fallback(
        "Mountain landscape at sunset with a lake in the foreground",
        GenerationConfig(
            provider_order=[Provider.STABLE_DIFFUSION],
            quality_threshold=0.6,
        ),
    )
    print(f"Success: {r2.success}, Provider: {r2.provider_used}, Score: {r2.quality_score}")
    print(f"Costs:\n{r2.cost_summary()}")

    print("\n--- Test 3: Without quality check ---")
    r3 = generate_with_fallback(
        "Modern user interface for a finance app",
        GenerationConfig(enable_quality_check=False),
    )
    print(f"Success: {r3.success}, Provider: {r3.provider_used}")
    print(f"Costs:\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"Success: {success_rate:.0%}, Total cost: ${total_cost:.4f}")


if __name__ == "__main__":
    main()

Extension 1: Batch Generation

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"Cost: ${self.total_cost:.4f} (avg: ${avg:.4f})\n"
            f"Time: {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 = [
    "Premium wireless headphones in matte black on a white background",
    "Sports smartwatch with a circular display and blue strap",
    "Compact mechanical keyboard with RGB lighting on wood",
]

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

Extension 2: Advanced Verification with Auto-improvement

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"Evaluate for: '{original_prompt}'\n"
                        "Criteria (0-1): subject_accuracy, style_match, composition, "
                        "technical_quality, no_artifacts, commercial_ready\n"
                        "If overall < 0.7, suggest an improved prompt.\n"
                        'JSON: {{"scores": {{...}}, "overall": 0-1, "issues": [...], '
                        '"suggestions": [...], "improved_prompt": "...or empty"}}'
                    ),
                },
                {"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(
    "Professional infographic about the water cycle, modern educational style",
    GenerationConfig(quality=Quality.HD, quality_threshold=0.75),
    max_rounds=2,
)
print(f"Score: {improved.quality_score}, Provider: {improved.provider_used}")
print(f"Costs:\n{improved.cost_summary()}")

Troubleshooting

All providers fail

Cause: Invalid or unconfigured API keys.

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 is very slow

Cause: Replicate has cold starts when a model hasn't been used recently.

Solution: Reduce parameters for the first attempt:

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

Quality check rejects good images

Solution: Lower quality_threshold to 0.6. If it persists, use more specific criteria in the evaluation prompt.

Content policy blocks legitimate prompts

Cause: Words that trigger filters by context ("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

Higher costs than expected

Solution: Audit with 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("Costs per step:")
    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 with all parameters configurable
  • GenerationResult with complete tracking of costs and attempts
  • Centralized COST_TABLE with prices per model
  • optimize_prompt() adapts to each provider's format
  • generate_dalle() with content policy and rate limit handling
  • generate_sd() with negative prompts and configurable parameters
  • check_quality() with Vision API and threshold
  • generate_with_fallback() orchestrates providers with retries
  • main() demonstrates 3 scenarios (standard, SD only, without QC)
  • generate_batch() with rate limiting and progress
  • detailed_quality_check() with suggestions and an improved prompt
  • generate_with_auto_improvement() regenerates with improved prompts
  • Logging configured with appropriate levels
  • API key validation
  • Troubleshooting for the 5 most common problems

Exercises

Exercise 1: Add an additional provider

Extend the system to support Stability AI directly (not via Replicate). Add the caller and update the fallback router without modifying the existing logic.

See solution 
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 = "All providers failed"
    result.latency_seconds = round(time.time() - start, 2)
    return result

Exercise 2: Generator metrics dashboard

Create a system that accumulates metrics from all generations (success/failure per provider, costs, latencies, scores) and generates a report to identify the most reliable and economical provider.

See solution 
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']}, Success: {r['success_rate']:.0%}, Cost: ${r['total_cost']:.4f}")
        for prov, s in r.get("provider_stats", {}).items():
            print(f"  {prov}: {s['success_rate']:.0%} success, ${s['avg_cost']:.4f}/img, {s['avg_latency']:.1f}s")
        print(f"Most reliable: {r['most_reliable']}, Most economical: {r['most_economical']}")


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

Summary

This project integrates all the module's techniques:

  • Typed dataclasses for configuration, results and costs
  • Multi-provider fallback (DALL-E → SD) with specific error handling
  • Prompt optimization adapted to each provider's format
  • Quality checking with the Vision API and a configurable threshold
  • Cost tracking granular by step and provider
  • Batch generation with rate limiting and progress
  • Auto-improvement that regenerates with improved prompts
  • Extensibility to add providers without modifying the router

The architecture follows the Chain of Responsibility pattern: each provider is a link. If it fails, it passes to the next. The quality check acts as a gatekeeper. The cost tracker observes each operation without interfering.

Next module: Module 5 — Audio Processing. You already generate images; now you'll master the other output modality: transcription with Whisper and voice synthesis with TTS.

Additional Resources

  1. OpenAI Images API — Complete reference
  2. Replicate Python Client — Official client
  3. OpenAI Pricing — DALL-E and GPT-4o prices
  4. Stability AI API — Direct API
  5. Python Dataclasses — Official documentation
← Previous7. Pipeline IntegrationNext →1. Introduction: Audio in AI