Phase 7: Multi-Model Pipeline Orchestration · 50 min · Python · Celery · Redis

A/B Testing Models & Batch Processing

Same prompt, multiple models, compare quality — that's A/B testing. 1000 products overnight — that's batch processing.

Hiring signal: A/B testing and batch processing at scale are production engineering skills that demonstrate you can operate generative media systems, not just build them.

What you will learn

Implement A/B testing models: same prompt, multiple models, compare quality scores
Implement A/B testing parameters: same model, different guidance/steps/samplers
Build batch processing: 1000 products through image pipeline overnight
Implement batch orchestration: job queuing, progress tracking, failure handling, resume

The Problem

A team needs to:

A/B test models: Is FLUX better than SDXL for product images? Same prompt, both models, compare quality scores
A/B test parameters: Does 30 steps produce better results than 20 steps? Same model, different parameters
Batch process: Generate images for 1000 products overnight — queue, track progress, handle failures, resume

These are production operations skills — running and operating generative media systems at scale.

What you'll build
Implement A/B testing for models (same prompt, different models, compare scores) and parameters (same model, different seeds/steps/cfg). Build batch processing with job queuing, progress tracking, failure handling, and resume.

A/B Testing Models

Same Prompt, Different Models

import asyncio
from dataclasses import dataclass, field

@dataclass
class ABTestResult:
    """Result of an A/B test."""
    model: str
    prompt: str
    output_url: str
    quality_score: float
    generation_time: float
    cost: float

async def ab_test_models(prompt: str, models: list[str]) -> list[ABTestResult]:
    """A/B test: same prompt, different models, compare quality."""
    # Generate with all models in parallel
    tasks = [generate_with_model(model, prompt) for model in models]
    results = await asyncio.gather(*tasks)
    
    # Sort by quality score
    results.sort(key=lambda r: r.quality_score, reverse=True)
    
    return results

async def generate_with_model(model: str, prompt: str) -> ABTestResult:
    """Mock: generate with a specific model."""
    import random, time
    t0 = time.time()
    await asyncio.sleep(random.uniform(1, 3))  # Simulate generation
    
    return ABTestResult(
        model=model,
        prompt=prompt,
        output_url=f"https://cdn.example.com/{model}_{hash(prompt)}.png",
        quality_score=random.uniform(0.6, 1.0),
        generation_time=time.time() - t0,
        cost={"flux": 0.04, "sdxl": 0.02, "dalle": 0.04}.get(model, 0.03),
    )

A/B Test Report

def generate_ab_report(results: list[ABTestResult]) -> str:
    """Generate an A/B test comparison report."""
    lines = ["A/B Test Results", "=" * 50, f"Prompt: {results[0].prompt}", ""]
    
    for i, r in enumerate(results):
        lines.append(f"  {i+1}. {r.model}")
        lines.append(f"     Quality: {r.quality_score:.2f}")
        lines.append(f"     Time: {r.generation_time:.1f}s")
        lines.append(f"     Cost: ${r.cost:.2f}")
        lines.append(f"     URL: {r.output_url}")
        lines.append("")
    
    best = results[0]
    lines.append(f"  Winner: {best.model} (score={best.quality_score:.2f})")
    return "\n".join(lines)

What is the correct methodology for A/B testing two image generation models (FLUX vs SDXL)?

Ask users which one they prefer without scoring

Unlock the full lesson

You've read the first 2 sections. The rest of this lesson covers A/B Testing Parameters, Batch Processing, Key Takeaways, What's Next — plus a hands-on lab, quiz, and project artifact.

Create a free account to unlock Phase 0 and Phase 1 of every course — no credit card.