Add the harness code: Transformers, vLLM and server clients, the schema and the visualizer

caption_visualizer.py

@@ -0,0 +1,574 @@
+"""Self-contained HTML visualization of structured scene descriptions.
+
+`CaptionVisualizer` renders an `SceneDescription` or a `SceneObjectList` beside its image:
+the description fields as collapsible cards, and the geometry (bounding boxes, masks,
+keypoints, text lines) as SVG overlays on the image. Overlays are toggled per object with
+CSS only, so the output needs no JavaScript and no network access, and it works inside a
+`gradio.HTML` component.
+
+    visualizer = CaptionVisualizer()
+    visualizer.save("description.html", image, description)
+"""
+
+from __future__ import annotations
+
+import base64
+import html
+import io
+import json
+import uuid
+from pathlib import Path
+from typing import Any
+
+from data_model import (
+    BaseObject,
+    BodyKeypoint,
+    BoundingBox,
+    Color,
+    GeneralDescription,
+    ObjectMask,
+    Person,
+    SceneDescription,
+    SceneObjectList,
+    Text,
+    _parse_color_token,
+    _parse_coord_token,
+)
+from PIL import Image
+
+#: Overlay colors for objects without a color of their own.
+PALETTE = ("#FF6B6B", "#F7B801", "#6BCB77", "#4D96FF", "#845EC2", "#FF9671", "#00C9A7", "#F9C74F")
+FACE_COLOR = "#FF69B4"
+
+#: Fields shown by dedicated renderers, or as nested objects, instead of as plain values.
+_SPECIAL_FIELDS = frozenset(
+    {
+        "name",
+        "kind",
+        "description",
+        "details",
+        "artifacts",
+        "relationships",
+        "colors",
+        "skin_color",
+        "bounding_boxes",
+        "face_bbox",
+        "keypoints",
+        "mask",
+        "lines",
+        "children",
+        "clothing",
+        "footwear",
+        "accessories",
+        "eyes",
+        "hair",
+        "facial_hair",
+    }
+)
+_NESTED_LISTS = (
+    ("children", "Children"),
+    ("clothing", "Clothing"),
+    ("footwear", "Footwear"),
+    ("accessories", "Accessories"),
+)
+_NESTED_SINGLE = (("eyes", "Eyes"), ("hair", "Hair"), ("facial_hair", "Facial hair"))
+
+_WEBP_QUALITY = 90
+
+#: Scoped with `#ROOT`, which each render replaces with its own id, so several renders can
+#: share one page.
+_FRAGMENT_CSS = """
+#ROOT { --cv-bg: #ffffff; --cv-card: #f6f7f9; --cv-card-head: #eceff4; --cv-text: #15181d; --cv-muted: #5b6270;
+  --cv-border: rgba(120, 130, 150, 0.35); --cv-chip: #e6e9ef; --cv-accent: #3b5bdb;
+  font-family: system-ui, -apple-system, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif;
+  font-size: 14px; line-height: 1.5; color: var(--cv-text); }
+@media (prefers-color-scheme: dark) {
+  :root:not([data-theme="light"]) #ROOT { --cv-bg: #16181d; --cv-card: #1e2127; --cv-card-head: #262a32;
+    --cv-text: #e7e9ee; --cv-muted: #9aa1ae; --cv-border: rgba(160, 170, 190, 0.25); --cv-chip: #2b3039;
+    --cv-accent: #8ea4ff; }
+}
+/* An explicit theme on the page (`data-theme`, or gradio's `.dark`) wins over the system one. */
+[data-theme="dark"] #ROOT, .dark #ROOT { --cv-bg: #16181d; --cv-card: #1e2127; --cv-card-head: #262a32; --cv-text: #e7e9ee; --cv-muted: #9aa1ae;
+  --cv-border: rgba(160, 170, 190, 0.25); --cv-chip: #2b3039; --cv-accent: #8ea4ff; }
+#ROOT, #ROOT * { box-sizing: border-box; }
+#ROOT .cv-form { margin: 0; background: var(--cv-bg); border-radius: 12px; padding: 1rem; }
+#ROOT .cv-toggle { display: none; }
+#ROOT .cv-layout { display: flex; flex-direction: column; gap: 1.25rem; }
+#ROOT .cv-image-panel { width: 100%; min-width: 0; }
+#ROOT .cv-image-box { position: relative; border-radius: 10px; overflow: hidden; background: #0d0f13; }
+/* The pixel grid the coordinates refer to, without EXIF rotation. */
+#ROOT .cv-image { display: block; width: 100%; height: auto; image-orientation: none; }
+#ROOT .cv-overlay { position: absolute; inset: 0; width: 100%; height: 100%; pointer-events: none; }
+#ROOT .cv-overlay g.cv-ov { display: none; }
+/* Keypoint names only for one open person: with all overlays shown they cover the image. */
+#ROOT .cv-overlay g.cv-name-label { display: none; }
+#ROOT .cv-controls { display: flex; flex-wrap: wrap; gap: 0.5rem; margin-top: 0.75rem; }
+#ROOT .cv-button { font: inherit; font-size: 0.82rem; color: var(--cv-text); background: var(--cv-chip);
+  border: 1px solid var(--cv-border); border-radius: 999px; padding: 0.25rem 0.8rem; cursor: pointer;
+  user-select: none; }
+#ROOT .cv-button:hover { border-color: var(--cv-accent); }
+#ROOT .cv-button .cv-on { display: none; }
+#ROOT .cv-tree { display: flex; flex-direction: column; gap: 1rem; min-width: 0; }
+#ROOT .cv-section h3 { margin: 0 0 0.5rem; font-size: 1rem; }
+#ROOT .cv-general { margin: 0; padding: 0; list-style: none; display: flex; flex-direction: column; gap: 0.4rem; }
+#ROOT .cv-object { border: 1px solid var(--cv-border); border-radius: 8px; background: var(--cv-card);
+  margin-bottom: 0.6rem; overflow: hidden; }
+#ROOT .cv-object:last-child { margin-bottom: 0; }
+#ROOT .cv-summary { display: flex; align-items: center; gap: 0.55rem; padding: 0.55rem 0.8rem; cursor: pointer;
+  background: var(--cv-card-head); }
+#ROOT .cv-summary::-webkit-details-marker { display: none; }
+#ROOT .cv-dot { flex: none; width: 11px; height: 11px; border-radius: 50%; box-shadow: 0 0 0 2px var(--cv-bg); }
+#ROOT .cv-kind { flex: none; font-size: 0.7rem; text-transform: uppercase; letter-spacing: 0.08em;
+  color: var(--cv-muted); }
+#ROOT .cv-name { font-weight: 600; overflow-wrap: anywhere; }
+#ROOT .cv-body { padding: 0.55rem 0.8rem 0.7rem; display: flex; flex-direction: column; gap: 0.45rem; }
+#ROOT .cv-field { display: flex; flex-wrap: wrap; align-items: baseline; gap: 0.2rem 0.55rem; }
+#ROOT .cv-label { font-size: 0.7rem; text-transform: uppercase; letter-spacing: 0.08em; font-weight: 600;
+  color: var(--cv-muted); }
+#ROOT .cv-value { flex: 1 1 12rem; min-width: 0; overflow-wrap: anywhere; }
+#ROOT .cv-list { margin: 0; padding-left: 1.1rem; }
+#ROOT .cv-list--plain { list-style: none; padding-left: 0; }
+#ROOT .cv-chips { list-style: none; margin: 0; padding: 0; display: flex; flex-wrap: wrap; gap: 0.3rem; }
+#ROOT .cv-chip { display: inline-flex; align-items: center; gap: 0.35rem; padding: 0.15rem 0.5rem; border-radius: 999px;
+  background: var(--cv-chip); font-size: 0.8rem; }
+#ROOT .cv-swatch { width: 14px; height: 14px; border-radius: 50%; border: 1px solid var(--cv-border); }
+#ROOT .cv-sub > h4 { margin: 0.2rem 0 0.35rem; font-size: 0.85rem; }
+#ROOT .cv-empty { margin: 0; color: var(--cv-muted); font-style: italic; }
+@media (min-width: 980px) {
+  #ROOT .cv-layout { flex-direction: row; align-items: flex-start; }
+  #ROOT .cv-image-panel { flex: 1 1 55%; position: sticky; top: 0.5rem; }
+  /* A tall image fits the screen height, so it stays in view beside a long description. */
+  #ROOT .cv-image-box { width: min(100%, calc((100vh - 5rem) * var(--cv-ar, 1))); margin: 0 auto; }
+  #ROOT .cv-tree { flex: 1 1 45%; }
+}
+"""
+
+_PAGE_CSS = """
+:root { color-scheme: light dark; }
+body { margin: 0; padding: 1.25rem 1rem 2.5rem; background: #eef0f4; color: #15181d;
+  font-family: system-ui, -apple-system, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif; }
+@media (prefers-color-scheme: dark) { body { background: #0d0f13; color: #e7e9ee; } }
+.cv-page { max-width: 1400px; margin: 0 auto; }
+.cv-page h1 { margin: 0 0 1rem; font-size: 1.2rem; }
+"""
+
+
+class CaptionVisualizer:
+    """Renders structured scene descriptions as HTML with image overlays."""
+
+    def render(
+        self,
+        image: Image.Image,
+        caption: SceneDescription | SceneObjectList | dict | str,
+        title: str | None = None,
+    ) -> str:
+        """A standalone HTML page for `caption` over `image`."""
+        heading = f"<h1>{html.escape(title)}</h1>" if title else ""
+        page_title = html.escape(title or "Structured scene description")
+        return (
+            "<!DOCTYPE html>\n"
+            '<html lang="en">\n<head>\n<meta charset="utf-8"/>\n'
+            '<meta name="viewport" content="width=device-width, initial-scale=1"/>\n'
+            f"<title>{page_title}</title>\n<style>{_PAGE_CSS}</style>\n</head>\n<body>\n"
+            f'<div class="cv-page">{heading}{self.render_fragment(image, caption)}</div>\n'
+            "</body>\n</html>\n"
+        )
+
+    def render_fragment(
+        self,
+        image: Image.Image,
+        caption: SceneDescription | SceneObjectList | dict | str,
+        title: str | None = None,
+    ) -> str:
+        """The visualization as an HTML fragment with its own scoped styles, for embedding."""
+        caption = self._parse(caption)
+        return _Render(image, caption, title).html()
+
+    def save(
+        self,
+        path: str | Path,
+        image: Image.Image,
+        caption: SceneDescription | SceneObjectList | dict | str,
+        title: str | None = None,
+    ) -> Path:
+        """Write the standalone page to `path` and return the path."""
+        path = Path(path)
+        path.write_text(self.render(image, caption, title), encoding="utf-8")
+        return path
+
+    @staticmethod
+    def _parse(caption: Any) -> SceneDescription | SceneObjectList:
+        if isinstance(caption, (SceneDescription, SceneObjectList)):
+            return caption
+        data = json.loads(caption) if isinstance(caption, (str, bytes)) else caption
+        if not isinstance(data, dict):
+            raise TypeError(f"unsupported caption type {type(caption).__name__}")
+        if "general" in data:
+            return SceneDescription.model_validate(data)
+        return SceneObjectList.model_validate(data)
+
+
+def _coord(value: Any) -> float:
+    return _parse_coord_token(value)
+
+
+def _rgb(color: Color) -> tuple[int, int, int]:
+    return _parse_color_token(color.red), _parse_color_token(color.green), _parse_color_token(color.blue)
+
+
+def _box(bbox: BoundingBox) -> tuple[float, float, float, float]:
+    """(left, top, right, bottom) in percent, corners ordered."""
+    x0, y0 = _coord(bbox.top_left.x), _coord(bbox.top_left.y)
+    x1, y1 = _coord(bbox.bottom_right.x), _coord(bbox.bottom_right.y)
+    return min(x0, x1), min(y0, y1), max(x0, x1), max(y0, y1)
+
+
+def _number(value: float) -> str:
+    return f"{value:.1f}"
+
+
+def _format_value(value: Any) -> str:
+    if isinstance(value, list):
+        if not value:
+            return ""
+        return '<ul class="cv-list">' + "".join(f"<li>{_format_value(item)}</li>" for item in value) + "</ul>"
+    if isinstance(value, dict):
+        items = "".join(
+            f'<li><span class="cv-label">{html.escape(str(key))}</span> {_format_value(item)}</li>'
+            for key, item in value.items()
+        )
+        return f'<ul class="cv-list cv-list--plain">{items}</ul>'
+    if isinstance(value, float):
+        return f"{value:.2f}".rstrip("0").rstrip(".")
+    return html.escape(str(value))
+
+
+def _image_data_url(image: Image.Image) -> str:
+    buffer = io.BytesIO()
+    image.convert("RGB").save(buffer, format="WEBP", quality=_WEBP_QUALITY)
+    return "data:image/webp;base64," + base64.b64encode(buffer.getvalue()).decode("ascii")
+
+
+class _Render:
+    """One render: collects the overlay groups while it walks the caption tree."""
+
+    def __init__(self, image: Image.Image, caption: SceneDescription | SceneObjectList, title: str | None):
+        self.image = image
+        self.caption = caption
+        self.title = title
+        self.root = f"cv-{uuid.uuid4().hex[:12]}"
+        self.width, self.height = image.size
+        self.unit = min(self.width, self.height) / 100.0
+        self.toggles: list[tuple[str, str, str]] = []  # (toggle id, overlay id, color)
+        self.overlays: list[str] = []
+        self.counter = 0
+
+    # -- page ------------------------------------------------------------------
+
+    def html(self) -> str:
+        if isinstance(self.caption, SceneDescription):
+            general = self._general(self.caption.general)
+            objects = self._section("Objects", self.caption.objects or [])
+        else:
+            general = ""
+            objects = self._section("Objects", self.caption.objects or [])
+        tree = general + objects or '<p class="cv-empty">The scene description has no content.</p>'
+
+        all_toggle = f"{self.root}-all"
+        inputs = f'<input type="checkbox" class="cv-toggle" id="{all_toggle}"/>' if self.toggles else ""
+        inputs += "".join(f'<input type="checkbox" class="cv-toggle" id="{toggle}"/>' for toggle, _, _ in self.toggles)
+        controls = ""
+        if self.toggles:
+            controls = (
+                '<div class="cv-controls">'
+                f'<label class="cv-button" for="{all_toggle}">'
+                '<span class="cv-off">Show all overlays</span><span class="cv-on">Hide all overlays</span></label>'
+                '<button type="reset" class="cv-button">Clear</button>'
+                "</div>"
+            )
+        overlay = (
+            f'<svg class="cv-overlay" viewBox="0 0 {self.width} {self.height}" preserveAspectRatio="none" '
+            f'xmlns="http://www.w3.org/2000/svg" aria-hidden="true">{"".join(self.overlays)}</svg>'
+        )
+        heading = f'<h3 class="cv-title">{html.escape(self.title)}</h3>' if self.title else ""
+        return (
+            f'<section id="{self.root}" class="cv-root">'
+            f"<style>{_FRAGMENT_CSS.replace('#ROOT', '#' + self.root)}{self._toggle_css(all_toggle)}</style>"
+            f'<form class="cv-form">{inputs}{heading}'
+            '<div class="cv-layout">'
+            f'<div class="cv-image-panel"><div class="cv-image-box" style="--cv-ar: {self.width / self.height:.4f}">'
+            f'<img class="cv-image" src="{_image_data_url(self.image)}" alt="described image"/>{overlay}'
+            f"</div>{controls}</div>"
+            f'<div class="cv-tree">{tree}</div>'
+            "</div></form></section>"
+        )
+
+    def _toggle_css(self, all_toggle: str) -> str:
+        root = f"#{self.root}"
+        rules = [
+            f"{root} input#{all_toggle}:checked ~ .cv-layout .cv-overlay g.cv-ov {{ display: inline; }}",
+            f'{root} input#{all_toggle}:checked ~ .cv-layout label[for="{all_toggle}"] .cv-off {{ display: none; }}',
+            f'{root} input#{all_toggle}:checked ~ .cv-layout label[for="{all_toggle}"] .cv-on {{ display: inline; }}',
+        ]
+        for toggle, overlay_id, color in self.toggles:
+            label = f'{root} input#{toggle}:checked ~ .cv-layout label[for="{toggle}"]'
+            group = f'{root} input#{toggle}:checked ~ .cv-layout .cv-overlay g[data-id="{overlay_id}"]'
+            rules += [
+                f"{group} {{ display: inline; }}",
+                f"{group} .cv-name-label {{ display: inline; }}",
+                f"{label} {{ border-color: {color}; box-shadow: inset 0 0 0 1px {color}; }}",
+                f"{label} .cv-off {{ display: none; }}",
+                f"{label} .cv-on {{ display: inline; }}",
+            ]
+        return "\n".join(rules)
+
+    # -- sections --------------------------------------------------------------
+
+    def _general(self, general: GeneralDescription | None) -> str:
+        if general is None:
+            return ""
+        data = general.model_dump(exclude_none=True)
+        items = []
+        if general.color_palette:
+            data.pop("color_palette", None)
+            items.append(self._field("Color palette", self._chips(general.color_palette), tag="li"))
+        for key, value in data.items():
+            items.append(self._field(key.replace("_", " ").capitalize(), _format_value(value), tag="li"))
+        if not items:
+            return ""
+        return f'<section class="cv-section"><h3>General</h3><ul class="cv-general">{"".join(items)}</ul></section>'
+
+    def _section(self, title: str, objects: list[BaseObject]) -> str:
+        if not objects:
+            return ""
+        body = "".join(self._object(obj, depth=0) for obj in objects)
+        return f'<section class="cv-section"><h3>{html.escape(title)} ({len(objects)})</h3>{body}</section>'
+
+    # -- objects ---------------------------------------------------------------
+
+    def _object(self, obj: BaseObject, depth: int) -> str:
+        self.counter += 1
+        obj_id = f"{self.root}-o{self.counter}"
+        color = self._object_color(obj)
+        line_text = getattr(obj, "text", None)
+        if line_text is None and getattr(obj, "lines", None):
+            # A text object has no name; its lines say more than its description.
+            line_text = " / ".join(line.text for line in obj.lines if line.text)
+            if len(line_text) > 64:
+                line_text = line_text[:61] + "..."
+        name = obj.name or ""
+        if not name and line_text:
+            name = f"“{line_text}”"
+        if not name and obj.description:
+            snippet = obj.description.strip()
+            name = snippet if len(snippet) <= 64 else snippet[:61] + "..."
+        name = name or obj.kind.replace("_", " ")
+
+        fields = []
+        if obj.description and name != obj.description.strip():
+            fields.append(self._field("Description", html.escape(obj.description)))
+        if line_text is not None and obj.name:
+            fields.append(self._field("Text", html.escape(line_text)))
+        for label, key in (("Details", "details"), ("Artifacts", "artifacts")):
+            if getattr(obj, key, None):
+                fields.append(self._field(label, _format_value(getattr(obj, key))))
+        if obj.relationships:
+            items = "".join(
+                f"<li><strong>{html.escape(rel.relationship)}</strong> → {html.escape(rel.to)}</li>"
+                for rel in obj.relationships
+            )
+            fields.append(self._field("Relationships", f'<ul class="cv-list">{items}</ul>'))
+        if obj.colors:
+            fields.append(self._field("Colors", self._chips(obj.colors)))
+        if isinstance(obj, Person) and obj.skin_color:
+            fields.append(self._field("Skin color", self._chips([obj.skin_color])))
+
+        data = obj.model_dump(exclude_none=True)
+        for key, value in data.items():
+            if key not in _SPECIAL_FIELDS:
+                fields.append(self._field(key.replace("_", " ").capitalize(), _format_value(value)))
+
+        fields += self._geometry_fields(obj, obj_id, color)
+
+        nested = []
+        for key, title in _NESTED_SINGLE:
+            item = getattr(obj, key, None)
+            if item is not None:
+                nested.append(self._subsection(title, self._object(item, depth + 1)))
+        for key, title in _NESTED_LISTS:
+            items = getattr(obj, key, None) or []
+            if items:
+                nested.append(self._subsection(title, "".join(self._object(item, depth + 1) for item in items)))
+        if isinstance(obj, Text) and obj.lines:
+            nested.append(self._subsection("Lines", "".join(self._object(line, depth + 1) for line in obj.lines)))
+
+        body = "".join(fields + nested) or '<p class="cv-empty">No other fields.</p>'
+        open_attr = " open" if depth == 0 else ""
+        return (
+            f'<details class="cv-object"{open_attr}>'
+            '<summary class="cv-summary">'
+            f'<span class="cv-dot" style="background:{color}"></span>'
+            f'<span class="cv-kind">{html.escape(obj.kind.replace("_", " "))}</span>'
+            f'<span class="cv-name">{html.escape(name)}</span>'
+            "</summary>"
+            f'<div class="cv-body">{body}</div>'
+            "</details>"
+        )
+
+    def _geometry_fields(self, obj: BaseObject, obj_id: str, color: str) -> list[str]:
+        fields = []
+        boxes = obj.bounding_boxes or []
+        if boxes:
+            shapes = "".join(
+                self._box_shape(bbox, color, index + 1 if len(boxes) > 1 else None) for index, bbox in enumerate(boxes)
+            )
+            items = "".join("<li>({}, {}) → ({}, {})</li>".format(*(_number(v) for v in _box(bbox))) for bbox in boxes)
+            title = "Bounding boxes" if len(boxes) > 1 else "Bounding box"
+            fields.append(
+                self._overlay_field(
+                    title, f"{obj_id}-box", color, shapes, f'<ul class="cv-list cv-list--plain">{items}</ul>'
+                )
+            )
+        if isinstance(obj, Person) and obj.face_bbox is not None:
+            left, top, right, bottom = _box(obj.face_bbox)
+            fields.append(
+                self._overlay_field(
+                    "Face",
+                    f"{obj_id}-face",
+                    FACE_COLOR,
+                    self._box_shape(obj.face_bbox, FACE_COLOR, None),
+                    f"({_number(left)}, {_number(top)}) → ({_number(right)}, {_number(bottom)})",
+                )
+            )
+        if isinstance(obj, Person) and obj.keypoints:
+            shapes = "".join(self._keypoint_shape(point, color) for point in obj.keypoints)
+            items = "".join(
+                f"<li>{html.escape(point.name)} ({_number(_coord(point.point.x))}, "
+                f"{_number(_coord(point.point.y))})</li>"
+                for point in obj.keypoints
+            )
+            fields.append(
+                self._overlay_field(
+                    f"Keypoints ({len(obj.keypoints)})",
+                    f"{obj_id}-kp",
+                    color,
+                    shapes,
+                    f'<details><summary>list</summary><ul class="cv-list cv-list--plain">{items}</ul></details>',
+                )
+            )
+        if obj.mask is not None:
+            shape, summary = self._mask_shape(obj.mask, color)
+            if shape:
+                fields.append(self._overlay_field("Mask", f"{obj_id}-mask", color, shape, summary))
+            else:
+                fields.append(self._field("Mask", summary))
+        return fields
+
+    def _overlay_field(self, label: str, overlay_id: str, color: str, shapes: str, content: str) -> str:
+        toggle = f"{overlay_id}-t"
+        self.toggles.append((toggle, overlay_id, color))
+        self.overlays.append(f'<g class="cv-ov" data-id="{overlay_id}">{shapes}</g>')
+        button = (
+            f'<label class="cv-button" for="{toggle}">'
+            '<span class="cv-off">Show</span><span class="cv-on">Hide</span></label>'
+        )
+        return (
+            '<div class="cv-field">'
+            f'<span class="cv-label">{html.escape(label)}</span>{button}'
+            f'<div class="cv-value">{content}</div>'
+            "</div>"
+        )
+
+    # -- overlay shapes (viewBox is the image in pixels) -----------------------
+
+    def _px(self, x: float, y: float) -> tuple[float, float]:
+        return x / 100.0 * self.width, y / 100.0 * self.height
+
+    def _box_shape(self, bbox: BoundingBox, color: str, index: int | None) -> str:
+        left, top, right, bottom = _box(bbox)
+        x0, y0 = self._px(left, top)
+        x1, y1 = self._px(right, bottom)
+        width, height = max(x1 - x0, 1.0), max(y1 - y0, 1.0)
+        stroke = max(self.unit * 0.45, 1.5)
+        # A white casing under the colored line keeps boxes visible on pixels of the same color.
+        shape = (
+            f'<rect x="{x0:.1f}" y="{y0:.1f}" width="{width:.1f}" height="{height:.1f}" fill="none" '
+            f'stroke="#ffffff" stroke-opacity="0.9" stroke-width="{stroke * 2:.1f}"/>'
+            f'<rect x="{x0:.1f}" y="{y0:.1f}" width="{width:.1f}" height="{height:.1f}" fill="{color}" '
+            f'fill-opacity="0.08" stroke="{color}" stroke-width="{stroke:.1f}"/>'
+        )
+        if index is not None:
+            shape += self._label(x0 + stroke, y0 + stroke, str(index))
+        return shape
+
+    def _keypoint_shape(self, point: BodyKeypoint, color: str) -> str:
+        x, y = self._px(_coord(point.point.x), _coord(point.point.y))
+        radius = max(self.unit * 0.7, 2.5)
+        return (
+            f'<circle cx="{x:.1f}" cy="{y:.1f}" r="{radius:.1f}" fill="{color}" stroke="#ffffff" '
+            f'stroke-width="{radius * 0.45:.1f}"><title>{html.escape(point.name)}</title></circle>'
+            '<g class="cv-name-label">'
+            + self._label(x + radius * 1.6, y - radius * 1.6, point.name, small=True)
+            + "</g>"
+        )
+
+    def _label(self, x: float, y: float, text: str, small: bool = False) -> str:
+        size = max(self.unit * (2.0 if small else 2.6), 9.0)
+        return (
+            f'<text x="{x:.1f}" y="{y + size:.1f}" font-size="{size:.1f}" font-family="system-ui, sans-serif" '
+            f'font-weight="600" fill="#111418" stroke="#ffffff" stroke-width="{size * 0.25:.1f}" '
+            f'paint-order="stroke">{html.escape(text)}</text>'
+        )
+
+    def _mask_shape(self, mask: ObjectMask, color: str) -> tuple[str, str]:
+        rings = []
+        for ring in mask.polygons or []:
+            values = [_coord(value) for value in ring]
+            points = [self._px(x, y) for x, y in zip(values[0::2], values[1::2], strict=False)]
+            if len(points) >= 3:
+                rings.append("M" + " L".join(f"{x:.1f},{y:.1f}" for x, y in points) + " Z")
+        if not rings:
+            return "", "empty mask"
+        vertices = sum(len(ring) // 2 for ring in mask.polygons or [])
+        stroke = max(self.unit * 0.35, 1.2)
+        path = " ".join(rings)
+        shape = (
+            f'<path d="{path}" fill="none" stroke="#ffffff" stroke-opacity="0.9" stroke-width="{stroke * 2:.1f}"/>'
+            f'<path d="{path}" fill="{color}" fill-opacity="0.35" fill-rule="evenodd" stroke="{color}" '
+            f'stroke-width="{stroke:.1f}"/>'
+        )
+        return shape, f"polygon mask, {len(rings)} ring(s), {vertices} vertices"
+
+    # -- small parts -----------------------------------------------------------
+
+    @staticmethod
+    def _field(label: str, content: str, tag: str = "div") -> str:
+        return (
+            f'<{tag} class="cv-field"><span class="cv-label">{html.escape(label)}</span>'
+            f'<div class="cv-value">{content}</div></{tag}>'
+        )
+
+    @staticmethod
+    def _subsection(title: str, content: str) -> str:
+        return f'<div class="cv-sub"><h4>{html.escape(title)}</h4>{content}</div>'
+
+    @staticmethod
+    def _chips(colors: list[Color]) -> str:
+        chips = []
+        for color in colors:
+            red, green, blue = _rgb(color)
+            label = f"{html.escape(color.name)} · " if color.name else ""
+            chips.append(
+                f'<li class="cv-chip"><span class="cv-swatch" style="background:rgb({red},{green},{blue})"></span>'
+                f"{label}rgb({red}, {green}, {blue})</li>"
+            )
+        return f'<ul class="cv-chips">{"".join(chips)}</ul>'
+
+    def _object_color(self, obj: BaseObject) -> str:
+        primary = (obj.colors or [None])[0]
+        if primary is None and isinstance(obj, Person):
+            primary = obj.skin_color
+        if primary is not None:
+            red, green, blue = _rgb(primary)
+            return f"rgb({red},{green},{blue})"
+        return PALETTE[self.counter % len(PALETTE)]

data_model.py

@@ -0,0 +1,640 @@
+from collections.abc import Iterator
+from typing import Literal
+
+import numpy as np
+from PIL import Image, ImageDraw
+from pydantic import BaseModel, Field
+
+#: Attributes that can hold nested scene objects.
+_NESTED_LIST_FIELDS = ("children", "clothing", "footwear", "accessories")
+_NESTED_SINGLE_FIELDS = ("eyes", "hair", "facial_hair")
+
+
+def _iter_nested(object, seen=None):
+    """Yield `object` and every scene object nested below it, parent first."""
+    if object is None:
+        return
+    if seen is None:
+        seen = set()
+    if id(object) in seen:
+        return
+    seen.add(id(object))
+    yield object
+    for field in _NESTED_LIST_FIELDS:
+        for item in getattr(object, field, None) or []:
+            yield from _iter_nested(item, seen)
+    for field in _NESTED_SINGLE_FIELDS:
+        yield from _iter_nested(getattr(object, field, None), seen)
+
+
+def _format_coord_token(value: float) -> str:
+    return f"<coord_{round(float(value), 1):.1f}>"
+
+
+def _parse_coord_token(value: float | str) -> float:
+    if isinstance(value, str) and value.startswith("<coord_") and value.endswith(">"):
+        return float(value[len("<coord_") : -1])
+    return float(value)
+
+
+def _format_color_token(value: int) -> int | str:
+    if value < 0 or value > 255:
+        raise ValueError("Color channels must be between 0 and 255")
+    return f"<color_{value}>"
+
+
+def _parse_color_token(value: int | str) -> int:
+    if isinstance(value, str) and value.startswith("<color_") and value.endswith(">"):
+        return int(value[len("<color_") : -1])
+    return int(value)
+
+
+class Color(BaseModel):
+    name: str | None = Field(None)
+    red: int | str = Field(...)
+    green: int | str = Field(...)
+    blue: int | str = Field(...)
+
+    def post_process(self):
+        if isinstance(self.red, str):
+            self.red = _parse_color_token(self.red)
+        if isinstance(self.green, str):
+            self.green = _parse_color_token(self.green)
+        if isinstance(self.blue, str):
+            self.blue = _parse_color_token(self.blue)
+
+
+class Point(BaseModel):
+    x: float | str = Field(...)
+    y: float | str = Field(...)
+
+    def post_process(self):
+        if isinstance(self.x, str):
+            self.x = _parse_coord_token(self.x)
+        if isinstance(self.y, str):
+            self.y = _parse_coord_token(self.y)
+
+
+class BoundingBox(BaseModel):
+    top_left: Point = Field(...)
+    bottom_right: Point = Field(...)
+
+    def post_process(self):
+        self.top_left.post_process()
+        self.bottom_right.post_process()
+
+
+class BodyKeypoint(BaseModel):
+    name: str = Field(..., description="The name of the keypoint")
+    point: Point = Field(...)
+
+    def post_process(self):
+        self.point.post_process()
+
+
+class ObjectMask(BaseModel):
+    polygons: list[list[float | str]] | None = Field(None)
+
+    def post_process(self):
+        if self.polygons is not None:
+            self.polygons = [[_parse_coord_token(value) for value in ring] for ring in self.polygons]
+
+    def decode(self, image_width: int, image_height: int) -> np.ndarray:
+        if image_width <= 0 or image_height <= 0:
+            raise ValueError("Image width and height must be positive")
+
+        canvas = Image.new("L", (image_width, image_height), 0)
+        draw = ImageDraw.Draw(canvas)
+        for ring in self.polygons or []:
+            values = [_parse_coord_token(value) for value in ring]
+            points = [(x * image_width / 100.0, y * image_height / 100.0) for x, y in zip(values[0::2], values[1::2])]
+            if len(points) >= 3:
+                draw.polygon(points, fill=255)
+        return np.asarray(canvas) >= 128
+
+
+class Relationship(BaseModel):
+    to: str = Field(...)
+    relationship: str = Field(...)
+
+
+class BaseObject(BaseModel):
+    name: str = Field(
+        "",
+        description="Short description of the object",
+    )
+    kind: Literal["base_object", "object", "person", "animal", "text"] = "base_object"
+    description: str | None = Field(
+        None,
+        description="A more detailed description of the object",
+    )
+    location: str | None = Field(None)
+    position: Literal["foreground", "midground", "background"] | None = Field(None)
+    relative_size: Literal["small", "medium", "large"] | None = Field(None)
+    bounding_boxes: list[BoundingBox] | None = Field(None)
+    num_objects: int | None = Field(None)
+    relative_distance: float | None = Field(
+        None,
+        ge=0,
+        le=100,
+        description="The relative distance of the object from the camera, 0 is at the camera, 100 is at the far edge of the image, None if not applicable",
+    )
+    colors: list[Color] | None = Field(None)
+    blur: str | None = Field(
+        None,
+        description="Description of the blurring of the object, if any",
+    )
+    details: list[str] | None = Field(
+        None,
+        description="Additional details about the object",
+    )
+    relationships: list[Relationship] | None = Field(None)
+    mask: ObjectMask | None = Field(None)
+    artifacts: list[str] | None = Field(
+        None,
+        description="Visual issues with the object",
+    )
+
+    def make_full_mask(self):
+        self.location = None
+        self.position = None
+        self.relative_size = None
+
+    def make_full(self):
+        self.make_full_mask()
+        self.mask = None
+
+    def make_long(self):
+        self.make_full()
+        self.add_location()
+        self.add_position()
+        self.add_relative_size()
+        self.bounding_boxes = None
+        self.relative_distance = None
+        self.mask = None
+
+    def make_short(self):
+        self.make_long()
+        self.relationships = None
+        self.details = None
+        self.colors = None
+        self.blur = None
+        self.artifacts = None
+
+    def drop_masks(self):
+        """Clear `mask` on this object and every object nested below it."""
+        for object in _iter_nested(self):
+            object.mask = None
+
+    def _keep_only(self, fields_to_keep: set[str]) -> None:
+        for field_name in type(self).model_fields:
+            if field_name not in fields_to_keep:
+                setattr(self, field_name, None)
+
+    def make_for_detect(self):
+        self._keep_only({"name", "kind", "bounding_boxes"})
+
+    def make_for_mask(self):
+        self._keep_only({"name", "kind", "bounding_boxes", "mask"})
+
+    def make_for_keypoints(self):
+        self._keep_only({"name", "kind", "keypoints", "bounding_boxes"})
+
+    def _box_extents(self) -> list[tuple[float, float, float, float]]:
+        """(left, top, right, bottom) of each bounding box, in 0-100 coordinates."""
+        extents = []
+        for bounding_box in self.bounding_boxes or []:
+            left, right = sorted(
+                (_parse_coord_token(bounding_box.top_left.x), _parse_coord_token(bounding_box.bottom_right.x))
+            )
+            top, bottom = sorted(
+                (_parse_coord_token(bounding_box.top_left.y), _parse_coord_token(bounding_box.bottom_right.y))
+            )
+            extents.append((left, top, right, bottom))
+        return extents
+
+    def add_location(self):
+        """Set `location` from the joint extent of the bounding boxes, on a grid of thirds."""
+        extents = self._box_extents()
+        if not extents:
+            return self
+        left = min(extent[0] for extent in extents)
+        top = min(extent[1] for extent in extents)
+        right = max(extent[2] for extent in extents)
+        bottom = max(extent[3] for extent in extents)
+        if left <= 1 and top <= 1 and right >= 99 and bottom >= 99:
+            self.location = "full frame"
+            return self
+
+        third, two_thirds, edge = 100 / 3, 200 / 3, 1.0
+
+        def classify(low: float, high: float) -> str:
+            # "full" is an object that spans the axis, such as sky or ground.
+            if low < third and high > two_thirds and high - low > 50:
+                return "full"
+            center = (low + high) / 2
+            if center < third:
+                return "low"
+            if center > two_thirds:
+                return "high"
+            return "mid"
+
+        horizontal, vertical = classify(left, right), classify(top, bottom)
+        h_name = {"low": "left", "mid": "center", "high": "right"}.get(horizontal)
+        v_name = {"low": "top", "mid": "middle", "high": "bottom"}.get(vertical)
+        if horizontal == "full" and vertical == "full":
+            location = "center"
+        elif horizontal == "full":
+            location = {"low": "top", "high": "bottom", "mid": "center"}[vertical]
+        elif vertical == "full":
+            location = {"low": "left", "high": "right", "mid": "center"}[horizontal]
+        elif horizontal == "low" and vertical == "low" and left <= edge and top <= edge:
+            location = "top-left corner"
+        elif horizontal == "high" and vertical == "low" and right >= 100 - edge and top <= edge:
+            location = "top-right corner"
+        elif horizontal == "low" and vertical == "high" and left <= edge and bottom >= 100 - edge:
+            location = "bottom-left corner"
+        elif horizontal == "high" and vertical == "high" and right >= 100 - edge and bottom >= 100 - edge:
+            location = "bottom-right corner"
+        elif horizontal == "mid" and vertical == "mid":
+            location = "center"
+        elif vertical == "mid":
+            location = f"{h_name} of center"
+        elif horizontal == "mid":
+            location = f"{v_name} center"
+        else:
+            location = f"{v_name} {h_name}"
+        self.location = location
+        return self
+
+    def add_position(self):
+        """Set `position` from `relative_distance`, in thirds of the depth."""
+        if self.relative_distance is None:
+            return self
+        if self.relative_distance < 100 / 3:
+            self.position = "foreground"
+        elif self.relative_distance < 200 / 3:
+            self.position = "midground"
+        else:
+            self.position = "background"
+        return self
+
+    def add_relative_size(self):
+        """Set `relative_size` from the summed area of the bounding boxes."""
+        extents = self._box_extents()
+        if not extents:
+            return self
+        ratio = sum((right - left) * (bottom - top) for left, top, right, bottom in extents) / (100.0 * 100.0)
+        if ratio < 0.05:
+            self.relative_size = "small"
+        elif ratio < 0.25:
+            self.relative_size = "medium"
+        else:
+            self.relative_size = "large"
+        return self
+
+    def post_process(self):
+        if self.bounding_boxes is not None:
+            for bbox in self.bounding_boxes:
+                bbox.post_process()
+        if self.colors is not None:
+            for color in self.colors:
+                color.post_process()
+        if self.mask is not None:
+            self.mask.post_process()
+
+
+class Object(BaseObject):
+    kind: Literal["object"] = "object"
+    children: list["SceneObject"] | None = Field(None)
+    material: str | None = Field(None)
+    texture: str | None = Field(None)
+    transparency: float | None = Field(
+        None,
+        ge=0,
+        le=100,
+        description="The transparency of the object, 0 is fully transparent, 100 is fully opaque",
+    )
+
+    def make_long(self):
+        super().make_long()
+        self.children = None
+
+    def make_short(self):
+        super().make_short()
+        self.material = None
+        self.texture = None
+        self.transparency = None
+
+    def post_process(self):
+        super().post_process()
+        if self.children is not None:
+            for child in self.children:
+                child.post_process()
+
+
+class TextLine(BaseObject):
+    text: str = Field(...)
+
+
+class Text(BaseObject):
+    kind: Literal["text"] = "text"
+    language: str | None = Field(None, pattern=r"^[A-Za-z]{2,3}(-[A-Za-z]{2})?$")
+    lines: list[TextLine] = Field(...)
+    font_family: str | None = Field(None)
+    font_description: str | None = Field(None)
+    font_weight: Literal["100", "200", "300", "400", "500", "600", "700", "800", "900"] | None = Field(None)
+    font_style: Literal["normal", "italic", "oblique"] | None = Field(None)
+    text_case: Literal["lowercase", "uppercase", "title", "sentence", "mixed"] | None = Field(None)
+    text_orientation: Literal["horizontal", "vertical"] | None = Field(None)
+    text_alignment: Literal["left", "center", "right", "justify"] | None = Field(None)
+    text_decoration: Literal["none", "underline", "line-through", "overline", "shadow"] | None = Field(None)
+    rotation: float | None = Field(None, ge=-180, le=180)
+
+    def make_long(self):
+        super().make_long()
+        for line in self.lines:
+            line.make_long()
+
+    def make_short(self):
+        super().make_short()
+        self.language = None
+        self.font_family = None
+        self.font_description = None
+        self.font_weight = None
+        self.font_style = None
+        self.text_case = None
+        self.text_orientation = None
+        self.text_alignment = None
+        self.text_decoration = None
+        self.rotation = None
+        for line in self.lines:
+            line.make_short()
+
+    def post_process(self):
+        super().post_process()
+        for line in self.lines:
+            line.post_process()
+
+
+class Person(BaseObject):
+    kind: Literal["person"] = "person"
+    age: str | None = Field(None)
+    gender: str | None = Field(None)
+    ethnicity: str | None = Field(None)
+    skin_color: Color | None = Field(None)
+    keypoints: list[BodyKeypoint] | None = Field(None)
+    activity: str | None = Field(None)
+    face_bbox: BoundingBox | None = Field(None)
+    expression: str | None = Field(None)
+    eyes: Object | None = Field(None, description="Description of the eyes")
+    mouth: str | None = Field(None, description="Description of the mouth")
+    nose: str | None = Field(None, description="Description of the nose")
+    hair: Object | None = Field(None, description="Description of the hair")
+    ears: str | None = Field(None, description="Description of the ears")
+    facial_hair: Object | None = Field(
+        None,
+        description="Description of the facial hair",
+    )
+    body: str | None = Field(None)
+    left_hand: str | None = Field(None)
+    right_hand: str | None = Field(None)
+    left_leg: str | None = Field(None)
+    right_leg: str | None = Field(None)
+    clothing: list[Object] | None = Field(None)
+    footwear: list[Object] | None = Field(None)
+    accessories: list[Object] | None = Field(None)
+    children: list[Object | Text] | None = Field(None)
+
+    def make_long(self):
+        super().make_long()
+        self.face_bbox = None
+        self.keypoints = None
+        self.children = None
+
+    def make_short(self):
+        super().make_short()
+        self.skin_color = None
+        self.activity = None
+        self.expression = None
+        self.eyes = None
+        self.mouth = None
+        self.nose = None
+        self.hair = None
+        self.ears = None
+        self.facial_hair = None
+        self.body = None
+        self.left_hand = None
+        self.right_hand = None
+        self.left_leg = None
+        self.right_leg = None
+        self.clothing = None
+        self.footwear = None
+        self.accessories = None
+        self.face_bbox = None
+        self.children = None
+
+    def post_process(self):
+        super().post_process()
+        if self.keypoints is not None:
+            for keypoint in self.keypoints:
+                keypoint.post_process()
+        if self.face_bbox is not None:
+            self.face_bbox.post_process()
+        if self.skin_color is not None:
+            self.skin_color.post_process()
+        if self.facial_hair is not None:
+            self.facial_hair.post_process()
+        if self.hair is not None:
+            self.hair.post_process()
+        if self.eyes is not None:
+            self.eyes.post_process()
+        if self.clothing is not None:
+            for clothing in self.clothing:
+                clothing.post_process()
+        if self.footwear is not None:
+            for footwear in self.footwear:
+                footwear.post_process()
+        if self.accessories is not None:
+            for accessory in self.accessories:
+                accessory.post_process()
+        if self.children is not None:
+            for child in self.children:
+                child.post_process()
+
+
+class Animal(BaseObject):
+    kind: Literal["animal"] = "animal"
+    children: list[Object | Text] | None = Field(None)
+    head: str | None = Field(None)
+    body: str | None = Field(None)
+    legs: str | None = Field(None)
+    hands: str | None = Field(None)
+    tail: str | None = Field(None)
+    wings: str | None = Field(None)
+
+    def make_long(self):
+        super().make_long()
+
+    def make_short(self):
+        super().make_short()
+        self.head = None
+        self.body = None
+        self.legs = None
+        self.hands = None
+        self.tail = None
+        self.wings = None
+
+    def post_process(self):
+        super().post_process()
+        if self.children is not None:
+            for child in self.children:
+                child.post_process()
+
+
+SceneObject = Object | Person | Animal | Text | BaseObject
+
+Object.model_rebuild()
+
+
+class SceneObjectList(BaseModel):
+    objects: list[SceneObject] = Field(...)
+
+    def post_process(self):
+        if self.objects is not None:
+            for object in self.objects:
+                object.post_process()
+
+
+class GeneralDescription(BaseModel):
+    user_prompt: str | None = Field(None, description="The user's prompt for the image")
+    image_type: str | None = Field(None, description="The type of the image")
+    style: str | None = Field(None, description="The style of the image")
+    medium: str | None = Field(None)
+    color_palette: list[Color] | None = Field(None)
+    aesthetic_score: float | None = Field(None, ge=0, le=10, description="The aesthetic score of the image")
+    atmosphere: str | None = Field(None, description="The overall feeling or emotion the image evokes")
+    setting: str | None = Field(None, description="The environment or context where the image takes place")
+    lighting: str | None = Field(None)
+    composition: str | None = Field(None)
+    depth_of_field: str | None = Field(None)
+    viewpoint: str | None = Field(None)
+    intent: str | None = Field(None, description="The purpose of the image")
+    description: str | None = Field(None, description="A short and comprehensive description of the image")
+    artifacts: list[str] | None = Field(None, description="Visual discrepancies or anomalies in the image")
+    camera_model: str | None = Field(None)
+    lens_model: str | None = Field(None)
+    focal_length: float | None = Field(None)
+    aperture: float | None = Field(None)
+    shutter_speed: float | None = Field(None)
+    iso: int | None = Field(None)
+    f_number: float | None = Field(None)
+    exposure_time_ms: float | None = Field(None)
+
+    def post_process(self):
+        if self.color_palette is not None:
+            for color in self.color_palette:
+                color.post_process()
+        return self
+
+
+class SceneDescription(BaseModel):
+    general: GeneralDescription = Field(...)
+    objects: list[SceneObject] | None = Field(None)
+
+    def iter_objects(self) -> Iterator[SceneObject]:
+        """Yield every object in the `children` tree, parent first. Wearables stay behind."""
+
+        def walk(objects, seen):
+            for object in objects or []:
+                if id(object) in seen:
+                    continue
+                seen.add(id(object))
+                yield object
+                yield from walk(getattr(object, "children", None), seen)
+
+        yield from walk(self.objects, set())
+
+    def iter_all_objects(self) -> Iterator[SceneObject]:
+        """Yield every object anywhere in the tree, wearables and face parts included."""
+        seen: set[int] = set()
+        for object in self.objects or []:
+            yield from _iter_nested(object, seen)
+
+    def _hoisted_wearables(self) -> list[SceneObject]:
+        """Wearables with a box, renamed "<item> on <person>", as standalone targets."""
+        hoisted = []
+        for object in self.iter_objects():
+            if object.kind != "person":
+                continue
+            for field in ("clothing", "footwear", "accessories"):
+                for item in getattr(object, field, None) or []:
+                    if not item.name or not item.bounding_boxes:
+                        continue
+                    item.name = f"{item.name} on {object.name}"
+                    hoisted.append(item)
+        return hoisted
+
+    def make_for_detect(self):
+        all_objects = list(self.iter_objects()) + self._hoisted_wearables()
+        for object in all_objects:
+            object.make_for_detect()
+        self.objects = all_objects
+
+    def make_for_mask(self):
+        all_objects = list(self.iter_objects()) + self._hoisted_wearables()
+        for object in all_objects:
+            object.make_for_mask()
+        self.objects = all_objects
+
+    def make_for_keypoints(self):
+        people = [object for object in self.iter_objects() if object.kind == "person" and object.keypoints is not None]
+        for object in people:
+            object.make_for_keypoints()
+        self.objects = people
+
+    def make_general(self):
+        self.objects = None
+
+    def make_long(self):
+        every_object = list(self.iter_all_objects())
+        flattened = list(self.iter_objects())
+        for object in every_object:
+            object.make_long()
+        self.objects = flattened
+
+    def make_short(self):
+        new_objects = []
+        for object in self.iter_objects():
+            new_objects.append(object)
+        for object in new_objects:
+            object.make_short()
+        self.objects = new_objects
+
+    def make_full_mask(self):
+        for object in self.iter_all_objects():
+            object.make_full_mask()
+
+    def make_full(self):
+        for object in self.iter_all_objects():
+            object.make_full()
+
+    def post_process(self):
+        if self.general is not None:
+            self.general.post_process()
+        for object in self.objects or []:
+            object.post_process()
+
+
+for _model in (
+    BaseObject,
+    Object,
+    Person,
+    Animal,
+    TextLine,
+    Text,
+    SceneObjectList,
+    GeneralDescription,
+    SceneDescription,
+):
+    _model.model_rebuild(force=True)

scene_analyzer.py

@@ -0,0 +1,845 @@
+"""Scene Analyzer [dev] with Hugging Face Transformers.
+
+`SceneAnalyzerModel` builds the task prompts the model was trained on, samples answers,
+and returns the best valid answer per input as a `data_model` object:
+
+    from scene_analyzer import SceneAnalyzerModel, SceneAnalyzerTask, DetailLevel
+    model = SceneAnalyzerModel("briaai/fibo-scene-analyzer")
+    [caption] = model.generate([image], SceneAnalyzerTask.CAPTION, detail_level=DetailLevel.FULL)
+
+The task logic in this module is shared by the other two harnesses, which replace only
+the engine: `scene_analyzer_vllm` (vLLM in-process, with schema-constrained decoding)
+and `scene_analyzer_server` (an OpenAI-compatible server).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass, field, replace
+from enum import Enum
+from typing import Any
+
+from data_model import BoundingBox, SceneDescription, SceneObjectList, _format_color_token, _format_coord_token
+from PIL import Image
+
+logger = logging.getLogger(__name__)
+
+VISION_TOKEN = "<|vision_start|><|image_pad|><|vision_end|>"
+IM_START_TOKEN = "<|im_start|>"
+IM_END_TOKEN = "<|im_end|>"
+EOS_TOKEN = "<|endoftext|>"
+STOP_TOKENS = (IM_END_TOKEN, EOS_TOKEN)
+THINK_END = "</think>"
+ASSISTANT_PREFIX = f"{IM_START_TOKEN}assistant\n<think>\n\n"
+# The trained transition from the reasoning trace to the answer includes both newlines.
+THINK_CLOSE = f"{THINK_END}\n\n"
+
+# Sampling settings the model is tuned for.
+TEMPERATURE = 0.6
+TOP_P = 0.97
+
+# Context held back for the answer when a reasoning trace is added to the prompt.
+ANSWER_RESERVE_TOKENS = 8192
+
+# A candidate that repeats one text line (3+ characters) this many times is a repetition
+# loop that happened to close its JSON, and is rejected.
+TEXT_LOOP_REPEATS = 25
+# The same for one object name (the description, for unnamed text objects) in a caption.
+# Repeated objects are grouped by design, so correct captions stay far below this.
+NAME_LOOP_REPEATS = 25
+
+# Two boxes overlapping above this IoU are the same box.
+DUPLICATE_BOX_IOU = 0.95
+
+
+class SceneAnalyzerTask(Enum):
+    CAPTION = "caption"
+    DESCRIBE = "describe"
+    DETECT = "detect"
+    KEYPOINTS = "keypoints"
+    OCR = "ocr"
+    EXPAND = "expand"
+    EXPAND_OBJECT = "expand_object"
+    MASK = "mask"
+    REFINE = "refine"
+    EDIT = "edit"
+
+
+class DetailLevel(Enum):
+    """The detail level tag of <caption>, <expand>, <expand_object>, <refine> and <edit>."""
+
+    GENERAL = "general"
+    SHORT = "short"
+    LONG = "long"
+    FULL = "full"
+    FULL_MASK = "full_mask"
+
+
+class ThinkingEffort(Enum):
+    MINIMAL = "minimal"
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+
+
+PLAIN_DETAIL_LEVELS = frozenset({DetailLevel.GENERAL, DetailLevel.SHORT, DetailLevel.LONG, DetailLevel.FULL})
+
+#: The detail levels each task accepts. <caption> accepts all; tasks without the tag ignore it.
+TASK_DETAIL_LEVELS = {
+    SceneAnalyzerTask.EXPAND: PLAIN_DETAIL_LEVELS,
+    SceneAnalyzerTask.EXPAND_OBJECT: frozenset(
+        {DetailLevel.SHORT, DetailLevel.LONG, DetailLevel.FULL, DetailLevel.FULL_MASK}
+    ),
+    SceneAnalyzerTask.REFINE: PLAIN_DETAIL_LEVELS,
+    SceneAnalyzerTask.EDIT: PLAIN_DETAIL_LEVELS,
+}
+
+#: Tasks that answer with a whole SceneDescription. All others answer with a
+#: SceneObjectList (<expand_object> with a one-element list).
+SCENE_DESCRIPTION_TASKS = frozenset(
+    {SceneAnalyzerTask.CAPTION, SceneAnalyzerTask.EXPAND, SceneAnalyzerTask.REFINE, SceneAnalyzerTask.EDIT}
+)
+
+#: Tasks that take no image.
+TEXT_ONLY_TASKS = frozenset({SceneAnalyzerTask.EXPAND, SceneAnalyzerTask.EXPAND_OBJECT, SceneAnalyzerTask.REFINE})
+
+#: Tasks whose answer is geometry for the image content. Their best sample is the most
+#: confident one: for them, more objects means more false positives, not a richer answer.
+GROUNDING_TASKS = frozenset({SceneAnalyzerTask.DETECT, SceneAnalyzerTask.MASK, SceneAnalyzerTask.KEYPOINTS, SceneAnalyzerTask.OCR})
+
+#: Tasks with a trained completion form (the prompt shows part of the answer and the model
+#: gives the missing objects), so they support revision rounds.
+ITERATIVE_TASKS = frozenset({SceneAnalyzerTask.CAPTION, *GROUNDING_TASKS})
+
+#: Grounding tasks whose prompt can name one target.
+NAMEABLE_TARGET_TASKS = frozenset({SceneAnalyzerTask.DETECT, SceneAnalyzerTask.MASK, SceneAnalyzerTask.KEYPOINTS})
+
+#: Nested lists that a revision round can graft onto an existing root object.
+NESTED_MERGE_FIELDS = ("children", "clothing", "footwear", "accessories")
+
+#: Tasks trained with reasoning traces. The other tasks always answer directly.
+THINKING_TASKS = SCENE_DESCRIPTION_TASKS
+
+#: The reasoning efforts trained for each detail level.
+DETAIL_LEVEL_EFFORTS = {
+    DetailLevel.GENERAL: (ThinkingEffort.MINIMAL,),
+    DetailLevel.SHORT: (ThinkingEffort.MINIMAL, ThinkingEffort.LOW),
+    DetailLevel.LONG: (ThinkingEffort.MINIMAL, ThinkingEffort.LOW, ThinkingEffort.MEDIUM),
+    DetailLevel.FULL: tuple(ThinkingEffort),
+    DetailLevel.FULL_MASK: tuple(ThinkingEffort),
+}
+
+
+@dataclass
+class Prompt:
+    """One model input in the trained chat format.
+
+    :param text: the user turn after the images: the task tags and the task input.
+    :param images: the images, in order.
+    :param thinking_effort: None for a direct answer; otherwise the effort of the
+        reasoning trace that the model writes before the answer.
+    :param continuation: assistant text after the opening, such as a finished trace.
+    """
+
+    text: str
+    images: list[Image.Image] = field(default_factory=list)
+    thinking_effort: ThinkingEffort | None = None
+    continuation: str = ""
+
+    def render(self) -> str:
+        """The prompt as the model sees it."""
+        opening = f"<{self.thinking_effort.value}>" if self.thinking_effort else THINK_CLOSE
+        vision = VISION_TOKEN * len(self.images)
+        return (
+            f"{IM_START_TOKEN}user\n{vision}{self.text}{IM_END_TOKEN}\n{ASSISTANT_PREFIX}{opening}{self.continuation}"
+        )
+
+    @property
+    def answers_directly(self) -> bool:
+        """Whether the next token starts the answer, not a reasoning trace."""
+        return self.thinking_effort is None or self.continuation.endswith(THINK_CLOSE)
+
+
+@dataclass
+class SamplingConfig:
+    n: int = 1
+    temperature: float = TEMPERATURE
+    top_p: float = TOP_P
+    max_tokens: int = 32768
+    stop: tuple[str, ...] = STOP_TOKENS
+    #: JSON schema for constrained decoding. Engines without constrained decoding ignore it.
+    json_schema: dict | None = None
+    seed: int | None = None
+
+
+@dataclass
+class Sample:
+    """One sampled completion. `text` excludes the stop token; `token_ids` includes it."""
+
+    text: str
+    token_ids: list[int]
+    cumulative_logprob: float
+
+
+@dataclass
+class Completion:
+    """The samples for one prompt, and the prompt length in tokens."""
+
+    samples: list[Sample]
+    prompt_tokens: int
+
+
+def _aspect_ratio_value(aspect_ratio: str) -> float:
+    """The "W:H" aspect ratio as the rounded number the prompts carry."""
+    width, height = aspect_ratio.split(":")
+    return round(int(width) / int(height), 2)
+
+
+def _confidence(sample: Sample) -> float:
+    """Length-normalised log probability of a sample."""
+    return sample.cumulative_logprob / max(len(sample.token_ids), 1) ** 0.6
+
+
+def prompt_json(description: SceneDescription) -> str:
+    """`description` as compact JSON in the trained prompt form, with coordinates and color
+    channels as tokens also after `post_process()` has converted them to numbers."""
+
+    def tokens(node: Any, key: str | None = None) -> Any:
+        if isinstance(node, dict):
+            if node.keys() == {"x", "y"}:
+                return {k: v if isinstance(v, str) else _format_coord_token(v) for k, v in node.items()}
+            if {"red", "green", "blue"} <= node.keys():
+                return {
+                    k: _format_color_token(v) if k in ("red", "green", "blue") and not isinstance(v, str) else v
+                    for k, v in node.items()
+                }
+            return {k: tokens(v, k) for k, v in node.items()}
+        if isinstance(node, list):
+            if key == "polygons":
+                return [[v if isinstance(v, str) else _format_coord_token(v) for v in ring] for ring in node]
+            return [tokens(v, key) for v in node]
+        return node
+
+    dumped = description.model_dump(mode="json", exclude_none=True, exclude_unset=True)
+    return json.dumps(tokens(dumped), ensure_ascii=False, separators=(",", ":"))
+
+
+_DETERMINERS = frozenset({"a", "an", "the", "some", "several", "many", "two", "three", "four", "five"})
+
+
+def _name_in_scene_text(scene: SceneDescription, name: str | None) -> SceneDescription:
+    if not name or not name.strip():
+        return scene
+    name = name.strip()
+    first = name.split()[0].lower()
+    article = "" if first in _DETERMINERS or first[0].isdigit() else ("an " if first[0] in "aeiou" else "a ")
+    general = scene.general.model_copy()
+    general.description = f"{(general.description or '').rstrip()} The scene also shows {article}{name}.".lstrip()
+    if general.user_prompt:
+        general.user_prompt = f"{general.user_prompt.rstrip().rstrip('.')}, with {article}{name}."
+    return scene.model_copy(update={"general": general})
+
+
+def _answer_text(text: str) -> str:
+    """The JSON answer in a completion: after the reasoning trace, without stop tokens."""
+    text = text.strip()
+    for token in STOP_TOKENS:
+        text = text.removesuffix(token).strip()
+    return text.split(THINK_END)[-1].strip()
+
+
+class SceneAnalyzerModel:
+    """Scene Analyzer on Hugging Face Transformers.
+
+    Transformers has no schema-constrained decoding, so an answer that is not valid JSON
+    for the schema is rejected and sampled again (see `retries` in `generate`). The vLLM
+    harness constrains the decoding and is much faster.
+    """
+
+    def __init__(
+        self,
+        model_name: str,
+        device_map: str | dict | None = "auto",
+        dtype: str = "bfloat16",
+        max_images: int = 5,
+        seed: int | None = None,
+        max_output_tokens: int = 32768,
+        max_model_len: int = 32768,
+        max_trace_tokens: int = 16384,
+        **model_kwargs: Any,
+    ):
+        """
+        :param model_name: path or HF id of the checkpoint.
+        :param device_map: passed to `from_pretrained`.
+        :param dtype: the model dtype.
+        :param max_images: cap on images per prompt (<edit> takes up to 5).
+        :param seed: sampling seed, for reproducible answers.
+        :param max_output_tokens: completion cap. A full caption can need tens of
+            thousands of tokens.
+        :param max_model_len: context length; the prompt and the answer must fit in it.
+        :param max_trace_tokens: cap on the reasoning trace.
+        :param model_kwargs: other `from_pretrained` arguments, e.g. `attn_implementation`.
+        """
+        import torch
+        from transformers import AutoModelForImageTextToText, AutoProcessor, GenerationConfig
+
+        self._torch = torch
+        self.processor = AutoProcessor.from_pretrained(model_name)
+        self.model = AutoModelForImageTextToText.from_pretrained(
+            model_name, dtype=getattr(torch, dtype), device_map=device_map, **model_kwargs
+        ).eval()
+        # Neutral defaults: `_LogprobRecorder` applies the temperature and top-p itself.
+        self.model.generation_config = GenerationConfig(do_sample=True, temperature=1.0, top_p=1.0, top_k=0)
+        self._setup(max_images, seed, max_output_tokens, max_model_len, max_trace_tokens)
+
+    def _setup(
+        self, max_images: int, seed: int | None, max_output_tokens: int, max_model_len: int, max_trace_tokens: int
+    ) -> None:
+        """Settings shared by every engine."""
+        self.max_images = max_images
+        self.seed = seed
+        self.max_output_tokens = max_output_tokens
+        self.max_model_len = max_model_len
+        self.max_trace_tokens = max_trace_tokens
+
+    # -- engine --------------------------------------------------------------------
+
+    def _sample(self, prompts: list[Prompt], configs: list[SamplingConfig]) -> list[Completion]:
+        """Sample completions for each prompt. Other engines override this method."""
+        return [self._sample_one(prompt, config) for prompt, config in zip(prompts, configs)]
+
+    def _decode(self, token_ids: list[int]) -> str:
+        return self.processor.tokenizer.decode(token_ids, skip_special_tokens=False)
+
+    def _sample_one(self, prompt: Prompt, config: SamplingConfig) -> Completion:
+        torch = self._torch
+        from transformers import LogitsProcessorList
+
+        tokenizer = self.processor.tokenizer
+        inputs = self.processor(text=[prompt.render()], images=prompt.images or None, return_tensors="pt")
+        inputs = inputs.to(self.model.device)
+        prompt_length = inputs["input_ids"].shape[1]
+        stop_ids = [tokenizer.convert_tokens_to_ids(token) for token in config.stop]
+        stop_ids = [token_id for token_id in stop_ids if token_id is not None and token_id != tokenizer.unk_token_id]
+        recorder = _LogprobRecorder(config.temperature, config.top_p, stop_ids)
+        if config.seed is not None:
+            torch.manual_seed(config.seed)
+        with torch.inference_mode():
+            sequences = self.model.generate(
+                **inputs,
+                max_new_tokens=max(1, min(config.max_tokens, self.max_model_len - prompt_length)),
+                num_return_sequences=config.n,
+                eos_token_id=stop_ids,
+                pad_token_id=stop_ids[0],
+                logits_processor=LogitsProcessorList([recorder]),
+            )
+        samples = []
+        for row, sequence in enumerate(sequences[:, prompt_length:].tolist()):
+            end = next((i for i, token_id in enumerate(sequence) if token_id in stop_ids), len(sequence))
+            text = self._decode(sequence[:end])
+            samples.append(Sample(text, sequence[: end + 1], recorder.total(row, sequence)))
+        return Completion(samples, prompt_length)
+
+    # -- prompts -------------------------------------------------------------------
+
+    def build_prompt(
+        self,
+        task: SceneAnalyzerTask,
+        image: Image.Image | list[Image.Image] | None,
+        thinking_effort: ThinkingEffort | None = None,
+        text_input: str | None = None,
+        aspect_ratio: str | None = None,
+        detail_level: DetailLevel = DetailLevel.FULL,
+        scene_context: SceneDescription | str | None = None,
+    ) -> Prompt:
+        """The prompt for one input.
+
+        :param image: the image; None for text-only tasks; for <edit>, 1-5 input images.
+        :param thinking_effort: None for a direct answer, or the effort of a reasoning trace.
+        :param text_input: the task's text: the target (<describe>, <detect>, <mask>,
+            <keypoints>; None for all targets), the prompt (<expand>), the object
+            (<expand_object>), or the instruction followed by the caption JSON (<refine>,
+            <edit>).
+        :param aspect_ratio: "W:H" of the output image, for <expand> and <edit>. <edit>
+            without it keeps the input frame and takes exactly one image.
+        :param scene_context: for <expand_object>, the scene the object belongs to, without
+            the object, as a `SceneDescription` or its JSON. The prompt adds one sentence
+            that names the object to the scene's general text (`_name_in_scene_text`).
+        """
+        level = detail_level.value
+        if scene_context is not None and task != SceneAnalyzerTask.EXPAND_OBJECT:
+            raise ValueError("only <expand_object> takes a scene context")
+        images = [] if image is None else list(image) if isinstance(image, list) else [image]
+        if task in TEXT_ONLY_TASKS and images:
+            raise ValueError(f"<{task.value}> takes no image")
+        if task not in TEXT_ONLY_TASKS and not images:
+            raise ValueError(f"<{task.value}> requires an image")
+        if len(images) > 1 and task != SceneAnalyzerTask.EDIT:
+            raise ValueError(f"<{task.value}> takes one image")
+        if len(images) > self.max_images:
+            raise ValueError(f"at most {self.max_images} images per prompt")
+
+        if task == SceneAnalyzerTask.CAPTION:
+            text = f"<caption><{level}>"
+        elif task == SceneAnalyzerTask.DESCRIBE:
+            text = f"<describe>{text_input}"
+        elif task in NAMEABLE_TARGET_TASKS:
+            text = f"<{task.value}>{text_input or ''}"
+        elif task == SceneAnalyzerTask.OCR:
+            text = "<ocr>"
+        elif task == SceneAnalyzerTask.EXPAND:
+            text = f"<expand><{level}>\n<aspect_ratio>{_aspect_ratio_value(aspect_ratio)}</aspect_ratio>\n{text_input}"
+        elif task == SceneAnalyzerTask.EXPAND_OBJECT:
+            text = f"<expand_object><{level}>{text_input}"
+            if scene_context is not None:
+                if isinstance(scene_context, str):
+                    scene_context = SceneDescription.model_validate_json(scene_context)
+                text += f"\n{prompt_json(_name_in_scene_text(scene_context, text_input))}"
+        elif task == SceneAnalyzerTask.REFINE:
+            text = f"<refine><{level}>{text_input}"
+        elif task == SceneAnalyzerTask.EDIT:
+            if aspect_ratio is None:
+                if len(images) != 1:
+                    raise ValueError("<edit> without an aspect ratio takes exactly one image")
+                text = f"<edit><{level}>{text_input}"
+            else:
+                text = (
+                    f"<edit><{level}>\n<aspect_ratio>{_aspect_ratio_value(aspect_ratio)}</aspect_ratio>\n{text_input}"
+                )
+        else:
+            raise ValueError(f"unsupported task {task}")
+        return Prompt(text, images, thinking_effort)
+
+    # -- generation ----------------------------------------------------------------
+
+    def generate(
+        self,
+        images: list[Image.Image | list[Image.Image] | None],
+        task: SceneAnalyzerTask,
+        thinking_effort: ThinkingEffort | None = None,
+        text_inputs: list[str | None] | None = None,
+        num_samples: int = 1,
+        retries: int = 3,
+        aspect_ratio: str | None = None,
+        detail_level: DetailLevel = DetailLevel.FULL,
+        num_iterations: int = 0,
+        scene_contexts: list[SceneDescription | str | None] | None = None,
+    ) -> list[SceneDescription | SceneObjectList | None]:
+        """Answer `task` for every input, one result (or None) per input.
+
+        :param images: one image per input (a list of images for <edit>, None for
+            text-only tasks).
+        :param thinking_effort: None for direct answers, or the effort of a reasoning
+            trace before each answer.
+        :param text_inputs: one text input per image (see `build_prompt`).
+        :param num_samples: candidates sampled per input; the best valid one is returned.
+        :param retries: extra attempts for inputs that produced no valid answer.
+        :param num_iterations: revision rounds after the first answer. Each round shows the
+            model its answer and adds the objects it reports as missing. An input stops
+            early when a round adds nothing. 0 is a single pass.
+        :param scene_contexts: for <expand_object>, one scene description per input, or
+            None: the rest of the scene, without the object. The model then describes the
+            object to fit the scene, with relationships to its objects.
+        :return: the answers, with coordinates and colors as `<coord_N.N>` / `<color_N>`
+            strings; `post_process()` on an answer converts them to numbers.
+        """
+        self._validate_arguments(task, text_inputs, detail_level, num_iterations, thinking_effort)
+        prompts = [
+            self.build_prompt(
+                task,
+                image,
+                thinking_effort,
+                self._text_input(text_inputs, i),
+                aspect_ratio,
+                detail_level,
+                self._text_input(scene_contexts, i),
+            )
+            for i, image in enumerate(images)
+        ]
+
+        results: list[Any] = [None] * len(prompts)
+        pending = list(range(len(prompts)))
+        for _ in range(1 + retries):
+            answers = self._select(task, [prompts[i] for i in pending], num_samples)
+            for i, answer in zip(pending, answers):
+                results[i] = answer
+            pending = [i for i in pending if results[i] is None]
+            if not pending:
+                break
+
+        active = [i for i, result in enumerate(results) if result is not None]
+        rounds_run = 0
+        while rounds_run < num_iterations and active:
+            rounds_run += 1
+            active = self._revise_round(results, active, prompts, task, detail_level, num_samples)
+        return results
+
+    @staticmethod
+    def _validate_arguments(
+        task: SceneAnalyzerTask,
+        text_inputs: list[str | None] | None,
+        detail_level: DetailLevel,
+        num_iterations: int,
+        thinking_effort: ThinkingEffort | None = None,
+    ) -> None:
+        """Reject argument combinations that give prompts the model was not trained on."""
+        if thinking_effort is not None:
+            if task not in THINKING_TASKS:
+                raise ValueError(f"<{task.value}> is trained without reasoning traces; use thinking_effort=None")
+            if thinking_effort not in DETAIL_LEVEL_EFFORTS[detail_level]:
+                raise ValueError(
+                    f"The detail level <{detail_level.value}> is trained with the efforts "
+                    f"{[effort.value for effort in DETAIL_LEVEL_EFFORTS[detail_level]]}"
+                )
+        if num_iterations < 0:
+            raise ValueError("num_iterations must not be negative")
+        if num_iterations > 0:
+            if task not in ITERATIVE_TASKS:
+                raise ValueError(
+                    f"<{task.value}> does not support num_iterations; "
+                    f"choose one of {sorted(item.value for item in ITERATIVE_TASKS)}"
+                )
+            if task == SceneAnalyzerTask.CAPTION and detail_level == DetailLevel.GENERAL:
+                raise ValueError("<caption><general> does not support num_iterations: it has no objects")
+            if task in NAMEABLE_TARGET_TASKS and any(text is not None for text in text_inputs or []):
+                raise ValueError(
+                    f"<{task.value}> with a named target does not support num_iterations; "
+                    "the completion form is trained only for the whole-list prompt"
+                )
+        allowed = TASK_DETAIL_LEVELS.get(task)
+        if allowed is not None and detail_level not in allowed:
+            raise ValueError(
+                f"<{task.value}> does not support the detail level <{detail_level.value}>; "
+                f"choose one of {sorted(level.value for level in allowed)}"
+            )
+
+    @staticmethod
+    def _text_input(text_inputs: list[str | None] | None, index: int) -> str | None:
+        if text_inputs is None or index >= len(text_inputs):
+            return None
+        return text_inputs[index]
+
+    def _select(self, task: SceneAnalyzerTask, prompts: list[Prompt], num_samples: int) -> list[Any]:
+        """Sample answers and select the best valid one per prompt.
+
+        A caption is scored by its object count, which gives the richest answer. That
+        also prefers repetition loops, so loops are rejected first. Grounding answers are
+        scored by confidence; other object lists by object count.
+        """
+        describes_image = task in SCENE_DESCRIPTION_TASKS
+        answer_type = SceneDescription if describes_image else SceneObjectList
+        results = []
+        for samples in self._generate_answers(prompts, answer_type.model_json_schema(), num_samples):
+            best, best_score = None, float("-inf")
+            for sample in samples:
+                try:
+                    answer = answer_type.model_validate_json(_answer_text(sample.text))
+                    if self.has_text_loop(answer):
+                        continue
+                    if describes_image:
+                        if not self.validate_duplicate_bboxes(answer) or self.has_name_loop(answer):
+                            continue
+                        score = len(list(answer.iter_all_objects()))
+                    elif task in GROUNDING_TASKS:
+                        score = _confidence(sample)
+                    else:
+                        score = len(answer.objects or [])
+                except Exception as error:
+                    logger.debug("rejected candidate: %s", error)
+                    continue
+                if score > best_score:
+                    best, best_score = answer, score
+            results.append(best)
+        return results
+
+    def _generate_answers(self, prompts: list[Prompt], json_schema: dict, num_samples: int) -> list[list[Sample]]:
+        """Sample `num_samples` answers per prompt.
+
+        A thinking prompt is generated in two passes. The trace is sampled without a
+        schema and stops at `</think>`. The answer is then sampled from the prompt plus
+        the trace plus `</think>\\n\\n`, under the schema. A schema in the same pass as the
+        trace makes the trained newlines after `</think>` unreachable.
+        """
+        answer_config = SamplingConfig(max_tokens=self.max_output_tokens, json_schema=json_schema, seed=self.seed)
+        needs_trace = [i for i, prompt in enumerate(prompts) if not prompt.answers_directly]
+        traces: dict[int, list[str]] = {}
+        if needs_trace:
+            trace_config = SamplingConfig(
+                n=num_samples, max_tokens=self.max_trace_tokens, stop=(*STOP_TOKENS, THINK_END), seed=self.seed
+            )
+            completions = self._sample([prompts[i] for i in needs_trace], [trace_config] * len(needs_trace))
+            for i, completion in zip(needs_trace, completions):
+                budget = self.max_model_len - completion.prompt_tokens - ANSWER_RESERVE_TOKENS
+                traces[i] = [self._trace_text(sample, budget) for sample in completion.samples]
+
+        # One answer request per trace. Direct prompts draw all samples from one request.
+        requests: list[Prompt] = []
+        configs: list[SamplingConfig] = []
+        owners: list[int] = []
+        for i, prompt in enumerate(prompts):
+            if i in traces:
+                for trace in traces[i]:
+                    requests.append(replace(prompt, continuation=prompt.continuation + trace + THINK_CLOSE))
+                    configs.append(answer_config)
+                    owners.append(i)
+            else:
+                requests.append(prompt)
+                configs.append(replace(answer_config, n=num_samples))
+                owners.append(i)
+
+        answers: list[list[Sample]] = [[] for _ in prompts]
+        for owner, completion in zip(owners, self._sample(requests, configs)):
+            answers[owner].extend(completion.samples)
+        return answers
+
+    def _trace_text(self, sample: Sample, budget: int) -> str:
+        """The reasoning trace, cut to `budget` tokens if it is longer.
+
+        A cut trace is still usable: the answer pass closes the think block regardless.
+        """
+        budget = max(budget, 0)
+        text = self._decode(sample.token_ids[:budget]) if len(sample.token_ids) > budget else sample.text
+        for token in STOP_TOKENS:
+            text = text.removesuffix(token)
+        return text.split(THINK_END)[0]
+
+    # -- answer checks -------------------------------------------------------------
+
+    @staticmethod
+    def _bbox_iou(a: BoundingBox, b: BoundingBox) -> float:
+        a, b = a.model_copy(deep=True), b.model_copy(deep=True)
+        a.post_process()
+        b.post_process()
+
+        def area(box: BoundingBox) -> float:
+            return max(0.0, box.bottom_right.x - box.top_left.x) * max(0.0, box.bottom_right.y - box.top_left.y)
+
+        width = max(0.0, min(a.bottom_right.x, b.bottom_right.x) - max(a.top_left.x, b.top_left.x))
+        height = max(0.0, min(a.bottom_right.y, b.bottom_right.y) - max(a.top_left.y, b.top_left.y))
+        intersection = width * height
+        union = area(a) + area(b) - intersection
+        return intersection / union if union > 0 else 0.0
+
+    @staticmethod
+    def has_text_loop(answer: SceneDescription | SceneObjectList) -> bool:
+        """Whether one text line (3+ characters) repeats TEXT_LOOP_REPEATS times or more."""
+        counts: dict[str, int] = {}
+
+        def walk(node: Any) -> None:
+            if isinstance(node, dict):
+                for line in node.get("lines") or []:
+                    text = (line.get("text") or "").strip() if isinstance(line, dict) else ""
+                    if len(text) >= 3:
+                        counts[text] = counts.get(text, 0) + 1
+                for value in node.values():
+                    walk(value)
+            elif isinstance(node, list):
+                for value in node:
+                    walk(value)
+
+        # Text can nest under any object, so walk the whole dumped answer.
+        walk(answer.model_dump())
+        return max(counts.values(), default=0) >= TEXT_LOOP_REPEATS
+
+    @staticmethod
+    def has_name_loop(image_description: SceneDescription) -> bool:
+        """Whether one object name (or description, when unnamed) repeats
+        NAME_LOOP_REPEATS times or more."""
+        counts: dict[str, int] = {}
+        for object in image_description.iter_all_objects():
+            name = ((object.name or "").strip() or (object.description or "").strip()).lower()
+            if name:
+                counts[name] = counts.get(name, 0) + 1
+        return max(counts.values(), default=0) >= NAME_LOOP_REPEATS
+
+    def validate_duplicate_bboxes(self, image_description: SceneDescription) -> bool:
+        """False when two objects with the same name have the same box."""
+        seen: list[tuple[BoundingBox, str]] = []
+        for object in image_description.iter_all_objects():
+            for bounding_box in object.bounding_boxes or []:
+                for other, other_name in seen:
+                    if object.name == other_name and self._bbox_iou(bounding_box, other) > DUPLICATE_BOX_IOU:
+                        return False
+                seen.append((bounding_box, object.name))
+        return True
+
+    # -- revision rounds -----------------------------------------------------------
+
+    def _revise_round(
+        self,
+        results: list[Any],
+        active: list[int],
+        prompts: list[Prompt],
+        task: SceneAnalyzerTask,
+        detail_level: DetailLevel,
+        num_samples: int,
+    ) -> list[int]:
+        """One revision round over `active`, merging new objects into `results` in place.
+
+        Returns the indices that gained objects. An empty answer means the model considers
+        the answer complete, so that index stops.
+        """
+        tag = f"<caption><{detail_level.value}>" if task == SceneAnalyzerTask.CAPTION else f"<{task.value}>"
+        requests = [
+            Prompt(f"{tag}\n{self._source_text(task, results[i])}", prompts[i].images, prompts[i].thinking_effort)
+            for i in active
+        ]
+        schema = SceneObjectList.model_json_schema()
+        gained = []
+        for i, samples in zip(active, self._generate_answers(requests, schema, num_samples)):
+            # The best completion is the one that adds the most after the merge rules.
+            best = None
+            for sample in samples:
+                try:
+                    candidate = SceneObjectList.model_validate_json(_answer_text(sample.text))
+                except Exception:
+                    continue
+                plan = self._plan_merge(results[i], candidate.objects or [])
+                if plan[2] > (best[2] if best else 0):
+                    best = plan
+            if best is not None and self._apply_merge(results[i], best):
+                gained.append(i)
+        logger.info("[revise] <%s>: asked %d, gained %d", task.value, len(active), len(gained))
+        return gained
+
+    @staticmethod
+    def _source_text(task: SceneAnalyzerTask, result: Any) -> str:
+        """An answer serialized as the source list of a completion prompt.
+
+        The source must have the same form as in training, so the task's own view is
+        applied to a copy of each object. A caption is serialized as answered: it is
+        already in the view of its detail level.
+        """
+        if task == SceneAnalyzerTask.CAPTION:
+            return result.model_dump_json(exclude_none=True, exclude_unset=True)
+        objects = [object.model_copy(deep=True) for object in (result.objects or [])]
+        for object in objects:
+            if task == SceneAnalyzerTask.DETECT:
+                object.make_for_detect()
+            elif task == SceneAnalyzerTask.MASK:
+                object.make_for_mask()
+            elif task == SceneAnalyzerTask.KEYPOINTS:
+                object.make_for_keypoints()
+            else:
+                # <ocr> keeps the text objects without their masks.
+                object.drop_masks()
+                for line in getattr(object, "lines", None) or []:
+                    line.drop_masks()
+        # model_construct, because validation would restore the fields the view cleared.
+        return SceneObjectList.model_construct(objects=objects).model_dump_json(exclude_none=True, exclude_unset=True)
+
+    @staticmethod
+    def _all_objects(result: Any) -> list[Any]:
+        """Every object in a caption or an object list."""
+        walk = getattr(result, "iter_all_objects", None)
+        return list(walk()) if walk is not None else list(result.objects or [])
+
+    def _plan_merge(self, result: Any, new_objects: list[Any]) -> tuple[list, list, int]:
+        """What of `new_objects` would be added to `result`, without changing it.
+
+        Returns `(new_roots, grafts, total)`. An object whose name matches a root is a
+        stub: its nested items are grafted onto that root. The rules:
+
+        * No addition may reuse an existing box (IoU above DUPLICATE_BOX_IOU), whatever
+          its name.
+        * A nested item is refused when its parent already lists that name. Nested items
+          often have no box, so the box rule alone does not stop repeats.
+        * An unnamed object is refused, except a root Text, which has no name.
+        """
+        by_name = {o.name: o for o in (result.objects or []) if o.name}
+        existing_boxes = [box for o in self._all_objects(result) for box in (o.bounding_boxes or [])]
+        claimed: list[BoundingBox] = []
+
+        def box_taken(object: Any) -> bool:
+            return any(
+                self._bbox_iou(box, other) > DUPLICATE_BOX_IOU
+                for box in object.bounding_boxes or []
+                for other in existing_boxes + claimed
+            )
+
+        new_roots: list[Any] = []
+        grafts: list[tuple] = []
+        for object in new_objects:
+            parent = by_name.get(object.name) if object.name else None
+            if parent is not None:
+                for field_name in NESTED_MERGE_FIELDS:
+                    taken = {item.name for item in (getattr(parent, field_name, None) or []) if item.name}
+                    keep = []
+                    for item in getattr(object, field_name, None) or []:
+                        if not item.name or item.name in taken or box_taken(item):
+                            continue
+                        taken.add(item.name)
+                        claimed.extend(item.bounding_boxes or [])
+                        keep.append(item)
+                    if keep:
+                        grafts.append((parent, field_name, keep))
+                continue
+            if not object.name and object.kind != "text":
+                continue
+            if box_taken(object):
+                continue
+            claimed.extend(object.bounding_boxes or [])
+            new_roots.append(object)
+        return new_roots, grafts, len(new_roots) + sum(len(items) for _, _, items in grafts)
+
+    @staticmethod
+    def _apply_merge(result: Any, plan: tuple[list, list, int]) -> int:
+        """Apply a plan from `_plan_merge` and return how many objects it added."""
+        new_roots, grafts, total = plan
+        for parent, field_name, items in grafts:
+            setattr(parent, field_name, list(getattr(parent, field_name, None) or []) + items)
+        if new_roots:
+            result.objects = list(result.objects or []) + new_roots
+        return total
+
+
+class _LogprobRecorder:
+    """A logits processor that applies the temperature and top-p, and sums the raw log
+    probability of each sampled token per sequence.
+
+    Generation calls it once per step, with the tokens sampled so far, so each call adds
+    the previous step's token. `total()` adds the last token.
+    """
+
+    def __init__(self, temperature: float, top_p: float, stop_ids: list[int]):
+        self.temperature = temperature
+        self.top_p = top_p
+        self.stop_ids = sorted(set(stop_ids))
+        self.sums = None
+        self.previous = None
+        self.finished = None
+
+    def __call__(self, input_ids, scores):
+        import torch
+
+        scores = scores.float()
+        logprobs = torch.log_softmax(scores, dim=-1)
+        if self.previous is None:
+            self.sums = torch.zeros(scores.shape[0], device=scores.device)
+            self.finished = torch.zeros(scores.shape[0], dtype=torch.bool, device=scores.device)
+        else:
+            last = input_ids[:, -1]
+            chosen = self.previous.gather(1, last[:, None]).squeeze(1)
+            self.sums += torch.where(self.finished, torch.zeros_like(chosen), chosen)
+            self.finished |= torch.isin(last, torch.tensor(self.stop_ids, device=last.device))
+        self.previous = logprobs
+
+        warped = scores / max(self.temperature, 1e-5)
+        if self.top_p < 1.0:
+            sorted_scores, order = torch.sort(warped, descending=True, dim=-1)
+            probabilities = torch.softmax(sorted_scores, dim=-1)
+            # Keep the smallest set of top tokens whose probability reaches top_p.
+            remove_sorted = probabilities.cumsum(dim=-1) - probabilities > self.top_p
+            remove = torch.zeros_like(remove_sorted).scatter(1, order, remove_sorted)
+            warped = warped.masked_fill(remove, float("-inf"))
+        return warped
+
+    def total(self, row: int, sequence: list[int]) -> float:
+        """The summed log probability of the completion in `row`."""
+        if self.sums is None:
+            return 0.0
+        total = float(self.sums[row])
+        if sequence and not bool(self.finished[row]):
+            total += float(self.previous[row, sequence[-1]])
+        return total

scene_analyzer_server.py

@@ -0,0 +1,262 @@
+"""`SceneAnalyzerModel` for an OpenAI-compatible server.
+
+Same interface as `scene_analyzer.SceneAnalyzerModel`; only the engine differs:
+
+    from scene_analyzer_server import SceneAnalyzerModel
+    model = SceneAnalyzerModel("http://localhost:8000/v1")
+
+The client sends standard chat-completion requests. The checkpoint's chat template
+renders the trained prompt, and two request fields outside the OpenAI schema are needed
+for full function (vLLM and SGLang support both):
+
+* `chat_template_kwargs.thinking_effort` selects a reasoning trace. A server that ignores
+  it gives direct answers. `enable_thinking` is also sent, for templates that know only it.
+* `skip_special_tokens: false`. The JSON structure and the coordinate and color values
+  are special tokens, which servers remove from the output by default.
+
+For example, with vLLM:
+
+    vllm serve briaai/fibo-scene-analyzer --max-model-len 32768 --limit-mm-per-prompt '{"image": 5}' \\
+        --reasoning-parser deepseek_r1
+
+Unlike the in-process harnesses, a thinking request is one pass without a schema: the
+server cannot continue an assistant turn in a standard request.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import concurrent.futures
+import inspect
+import io
+import logging
+from pathlib import Path
+from typing import Any
+
+from PIL import Image
+
+import scene_analyzer as _base
+from scene_analyzer import (
+    Completion,
+    DetailLevel,
+    SceneAnalyzerTask,
+    Prompt,
+    Sample,
+    SamplingConfig,
+    ThinkingEffort,
+)
+
+__all__ = ["DetailLevel", "SceneAnalyzerModel", "SceneAnalyzerTask", "ThinkingEffort"]
+
+logger = logging.getLogger(__name__)
+
+#: The chat template of the checkpoint, for servers that serve an older copy of it.
+CHAT_TEMPLATE_PATH = Path(__file__).with_name("chat_template.jinja")
+
+
+class SceneAnalyzerModel(_base.SceneAnalyzerModel):
+    """Scene Analyzer on an OpenAI-compatible server."""
+
+    def __init__(
+        self,
+        base_url: str,
+        model: str | None = None,
+        api_key: str = "EMPTY",
+        max_images: int = 5,
+        seed: int | None = None,
+        max_output_tokens: int = 32768,
+        max_model_len: int = 32768,
+        request_timeout: float = 3600.0,
+        max_concurrency: int = 64,
+        send_chat_template: bool = False,
+    ):
+        """
+        :param base_url: the server's OpenAI root, e.g. ``http://localhost:8000/v1``.
+        :param model: served model name. Defaults to the server's first model.
+        :param max_images: cap on images per prompt. Must not exceed the server's limit.
+        :param max_output_tokens: completion cap. A cap that does not fit beside the prompt
+            is not sent, and the server then uses the rest of its context.
+        :param max_model_len: the server's context length.
+        :param max_concurrency: inputs in flight.
+        :param send_chat_template: send `chat_template.jinja` with each request, for a
+            server whose checkpoint has another template. vLLM needs
+            ``--trust-request-chat-template`` for this.
+
+        Engine settings (GPU memory, batch size, speculative decoding) are set on the
+        server command line.
+        """
+        from openai import OpenAI
+
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.request_timeout = request_timeout
+        self.max_concurrency = max(1, max_concurrency)
+        self.chat_template = CHAT_TEMPLATE_PATH.read_text() if send_chat_template else None
+        self.served_model = model or OpenAI(base_url=self.base_url, api_key=api_key).models.list().data[0].id
+        self._setup(max_images, seed, max_output_tokens, max_model_len, max_trace_tokens=0)
+
+    # -- engine --------------------------------------------------------------------
+
+    def _generate_answers(self, prompts: list[Prompt], json_schema: dict, num_samples: int) -> list[list[Sample]]:
+        """One pass per prompt. A thinking prompt is sampled without the schema, and the
+        answer is the text after the trace."""
+        configs = [
+            SamplingConfig(
+                n=num_samples,
+                max_tokens=self.max_output_tokens,
+                json_schema=json_schema if prompt.answers_directly else None,
+                seed=self.seed,
+            )
+            for prompt in prompts
+        ]
+        return [completion.samples for completion in self._sample(prompts, configs)]
+
+    def _sample(self, prompts: list[Prompt], configs: list[SamplingConfig]) -> list[Completion]:
+        return _run_coroutine(self._sample_async(prompts, configs))
+
+    async def _sample_async(self, prompts: list[Prompt], configs: list[SamplingConfig]) -> list[Completion]:
+        from openai import AsyncOpenAI
+
+        semaphore = asyncio.Semaphore(self.max_concurrency)
+        # A new client per call, because a client binds to the event loop it first runs on.
+        async with AsyncOpenAI(base_url=self.base_url, api_key=self.api_key, max_retries=3) as client:
+
+            async def one(prompt: Prompt, config: SamplingConfig) -> Completion:
+                async with semaphore:
+                    # Built inside the semaphore, so only the requests in flight hold
+                    # encoded images in memory.
+                    request = self._request(prompt, config)
+                    response = await client.chat.completions.create(timeout=self.request_timeout, **request)
+                return self._completion(response)
+
+            return list(await asyncio.gather(*(one(prompt, config) for prompt, config in zip(prompts, configs))))
+
+    def _request(self, prompt: Prompt, config: SamplingConfig) -> dict[str, Any]:
+        if prompt.continuation:
+            raise ValueError("an OpenAI-compatible request cannot continue an assistant turn")
+        content: list[dict[str, Any]] = [
+            {"type": "image_url", "image_url": {"url": _data_url(image)}} for image in prompt.images
+        ]
+        content.append({"type": "text", "text": prompt.text})
+        # `enable_thinking` is for templates without `thinking_effort`, such as the stock one.
+        template_kwargs: dict[str, Any] = {"enable_thinking": prompt.thinking_effort is not None}
+        if prompt.thinking_effort is not None:
+            template_kwargs["thinking_effort"] = prompt.thinking_effort.value
+        extra_body: dict[str, Any] = {"skip_special_tokens": False, "chat_template_kwargs": template_kwargs}
+        if self.chat_template is not None:
+            extra_body["chat_template"] = self.chat_template
+        request: dict[str, Any] = {
+            "model": self.served_model,
+            "messages": [{"role": "user", "content": content}],
+            "n": config.n,
+            "temperature": config.temperature,
+            "top_p": config.top_p,
+            "stop": list(config.stop),
+            "logprobs": True,
+            "extra_body": extra_body,
+        }
+        if config.max_tokens < self.max_model_len:
+            request["max_tokens"] = config.max_tokens
+        if config.seed is not None:
+            request["seed"] = config.seed
+        if config.json_schema is not None:
+            request["response_format"] = {
+                "type": "json_schema",
+                "json_schema": {"name": "answer", "schema": config.json_schema},
+            }
+        return request
+
+    @staticmethod
+    def _completion(response: Any) -> Completion:
+        samples = []
+        for choice in response.choices:
+            entries = (choice.logprobs.content or []) if choice.logprobs else []
+            samples.append(
+                Sample(
+                    text=choice.message.content or "",
+                    # The ids are used only for their count.
+                    token_ids=[0] * len(entries),
+                    cumulative_logprob=sum(entry.logprob for entry in entries),
+                )
+            )
+        usage = getattr(response, "usage", None)
+        return Completion(samples, getattr(usage, "prompt_tokens", 0) or 0)
+
+    # -- generation ----------------------------------------------------------------
+
+    def generate(self, images, task, *args, on_result=None, **kwargs):
+        """The inherited `generate`, run separately per input with `max_concurrency`
+        inputs in flight.
+
+        Generation lengths vary much between inputs, and one batched call waits for its
+        longest input. Per-input calls keep the server busy.
+
+        `images` and `text_inputs` can be lists or iterators. Inputs are read only when a
+        slot is free, so a lazy iterator keeps at most `max_concurrency` images in memory.
+
+        :param on_result: optional `on_result(index, result)`, called on the caller's
+            thread as each input finishes, in completion order.
+        :return: the results in input order. Argument errors (ValueError, TypeError) are
+            raised; other failures give None for that input only.
+        """
+        bound = inspect.signature(_base.SceneAnalyzerModel.generate).bind(self, images, task, *args, **kwargs)
+        arguments = dict(bound.arguments)
+        arguments.pop("self")
+        inputs = arguments.pop("images")
+        text_inputs = arguments.pop("text_inputs", None)
+        scene_contexts = iter(arguments.pop("scene_contexts", None) or ())
+        pairs = zip(inputs, text_inputs) if text_inputs is not None else ((image, None) for image in inputs)
+        work = ((image, text_input, next(scene_contexts, None)) for image, text_input in pairs)
+
+        def one(image, text_input, scene_context):
+            try:
+                return super(SceneAnalyzerModel, self).generate(
+                    [image],
+                    text_inputs=None if text_input is None else [text_input],
+                    scene_contexts=None if scene_context is None else [scene_context],
+                    **arguments,
+                )[0]
+            except (ValueError, TypeError):
+                raise
+            except Exception as error:  # noqa: BLE001 - one failed input must not fail the others
+                logger.warning("generate: input failed: %r", error)
+                return None
+
+        results: dict[int, Any] = {}
+        pending: dict[concurrent.futures.Future, int] = {}
+
+        def collect(futures) -> None:
+            for future in futures:
+                index = pending.pop(future)
+                results[index] = future.result()
+                if on_result is not None:
+                    on_result(index, results[index])
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_concurrency) as pool:
+            for index, (image, text_input, scene_context) in enumerate(work):
+                if len(pending) >= self.max_concurrency:
+                    done, _ = concurrent.futures.wait(pending, return_when=concurrent.futures.FIRST_COMPLETED)
+                    collect(done)
+                pending[pool.submit(one, image, text_input, scene_context)] = index
+                # Report finished inputs now, not only when the window is full.
+                collect([future for future in pending if future.done()])
+            collect(concurrent.futures.as_completed(list(pending)))
+        return [results[index] for index in range(len(results))]
+
+
+def _data_url(image: Image.Image) -> str:
+    # PNG, because a lossy format would change the pixels the model describes.
+    buffer = io.BytesIO()
+    image.convert("RGB").save(buffer, format="PNG")
+    return "data:image/png;base64," + base64.b64encode(buffer.getvalue()).decode()
+
+
+def _run_coroutine(coroutine):
+    """Run `coroutine` to completion, also when the caller is already in an event loop."""
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(coroutine)
+    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+        return pool.submit(asyncio.run, coroutine).result()

scene_analyzer_vllm.py

@@ -0,0 +1,130 @@
+"""`SceneAnalyzerModel` on an in-process vLLM engine, with schema-constrained decoding.
+
+Same interface as `scene_analyzer.SceneAnalyzerModel`; only the engine differs:
+
+    from scene_analyzer_vllm import SceneAnalyzerModel
+    model = SceneAnalyzerModel("briaai/fibo-scene-analyzer", gpu_memory_utilization=0.9)
+
+All inputs of one `generate` call run as one batch.
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from vllm import LLM, SamplingParams
+from vllm.config import StructuredOutputsConfig
+from vllm.sampling_params import StructuredOutputsParams
+
+import scene_analyzer as _base
+from scene_analyzer import (
+    Completion,
+    DetailLevel,
+    SceneAnalyzerTask,
+    Prompt,
+    Sample,
+    SamplingConfig,
+    ThinkingEffort,
+)
+
+__all__ = ["DetailLevel", "SceneAnalyzerModel", "SceneAnalyzerTask", "ThinkingEffort"]
+
+# Pixel budget per image: 8192 visual tokens of 32x32 px (patch 16, merge 2), the same as
+# the processor config of the checkpoint.
+MAX_PIXELS = 8192 * 1024
+MIN_PIXELS = 4 * 28 * 28
+
+
+class SceneAnalyzerModel(_base.SceneAnalyzerModel):
+    """Scene Analyzer on vLLM."""
+
+    def __init__(
+        self,
+        model_name: str,
+        gpu_memory_utilization: float = 0.9,
+        max_images: int = 5,
+        seed: int | None = None,
+        max_output_tokens: int = 32768,
+        max_model_len: int = 32768,
+        max_num_seqs: int = 256,
+        max_trace_tokens: int = 16384,
+        speculative_model: str | None = None,
+        num_speculative_tokens: int | None = None,
+        **engine_kwargs: Any,
+    ):
+        """
+        :param model_name: path or HF id of the checkpoint.
+        :param max_num_seqs: concurrent sequences. The hybrid attention layers keep a
+            recurrent state per sequence, and vLLM fails at startup when fewer state
+            blocks fit on the GPU than this value. 256 fits on an 80 GB GPU.
+        :param speculative_model: path or HF id of a draft model for speculative decoding.
+        :param num_speculative_tokens: draft tokens per step, with ``speculative_model``.
+        :param engine_kwargs: other `vllm.LLM` arguments, e.g. ``tensor_parallel_size``.
+
+        The other parameters are as in `scene_analyzer.SceneAnalyzerModel`.
+        """
+        if speculative_model is not None:
+            speculative_config: dict[str, Any] = {"model": speculative_model}
+            if num_speculative_tokens is not None:
+                speculative_config["num_speculative_tokens"] = num_speculative_tokens
+            engine_kwargs["speculative_config"] = speculative_config
+        self.llm = LLM(
+            model=model_name,
+            dtype="bfloat16",
+            mm_processor_kwargs={"max_pixels": MAX_PIXELS, "min_pixels": MIN_PIXELS},
+            max_model_len=max_model_len,
+            max_num_seqs=max_num_seqs,
+            gpu_memory_utilization=gpu_memory_utilization,
+            limit_mm_per_prompt={"image": max_images},
+            # Answer prompts always end with a closed think block, so the reasoning parser
+            # reports the reasoning as ended and the schema applies from the first token.
+            structured_outputs_config=StructuredOutputsConfig(
+                reasoning_parser="deepseek_r1", enable_in_reasoning=False
+            ),
+            **engine_kwargs,
+        )
+        self.tokenizer = self.llm.get_tokenizer()
+        self._setup(max_images, seed, max_output_tokens, max_model_len, max_trace_tokens)
+
+    def _decode(self, token_ids: list[int]) -> str:
+        return self.tokenizer.decode(token_ids, skip_special_tokens=False)
+
+    def _sample(self, prompts: list[Prompt], configs: list[SamplingConfig]) -> list[Completion]:
+        inputs = [self._engine_input(prompt) for prompt in prompts]
+        params = [self._sampling_params(config) for config in configs]
+        outputs = self.llm.generate(inputs, sampling_params=params, use_tqdm=False)
+        completions = []
+        for output in outputs:
+            samples = []
+            for sample in output.outputs:
+                samples.append(Sample(sample.text, list(sample.token_ids), sample.cumulative_logprob or 0.0))
+            completions.append(Completion(samples, len(output.prompt_token_ids or [])))
+        return completions
+
+    @staticmethod
+    def _engine_input(prompt: Prompt) -> dict[str, Any] | str:
+        if not prompt.images:
+            return prompt.render()
+        return {
+            "prompt": prompt.render(),
+            "multi_modal_data": {"image": list(prompt.images)},
+            # A fresh uuid per image, so the engine's cache never matches another image.
+            "multi_modal_uuids": {"image": [str(uuid.uuid4()) for _ in prompt.images]},
+        }
+
+    @staticmethod
+    def _sampling_params(config: SamplingConfig) -> SamplingParams:
+        return SamplingParams(
+            n=config.n,
+            temperature=config.temperature,
+            top_p=config.top_p,
+            max_tokens=config.max_tokens,
+            stop=list(config.stop),
+            # The JSON structure and the coordinate and color values are special tokens.
+            skip_special_tokens=False,
+            # Needed for cumulative_logprob.
+            logprobs=1,
+            seed=config.seed,
+            structured_outputs=StructuredOutputsParams(json=config.json_schema) if config.json_schema else None,
+        )