搭建多模态视觉模型自动评测流水线

人工智能

在本教程中，我们为 PerceptionBench 设计了一套端到端评测工作流。这个多模态基准用于衡量模型在 OCR、计数、定位、上下文推理、比较、深度理解和幻觉检测等任务上的细粒度视觉感知能力。首先，我们配置一个兼容 Colab 的环境，安装所需库，并通过稳健的多阶段流式加载与下载策略载入数据集的均衡子集。随后，我们解码以 base64 编码的图像，解析交错排列的图像占位符，将每个样本规范化为一致的记录格式，并分析数据集的能力分布、图像要求、答案类型和来源基准。在此基础上，我们构建一个统一的评测框架，支持盲先验基线、兼容 OpenAI 的多模态 API，以及本地 Hugging Face 视觉语言模型。我们还实现了基于规则和可选的 LLM 辅助评判，计算 bootstrap 置信区间，考察不同难度切片上的性能，将能力画像与随附的排行榜进行比较，并导出可复现的预测结果和报告产物。

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")

我们配置 PerceptionBench 环境，定义数据集、后端、图像处理、评判和输出设置，并初始化可复现的随机行为。我们安装数据集加载、数值分析、可视化、HTTP 通信和图像处理所需的库。此外，我们还配置 Matplotlib 并准备输出目录，以确保后续评测工作流能够在 Google Colab 或本地环境中一致运行。

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"])

我们实现了一个具备容错能力的数据集加载器：它首先尝试以流式方式加载转换后的 Parquet 文件；如果失败，则回退到以流式方式加载原始文件；必要时，最终执行完整下载。我们在限制已处理行数的同时扫描数据集，并使用 error_category 字段将样本组织到特定能力对应的存储桶中。随后，我们创建一个均衡且经过随机打乱的子集，使每种视觉能力在评测问题中所占的数量大致相当。

DATA_URI_RE = re.compile(r"^data:(image/[A-Za-z0-9.+-]+);base64,(.*)$", re.S)
PLACEHOLDER_RE = re.compile(r"<\|image[ _](\d+)\|>")
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 =

我们将 data URI、原始 Base64 字符串、字节数组、PIL 对象以及 Hugging Face 图像字典中的图像解码为统一的 RGB 格式。我们把数据集中的每一行规范化为结构化记录，其中包含问题文本、答案、图像、能力标签、尺寸、占位符数量及来源信息。随后，我们分析能力覆盖范围、答案格式、图像数量、分辨率特征和来源基准，并将生成的数据集概况可视化。

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}]

我们将相关图像排列成易于阅读的网格，并同时展示每个问题对应的能力、参考答案和来源，以此呈现具有代表性的基准测试样例。我们定义了严格的多模态系统提示词，要求被评估模型检查所有图像，并以一致的格式返回简洁的最终答案。我们还会调整图像尺寸，保留它们相对于问题占位符的位置，并将生成的内容转换为与 OpenAI 兼容的多模态消息。

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_tokens, "temperature": 0}
       for a in range(retries):
           try:
               r = requests.post(f"{self.base}/chat/completions",
                                 headers={"Authorization": f"Bearer {self.key}",
                                          "Content-Type": "application/json"},
                                 json=body, timeout=180)
               if r.status_code in (429, 500, 502, 503, 529):
                   time.sleep(2 ** a + random.random()); continue
               r.raise_for_status()
               return r.json()["choices"][0]["message"]["content"]
           except Exception as e:
               if a == retries - 1:
                   return f"__ERROR__ {type(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.workers) as ex:
           futs = {ex.submit(self._one, r): i for i, r in enumerate(recs)}
           done = 0
           for f in as_completed(futs):
               out[futs[f]] = f.result(); done += 1
               print(f"   [api] {done}/{len(recs)}", end="\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.name = f"local:{model_id.split('/')[-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.model = AutoModelForImageTextToText.from_pretrained(
           model_id, torch_dtype=dtype,
           device_map="auto" if torch.cuda.is_available() else None)
       self.model.eval()
   def predict(se

我们创建了统一的后端接口，并实现了盲先验、兼容 OpenAI API 以及本地 Hugging Face 视觉语言模型后端。我们构建了答案提取与规范化工具，用于处理数字、英文数字单词、标点符号、格式化响应、布尔值答案和短语。随后，我们采用基于规则或可选的 LLM 辅助判定方式，并运行离线自测，以验证评估器能否正确处理常见的回答变体。

def run_eval(records, backend):
   print(f"\n[eval] backend = {backend.name} on {len(records)} questions")
   t0 = time.time()
   preds = backend.predict_batch(records)
   rows = []
   for rec, raw in zip(records, preds):
       if CFG["JUDGE"] == "llm" and CFG["API_KEY"]:
           ok, how = llm_judge(raw, rec["answer"], rec["problem"])
       else:
           ok, how = rule_judge(raw, rec["answer"], CFG["NUM_REL_TOL"])
       rows.append(dict(index=rec["index"], code=rec["code"], category=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(raw),
                        raw=str(raw)[:2000], correct=ok, how=how))
   print(f"[eval] done in {time.time()-t0:.1f}s")
   return pd.DataFrame(rows)
def bootstrap_ci(vals, n_boot=4000, seed=0):
   a = np.asarray(vals, dtype=float)
   if a.size == 0:
       return (float("nan"), float("nan"))
   rng = np.random.default_rng(seed)
   means = a[rng.integers(0, a.size, (n_boot, a.size))].mean(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, hi = bootstrap_ci(res.correct)
   print(f"\nOVERALL accuracy: {res.correct.mean()*100:5.1f}%   "
         f"95% CI [{lo:.1f}, {hi:.1f}]   (n={len(res)})")
   print("(card: no frontier model exceeds 60% overall)\n")
   print("-- per atomic capability --")
   tab = []
   for code, g in res.groupby("code"):
       l, h = bootstrap_ci(g.correct)
       tab.append(dict(code=code, capability=CODE_FULL.get(code, code),
                       n=len(g), acc=g.correct.mean() * 100, lo=l, hi=h))
   t = pd.DataFrame(tab).sort_values("acc", ascending=False)
   print(t.to_string(index=False, float_format=lambda x: f"{x:6.1f}"))
   print("\n-- difficulty slices --")
   res = res.copy()
   res["img_bucket"] = np.where(res.n_images > 1, "multi-image", "single-image")
   res["res_bucket"] = pd.cut(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, observed=True).correct.agg(["size", "mean"])
       s["mean"] = (s["mean"] * 100).round(1)
       print(f"\n  by {col}:\n{s.rename(columns={'size':'n','mean':'acc%'}).to_string()}")
   print("\n-- judge decision breakdown --")
   print(res.how.value_counts().to_string())
   errs = res[res.correct == 0]
   if len(errs):
       print("\n-- sample 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.name)
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.DataFrame(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 finding from the card: Hallu is the weakest column "
     "almost everywhere,\nand models with

我们在所有准备好的记录上运行选定的后端，对每个预测结果进行评判，并将结果存储到结构化的 DataFrame 中以供分析。我们计算总体准确率和各能力维度的准确率、Bootstrap 置信区间、不同难度分组的结果、失败案例，以及与基准测试所附排行榜的对比结果。最后，我们生成能力维度的可视化图表，并将预测结果、能力报告、配置元数据、准确率统计数据和可复现性详情导出为 JSONL、CSV 和 JSON 文件。

总之，我们建立了一套模块化且可复现的框架，用于在 PerceptionBench 上评估多模态模型。我们覆盖了完整的工作流，包括稳健的数据集加载、图像预处理、提示词构建、后端执行、答案提取、自动化评判、统计分析、可视化以及产物导出。借助盲先验基线，我们无需 API 密钥或 GPU 即可运行该流水线；只需更改后端配置，便可切换到由 API 托管的视觉语言模型或本地视觉语言模型。生成的报告使我们能够超越单一的总体准确率分数，深入检查每个模型在各项视觉能力、多图像问题、图像分辨率和答