Moonshot PerceptionBench、多モーダル視覚モデル評価ワークフローを公開

In this tutorial, we design an end-to-end evaluation workflow for PerceptionBench. This multimodal benchmark measures fine-grained visual perception capabilities across tasks such as OCR, counting, localization, contextual reasoning, comparison, depth understanding, and hallucination detection. We begin by configuring a Colab-compatible environment, installing the required libraries, and loading a balanced subset of the dataset through a robust multi-stage streaming and download strategy. We then decode base64-encoded images, parse interleaved image placeholders, normalize each example into a consistent record format, and analyze the dataset’s capability distribution, image requirements, answer types, and source benchmarks. From there, we construct a unified evaluation harness that supports a blind-prior baseline, OpenAI-compatible multimodal APIs, and local Hugging Face vision-language models. We also implement rule-based and optional LLM-assisted judging, calculate bootstrap confidence intervals, examine performance across difficulty slices, compare capability profiles with the included leaderboard, and export reproducible prediction and reporting artifacts.

Copy CodeCopiedUse a different Browser

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 = "train",

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/perceptionbench_out" if os.path.isdir("/content") 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],

check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

if CFG["INSTALL_DEPS"]:

print("[setup] installing dependencies (quiet, ~30s on a cold Colab)…")

_sh(["datasets>=3.0.0", "huggingface_hub>=0.25.0", "pillow", "pandas",

"numpy", "matplotlib", "requests", "pyarrow"])

if CFG["BACKEND"] == "local":

_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.update({"figure.dpi": 110, "font.size": 9, "axes.grid": True,

"grid.alpha": .25, "axes.spines.top": False,

"axes.spines.right": False})

print("[setup] ready\n")

We configure the PerceptionBench environment, define the dataset, backend, image-processing, judging, and output settings, and initialize reproducible random behavior. We install the required libraries for dataset loading, numerical analysis, visualization, HTTP communication, and image processing. We also configure Matplotlib and prepare the output directory so the remaining evaluation workflow runs consistently in Google Colab or a local environment.

Copy CodeCopiedUse a different Browser

def _iter_rows(repo, split, mode, max_scan):

"""Yield dict rows, trying progressively heavier strategies."""

from datasets import load_dataset

if mode == "full":

print("[load] full download (~1.63 GB) …")

ds = load_dataset(repo, split=split)

for i, r in enumerate(ds):

if i >= max_scan:

return

yield r

return

try:

from huggingface_hub import HfApi, hf_hub_url

api = HfApi()

files = api.list_repo_files(repo, repo_type="dataset", revision="refs/convert/parquet")

pq = sorted(f for f in files if f.endswith(".parquet") and f"/{split}/" 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, split="train", streaming=True)

for i, r in enumerate(ds):

if i >= max_scan:

return

yield r

return

except Exception as e:

print(f"[load] parquet stream unavailable ({type(e).__name__}: {e}); falling back")

try:

print("[load] streaming original data files")

ds = load_dataset(repo, split=split, streaming=True)

for i, r in enumerate(ds):

if i >= max_scan:

return

yield r

return

except Exception as e:

print(f"[load] json stream failed ({type(e).__name__}); doing a full download")

ds = load_dataset(repo, split=split)

for i, r in enumerate(ds):

if i >= max_scan:

return

yield r

def stratified_subset(repo, split, n_per_cat, max_scan, mode):

"""Balanced sample across error_category — the ten atomic capabilities.

Balancing matters: the benchmark reports a *capability profile*, and an

unbalanced sample makes the overall number a weighted average of whichever

capabilities happened to appear first in the shard.

"""

buckets, scanned, t0 = defaultdict(list), 0, time.time()

for row in _iter_rows(repo, split, 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:

filled = sum(len(v) >= n_per_cat for v in buckets.values())

print(f" scanned={scanned:5d} categories={len(buckets):2d} "

f"filled={filled:2d} {time.time()-t0:5.1f}s", end="\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 -> kept {len(rows)} across "

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 attempts converted Parquet streaming, then falls back to streaming the original files, and finally performs a full download when necessary. We scan the dataset while limiting the number of processed rows and organize examples into capability-specific buckets using the error_category field. We then create a balanced, shuffled subset so each visual capability contributes a comparable number of evaluation questions.

Copy CodeCopiedUse a different Browser

DATA_URI_RE = re.compile(r"^data:(image/[A-Za-z0-9.+-]+);base64,(.*)$", re.S)

PLACEHOLDER_RE = re.compile(r"<\|image _\|>")

def decode_image(entry):

"""data-URI string | raw 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("image") or []

if isinstance(imgs, (str, bytes, dict)):

imgs = [imgs]

out = []

for e in imgs:

try:

out.append(decode_image(e))

except Exception as err:

print(f" [warn] undecodable image on idx={row.get('index')}: {err}")

return out

def shrink(img, max_side, quality):

"""Downscale + re-encode. Returns (PIL, data_uri). Controls the token bill:

a 3000px screenshot can cost >2k vision tokens per image, and these

questions carry up to 8 images each."""

w, h = img.size

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", quality=quality)

uri = "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()

return img, uri

def split_on_placeholders(problem, n_images):

"""<|image_1|>text<|image_2|>? -> [('image',0),('text','…'),('image',1)…]

Any image never referenced by a placeholder is appended at the end, so we

never silently drop visual evidence."""

parts, last = [], 0

for m in PLACEHOLDER_RE.finditer(problem):

chunk = problem[last:m.start()].strip()

if chunk:

parts.append(("text", chunk))

i = int(m.group(1)) - 1

if 0 <= i < n_images:

parts.append(("image", i))

last = m.end()

tail = problem[last:].strip()

if tail:

parts.append(("text", tail))

used = {p[1] for p in parts if p[0] == "image"}

for i in range(n_images):

if i not in used:

parts.append(("image", i))

return parts

def to_record(row):

imgs = load_images(row)

problem = (row.get("problem") or "").strip()

return dict(

index = row.get("index"),

problem = problem,

answer = str(row.get("answer", "")).strip(),

hint = (row.get("hint") or "").strip(),

category = row.get("error_category") or "unknown",

source_bmk = row.get("source_bmk") or "NA",

source_idx = row.get("source_idx"),

images = imgs,

n_images = len(imgs),

n_placeholders= len(PLACEHOLDER_RE.findall(problem)),

q_chars = len(problem),

px_total = sum(w * h for w, h in (im.size for im in imgs)),

max_side = max([max(im.size) for im in imgs], default=0),

)

print("[decode] decoding images…")

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 records\n")

def cat_code(cat):

c = (cat or "").lower()

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": "visual relation", "Count": "counting", "Attr": "attribute",

"Depth": "depth & 3D", "Loc": "localization", "Comp": "comparison",

"FGR": "fine-grained recog.", "Ctx": "contextual integration",

"OCR": "OCR", "Hallu": "perception hallucination"}

for r in RECORDS:

r["code"] = cat_code(r["category"])

df = pd.DataFrame([{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)(yes|no|true|false)", a): return "boolean"

if re.fullmatch(r"(?i)[A-H]", a): return "letter"

if len(a.split()) == 1: return "single-word"

return "phrase"

df["ans_type"] = df["answer"].map(answer_type)

print("=" * 78)

print("§4 DATASET PROFILE (stratified subset — the card reports 3,000 total)")

print("=" * 78)

print("\n-- atomic capabilities present --")

print(df.groupby("code").agg(n=("index", "size"),

mean_imgs=("n_images", "mean"),

mean_qlen=("q_chars", "mean")).round(2).to_string())

print("\n-- answer surface forms --")

print(df["ans_type"].value_counts().to_string())

print("\n-- images per question --")

print(df["n_images"].value_counts().sort_index().to_string())

print(f"\n-- multi-image questions: {(df.n_images > 1).mean():.1%} of the subset")

print(f"-- median longest image edge: {df.max_side.median():.0f}px "

f"(max {df.max_side.max():.0f}px)")

print("\n-- provenance (top source benchmarks) --")

print(df["source_bmk"].value_counts().head(8).to_string())

print(f"\n-- newly-authored (source_bmk == 'NA'): {(df.source_bmk=='NA').mean():.1%} "

f"(card: 40% authored / 60% decomposed)\n")

def hf_full_stats(repo, split="train", config="default"):

try:

r = requests.get("https://datasets-server.huggingface.co/statistics",

params={"dataset": repo, "config": config, "split": split},

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 capability distribution (3,000 rows, via datasets-server) --")

for k, v in sorted(freq.items(), key=lambda x: -x[1]):

print(f" {cat_code(k):6s} {k:34s} {v:5d} {v/tot:6.1%}")

print()

return freq

except Exception as e:

print(f"[stats] datasets-server unavailable ({type(e).__name__}); "

f"using subset statistics only\n")

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], color="#4C72B0")

ax[0].set_title("Questions per atomic capability"); ax[0].set_xlabel("")

df.n_images.value_counts().sort_index().plot.bar(ax=ax[1], color="#DD8452")

ax[1].set_title("Images per question"); ax[1].set_xlabel("# images")

df.ans_type.value_counts().plot.barh(ax=ax[2], color="#55A868")

ax[2].set_title("Answer surface form")

plt.tight_layout(); plt.show()

We decode images from data URIs, raw base64 strings, byte arrays, PIL objects, and Hugging Face image dictionaries into a consistent RGB format. We normalize every dataset row into a structured record containing question text, answers, images, capability labels, dimensions, placeholder counts, and source information. We then analyze capability coverage, answer formats, image counts, resolution characteristics, and source benchmarks while visualizing the resulting dataset profile.

Copy CodeCopiedUse a different Browser

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.show()

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 careful visual perception assistant. Examine the image(s) closely "

"before answering. Every question has a short, uniquely determined answer.\n"

"Reason briefly if needed, then end your reply with exactly one line:\n"

"Answer: <your final short answer>\n"

"Give only the value (a number, word, or short phrase) after 'Answer:' — "

"no units, no explanation, no full sentence."

)

def build_payload(rec, max_side, quality):

"""Returns (interleaved_parts, resized_pils, data_uris)."""

resized, uris = [], []

for im in rec["images"]:

pil, uri = shrink(im, max_side, quality)

resized.append(pil); uris.append(uri)

parts = split_on_placeholders(rec["problem"], len(resized))

if rec["hint"]:

parts.append(("text", f"Hint: {rec['hint']}"))

return parts, resized, uris

def parts_to_openai(parts, uris):

content = []

for kind, val in parts:

if kind == "text":

content.append({"type": "text", "text": val})

else:

content.append({"type": "image_url", "image_url": {"url": uris[val]}})

return [{"role": "system", "content": SYSTEM_PROMPT},

{"role": "user", "content": content}]

We display representative benchmark examples by arranging the associated images into readable grids and presenting each question with its capability, reference answer, and source. We define a strict multimodal system prompt that instructs the evaluated model to inspect all images and return a concise final answer in a consistent format. We also resize images, preserve their placement relative to question placeholders, and convert the resulting content into OpenAI-compatible multimodal messages.

Copy CodeCopiedUse a different Browser

class Backend:

name = "base"

def predict(self, rec): raise NotImplementedError

def predict_batch(self, recs):

return [self.predict(r) for r in recs]

class BlindPriorBackend(Backend):

"""Text-only floor. Answers using the *answer prior* conditioned on the

surface form the question implies — no pixels are ever read.

This is the control condition that makes an accuracy number meaningful:

'How many hinges?' has a guessable prior (small integers dominate). If a

vision model barely beats this, it isn't perceiving, it's guessing."""

name = "blind-prior"

def __init__(self, records, seed=0):

self.rng = random.Random(seed)

self.by_type = defaultdict(list)

for r in records:

self.by_type[answer_type(r["answer"])].append(r["answer"])

self.all = [r["answer"] for r in records]

def predict(self, rec):

q = rec["problem"].lower()

if re.search(r"how many|number of|count", q):

pool = self.by_type.get("integer") or self.all

elif re.search(r"\bis\b.*\?|does |are there", q):

pool = self.by_type.get("boolean") or self.all

else:

pool = self.all

return f"Answer: {self.rng.choice(pool)}"

class OpenAICompatBackend(Backend):

"""Works with OpenAI, Moonshot/Kimi, OpenRouter, Together, vLLM, LM Studio…

anything exposing POST {base}/chat/completions with image_url content."""

def __init__(self, base, key, model, max_tokens, workers, max_side, quality):

self.base, self.key, self.model = base.rstrip("/"), key, model

self.max_tokens, self.workers = max_tokens, workers

self.max_side, self.quality = max_side, quality

self.name = f"api:{model}"

def _one(self, rec, retries=4):

parts, _, uris = build_payload(rec, self.max_side, self.quality)

body = {"model": self.model, "messages": parts_to_openai(parts, uris),

"max_tokens": self.max_to