| import json |
| import math |
| import os |
| import statistics |
| import time |
| from pathlib import Path |
|
|
| from huggingface_hub import InferenceClient |
| from PIL import ImageFilter, ImageStat |
|
|
| HF_TOKEN = os.environ.get("HF_TOKEN") |
| if not HF_TOKEN: |
| raise RuntimeError("HF_TOKEN is required for the real inference benchmark") |
|
|
| OUT = Path("benchmark-results") |
| OUT.mkdir(parents=True, exist_ok=True) |
|
|
| BASELINE_MODEL = "black-forest-labs/FLUX.1-schnell" |
| CATEGORIES = { |
| "photo": { |
| "prompt": ( |
| "professional editorial portrait photo of a Thai man in Bangkok, natural skin texture, " |
| "85mm lens, soft window light, realistic fabric, shallow depth of field, high detail" |
| ), |
| "candidates": ["black-forest-labs/FLUX.1-Krea-dev", BASELINE_MODEL], |
| }, |
| "anime": { |
| "prompt": ( |
| "animefal, anime swordswoman standing in rainy neon Bangkok at night, clean line art, expressive eyes, " |
| "polished cel shading, coherent anatomy, detailed city background" |
| ), |
| "candidates": ["falanaja/animefal", BASELINE_MODEL], |
| }, |
| "design": { |
| "prompt": ( |
| "minimal geometric rabbit head logo for a modern streetwear brand, black and white, centered, " |
| "clean vector-like silhouette, crisp edges, no mockup, no extra objects" |
| ), |
| "candidates": ["Qwen/Qwen-Image", BASELINE_MODEL], |
| }, |
| } |
|
|
| client = InferenceClient(provider="auto", api_key=HF_TOKEN) |
|
|
|
|
| def visual_signal_score(image): |
| """Non-semantic sanity proxy in [0, 100] based on contrast and edge energy.""" |
| gray = image.convert("L").resize((256, 256)) |
| contrast = float(ImageStat.Stat(gray).stddev[0]) |
| edges = gray.filter(ImageFilter.FIND_EDGES) |
| edge_mean = float(ImageStat.Stat(edges).mean[0]) |
| contrast_component = min(1.0, contrast / 64.0) |
| edge_component = min(1.0, edge_mean / 48.0) |
| return round(100.0 * (0.55 * contrast_component + 0.45 * edge_component), 2) |
|
|
|
|
| def safe_name(model): |
| return model.replace("/", "__").replace(":", "_") |
|
|
|
|
| def is_credit_error(exc): |
| text = str(exc).lower() |
| return "402" in text or "payment required" in text or ("depleted" in text and "credits" in text) |
|
|
|
|
| def generate(category, model, prompt): |
| started = time.perf_counter() |
| try: |
| image = client.text_to_image( |
| prompt=prompt, |
| model=model, |
| width=768, |
| height=768, |
| seed=20260830, |
| ) |
| latency = round(time.perf_counter() - started, 3) |
| path = OUT / f"{category}__{safe_name(model)}.png" |
| image.save(path) |
| return { |
| "category": category, |
| "model": model, |
| "success": True, |
| "blocked_by_credits": False, |
| "latency_seconds": latency, |
| "width": image.width, |
| "height": image.height, |
| "visual_signal_score": visual_signal_score(image), |
| "image": str(path), |
| "error": None, |
| } |
| except Exception as exc: |
| latency = round(time.perf_counter() - started, 3) |
| return { |
| "category": category, |
| "model": model, |
| "success": False, |
| "blocked_by_credits": is_credit_error(exc), |
| "latency_seconds": latency, |
| "width": None, |
| "height": None, |
| "visual_signal_score": 0.0, |
| "image": None, |
| "error": f"{type(exc).__name__}: {exc}", |
| } |
|
|
|
|
| def choose_model(results): |
| successful = [item for item in results if item["success"]] |
| if not successful: |
| return None, "no candidate succeeded in this run" |
|
|
| baseline = next((x for x in successful if x["model"] == BASELINE_MODEL), None) |
| specialist = next((x for x in successful if x["model"] != BASELINE_MODEL), None) |
|
|
| if specialist and baseline: |
| signal_ok = specialist["visual_signal_score"] >= baseline["visual_signal_score"] * 0.70 |
| latency_ok = specialist["latency_seconds"] <= max(60.0, baseline["latency_seconds"] * 4.0) |
| if signal_ok and latency_ok: |
| return specialist["model"], "specialist healthy versus FLUX.1-schnell baseline" |
| return baseline["model"], "specialist failed health/latency guard; keep baseline" |
|
|
| chosen = specialist or baseline or min(successful, key=lambda x: x["latency_seconds"]) |
| return chosen["model"], "only one healthy candidate available" |
|
|
|
|
| def main(): |
| all_results = [] |
| recommendations = {} |
| credits_blocked = False |
|
|
| for category, config in CATEGORIES.items(): |
| if credits_blocked: |
| recommendations[category] = { |
| "model": None, |
| "reason": "not executed because Inference Provider credits were depleted", |
| } |
| continue |
|
|
| category_results = [] |
| for model in config["candidates"]: |
| print(f"BENCHMARK_START category={category} model={model}", flush=True) |
| result = generate(category, model, config["prompt"]) |
| category_results.append(result) |
| all_results.append(result) |
| print( |
| "BENCHMARK_RESULT " |
| + json.dumps( |
| { |
| "category": category, |
| "model": model, |
| "success": result["success"], |
| "blocked_by_credits": result["blocked_by_credits"], |
| "latency_seconds": result["latency_seconds"], |
| "visual_signal_score": result["visual_signal_score"], |
| "error": result["error"], |
| }, |
| ensure_ascii=False, |
| ), |
| flush=True, |
| ) |
| if result["blocked_by_credits"]: |
| credits_blocked = True |
| break |
|
|
| model, reason = choose_model(category_results) |
| if credits_blocked and not model: |
| reason = "benchmark blocked by depleted Inference Provider credits" |
| recommendations[category] = {"model": model, "reason": reason} |
| print(f"RECOMMEND category={category} model={model} reason={reason}", flush=True) |
|
|
| success_latencies = [r["latency_seconds"] for r in all_results if r["success"]] |
| summary = { |
| "baseline_model": BASELINE_MODEL, |
| "resolution": "768x768", |
| "seed": 20260830, |
| "benchmark_status": "blocked_by_credits" if credits_blocked else "complete", |
| "quality_metric_note": ( |
| "visual_signal_score is a non-semantic sanity proxy for contrast/edge energy, not an aesthetic score" |
| ), |
| "results": all_results, |
| "recommendations": recommendations, |
| "successful_calls": sum(1 for r in all_results if r["success"]), |
| "total_calls_attempted": len(all_results), |
| "mean_latency_seconds": round(statistics.mean(success_latencies), 3) if success_latencies else None, |
| "p95_latency_seconds": ( |
| round(sorted(success_latencies)[max(0, math.ceil(len(success_latencies) * 0.95) - 1)], 3) |
| if success_latencies |
| else None |
| ), |
| } |
|
|
| (OUT / "benchmark.json").write_text( |
| json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8" |
| ) |
|
|
| lines = [ |
| "# Phase 1.5 Model Benchmark", |
| "", |
| f"Status: **{summary['benchmark_status']}**", |
| "", |
| "Real Hugging Face Inference Provider calls at 768x768 with a fixed seed.", |
| "", |
| "| Category | Model | Success | Credits blocked | Latency (s) | Visual signal |", |
| "|---|---|---:|---:|---:|---:|", |
| ] |
| for r in all_results: |
| lines.append( |
| f"| {r['category']} | `{r['model']}` | {r['success']} | {r['blocked_by_credits']} | " |
| f"{r['latency_seconds']} | {r['visual_signal_score']} |" |
| ) |
| lines.extend(["", "## Recommendations", ""]) |
| for category, rec in recommendations.items(): |
| lines.append(f"- **{category}**: `{rec['model']}`. {rec['reason']}") |
| lines.extend( |
| [ |
| "", |
| "> `visual_signal_score` is only a technical sanity proxy. Final visual quality should also be reviewed manually.", |
| ] |
| ) |
| (OUT / "benchmark.md").write_text("\n".join(lines) + "\n", encoding="utf-8") |
|
|
| print("BENCHMARK_SUMMARY " + json.dumps(summary, ensure_ascii=False), flush=True) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
