AI画像APIのベンチマーク方法：忠実度、レイテンシ、誤拒否

import numpy as np
from PIL import Image

def drift(src_path: str, out_path: str, mask_path: str) -> float:
    """Mean absolute difference (0-255) outside the mask; mask white = allowed to change."""
    src = Image.open(src_path).convert("RGB")
    out = np.asarray(Image.open(out_path).convert("RGB").resize(src.size), float)
    keep = np.asarray(Image.open(mask_path).convert("L").resize(src.size)) < 128
    return float(np.abs(np.asarray(src, float) - out)[keep].mean())

cost_per_usable_result = total_spend / usable_images

"""Sequential benchmark runner for SpicyAPI image tasks (stdlib + requests).

export SPICY_API_KEY=...   # your own key
python bench.py run results.csv MODEL_ID MED-01 5 '{"prompt": "...", "aspect_ratio": "4:3"}'
# fill the "review" column with usable / softened / off_brief for each returned image
python bench.py summary results.csv
"""
import csv, json, math, os, sys, time, uuid
import requests

BASE = "https://api.spicyapi.ai/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['SPICY_API_KEY']}"}
FIELDS = ["case_id", "model", "run", "request_key", "task_id", "outcome", "error_code",
          "seconds_to_file", "cost", "settled", "file", "review"]


def run_once(model: str, case_id: str, run: int, task_input: dict) -> dict:
    key = str(uuid.uuid4())
    t0 = time.monotonic()
    row = dict(case_id=case_id, model=model, run=run, request_key=key, task_id="",
               outcome="", error_code="", seconds_to_file="", cost="", settled="",
               file="", review="")
    os.makedirs("bench-journal", exist_ok=True)
    journal = f"bench-journal/{key}.json"

    def save() -> None:
        # Save before submit and after receiving an ID, even if this process stops.
        with open(journal, "w") as f:
            json.dump({"row": row, "input": task_input}, f)

    save()
    try:
        res = requests.post(f"{BASE}/jobs/createTask?wait=60",
                            headers={**HEADERS, "Idempotency-Key": key},
                            json={"model": model, "input": task_input}, timeout=90)
        body = res.json()
    except (requests.RequestException, ValueError) as exc:
        row.update(outcome="submission_unknown", error_code=type(exc).__name__)
        save()
        return row  # reconcile this request before creating another paid run
    if res.status_code not in (200, 202) or body.get("code") != 200:
        row.update(outcome="create_error", error_code=str(body.get("code", res.status_code)))
        save()
        return row
    data = body.get("data") or {}
    if not data.get("taskId"):
        row.update(outcome="submission_unknown", error_code="missing_task_id")
        save()
        return row
    row["task_id"] = data["taskId"]
    save()
    delay = 2
    while data.get("state") in ("queued", "running"):
        if time.monotonic() - t0 > 300:
            row["outcome"] = "client_timeout"
            save()
            return row  # task may still finish and incur a charge
        time.sleep(delay)
        delay = min(delay * 1.5, 10)
        try:
            poll = requests.get(f"{BASE}/jobs/recordInfo", params={"taskId": row["task_id"]},
                                headers=HEADERS, timeout=30)
            poll.raise_for_status()
            candidate = poll.json().get("data") or {}
            if candidate.get("taskId") != row["task_id"] or not candidate.get("state"):
                continue
            data = candidate
        except (requests.RequestException, ValueError):
            continue  # poll the saved task until the local deadline; never resubmit
    row.update(cost=data.get("cost", ""), settled=data.get("settled", ""))
    if data.get("state") == "succeeded":
        assets = (data.get("output") or {}).get("assets") or []
        if not assets or any(a.get("pending") or a.get("unavailable") or not a.get("url") for a in assets):
            row.update(outcome="delivery_pending", error_code="no_ready_asset")
        else:
            asset = assets[0]
            try:
                img = requests.get(asset["url"], timeout=60)
                img.raise_for_status()
                mime = img.headers.get("Content-Type", "").split(";")[0].lower()
                if not mime.startswith("image/") or not img.content:
                    raise ValueError("response_is_not_an_image")
                os.makedirs("out", exist_ok=True)
                path = f"out/{key}.{mime.split('/')[-1]}"
                with open(path, "wb") as out:
                    out.write(img.content)
                row.update(file=path, seconds_to_file=round(time.monotonic() - t0, 1),
                           outcome="returned")  # human review still decides usable/off-brief
            except (requests.RequestException, ValueError) as exc:
                row.update(outcome="download_error", error_code=type(exc).__name__)
    else:
        row["outcome"] = "refused" if data.get("errorCode") == "content_rejected" else "error"
        row["error_code"] = data.get("errorCode", data.get("state", "missing_state"))
    save()
    return row


def wilson(k: int, n: int, z: float = 1.96) -> tuple[float, float]:
    if n == 0:
        return 0.0, 1.0
    p, d = k / n, 1 + z * z / n
    centre = (p + z * z / (2 * n)) / d
    half = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / d
    return max(0.0, centre - half), min(1.0, centre + half)


def pct(values: list[float], q: float) -> float:
    s = sorted(values)
    return s[min(len(s) - 1, math.ceil(q * len(s)) - 1)]  # nearest rank


def summarise(path: str) -> None:
    with open(path, newline="") as f:
        rows = list(csv.DictReader(f))
    for model in sorted({r["model"] for r in rows}):
        mine = [r for r in rows if r["model"] == model]
        scored = [r for r in mine if r["outcome"] in ("returned", "refused")]
        refused = sum(r["outcome"] == "refused" for r in scored)
        lo, hi = wilson(refused, len(scored))
        times = [float(r["seconds_to_file"]) for r in mine if r["seconds_to_file"]]
        usable = sum(r["review"] == "usable" for r in mine)
        known_costs = [float(r["cost"]) for r in mine if r["cost"] != ""]
        spend = sum(known_costs)
        unresolved = sum(r["cost"] == "" or str(r["settled"]).lower() != "true" for r in mine)
        print(f"{model}: {len(mine)} runs, refusal rate {refused}/{len(scored)} "
              f"(95% interval {lo:.0%}-{hi:.0%}), other errors "
              f"{sum(r['outcome'] not in ('returned', 'refused') for r in mine)}")
        if times:
            print(f"  p50 {pct(times, .5):.0f}s  p90 {pct(times, .9):.0f}s  known spend ${spend:.3f}")
        if unresolved:
            print(f"  {unresolved} unsettled/unknown costs: reconcile journal/task IDs before final cost comparisons")
        if usable and not unresolved:
            print(f"  cost per usable result ${spend / usable:.3f} ({usable} usable)")


if __name__ == "__main__":
    if sys.argv[1] == "run":  # bench.py run results.csv MODEL CASE_ID RUNS 'JSON input'
        path, model, case_id, runs, task_input = sys.argv[2], sys.argv[3], sys.argv[4], int(sys.argv[5]), sys.argv[6]
        import json
        new_file = not os.path.exists(path)
        with open(path, "a", newline="") as f:
            w = csv.DictWriter(f, FIELDS)
            if new_file:
                w.writeheader()
            for i in range(1, runs + 1):
                row = run_once(model, case_id, i, json.loads(task_input))
                w.writerow(row)
                f.flush()
                # Stop on an unknown submission/settlement rather than compound the gap.
                # Inspect the last CSV row and bench-journal before starting another run.
                if row["outcome"] in ("submission_unknown", "client_timeout", "delivery_pending"):
                    break
    else:  # bench.py summary results.csv
        summarise(sys.argv[2])

# CLI
npx @spicyapi/cli tasks create --model alibaba/z-image-turbo/text-to-image \
  --input-json '{"prompt": "Textbook cross-section of a shoulder joint injury, neutral white background", "aspect_ratio": "4:3"}' --wait

# MCP server: the command your assistant's MCP config runs
npx --yes --package=@spicyapi/mcp spicyapi-mcp

# Agent Skill
npx skills add spicyapi-ai/spicy-skill