Moonshot PerceptionBench 多模态视觉模型评估教程
Evaluating Multimodal Vision Models with Moonshot PerceptionBench Using Robust Data Loading and Automated Judging
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.
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.
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.
更进一步：量化金融体系
看懂新闻只是起点——沿量化金融路径，把它变成能交付的工程能力