翻訳待ち：Evaluating Multimodal Vision Models with Moonshot PerceptionBench Using Robust Data Loading and Automated Judging

AI サービスが一時的に利用できないため、復旧後に翻訳を補完します。

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 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"") 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): """text? -> [('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 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: \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) [truncated for AI cost control]