"""Result-panel rendering for the Polygen Qwen2-VL Space.
Pure presentation + scoring logic, kept out of `app.py` so it is unit-testable
without a Gradio runtime or a live endpoint. Everything here is honest-by-
construction:
- Token counts prefer the worker's MEASURED `n_polygen_visual_tokens` and only
fall back to the length law when the worker did not surface it.
- Correctness uses word-boundary token matching, NOT substring matching, so a
ground-truth token like ``"red"`` does not spuriously match ``"covered"``.
- The verdict is correctness-agreement when ground truth exists, so two
differently-worded-but-both-correct answers read as "Same answer" instead of
the near-always "Different" that strict string equality produces.
- Compute claims are FLOP-count reductions (attention is $O(L^2)$ in sequence
length), never wall-clock. No "faster" framing anywhere.
"""
import re
from html import escape
from typing import Optional
# ---------------------------------------------------------------------------
# Length law + FLOP math.
# ---------------------------------------------------------------------------
# Per-tier storage compression from the SDK tier registry
# (polygen/integrations/qwen2vl/tiers.py). Fallback only: the worker echoes the
# real value as `compression_ratio`. Lower tier keeps fewer lossless tiles, so
# it compresses storage harder (T=2 = 7.5x > T=3 = 5.1x).
_TIER_STORAGE = {0: 283.0, 1: 14.3, 2: 7.5, 3: 5.1, 4: 3.9}
def bypass_n_polygen_visual_tokens(
n_visual: int,
tier: int,
M: int = 4,
K: int = 3,
) -> int:
"""Post-bypass visual-token count -- FALLBACK ONLY.
Prefer the worker's measured `n_polygen_visual_tokens`; this reproduces the
SDK length law for the case the worker did not surface it. Bypass keeps
``tier`` of ``M`` tiles at full length and collapses the remaining
``M - tier`` tiles to ``K`` coefficient tokens each, so
``L_v_eff = sum(kept tile rows) + (M - tier) * K``. Tiles come from
``np.array_split``, so the first ``n_visual % M`` tiles carry one extra row;
this matters by at most 1 token versus the measured value. Verified against
the decomp_7B points (121->94, 676->510, 900->678, 1089->820, 1369->1030).
"""
if tier >= M:
return n_visual
if tier <= 0:
return M * K
base, rem = divmod(n_visual, M)
sizes = [base + 1] * rem + [base] * (M - rem)
kept = sum(sorted(sizes, reverse=True)[:tier])
return kept + (M - tier) * K
def attention_flops_reduction_pct(n_visual: int, n_poly: int) -> int:
"""Percent reduction in visual self-attention FLOPs ($O(L^2)$).
FLOP-count claim, not wall-clock. Returns 0 when the sequence did not
shorten (hook path, T=4, or a missing measurement).
"""
if n_visual <= 0 or n_poly <= 0 or n_poly >= n_visual:
return 0
return round(100 * (1 - (n_poly / n_visual) ** 2))
# ---------------------------------------------------------------------------
# Matching + correctness.
# ---------------------------------------------------------------------------
_WORD = re.compile(r"[a-z0-9]+")
def _tokens(s: str) -> list[str]:
return _WORD.findall((s or "").lower())
def _norm_phrase(s: str) -> str:
return " ".join(_tokens(s))
def is_correct(answer: str, ground_truth: Optional[list[str]]) -> Optional[bool]:
"""Score an answer against ground truth. ``None`` when no ground truth.
Word-boundary matching, NOT substring: a single-token ground-truth variant
must appear as a whole token (so ``"red"`` does not match ``"covered"``); a
multi-token variant must appear as a contiguous token phrase. Keep
ground-truth variants short and unambiguous; constrain the gallery questions
so the answer is a single noun / number / color. Residual risk: negation
("there is no cat") still tokenizes ``cat`` -- the smoke test's adversarial
wrong-answer check is the guard for that.
"""
if not ground_truth:
return None
ans_tokens = set(_tokens(answer))
ans_phrase = " ".join(_tokens(answer))
for variant in ground_truth:
gt = _tokens(variant)
if not gt:
continue
if len(gt) == 1:
if gt[0] in ans_tokens:
return True
elif " ".join(gt) in ans_phrase:
return True
return False
def answers_match(
baseline: str,
polygen: str,
b_correct: Optional[bool] = None,
p_correct: Optional[bool] = None,
) -> bool:
"""Did polygen preserve the baseline's answer?
With ground truth, this is correctness-agreement: both correct reads as
preserved even when the wording differs ("A cat" vs "It's a cat"); a
correct/incorrect split reads as diverged. Without ground truth (user
uploads) or when both are wrong, fall back to normalized equality then
bidirectional containment. Containment can overmatch on long answers
("a cat in the garden" contains "a cat"); for VQA-short answers that is
fine, and the smoke test confirms longer upload-style answers don't trip a
spurious match. If they do, tighten to token-set Jaccard or first-noun
match.
"""
if b_correct is not None and p_correct is not None:
if b_correct and p_correct:
return True
if b_correct != p_correct:
return False
# both wrong: did they at least produce the same wrong answer?
a, b = _norm_phrase(baseline), _norm_phrase(polygen)
if not a or not b:
return False
if a == b:
return True
return a in b or b in a
# ---------------------------------------------------------------------------
# HTML render.
# ---------------------------------------------------------------------------
def render_result_panel(
*,
baseline_answer: str,
polygen_answer: str,
n_visual_tokens: int,
n_polygen_tokens: int, # measured (worker); <= 0 means fall back to law
splice_mode_resolved: str, # "hook" | "bypass"
tier: int,
storage_compression_x: Optional[float],
ground_truth: Optional[list[str]], # None for uploads
) -> str:
"""Build the combined result panel HTML.
Per-result correctness (the verdict and the per-card chips) is scored
against ``ground_truth`` when present; uploads carry none and are unscored.
"""
b_correct = is_correct(baseline_answer, ground_truth)
p_correct = is_correct(polygen_answer, ground_truth)
matched = answers_match(baseline_answer, polygen_answer, b_correct, p_correct)
storage_x = (
float(storage_compression_x)
if storage_compression_x
else _TIER_STORAGE.get(tier, 0.0)
)
return f"""
""".strip()
def _resolve_n_poly(n_visual: int, n_poly_measured: int, tier: int) -> int:
"""Prefer the worker's measured value; fall back to the length law."""
if n_poly_measured and n_poly_measured > 0:
return int(n_poly_measured)
return bypass_n_polygen_visual_tokens(n_visual, tier)
def _verdict_bar(matched: bool, mode: str) -> str:
if matched:
sub = {
"hook": "Polygen preserved the answer with a length-preserving "
"compressed representation.",
"bypass": "Polygen preserved the answer with a shorter visual sequence.",
}.get(mode, "Polygen preserved the answer with a compressed representation.")
return f"""
✓Same answer{escape(sub)}
"""
return """
⚠Different answerAt this compression tier the two paths
diverged on this input. Surfaced honestly; the per-card chips show which
one matched the ground truth.
"""
def _correctness_chip(correct: Optional[bool]) -> str:
if correct is None:
return ""
if correct:
return '✓ correct'
return '✗ incorrect'
def _baseline_card(answer: str, n_visual: int, correct: Optional[bool]) -> str:
answer_html = (
escape(answer)
if answer
else '(run a comparison)'
)
return f"""Baseline · Qwen2-VL