Spaces:
Running on Zero
Running on Zero
File size: 4,780 Bytes
fed954e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | """
stub_runner.py — A CPU, torch-free fake VLM for tests and dry runs.
`StubVLMRunner` produces deterministic outputs so the whole orchestrator —
scoring, durability, leaderboard, verdict — can be exercised with no GPU and no
torch import. Three behaviours:
perfect : schema-valid output that MATCHES the ground truth (high accuracy) and
is emitted as clean bare JSON (json_robust = True).
fragile : the SAME content, but wrapped in a ```json fence so it only parses
after repair (json_robust = False) — used to prove that the
labeler_score penalizes fragile JSON even at equal accuracy.
random : emits invalid / off output for a fraction of samples, to exercise the
reject sidecar and the schema_valid_rate path.
Note: "fragile" uses a PROSE wrapper (structural repair), not a markdown fence —
fence-stripping is benign/deterministic and no longer counts against robustness,
so a fenced output would (correctly) still score as robust.
It is a test double, so it is allowed to peek at the ground truth (a real runner
never does).
"""
from __future__ import annotations
import json
from typing import Optional
from .coords import XYWH, XYXY, CoordSpace, from_canonical, to_canonical
from .runner_types import VLMResult
from .tasks_vision import VisionTaskSpec, resolved_system_prompt
def _default_value(fs):
"""A schema-valid placeholder for one field spec."""
if fs.cardinality == "list":
return [] # lists always validate (default_factory=list)
if fs.nested_fields:
return {f.name: _default_value(f) for f in fs.nested_fields}
if fs.vocabulary == "closed":
return fs.closed_values[0]
vk = fs.value_kind
if vk == "number":
return 0.0
if vk == "integer":
return 0
if vk == "bbox":
return [0.0, 0.0, 0.0, 0.0]
if vk == "point":
return [0.0, 0.0]
return "x"
def _synthesize_valid(spec: VisionTaskSpec) -> dict:
return {name: _default_value(fs) for name, fs in spec.fields.items()}
def _gt_to_prediction(spec: VisionTaskSpec, gt, image_size) -> Optional[dict]:
"""Build a GT-matching, schema-valid prediction for the pilot categories.
Returns None if the category has no GT-driven construction (use synthesize)."""
cat = spec.category
if cat == "image_classification" and isinstance(gt, dict):
label = (gt.get("labels") or [gt.get("label", "x")])[0]
return {"label": label, "confidence": 0.95,
"top5": [{"label": label, "score": 0.95}]}
if cat == "bbox_grounding" and isinstance(gt, dict):
dets = []
for b in gt.get("boxes", []):
canon = to_canonical(b["bbox"], CoordSpace.PIXEL_ABS, image_size, fmt=b.get("fmt", XYWH))
box = from_canonical(canon, spec.coord_space, image_size, fmt=XYXY)
dets.append({"label": b.get("label", "x"), "box": [round(c, 2) for c in box], "score": 0.95})
return {"detections": dets, "count": len(dets)}
if cat == "ocr_text" and isinstance(gt, dict):
text = gt.get("text", "")
return {"full_text": text, "lines": [{"text": text}]}
return None
class StubVLMRunner:
"""Drop-in fake runner. Matches the VLMRunner.generate signature."""
def __init__(self, model_id: str = "stub", behavior: str = "perfect",
reasoning: str = "instruct", **_kwargs):
self.model_id = model_id
self.behavior = behavior
self.reasoning = reasoning
self._n = 0
def close(self) -> None: # symmetry with VLMRunner
pass
def generate(self, spec: VisionTaskSpec, image, mode: str, *,
image_id: str = "", image_size=(64, 64), gt=None, user_prompt=None) -> VLMResult:
self._n += 1
# Build content
pred = _gt_to_prediction(spec, gt, image_size)
if pred is None:
pred = _synthesize_valid(spec)
if self.behavior == "random" and (self._n % 4 == 0):
raw = "I cannot answer that." # invalid → exercises reject sidecar
return VLMResult(mode, raw, "stub", 8, 6, 0.001, image_id, grammar_conformant=False)
body = json.dumps(pred)
if self.behavior == "fragile":
# prose wrapper = STRUCTURAL repair (not a benign fence) → not robust
raw = f"Sure! Here is the structured result you requested: {body} — hope that helps!"
else:
raw = body # clean bare JSON
# touch the resolved prompt so prompt wiring is exercised
_ = resolved_system_prompt(spec)
grammar = (mode == "constrained")
return VLMResult(mode, raw, "stub", 12, max(1, len(body) // 4), 0.001,
image_id, grammar_conformant=grammar)
|