Evaluating Multimodal Vision Models with Moonshot PerceptionBench Using Robust Data Loading and Automated Judging
In this tutorial, we design an end-to-end analysis workflow for PerceptionBench. This multimodal benchmark measures fine-grained visible notion capabilities throughout duties resembling OCR, counting, localization, contextual reasoning, comparability, depth understanding, and hallucination detection. We start by configuring a Colab-compatible atmosphere, putting in the required libraries, and loading a balanced subset of the dataset by a sturdy multi-stage streaming and obtain technique. We then decode base64-encoded photographs, parse interleaved picture placeholders, normalize every instance right into a constant document format, and analyze the dataset’s functionality distribution, picture necessities, reply varieties, and supply benchmarks. From there, we assemble a unified analysis harness that helps a blind-prior baseline, OpenAI-compatible multimodal APIs, and native Hugging Face vision-language fashions. We additionally implement rule-based and non-compulsory LLM-assisted judging, calculate bootstrap confidence intervals, study efficiency throughout issue slices, evaluate functionality profiles with the included leaderboard, and export reproducible prediction and reporting artifacts.
import os, sys, io, re, json, time, math, base64, random, hashlib, subprocess, warnings
from collections import Counter, defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
warnings.filterwarnings("ignore")
CFG = dict(
REPO = "moonshotai/PerceptionBench",
SPLIT = "practice",
N_PER_CATEGORY = 12,
MAX_SCAN = 1200,
SEED = 0,
LOAD_MODE = "stream",
BACKEND = "blind",
API_BASE = os.environ.get("PB_API_BASE", "https://api.openai.com/v1"),
API_KEY = os.environ.get("PB_API_KEY", ""),
API_MODEL = os.environ.get("PB_API_MODEL", "gpt-4o-mini"),
API_WORKERS = 4,
API_MAX_TOKENS = 512,
LOCAL_MODEL = "HuggingFaceTB/SmolVLM2-2.2B-Instruct",
LOCAL_MAX_NEW = 128,
MAX_IMAGE_SIDE = 1024,
JPEG_QUALITY = 90,
JUDGE = "rule",
NUM_REL_TOL = 0.0,
OUT_DIR = "/content material/perceptionbench_out" if os.path.isdir("/content material") else "./perceptionbench_out",
INSTALL_DEPS = True,
SHOW_PLOTS = True,
)
random.seed(CFG["SEED"])
os.makedirs(CFG["OUT_DIR"], exist_ok=True)
def _sh(pkgs):
subprocess.run([sys.executable, "-m", "pip", "install", "-q", *pkgs],
test=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
if CFG["INSTALL_DEPS"]:
print("[setup] putting in dependencies (quiet, ~30s on a chilly Colab)…")
_sh(["datasets>=3.0.0", "huggingface_hub>=0.25.0", "pillow", "pandas",
"numpy", "matplotlib", "requests", "pyarrow"])
if CFG["BACKEND"] == "native":
_sh(["transformers>=4.51.0", "accelerate", "torch", "num2words"])
import numpy as np
import pandas as pd
import requests
import matplotlib
import matplotlib.pyplot as plt
from PIL import Image
matplotlib.rcParams.replace({"determine.dpi": 110, "font.dimension": 9, "axes.grid": True,
"grid.alpha": .25, "axes.spines.high": False,
"axes.spines.proper": False})
print("[setup] readyn")
We configure the PerceptionBench atmosphere, outline the dataset, backend, image-processing, judging, and output settings, and initialize reproducible random habits. We set up the required libraries for dataset loading, numerical evaluation, visualization, HTTP communication, and picture processing. We additionally configure Matplotlib and put together the output listing so the remaining analysis workflow runs persistently in Google Colab or an area atmosphere.
def _iter_rows(repo, cut up, mode, max_scan):
"""Yield dict rows, attempting progressively heavier methods."""
from datasets import load_dataset
if mode == "full":
print("[load] full obtain (~1.63 GB) …")
ds = load_dataset(repo, cut up=cut up)
for i, r in enumerate(ds):
if i >= max_scan:
return
yield r
return
attempt:
from huggingface_hub import HfApi, hf_hub_url
api = HfApi()
recordsdata = api.list_repo_files(repo, repo_type="dataset", revision="refs/convert/parquet")
pq = sorted(f for f in recordsdata if f.endswith(".parquet") and f"/{cut up}/" in f)
if pq:
urls = [hf_hub_url(repo, f, repo_type="dataset", revision="refs/convert/parquet") for f in pq]
print(f"[load] streaming {len(urls)} parquet shard(s) from refs/convert/parquet")
ds = load_dataset("parquet", data_files=urls, cut up="practice", streaming=True)
for i, r in enumerate(ds):
if i >= max_scan:
return
yield r
return
besides Exception as e:
print(f"[load] parquet stream unavailable ({sort(e).__name__}: {e}); falling again")
attempt:
print("[load] streaming unique information recordsdata")
ds = load_dataset(repo, cut up=cut up, streaming=True)
for i, r in enumerate(ds):
if i >= max_scan:
return
yield r
return
besides Exception as e:
print(f"[load] json stream failed ({sort(e).__name__}); doing a full obtain")
ds = load_dataset(repo, cut up=cut up)
for i, r in enumerate(ds):
if i >= max_scan:
return
yield r
def stratified_subset(repo, cut up, n_per_cat, max_scan, mode):
"""Balanced pattern throughout `error_category` — the ten atomic capabilities.
Balancing issues: the benchmark experiences a *functionality profile*, and an
unbalanced pattern makes the general quantity a weighted common of whichever
capabilities occurred to seem first within the shard.
"""
buckets, scanned, t0 = defaultdict(record), 0, time.time()
for row in _iter_rows(repo, cut up, mode, max_scan):
scanned += 1
cat = row.get("error_category") or "unknown"
if len(buckets[cat]) < n_per_cat:
buckets[cat].append(row)
if scanned % 100 == 0:
stuffed = sum(len(v) >= n_per_cat for v in buckets.values())
print(f" scanned={scanned:5d} classes={len(buckets):second} "
f"stuffed={stuffed:second} {time.time()-t0:5.1f}s", finish="r")
if scanned >= 250 and len(buckets) >= 10 and all(len(v) >= n_per_cat for v in buckets.values()):
break
rows = [r for v in buckets.values() for r in v]
random.Random(CFG["SEED"]).shuffle(rows)
print(f"n[load] scanned {scanned} rows -> saved {len(rows)} throughout "
f"{len(buckets)} capabilities ({time.time()-t0:.1f}s)")
return rows, scanned
ROWS, N_SCANNED = stratified_subset(
CFG["REPO"], CFG["SPLIT"], CFG["N_PER_CATEGORY"], CFG["MAX_SCAN"], CFG["LOAD_MODE"])
We implement a resilient dataset loader that first makes an attempt transformed Parquet streaming, then falls again to streaming the unique recordsdata, and lastly performs a full obtain when obligatory. We scan the dataset whereas limiting the variety of processed rows and manage examples into capability-specific buckets utilizing the error_category discipline. We then create a balanced, shuffled subset so every visible functionality contributes a comparable variety of analysis questions.
DATA_URI_RE = re.compile(r"^information:(picture/[A-Za-z0-9.+-]+);base64,(.*)$", re.S)
PLACEHOLDER_RE = re.compile(r"<|picture[ _](d+)|>")
def decode_image(entry):
"""data-URI string | uncooked b64 | bytes | HF Image dict -> PIL.Image (RGB)."""
if isinstance(entry, Image.Image):
return entry.convert("RGB")
if isinstance(entry, dict):
if entry.get("bytes"):
return Image.open(io.BytesIO(entry["bytes"])).convert("RGB")
if entry.get("path"):
return Image.open(entry["path"]).convert("RGB")
if isinstance(entry, (bytes, bytearray)):
return Image.open(io.BytesIO(entry)).convert("RGB")
s = str(entry).strip()
m = DATA_URI_RE.match(s)
b64 = m.group(2) if m else s
b64 = re.sub(r"s+", "", b64)
b64 += "=" * (-len(b64) % 4)
return Image.open(io.BytesIO(base64.b64decode(b64))).convert("RGB")
def load_images(row):
imgs = row.get("picture") or []
if isinstance(imgs, (str, bytes, dict)):
imgs = [imgs]
out = []
for e in imgs:
attempt:
out.append(decode_image(e))
besides Exception as err:
print(f" [warn] undecodable picture on idx={row.get('index')}: {err}")
return out
def shrink(img, max_side, high quality):
"""Downscale + re-encode. Returns (PIL, data_uri). Controls the token invoice:
a 3000px screenshot can price >2k imaginative and prescient tokens per picture, and these
questions carry as much as 8 photographs every."""
w, h = img.dimension
if max(w, h) > max_side:
s = max_side / max(w, h)
img = img.resize((max(1, int(w * s)), max(1, int(h * s))), Image.LANCZOS)
buf = io.BytesIO()
img.save(buf, format="JPEG", high quality=high quality)
uri = "information:picture/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()
return img, uri
def split_on_placeholders(downside, n_images):
"""`<|image_1|>textual content<|image_2|>?` -> [('image',0),('text','…'),('image',1)…]
Any picture by no means referenced by a placeholder is appended on the finish, so we
by no means silently drop visible proof."""
elements, final = [], 0
for m in PLACEHOLDER_RE.finditer(downside):
chunk = downside[last:m.start()].strip()
if chunk:
elements.append(("textual content", chunk))
i = int(m.group(1)) - 1
if 0 <= i < n_images:
elements.append(("picture", i))
final = m.finish()
tail = downside[last:].strip()
if tail:
elements.append(("textual content", tail))
used = {p[1] for p in elements if p[0] == "picture"}
for i in vary(n_images):
if i not in used:
elements.append(("picture", i))
return elements
def to_record(row):
imgs = load_images(row)
downside = (row.get("downside") or "").strip()
return dict(
index = row.get("index"),
downside = downside,
reply = str(row.get("reply", "")).strip(),
trace = (row.get("trace") or "").strip(),
class = row.get("error_category") or "unknown",
source_bmk = row.get("source_bmk") or "NA",
source_idx = row.get("source_idx"),
photographs = imgs,
n_images = len(imgs),
n_placeholders= len(PLACEHOLDER_RE.findall(downside)),
q_chars = len(downside),
px_total = sum(w * h for w, h in (im.dimension for im in imgs)),
max_side = max([max(im.size) for im in imgs], default=0),
)
print("[decode] decoding photographs…")
RECORDS = [to_record(r) for r in ROWS]
RECORDS = [r for r in RECORDS if r["images"] and r["answer"]]
print(f"[decode] {len(RECORDS)} usable recordsn")
def cat_code(cat):
c = (cat or "").decrease()
for key, code in [("hallucin", "Hallu"), ("ocr", "OCR"), ("context", "Ctx"),
("fine_grain", "FGR"), ("fine-grain", "FGR"),
("compar", "Comp"), ("local", "Loc"), ("position", "Loc"),
("depth", "Depth"), ("3d", "Depth"),
("attribut", "Attr"), ("count", "Count"),
("relation", "VRel")]:
if key in c:
return code
return cat[:6].title()
CODE_ORDER = ["VRel", "Count", "Attr", "Depth", "Loc", "Comp", "FGR", "Ctx", "OCR", "Hallu"]
CODE_FULL = {"VRel": "visible relation", "Count": "counting", "Attr": "attribute",
"Depth": "depth & 3D", "Loc": "localization", "Comp": "comparability",
"FGR": "fine-grained recog.", "Ctx": "contextual integration",
"OCR": "OCR", "Hallu": "notion hallucination"}
for r in RECORDS:
r["code"] = cat_code(r["category"])
df = pd.DataBody([{k: v for k, v in r.items() if k != "images"} for r in RECORDS])
def answer_type(a):
a = a.strip()
if re.fullmatch(r"-?d+", a): return "integer"
if re.fullmatch(r"-?d*.d+", a): return "decimal"
if re.fullmatch(r"(?i)(sure|no|true|false)", a): return "boolean"
if re.fullmatch(r"(?i)[A-H]", a): return "letter"
if len(a.cut up()) == 1: return "single-word"
return "phrase"
df["ans_type"] = df["answer"].map(answer_type)
print("=" * 78)
print("§4 DATASET PROFILE (stratified subset — the cardboard experiences 3,000 whole)")
print("=" * 78)
print("n-- atomic capabilities current --")
print(df.groupby("code").agg(n=("index", "dimension"),
mean_imgs=("n_images", "imply"),
mean_qlen=("q_chars", "imply")).spherical(2).to_string())
print("n-- reply floor types --")
print(df["ans_type"].value_counts().to_string())
print("n-- photographs per query --")
print(df["n_images"].value_counts().sort_index().to_string())
print(f"n-- multi-image questions: {(df.n_images > 1).imply():.1%} of the subset")
print(f"-- median longest picture edge: {df.max_side.median():.0f}px "
f"(max {df.max_side.max():.0f}px)")
print("n-- provenance (high supply benchmarks) --")
print(df["source_bmk"].value_counts().head(8).to_string())
print(f"n-- newly-authored (source_bmk == 'NA'): {(df.source_bmk=='NA').imply():.1%} "
f"(card: 40% authored / 60% decomposed)n")
def hf_full_stats(repo, cut up="practice", config="default"):
attempt:
r = requests.get("https://datasets-server.huggingface.co/statistics",
params={"dataset": repo, "config": config, "cut up": cut up},
timeout=30)
r.raise_for_status()
for col in r.json().get("statistics", []):
if col["column_name"] == "error_category":
freq = col["column_statistics"].get("frequencies", {})
if freq:
tot = sum(freq.values())
print("-- FULL-CORPUS functionality distribution (3,000 rows, through datasets-server) --")
for okay, v in sorted(freq.objects(), key=lambda x: -x[1]):
print(f" {cat_code(okay):6s} {okay:34s} {v:5d} {v/tot:6.1%}")
print()
return freq
besides Exception as e:
print(f"[stats] datasets-server unavailable ({sort(e).__name__}); "
f"utilizing subset statistics onlyn")
return None
FULL_FREQ = hf_full_stats(CFG["REPO"], CFG["SPLIT"])
if CFG["SHOW_PLOTS"]:
fig, ax = plt.subplots(1, 3, figsize=(13, 3.4))
order = [c for c in CODE_ORDER if c in set(df.code)] +
[c for c in sorted(set(df.code)) if c not in CODE_ORDER]
df.code.value_counts().reindex(order).plot.bar(ax=ax[0], coloration="#4C72B0")
ax[0].set_title("Questions per atomic functionality"); ax[0].set_xlabel("")
df.n_images.value_counts().sort_index().plot.bar(ax=ax[1], coloration="#DD8452")
ax[1].set_title("Images per query"); ax[1].set_xlabel("# photographs")
df.ans_type.value_counts().plot.barh(ax=ax[2], coloration="#55A868")
ax[2].set_title("Answer floor type")
plt.tight_layout(); plt.present()
We decode photographs from information URIs, uncooked base64 strings, byte arrays, PIL objects, and Hugging Face picture dictionaries right into a constant RGB format. We normalize each dataset row right into a structured document containing query textual content, solutions, photographs, functionality labels, dimensions, placeholder counts, and supply info. We then analyze functionality protection, reply codecs, picture counts, decision traits, and supply benchmarks whereas visualizing the ensuing dataset profile.
def show_record(rec, max_imgs=4):
imgs = rec["images"][:max_imgs]
n = len(imgs)
fig, axes = plt.subplots(1, n, figsize=(4.2 * n, 4.2))
axes = np.atleast_1d(axes)
for a, im in zip(axes, imgs):
a.imshow(im); a.axis("off")
q = re.sub(r"s+", " ", rec["problem"])
q = (q[:150] + "…") if len(q) > 150 else q
fig.suptitle(f"[{rec['code']} · {rec['category']}] {q}n"
f"gold = {rec['answer']!r} | src = {rec['source_bmk']}",
fontsize=9, y=1.06)
plt.tight_layout(); plt.present()
if CFG["SHOW_PLOTS"]:
print("=" * 78); print("§5 ONE EXEMPLAR PER CAPABILITY"); print("=" * 78)
seen = set()
for rec in RECORDS:
if rec["code"] not in seen:
seen.add(rec["code"]); show_record(rec)
if len(seen) >= 4:
break
SYSTEM_PROMPT = (
"You are a cautious visible notion assistant. Examine the picture(s) intently "
"earlier than answering. Every query has a brief, uniquely decided reply.n"
"Reason briefly if wanted, then finish your reply with precisely one line:n"
"Answer: <your ultimate brief reply>n"
"Give solely the worth (a quantity, phrase, or brief phrase) after 'Answer:' — "
"no items, no clarification, no full sentence."
)
def build_payload(rec, max_side, high quality):
"""Returns (interleaved_parts, resized_pils, data_uris)."""
resized, uris = [], []
for im in rec["images"]:
pil, uri = shrink(im, max_side, high quality)
resized.append(pil); uris.append(uri)
elements = split_on_placeholders(rec["problem"], len(resized))
if rec["hint"]:
elements.append(("textual content", f"Hint: {rec['hint']}"))
return elements, resized, uris
def parts_to_openai(elements, uris):
content material = []
for sort, val in elements:
if sort == "textual content":
content material.append({"sort": "textual content", "textual content": val})
else:
content material.append({"sort": "image_url", "image_url": {"url": uris[val]}})
return [{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": content}]
We show consultant benchmark examples by arranging the related photographs into readable grids and presenting every query with its functionality, reference reply, and supply. We outline a strict multimodal system immediate that instructs the evaluated mannequin to examine all photographs and return a concise ultimate reply in a constant format. We additionally resize photographs, protect their placement relative to query placeholders, and convert the ensuing content material into OpenAI-compatible multimodal messages.
class Backend:
identify = "base"
def predict(self, rec): elevate NotImplementedError
def predict_batch(self, recs):
return [self.predict(r) for r in recs]
class BlindPriorBackend(Backend):
"""Text-only ground. Answers utilizing the *reply prior* conditioned on the
floor type the query implies — no pixels are ever learn.
This is the management situation that makes an accuracy quantity significant:
'How many hinges?' has a guessable prior (small integers dominate). If a
imaginative and prescient mannequin barely beats this, it is not perceiving, it is guessing."""
identify = "blind-prior"
def __init__(self, data, seed=0):
self.rng = random.Random(seed)
self.by_type = defaultdict(record)
for r in data:
self.by_type[answer_type(r["answer"])].append(r["answer"])
self.all = [r["answer"] for r in data]
def predict(self, rec):
q = rec["problem"].decrease()
if re.search(r"what number of|variety of|depend", q):
pool = self.by_type.get("integer") or self.all
elif re.search(r"bisb.*?|does |are there", q):
pool = self.by_type.get("boolean") or self.all
else:
pool = self.all
return f"Answer: {self.rng.selection(pool)}"
class OpenAICompatBackend(Backend):
"""Works with OpenAI, Moonshot/Kimi, OpenRouter, Together, vLLM, LM Studio…
something exposing POST {base}/chat/completions with image_url content material."""
def __init__(self, base, key, mannequin, max_tokens, employees, max_side, high quality):
self.base, self.key, self.mannequin = base.rstrip("/"), key, mannequin
self.max_tokens, self.employees = max_tokens, employees
self.max_side, self.high quality = max_side, high quality
self.identify = f"api:{mannequin}"
def _one(self, rec, retries=4):
elements, _, uris = build_payload(rec, self.max_side, self.high quality)
physique = {"mannequin": self.mannequin, "messages": parts_to_openai(elements, uris),
"max_tokens": self.max_tokens, "temperature": 0}
for a in vary(retries):
attempt:
r = requests.submit(f"{self.base}/chat/completions",
headers={"Authorization": f"Bearer {self.key}",
"Content-Type": "utility/json"},
json=physique, timeout=180)
if r.status_code in (429, 500, 502, 503, 529):
time.sleep(2 ** a + random.random()); proceed
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
besides Exception as e:
if a == retries - 1:
return f"__ERROR__ {sort(e).__name__}: {e}"
time.sleep(2 ** a + random.random())
return "__ERROR__ exhausted"
def predict(self, rec):
return self._one(rec)
def predict_batch(self, recs):
out = [None] * len(recs)
with ThreadPoolExecutor(max_workers=self.employees) as ex:
futs = {ex.submit(self._one, r): i for i, r in enumerate(recs)}
achieved = 0
for f in as_completed(futs):
out[futs[f]] = f.end result(); achieved += 1
print(f" [api] {achieved}/{len(recs)}", finish="r")
print()
return out
class LocalVLMBackend(Backend):
"""Small open VLM on a Colab GPU (T4 works for ~2-3B in fp16)."""
def __init__(self, model_id, max_new, max_side):
import torch
from transformers import AutoProcessor, AutoModelForImageTextToText
self.torch, self.max_new, self.max_side = torch, max_new, max_side
self.identify = f"native:{model_id.cut up('/')[-1]}"
dtype = torch.float16 if torch.cuda.is_available() else torch.float32
print(f"[local] loading {model_id} ({dtype})…")
self.proc = AutoProcessor.from_pretrained(model_id)
self.mannequin = AutoModelForImageTextToText.from_pretrained(
model_id, torch_dtype=dtype,
device_map="auto" if torch.cuda.is_available() else None)
self.mannequin.eval()
def predict(self, rec):
elements, pils, _ = build_payload(rec, self.max_side, 90)
content material = [{"type": "image", "image": pils[v]} if okay == "picture"
else {"sort": "textual content", "textual content": v} for okay, v in elements]
msgs = [{"role": "system", "content": [{"type": "text", "text": SYSTEM_PROMPT}]},
{"function": "person", "content material": content material}]
attempt:
inputs = self.proc.apply_chat_template(
msgs, add_generation_prompt=True, tokenize=True,
return_dict=True, return_tensors="pt").to(self.mannequin.gadget)
besides Exception:
textual content = self.proc.apply_chat_template(msgs, add_generation_prompt=True)
inputs = self.proc(textual content=[text], photographs=pils, return_tensors="pt").to(self.mannequin.gadget)
with self.torch.inference_mode():
ids = self.mannequin.generate(**inputs, max_new_tokens=self.max_new, do_sample=False)
gen = ids[0][inputs["input_ids"].form[-1]:]
return self.proc.decode(gen, skip_special_tokens=True)
def predict_batch(self, recs):
out = []
for i, r in enumerate(recs):
out.append(self.predict(r))
print(f" [local] {i+1}/{len(recs)}", finish="r")
print()
return out
def make_backend():
b = CFG["BACKEND"]
if b == "api":
assert CFG["API_KEY"], "Set CFG['API_KEY'] (or the PB_API_KEY env var)."
return OpenAICompatBackend(CFG["API_BASE"], CFG["API_KEY"], CFG["API_MODEL"],
CFG["API_MAX_TOKENS"], CFG["API_WORKERS"],
CFG["MAX_IMAGE_SIDE"], CFG["JPEG_QUALITY"])
if b == "native":
return LocalVLMBackend(CFG["LOCAL_MODEL"], CFG["LOCAL_MAX_NEW"], CFG["MAX_IMAGE_SIDE"])
return BlindPriorBackend(RECORDS, CFG["SEED"])
WORD2NUM = {w: i for i, w in enumerate(
"zero one two three 4 5 six seven eight 9 ten eleven twelve "
"13 fourteen fifteen sixteen seventeen eighteen nineteen twenty".cut up())}
ARTICLES = {"a", "an", "the", "is", "are", "there", "it", "of"}
def extract_answer(uncooked):
"""Pull the ultimate brief reply out of free-form mannequin textual content."""
if uncooked is None:
return ""
t = str(uncooked).strip()
if t.startswith("__ERROR__"):
return ""
m = re.findall(r"boxed{([^}]*)}", t)
if m:
return m[-1].strip()
m = re.findall(r"(?i)bfinal solutions*[:-]s*(.+)", t)
if m:
return m[-1].strip().cut up("n")[0].strip()
m = re.findall(r"(?i)^s*solutions*[:-]s*(.+)$", t, re.M)
if m:
return m[-1].strip()
strains = [l.strip() for l in t.split("n") if l.strip()]
return strains[-1] if strains else ""
def normalize(s):
s = str(s).strip().decrease()
s = re.sub(r"^**|**$", "", s)
s = re.sub(r"[u2018u2019u201cu201d]", "'", s)
s = re.sub(r"[.,;:!?'"()[]]+$", "", s)
s = re.sub(r"^[.,;:!?'"()[]]+", "", s)
s = s.substitute("%", " % ").substitute("$", " greenback ")
s = re.sub(r"(d),(d{3})b", r"12", s)
s = re.sub(r"s+", " ", s).strip()
toks = [WORD2NUM.get(t, t) for t in s.split()]
toks = [str(t) for t in toks if str(t) not in ARTICLES]
return " ".be part of(toks).strip()
def as_number(s):
s = normalize(s)
m = re.fullmatch(r"-?d+(?:.d+)?", s)
if m:
return float(s)
m = re.findall(r"-?d+(?:.d+)?", s)
return float(m[0]) if len(m) == 1 else None
def rule_judge(pred_raw, gold, rel_tol=0.0):
pred = extract_answer(pred_raw)
if not pred:
return 0, "empty"
p, g = normalize(pred), normalize(gold)
if p == g:
return 1, "precise"
pn, gn = as_number(p), as_number(g)
if pn just isn't None and gn just isn't None:
if pn == gn:
return 1, "numeric"
if rel_tol > 0 and gn != 0 and abs(pn - gn) / abs(gn) <= rel_tol:
return 1, "numeric~tol"
return 0, "numeric-mismatch"
if re.fullmatch(r"(sure|true)", g) and re.search(r"b(sure|true)b", p): return 1, "bool"
if re.fullmatch(r"(no|false)", g) and re.search(r"b(no|false)b", p): return 1, "bool"
if len(g.cut up()) <= 4 and re.search(rf"(?<!w){re.escape(g)}(?!w)", p):
return 1, "comprises"
return 0, "mismatch"
LLM_JUDGE_PROMPT = (
"You grade a visual-question reply. Reply with precisely one token: "
"CORRECT or INCORRECT.nQuestion: {q}nReference reply: {g}n"
"Model reply: {p}nSemantically equal to the reference (ignoring "
"phrasing, items, formatting)?")
def llm_judge(pred_raw, gold, query):
"""Mirrors the paper's protocol (they use GPT-oss-120B because the choose,
reporting 99.7% settlement with people on a 300-sample audit)."""
pred = extract_answer(pred_raw)
if not pred:
return 0, "empty"
okay, why = rule_judge(pred_raw, gold)
if okay:
return 1, "rule-shortcut"
attempt:
r = requests.submit(f"{CFG['API_BASE'].rstrip('/')}/chat/completions",
headers={"Authorization": f"Bearer {CFG['API_KEY']}"},
json={"mannequin": CFG["API_MODEL"], "temperature": 0, "max_tokens": 5,
"messages": [{"role": "user", "content":
LLM_JUDGE_PROMPT.format(q=question, g=gold, p=pred)}]},
timeout=60)
v = r.json()["choices"][0]["message"]["content"].strip().higher()
return (1, "llm") if v.startswith("CORRECT") else (0, "llm")
besides Exception:
return okay, why + "+judge-failed"
_JUDGE_TESTS = [
("Answer: 15400", "15400", 1), ("The score is 15,400.", "15400", 1),
("Answer: three", "3", 1), ("Answer: 4 hinges", "4", 1),
("boxed{8}", "8", 1), ("Answer: 7", "8", 0),
("Answer: Yes", "yes", 1), ("I cannot tell.", "5", 0),
("Final answer: blue", "Blue", 1),
]
_fails = [(p, g, e) for p, g, e in _JUDGE_TESTS if rule_judge(p, g)[0] != e]
print(f"[judge] self-test: {len(_JUDGE_TESTS)-len(_fails)}/{len(_JUDGE_TESTS)} handed"
+ (f" FAILURES: {_fails}" if _fails else ""))
We create a shared backend interface and implement blind-prior, OpenAI-compatible API, and native Hugging Face vision-language mannequin backends. We construct answer-extraction and normalization utilities that deal with numbers, written quantity phrases, punctuation, formatted responses, Boolean solutions, and brief phrases. We then apply rule-based or non-compulsory LLM-assisted judging and run offline self-tests to confirm that the evaluator handles frequent response variations appropriately.
def run_eval(data, backend):
print(f"n[eval] backend = {backend.identify} on {len(data)} questions")
t0 = time.time()
preds = backend.predict_batch(data)
rows = []
for rec, uncooked in zip(data, preds):
if CFG["JUDGE"] == "llm" and CFG["API_KEY"]:
okay, how = llm_judge(uncooked, rec["answer"], rec["problem"])
else:
okay, how = rule_judge(uncooked, rec["answer"], CFG["NUM_REL_TOL"])
rows.append(dict(index=rec["index"], code=rec["code"], class=rec["category"],
source_bmk=rec["source_bmk"], n_images=rec["n_images"],
q_chars=rec["q_chars"], max_side=rec["max_side"],
ans_type=answer_type(rec["answer"]),
gold=rec["answer"], pred=extract_answer(uncooked),
uncooked=str(uncooked)[:2000], appropriate=okay, how=how))
print(f"[eval] achieved in {time.time()-t0:.1f}s")
return pd.DataBody(rows)
def bootstrap_ci(vals, n_boot=4000, seed=0):
a = np.asarray(vals, dtype=float)
if a.dimension == 0:
return (float("nan"), float("nan"))
rng = np.random.default_rng(seed)
means = a[rng.integers(0, a.size, (n_boot, a.size))].imply(axis=1)
return tuple(np.percentile(means, [2.5, 97.5]) * 100)
def report(res, label):
print("n" + "=" * 78)
print(f"§9 RESULTS — {label}")
print("=" * 78)
lo, hello = bootstrap_ci(res.appropriate)
print(f"nOVERALL accuracy: {res.appropriate.imply()*100:5.1f}% "
f"95% CI [{lo:.1f}, {hi:.1f}] (n={len(res)})")
print("(card: no frontier mannequin exceeds 60% general)n")
print("-- per atomic functionality --")
tab = []
for code, g in res.groupby("code"):
l, h = bootstrap_ci(g.appropriate)
tab.append(dict(code=code, functionality=CODE_FULL.get(code, code),
n=len(g), acc=g.appropriate.imply() * 100, lo=l, hello=h))
t = pd.DataBody(tab).sort_values("acc", ascending=False)
print(t.to_string(index=False, float_format=lambda x: f"{x:6.1f}"))
print("n-- issue slices --")
res = res.copy()
res["img_bucket"] = np.the place(res.n_images > 1, "multi-image", "single-image")
res["res_bucket"] = pd.reduce(res.max_side, [0, 800, 1600, 10**6],
labels=["<800px", "800-1600px", ">1600px"])
for col in ["img_bucket", "res_bucket", "ans_type"]:
s = res.groupby(col, noticed=True).appropriate.agg(["size", "mean"])
s["mean"] = (s["mean"] * 100).spherical(1)
print(f"n by {col}:n{s.rename(columns={'dimension':'n','imply':'acc%'}).to_string()}")
print("n-- choose choice breakdown --")
print(res.how.value_counts().to_string())
errs = res[res.correct == 0]
if len(errs):
print("n-- pattern failures --")
for _, r in errs.head(5).iterrows():
print(f" [{r.code}] gold={r.gold!r:>14} pred={r['pred']!r:>20} ({r.how})")
return t
BACKEND = make_backend()
RES = run_eval(RECORDS, BACKEND)
PER_CAP = report(RES, BACKEND.identify)
LEADERBOARD = {
"GPT-5.6-Sol": [59.7, 69.7, 62.4, 62.1, 55.5, 76.7, 67.0, 55.9, 60.0, 54.9, 26.9],
"Kimi K3": [58.5, 68.2, 59.7, 59.4, 52.4, 70.3, 59.1, 55.9, 53.3, 61.2, 41.7],
"Claude-Fable-5": [57.2, 58.5, 52.9, 60.9, 51.5, 70.4, 56.1, 51.6, 59.8, 64.3, 45.0],
"Gemini-3.1-Pro": [56.2, 58.8, 56.9, 61.8, 50.0, 52.7, 61.7, 54.8, 61.2, 64.3, 40.6],
"Seed-2.1-Pro": [55.0, 57.6, 51.2, 58.2, 43.6, 50.0, 59.5, 56.6, 60.4, 66.7, 49.8],
"Qwen3.5-397B-A17B":[47.5, 55.2, 49.1, 53.0, 44.6, 46.7, 49.8, 44.8, 50.2, 52.9, 26.9],
"Gemma-4-31B": [40.7, 42.7, 33.9, 40.3, 39.1, 44.9, 43.7, 39.0, 45.9, 46.7, 32.1],
"GLM-4.6V": [32.5, 35.2, 31.8, 35.2, 29.1, 30.6, 34.8, 29.3, 33.7, 39.2, 26.9],
}
LB = pd.DataBody(LEADERBOARD, index=["Overall"] + CODE_ORDER).T
print("n" + "=" * 78)
print("§10 OFFICIAL LEADERBOARD (subset, accuracy %)")
print("=" * 78)
print(LB.to_string())
print("nNote the structural discovering from the cardboard: Hallu is the weakest column "
"nearly in all places,nand fashions with near-identical Overall scores have "
"very completely different functionality profiles.")
def radar(per_cap_df, label, evaluate=("GPT-5.6-Sol", "Gemma-4-31B")):
codes = [c for c in CODE_ORDER if c in set(per_cap_df.code)]
if len(codes) < 3:
print("[radar] want >=3 capabilities"); return
vals = per_cap_df.set_index("code").acc.reindex(codes).fillna(0).tolist()
ang = np.linspace(0, 2 * np.pi, len(codes), endpoint=False).tolist()
shut = lambda v: v + v[:1]
fig, ax = plt.subplots(figsize=(6.4, 6.4), subplot_kw=dict(polar=True))
ax.plot(shut(ang), shut(vals), lw=2.4, coloration="#C44E52", label=label)
ax.fill(shut(ang), shut(vals), alpha=.18, coloration="#C44E52")
for m in evaluate:
if m in LB.index:
v = LB.loc[m, codes].tolist()
ax.plot(shut(ang), shut(v), lw=1.3, ls="--", alpha=.85, label=m)
ax.set_xticks(ang)
ax.set_xticklabels(codes)
ax.set_ylim(0, 100)
ax.set_yticks([20, 40, 60, 80])
ax.set_title("PerceptionBench functionality profile", pad=24)
ax.legend(loc="higher proper", bbox_to_anchor=(1.32, 1.12), fontsize=8)
plt.tight_layout(); plt.present()
if CFG["SHOW_PLOTS"]:
radar(PER_CAP, BACKEND.identify)
fig, ax = plt.subplots(figsize=(7, 3.2))
s = LB["Overall"].sort_values()
ax.barh(s.index, s.values, coloration="#8C8C8C")
ax.barh([BACKEND.name], [RES.correct.mean() * 100], coloration="#C44E52")
ax.axvline(60, ls="--", c="okay", lw=1)
ax.textual content(60.5, -.4, "60% ceiling: unbeaten", fontsize=8)
ax.set_xlabel("Overall accuracy (%)"); ax.set_title("Your run vs. the leaderboard")
plt.tight_layout(); plt.present()
tag = re.sub(r"[^A-Za-z0-9_.-]", "_", BACKEND.identify)
p_pred = os.path.be part of(CFG["OUT_DIR"], f"predictions_{tag}.jsonl")
p_cap = os.path.be part of(CFG["OUT_DIR"], f"per_capability_{tag}.csv")
p_meta = os.path.be part of(CFG["OUT_DIR"], f"run_meta_{tag}.json")
with open(p_pred, "w") as f:
for _, r in RES.iterrows():
f.write(json.dumps(r.to_dict(), default=str) + "n")
PER_CAP.to_csv(p_cap, index=False)
json.dump({"config": {okay: v for okay, v in CFG.objects() if "KEY" not in okay},
"backend": BACKEND.identify, "n_questions": len(RES),
"rows_scanned": N_SCANNED,
"overall_acc": float(RES.appropriate.imply() * 100),
"ci95": record(bootstrap_ci(RES.appropriate)),
"timestamp": time.strftime("%Y-%m-%dTpercentH:%M:%S")},
open(p_meta, "w"), indent=2)
print(f"n[export] {p_pred}n[export] {p_cap}n[export] {p_meta}")
print("n" + "=" * 78)
print("DONE. Next steps:")
print(" 1) CFG['BACKEND']='api' + PB_API_KEY/PB_API_BASE/PB_API_MODEL -> rating an actual MLLM")
print(" 2) CFG['BACKEND']='native' on a GPU runtime -> rating an open 2-3B VLM")
print(" 3) CFG['JUDGE']='llm' -> reproduce the paper's LLM-as-judge protocol")
print(" 4) Raise N_PER_CATEGORY / MAX_SCAN, or LOAD_MODE='full' for all 3,000 rows")
print(" 5) Ablations price operating: crop-to-region vs. full picture, picture decision")
print(" sweep (MAX_IMAGE_SIDE 512/1024/2048), and CoT-on vs. CoT-off prompts")
print("=" * 78)
We execute the chosen backend throughout all ready data, choose each prediction, and retailer the leads to a structured DataBody for evaluation. We calculate general and per-capability accuracy, bootstrap confidence intervals, issue slices, failure examples, and comparisons with the included benchmark leaderboard. We lastly generate functionality visualizations and export predictions, functionality experiences, configuration metadata, accuracy statistics, and reproducibility particulars as JSONL, CSV, and JSON recordsdata.
In conclusion, we established a modular and reproducible framework for evaluating multimodal fashions on PerceptionBench. We dealt with the complete workflow from resilient dataset ingestion and picture preprocessing to immediate development, backend execution, reply extraction, automated judging, statistical evaluation, visualization, and artifact export. We can run the pipeline with out an API key or GPU utilizing the blind-prior baseline, and we are able to change to an API-hosted or native vision-language mannequin by altering the backend configuration. The ensuing experiences permit us to maneuver past a single general accuracy rating and examine how every mannequin performs throughout particular person visible capabilities, multi-image questions, picture resolutions, and reply codecs. We additionally created a basis for additional experiments involving immediate variations, image-resolution sweeps, cropping methods, choose comparisons, and full-dataset evaluations.
Check out the Full Codes here. Also, be happy to comply with us on Twitter and don’t overlook to hitch our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to accomplice with us for selling your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar and many others.? Connect with us
The submit Evaluating Multimodal Vision Models with Moonshot PerceptionBench Using Robust Data Loading and Automated Judging appeared first on MarkTechPost.
