RISEBench_Leaderboard / server /validation.py
renjianzhexuejia's picture
Update leaderboard, detailed results, figures and submission interface
9ad07ca verified
Raw History Blame
8.37 kB
"""Accept named score fields only; never execute or extract uploaded files."""
import copy
import math
import re
from urllib.parse import urlsplit
class Invalid(ValueError):
pass
def text(value, label, maximum=200, required=True):
if not isinstance(value, str) or len(value.strip()) > maximum:
raise Invalid(f"{label} must be text, at most {maximum} characters.")
value = value.strip()
if (required and not value) or any(ord(c) < 32 and c not in '\n\t' for c in value):
raise Invalid(f"Please provide a valid {label.lower()}.")
return value
def url(value, label):
value = text(value, label, 1000)
try:
parsed = urlsplit(value)
if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
raise ValueError()
except ValueError:
raise Invalid(f"{label} must be an https:// link without credentials.")
return value
def score(value, label):
if value is None:
return None
if type(value) not in (int, float) or not math.isfinite(value) or not 0 <= value <= 100:
raise Invalid(f"{label} must be a number between 0 and 100, or null.")
if abs(value * 10 - round(value * 10)) > 1e-7:
raise Invalid(f"{label} must use at most one decimal place, as in the published tables.")
return float(value)
def simple_scores(raw, columns, label):
if not isinstance(raw, dict):
raise Invalid(f"{label} must be an object with named scores.")
lookup = {c["label"]: c["key"] for c in columns}
lookup.update({c["key"]: c["key"] for c in columns})
result = {c["key"]: None for c in columns}
seen = set()
for key, value in raw.items():
if key not in lookup or lookup[key] in seen:
raise Invalid(f"Unknown or duplicate {label} field: {key}")
seen.add(lookup[key])
result[lookup[key]] = score(value, key)
return result
def nested_scores(raw, columns, label):
if not isinstance(raw, dict):
raise Invalid(f"{label} must group full score names by reasoning dimension.")
result = {c["key"]: None for c in columns}
seen = set()
for dimension, values in raw.items():
subset = [c for c in columns if c["dimension"].casefold() == dimension.casefold()]
if not subset or dimension.casefold() in seen:
raise Invalid(f"Unknown or duplicate {label} dimension: {dimension}")
seen.add(dimension.casefold())
result.update(simple_scores(values, subset, label))
return result
def validate_submission(raw, data):
if not isinstance(raw, dict):
raise Invalid("The submission must be a JSON object.")
allowed = {"model_name", "project_url", "evidence_url", "team", "email", "group", "benchmark",
"language", "multi_image", "revision_of", "notes", "full_dataset", "results"}
if set(raw) - allowed:
raise Invalid("Unknown submission fields: " + ", ".join(sorted(set(raw) - allowed)))
benchmark = next((b for b in data["benchmarks"] if b["id"] == raw.get("benchmark")), None)
if not benchmark or raw.get("language") not in benchmark["results_by_language"]:
raise Invalid("Select a supported benchmark and instruction language.")
if raw.get("full_dataset") is not True:
raise Invalid("Confirm that results use the full benchmark, not the demonstration data.")
if type(raw.get("multi_image")) is not bool:
raise Invalid("Select whether the model supports multiple input images.")
if benchmark["id"] == "risebench" and raw["multi_image"]:
raise Invalid("RISEBench evaluates single-image inputs only.")
if raw.get("group") not in {"closed", "open", "agentic"}:
raise Invalid("Select a model category.")
result = {key: raw[key] for key in ("benchmark", "language", "group", "multi_image")}
for key, label, size, required in [("model_name", "Model name", 160, True), ("team", "Team", 120, False),
("notes", "Notes", 3000, False), ("revision_of", "Existing model identifier", 160, False)]:
result[key] = text(raw.get(key, ""), label, size, required)
result["project_url"] = url(raw.get("project_url"), "Project link")
result["evidence_url"] = url(raw.get("evidence_url"), "Evaluation artifacts link")
result["email"] = text(raw.get("email"), "Contact email", 254)
if not re.fullmatch(r"[^\s@]+@[^\s@]+\.[^\s@]+", result["email"]):
raise Invalid("Provide a valid contact email.")
uploaded = raw.get("results")
if not isinstance(uploaded, dict) or set(uploaded) - {"accuracy", "metrics", "subcategories", "task_types"}:
raise Invalid("Upload a result JSON matching the downloadable template.")
result["accuracy"] = simple_scores(uploaded.get("accuracy"), benchmark["accuracy_columns"], "accuracy")
required_keys = [d["key"] for d in benchmark["dimensions"]]
if benchmark["id"] == "risebench-plusplus":
required_keys += ["single"] + (["multi", "total"] if result["multi_image"] else [])
if not result["multi_image"] and any(result["accuracy"][key] is not None for key in ("multi", "total")):
raise Invalid("Single-image-only models must leave Total and Multi null.")
else:
required_keys += ["total"]
if any(result["accuracy"][key] is None for key in required_keys):
raise Invalid("Fill in every reasoning-dimension accuracy and the applicable summary scores.")
result["metrics"] = simple_scores(uploaded.get("metrics", {}), benchmark["metrics"], "evaluation score")
for field, columns in (("subcategories", "subcategory_columns"), ("task_types", "task_type_columns")):
if columns not in benchmark:
if uploaded.get(field):
raise Invalid("Fine-grained submissions are supported only for RISEBench++.")
continue
result[field] = nested_scores(uploaded.get(field, {}), benchmark[columns], field)
if benchmark["id"] == "risebench-plusplus" and not result["multi_image"]:
for column in benchmark["task_type_columns"]:
if column["case_count"] == column["multi_image_case_count"] and result["task_types"][column["key"]] is not None:
raise Invalid(f"{column['label']} is a multi-image-only task; leave it null for this model.")
result["full_dataset"] = True
return result
def template(benchmark):
result = {"accuracy": {c["label"]: None for c in benchmark["accuracy_columns"]},
"metrics": {c["label"]: None for c in benchmark["metrics"]}}
for field, key in (("subcategories", "subcategory_columns"), ("task_types", "task_type_columns")):
if key in benchmark:
result[field] = {}
for c in benchmark[key]:
result[field].setdefault(c["dimension"].title(), {})[c["label"]] = None
return result
def public_leaderboard(base, state):
data = copy.deepcopy(base)
for b in data["benchmarks"]:
for language, entry in b["results_by_language"].items():
models = {m["id"]: m for m in entry["models"]}
approved = sorted((r for r in state["submissions"].values() if r["status"] == "approved"
and r["payload"]["benchmark"] == b["id"] and r["payload"]["language"] == language),
key=lambda r: r["reviewed_at"])
for record in approved:
p = record["payload"]
model_id = p["revision_of"] or "community-" + record["id"]
previous = models.get(model_id)
model = {"id": model_id, "name": p["model_name"], "group": p["group"], "multi_image": p["multi_image"],
"source_order": previous["source_order"] if previous else len(models) + 1}
# An explicit public allowlist: no email, notes, receipt tokens or review history.
for field in ("accuracy", "metrics", "subcategories", "task_types"):
if field in p:
model[field] = copy.deepcopy(p[field])
models[model_id] = model
entry["models"] = list(models.values())
entry["model_count"] = len(models)
for group in entry["groups"]:
group["count"] = sum(m["group"] == group["id"] for m in models.values())
return data