Spaces:
Running on Zero
Running on Zero
| from __future__ import annotations | |
| import io | |
| import hashlib | |
| import json | |
| import logging | |
| import time | |
| from collections import deque | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from threading import RLock, BoundedSemaphore | |
| from typing import Any | |
| # Import Spaces before Gradio so ZeroGPU can initialize its integration first. | |
| import spaces | |
| import gradio as gr | |
| from fastapi import HTTPException, Request | |
| from fastapi.responses import HTMLResponse, Response, JSONResponse | |
| from starlette.concurrency import run_in_threadpool | |
| from starlette.datastructures import UploadFile | |
| from src.leaderboard.build_rankings import build_rankings, frame_for_download | |
| from src.leaderboard.schema import ACTIVE_BENCHMARK_VERSION, METRIC_OPTIONS, LABELS | |
| logging.basicConfig(level=logging.INFO) | |
| LOGGER = logging.getLogger("ark-leaderboard") | |
| ROOT = Path(__file__).parent | |
| STATIC_DIR = ROOT / "static" | |
| SCORES_DIR = ROOT / "scores" | |
| DATA_LOCK = RLock() | |
| DATA: dict[str, Any] = {} | |
| VALID_METRICS = {(family, metric) for family, metric, _ in METRIC_OPTIONS} | |
| EVALUATION_SERVICE = None | |
| EVALUATION_LOCK = RLock() | |
| SUBMISSION_SLOT = BoundedSemaphore(1) | |
| SUBMISSION_ATTEMPTS: deque[float] = deque() | |
| # Only explicitly decorated functions may request GPU time. Ordinary browsing | |
| # must remain CPU-only even if the hosting environment enables auto-wrapping. | |
| spaces.disable_gradio_auto_wrap() | |
| def _empty_payload(message: str | None = None) -> dict[str, Any]: | |
| return { | |
| "views": {"overview": [], "knowledge_domain": [], "cognitive_demand": []}, | |
| "metric_options": [{"family": f, "metric": m, "label": l} for f, m, l in METRIC_OPTIONS], | |
| "active_benchmark_version": ACTIVE_BENCHMARK_VERSION, | |
| "updated_at": datetime.now(timezone.utc).isoformat(), | |
| "copy": { | |
| "title": "ARK-Bench Leaderboard", | |
| "description": "Knowledge- and reasoning-intensive multimodal retrieval benchmark for embedding models.", | |
| "about": ( | |
| "## About ARK-Bench\n\n" | |
| "ARK-Bench evaluates multimodal embedding models on 17 knowledge subtypes and six reasoning skills. " | |
| "Knowledge macro is the equal-weight average over the 17 subtypes; reasoning macro is the " | |
| "equal-weight average over KR, SR, FGVR, LR, SyR, and CA.\n\n" | |
| "Overview Avg is the leaderboard-defined `(Knowledge Macro + Reasoning Macro) / 2`. Cognitive Demand " | |
| "defaults to the hidden equal-weight mean of PO, KI, RI, and KRI.\n\n" | |
| "Current benchmark version: 1.0. Read the [paper](https://arxiv.org/pdf/2602.09839) for the protocol." | |
| ), | |
| "submission": ( | |
| "## Submit retrieval predictions\n\n" | |
| "1. Download the template and rank 20 candidates from each query's topic Gallery.\n" | |
| "2. Select Evaluate to calculate and inspect aggregate scores against hidden labels.\n" | |
| "3. Submit the report to notify maintainers through a private repository Issue.\n" | |
| "4. A maintainer reviews the report before publication on the leaderboard." | |
| ), | |
| }, | |
| "error": message, | |
| } | |
| def _build_payload(family: str, metric: str) -> dict[str, Any]: | |
| ranking = build_rankings(SCORES_DIR, family, metric, ACTIVE_BENCHMARK_VERSION) | |
| return { | |
| "views": {key: ranking[key] for key in ("overview", "knowledge_domain", "reasoning_skill", "cognitive_demand")}, | |
| "errors": ranking["errors"], | |
| "labels": LABELS, | |
| "score_date": max((row["date"] for row in ranking["overview"]), default=None), | |
| } | |
| def refresh_data() -> dict[str, Any]: | |
| """Load reviewed score files and build the default leaderboard views.""" | |
| payload = _empty_payload() | |
| try: | |
| SCORES_DIR.mkdir(parents=True, exist_ok=True) | |
| result = _build_payload("Recall", "R@1") | |
| payload.update(result) | |
| if result["errors"]: | |
| payload["error"] = "; ".join(result["errors"]) | |
| except Exception as exc: | |
| LOGGER.exception("Could not build leaderboard data") | |
| payload["error"] = f"Could not build leaderboard data: {exc}" | |
| with DATA_LOCK: | |
| DATA.clear() | |
| DATA.update(payload) | |
| return dict(DATA) | |
| server = gr.Server(title="ARK-Bench Leaderboard", description="ARK-Bench embedding model leaderboard.", version="1.0.0") | |
| def evaluation_service(): | |
| global EVALUATION_SERVICE | |
| with EVALUATION_LOCK: | |
| if EVALUATION_SERVICE is None: | |
| from src.evaluation.service import EvaluationService | |
| EVALUATION_SERVICE = EvaluationService(ROOT) | |
| return EVALUATION_SERVICE | |
| def evaluation_protocol(): | |
| return JSONResponse(evaluation_service().protocol(), headers={"Cache-Control": "no-store"}) | |
| def evaluation_template(): | |
| return Response(evaluation_service().template(), media_type="application/x-ndjson", headers={ | |
| "Content-Disposition": 'attachment; filename="ark-predictions-template.jsonl"', | |
| "Cache-Control": "no-store", "X-Content-Type-Options": "nosniff", | |
| }) | |
| async def _read_evaluation_form(request: Request, *, include_receipt: bool = False): | |
| from src.evaluation.service import MAX_UPLOAD_BYTES | |
| if not request.headers.get("content-type", "").lower().startswith("multipart/form-data;"): | |
| raise HTTPException(415, "Upload predictions using a multipart form.") | |
| body = bytearray() | |
| async for chunk in request.stream(): | |
| body.extend(chunk) | |
| if len(body) > MAX_UPLOAD_BYTES + 65536: | |
| raise HTTPException(413, "Upload exceeds the 2 MiB file limit.") | |
| request._body = bytes(body) | |
| async with request.form(max_files=1, max_fields=6 if include_receipt else 5, | |
| max_part_size=16384) as form: | |
| fields = {} | |
| upload = None | |
| allowed = {"model_name", "model_url", "params_b", "benchmark_version", "dataset_revision"} | |
| if include_receipt: | |
| allowed.add("evaluation_id") | |
| seen = set() | |
| for key, value in form.multi_items(): | |
| if key in seen: | |
| raise HTTPException(422, "Duplicate form fields are not allowed.") | |
| seen.add(key) | |
| if key == "file" and isinstance(value, UploadFile): | |
| upload = value | |
| elif key in allowed and isinstance(value, str): | |
| fields[key] = value | |
| else: | |
| raise HTTPException(422, "Unexpected submission field.") | |
| if upload is None: | |
| raise HTTPException(422, "Select a predictions JSONL file.") | |
| content = await upload.read(MAX_UPLOAD_BYTES + 1) | |
| if len(content) > MAX_UPLOAD_BYTES: | |
| raise HTTPException(413, "Upload exceeds the 2 MiB file limit.") | |
| return content, fields | |
| async def evaluate_predictions(request: Request): | |
| from src.evaluation.scoring import SubmissionError | |
| from src.evaluation.service import EvaluationUnavailable | |
| if not SUBMISSION_SLOT.acquire(blocking=False): | |
| raise HTTPException(429, "Another submission is being evaluated. Please retry shortly.", headers={"Retry-After": "10"}) | |
| try: | |
| # A global ceiling bounds anonymous scoring probes and private repository writes. | |
| # This is an in-process guard; use authenticated quotas for larger competitions. | |
| now = time.monotonic() | |
| while SUBMISSION_ATTEMPTS and SUBMISSION_ATTEMPTS[0] <= now - 3600: | |
| SUBMISSION_ATTEMPTS.popleft() | |
| if len(SUBMISSION_ATTEMPTS) >= 30: | |
| raise HTTPException(429, "The evaluation service has reached its hourly submission limit. Please try again later.", | |
| headers={"Retry-After": str(max(1, int(3600 - now + SUBMISSION_ATTEMPTS[0])))}) | |
| content, fields = await _read_evaluation_form(request) | |
| SUBMISSION_ATTEMPTS.append(now) | |
| result = await run_in_threadpool(evaluation_service().evaluate, content, fields) | |
| return JSONResponse(result, headers={"Cache-Control": "no-store"}) | |
| except SubmissionError as exc: | |
| raise HTTPException(422, str(exc)) from None | |
| except EvaluationUnavailable as exc: | |
| raise HTTPException(503, str(exc)) from None | |
| except HTTPException: | |
| raise | |
| except Exception: | |
| LOGGER.error("Evaluation request failed") | |
| raise HTTPException(503, "Evaluation is temporarily unavailable. Please retry later.") from None | |
| finally: | |
| SUBMISSION_SLOT.release() | |
| async def submit_evaluation_report(request: Request): | |
| from src.evaluation.scoring import SubmissionError | |
| from src.evaluation.service import EvaluationUnavailable | |
| if not SUBMISSION_SLOT.acquire(blocking=False): | |
| raise HTTPException(429, "Another evaluation request is being processed. Please retry shortly.", | |
| headers={"Retry-After": "10"}) | |
| try: | |
| content, fields = await _read_evaluation_form(request, include_receipt=True) | |
| result = await run_in_threadpool(evaluation_service().submit_report, content, fields) | |
| return JSONResponse(result, headers={"Cache-Control": "no-store"}) | |
| except SubmissionError as exc: | |
| raise HTTPException(422, str(exc)) from None | |
| except EvaluationUnavailable as exc: | |
| raise HTTPException(503, str(exc)) from None | |
| except Exception: | |
| LOGGER.error("Evaluation report submission failed") | |
| raise HTTPException(503, "Could not submit this report for review. Please retry later.") from None | |
| finally: | |
| SUBMISSION_SLOT.release() | |
| def zerogpu_check() -> dict[str, Any]: | |
| """Opt-in GPU diagnostic; never called by startup or leaderboard browsing. | |
| Registration must happen at module import for ZeroGPU startup discovery. | |
| Calling this endpoint on ZeroGPU requests GPU time and may consume quota. | |
| """ | |
| import torch | |
| if not torch.cuda.is_available(): | |
| return {"status": "unavailable", "message": "No CUDA GPU on this local host; leaderboard remains CPU-only."} | |
| checksum = torch.tensor([1.0, 2.0, 3.0], device="cuda").sum().item() | |
| return {"status": "ok", "device": torch.cuda.get_device_name(0), "checksum": checksum} | |
| async def homepage() -> HTMLResponse: | |
| html = (STATIC_DIR / "index.html").read_text(encoding="utf-8") | |
| for filename in ("styles.css", "app.js"): | |
| digest = hashlib.sha256((STATIC_DIR / filename).read_bytes()).hexdigest()[:20] | |
| html = html.replace(f"/static/{filename}", f"/ark-assets/{digest}/{filename}") | |
| return HTMLResponse(html, headers={"Cache-Control": "no-cache, max-age=0, must-revalidate"}) | |
| async def frontend_asset(version: str, filename: str) -> Response: | |
| # Separate namespace from Gradio; content-addressed URLs cannot reuse old CSS/JS. | |
| media_types = {"styles.css": "text/css", "app.js": "application/javascript"} | |
| if filename not in media_types: | |
| raise HTTPException(status_code=404) | |
| content = (STATIC_DIR / filename).read_bytes() | |
| digest = hashlib.sha256(content).hexdigest()[:20] | |
| if version != digest: | |
| raise HTTPException(status_code=404) | |
| return Response(content, media_type=media_types[filename], headers={ | |
| "Cache-Control": "public, max-age=31536000, immutable", | |
| "X-Content-Type-Options": "nosniff", | |
| }) | |
| def leaderboard_data(family: str = "Recall", metric: str = "R@1") -> dict[str, Any]: | |
| """Return all four leaderboard views for one selected metric.""" | |
| if (family, metric) not in VALID_METRICS: | |
| return _empty_payload(f"Unsupported metric: {family}/{metric}") | |
| try: | |
| result = _build_payload(family, metric) | |
| with DATA_LOCK: | |
| payload = dict(DATA) if DATA else _empty_payload() | |
| payload.update(result) | |
| payload["selected_metric"] = {"family": family, "metric": metric} | |
| payload["updated_at"] = datetime.now(timezone.utc).isoformat() | |
| return payload | |
| except Exception as exc: | |
| LOGGER.exception("Could not load leaderboard data") | |
| return _empty_payload(f"Could not load leaderboard data: {exc}") | |
| def download_ranking(view: str = "overview", family: str = "Recall", metric: str = "R@1") -> dict[str, str]: | |
| """Return the selected ranking as CSV and JSON strings.""" | |
| payload = leaderboard_data(family, metric) | |
| rows = payload.get("views", {}).get(view, []) | |
| frame = frame_for_download(rows) | |
| csv_buffer = io.StringIO() | |
| frame.to_csv(csv_buffer, index=False) | |
| download_rows = [{key: value for key, value in row.items() if key != "cognitive_avg"} for row in rows] | |
| return {"csv": csv_buffer.getvalue(), "json": json.dumps(download_rows, indent=2)} | |
| if __name__ == "__main__": | |
| refresh_data() | |
| server.launch(blocked_paths=[str(ROOT / name) for name in | |
| ("evaluation", "evaluation-cache", "submissions")], max_file_size="2mb") | |