feat(d3): image input for d3:lite (draft; stacked on #817)

.github/script/sweep_common.py

@@ -223,6 +223,8 @@ class SweepTask:
     name = "sweep"
     needs_asr = False
     needs_embed = False
+    # A decision model tag (e.g. "d3:lite") served with --decision; empty for none.
+    needs_decision = ""
     csv_header: list[str] = []
 
     @staticmethod
@@ -320,6 +322,8 @@ def start_server(self, append_log: bool = False) -> None:
             "-a", "1" if self.needs_asr else "0",
             "-e", "1" if self.needs_embed else "0",
         ]
+        if self.needs_decision:
+            cmd += ["--decision", self.needs_decision]
         self.log_event(
             "server_start",
             detail=" ".join(cmd),

.github/script/sweep_decision.py

@@ -0,0 +1,233 @@
+#!/usr/bin/env python3
+"""Sweep the decision models through POST /v1/systemone.
+
+Per model (served with `--decision <tag>`):
+  goldens     stored requests answered by the upstream BF16 model (assets/d3_goldens.jsonl): every answer has
+              the upstream shape, top-1 agrees on at least --min-agreement of them, mean |dp| at most --max-dp
+  types       one choice, one noul, one score, and all three in one request
+  two-letter  a 27-option choice (the 27th option's code is "AA")
+  invalid     a malformed question -> 422 {"detail": {"invalid_questions": ...}}, nothing run
+  too-long    a prompt over the input limit -> max_length_exceeded inside a 200 reply
+  image       a request with a PNG image (yellow box on blue) answered, and a broken image -> 422 "images[0]: ..."
+  docs        the example request of docs/docs/models/d3.md: the server's reply must decide what the documented
+              reply decides (choice, noul side, most likely score), so the docs cannot drift from the server
+
+Decision models are absent from /v1/models, so the model set comes from `flm list` (family "d3"). Each decision
+model gets its own server (--decision takes one tag).
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+from sweep_common import SweepTask, run_task
+
+REPO = Path(__file__).resolve().parents[2]
+STATE = "The order arrived damaged yesterday. The customer has a receipt and asks for a replacement today."
+
+
+def answer_shape_ok(answer: dict) -> bool:
+    kind = answer.get("type")
+    if "error" in answer:
+        return set(answer) == {"type", "error", "message"}
+    if kind == "choice":
+        return list(answer) == ["type", "choice", "probabilities", "confidence"]
+    if kind == "noul":
+        return list(answer) == ["type", "noul"]
+    if kind == "score":
+        return list(answer) == ["type", "score", "probabilities", "confidence", "legend"]
+    return False
+
+
+class DecisionSweep(SweepTask):
+    name = "decision"
+    csv_header = ["Model", "Case", "Detail", "Latency (s)", "Status", "Error"]
+
+    def __init__(self, args):
+        super().__init__(args)
+        # One server per decision model: the tag is the server's --decision argument. run() restarts
+        # the server for every further model.
+        self.needs_decision = (args.models or ["d3:lite"])[0]
+
+    @staticmethod
+    def add_arguments(parser) -> None:
+        parser.add_argument("--min-agreement", type=float, default=0.85,
+                            help="Share of golden answers whose top-1 must match the upstream BF16 model.")
+        parser.add_argument("--max-dp", type=float, default=0.05,
+                            help="Largest allowed mean |probability difference| against the upstream BF16 model.")
+
+    def select_models(self, catalog: list[dict]) -> list[str]:
+        return [entry["id"] for entry in catalog if entry["family"] == "d3"] or ["d3:lite"]
+
+    # -- http -------------------------------------------------------------
+
+    @staticmethod
+    def _json_or_raw(raw: bytes) -> dict:
+        """The reply body as JSON; a body that is not JSON (a proxy's or crash page's 5xx) becomes
+        {"_raw": its start}, so the case fails instead of the whole sweep."""
+        text = raw.decode("utf-8", errors="replace")
+        try:
+            body = json.loads(text or "{}")
+        except ValueError:
+            return {"_raw": text[:200]}
+        return body if isinstance(body, dict) else {"_raw": text[:200]}
+
+    def post(self, body) -> tuple[int, dict, float]:
+        data = body if isinstance(body, bytes) else json.dumps(body).encode("utf-8")
+        req = urllib.request.Request(f"{self.base_url}/systemone", data=data,
+                                     headers={"Content-Type": "application/json"}, method="POST")
+        started = time.monotonic()
+        try:
+            with urllib.request.urlopen(req, timeout=self.args.request_timeout) as resp:
+                return resp.status, self._json_or_raw(resp.read()), time.monotonic() - started
+        except urllib.error.HTTPError as err:
+            return err.code, self._json_or_raw(err.read()), time.monotonic() - started
+
+    def case(self, writer, model: str, case: str, ok: bool, detail: str, elapsed: float, error: str = "") -> None:
+        self.record(writer, [model, case, detail, f"{elapsed:.2f}", "ok" if ok else "error", "" if ok else error],
+                    model=model, error=None if ok else f"{case}: {error or detail}")
+        print(f"  {case}: {detail} -> {'ok' if ok else 'FAILED ' + error}")
+
+    # -- cases ------------------------------------------------------------
+
+    def _goldens(self, writer, model: str) -> None:
+        agree = total = 0
+        dps: list[float] = []
+        bad_shape: list[str] = []
+        elapsed = 0.0
+        for line in open(self.asset("d3_goldens.jsonl"), encoding="utf-8"):
+            rec = json.loads(line)
+            status, body, t = self.post(rec["request"])
+            elapsed += t
+            if status != 200:
+                bad_shape.append(f"{rec['id']}: status {status}")
+                continue
+            for key, want in rec["response"]["answers"].items():
+                got = body["answers"].get(key, {})
+                if not answer_shape_ok(got) or list(body["answers"]) != list(rec["response"]["answers"]):
+                    bad_shape.append(f"{rec['id']}/{key}")
+                    continue
+                if "probabilities" in want:
+                    p, q = list(want["probabilities"].values()), list(got["probabilities"].values())
+                    dps.append(sum(abs(a - b) for a, b in zip(p, q)) / len(p))
+                    agree += max(range(len(p)), key=p.__getitem__) == max(range(len(q)), key=q.__getitem__)
+                    total += 1
+                elif "noul" in want:
+                    dps.append(abs(want["noul"] - got["noul"]))
+                    agree += (want["noul"] >= 0.5) == (got["noul"] >= 0.5)
+                    total += 1
+        share = agree / max(total, 1)
+        mdp = sum(dps) / max(len(dps), 1)
+        ok = not bad_shape and share >= self.args.min_agreement and mdp <= self.args.max_dp
+        self.case(writer, model, "goldens", ok, f"top-1 {agree}/{total}, mean |dp| {mdp:.4f}", elapsed,
+                  "; ".join(bad_shape[:5]) or f"agreement {share:.3f} / mean |dp| {mdp:.4f} outside limits")
+
+    def _types(self, writer, model: str) -> None:
+        questions = {
+            "route": {"type": "choice", "instructions": "Which team should handle this?",
+                      "criteria": {"returns": "Refunds and replacements", "billing": "Payments", "general": None}},
+            "urgent": {"type": "noul", "instructions": "Does this need a reply today?"},
+            "anger": {"type": "score", "instructions": "How upset is the customer?", "criteria": ["calm", "annoyed", "angry"]},
+        }
+        status, body, t = self.post({"state": STATE, "questions": questions})
+        ok = (status == 200 and list(body.get("answers", {})) == list(questions)
+              and all(answer_shape_ok(a) and "error" not in a for a in body["answers"].values())
+              and body["usage"]["input_tokens"] > 0 and body["usage"]["output_tokens"] == 0)
+        self.case(writer, model, "types", ok, f"status {status}, {len(body.get('answers', {}))} answers", t, json.dumps(body)[:200])
+
+    def _two_letter(self, writer, model: str) -> None:
+        criteria = {f"option{i:02d}": f"Option number {i}" for i in range(27)}
+        status, body, t = self.post({"state": STATE, "questions": {"q": {"type": "choice", "criteria": criteria}}})
+        a = body.get("answers", {}).get("q", {})
+        ok = status == 200 and answer_shape_ok(a) and len(a.get("probabilities", {})) == 27
+        self.case(writer, model, "two-letter", ok, f"status {status}, {len(a.get('probabilities', {}))} options", t, json.dumps(body)[:200])
+
+    def _invalid(self, writer, model: str) -> None:
+        status, body, t = self.post({"questions": {"ok": {"type": "noul"}, "bad": {"type": "rank"}}})
+        ok = status == 422 and body == {"detail": {"invalid_questions": {
+            "bad": "unsupported question type 'rank' (choice, noul or score)"}}}
+        self.case(writer, model, "invalid", ok, f"status {status}", t, json.dumps(body)[:200])
+        status, body, t = self.post(b'{"questions": ')
+        ok = status == 422 and str(body.get("detail", "")).startswith("could not read the body as JSON")
+        self.case(writer, model, "not-json", ok, f"status {status}", t, json.dumps(body)[:200])
+
+    def _too_long(self, writer, model: str) -> None:
+        status, body, t = self.post({"state": "word " * 33000, "questions": {"q": {"type": "noul"}}})
+        a = body.get("answers", {}).get("q", {})
+        ok = status == 200 and a.get("error") == "max_length_exceeded" and body["usage"]["input_tokens"] == 0
+        self.case(writer, model, "too-long", ok, f"status {status}, {a.get('error')}", t, json.dumps(body)[:200])
+
+    # 96 x 64 PNG: a yellow box on a blue background
+    IMAGE = ("data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAGAAAABACAIAAABqVuVZAAAAiElEQVR42u3bMRXAIAxAQcKrjk5V1QkZlYAM"
+             "5FQEYmoApi4Z7u8ZuJeMxHm/RfsqAkCAAAECBAgQIK06/gzPp+V81dWHDXJigAABAiRAgAABAgQIECBAAgQIECBAgAABEiBAgAABAgQIkAAB"
+             "AgQIECBAgABpXfiSaYMAAQIECBAgQAIECBCgbH3ZCwYBFiTfPwAAAABJRU5ErkJggg==")
+
+    def _image(self, writer, model: str) -> None:
+        q = {"type": "choice", "instructions": "What is the main colour of the background?",
+             "criteria": {"blue": "Blue", "red": "Red", "green": "Green"}}
+        status, body, t = self.post({"questions": {"bg": q}, "images": [self.IMAGE]})
+        a = body.get("answers", {}).get("bg", {})
+        ok = status == 200 and answer_shape_ok(a) and "error" not in a and body["usage"]["input_tokens"] > 100
+        self.case(writer, model, "image", ok, f"status {status}, choice {a.get('choice')}", t, json.dumps(body)[:200])
+        status, body, t = self.post({"questions": {"bg": q}, "images": [self.IMAGE[:120]]})
+        ok = status == 422 and str(body.get("detail", "")).startswith("images[0]: ")
+        self.case(writer, model, "image-broken", ok, f"status {status}", t, json.dumps(body)[:200])
+
+    @staticmethod
+    def _decision(answer: dict):
+        """What an answer decides: the choice, the side of a noul, the most likely score level."""
+        if answer.get("type") == "noul":
+            return answer.get("noul", 0) >= 0.5
+        probabilities = answer.get("probabilities") or {}
+        return max(probabilities, key=probabilities.get) if probabilities else None
+
+    def _docs(self, writer, model: str) -> None:
+        doc = (REPO / "docs/docs/models/d3.md").read_text(encoding="utf-8")
+        m = re.search(r"-d '(\{.*?\})'", doc, re.S)
+        r = re.search(r"```json\s*(\{.*?\})\s*```", doc[m.end():], re.S) if m else None
+        if not m or not r:
+            self.case(writer, model, "docs", False, "no example request and reply in docs/docs/models/d3.md", 0.0, "missing")
+            return
+        request, documented = json.loads(m.group(1)), json.loads(r.group(1))
+        status, body, t = self.post(request)
+        answers = body.get("answers", {})
+        ok = (status == 200 and list(answers) == list(request["questions"])
+              and all(answer_shape_ok(a) and "error" not in a for a in answers.values()))
+        # the documented reply must decide what the server decides; its numbers are rounded
+        drift = [k for k, a in documented.get("answers", {}).items()
+                 if self._decision(a) != self._decision(answers.get(k, {}))]
+        ok = ok and not drift
+        self.case(writer, model, "docs", ok, f"status {status}" + (f", docs decide differently on {drift}" if drift else ""),
+                  t, json.dumps(body)[:200])
+
+    def run(self) -> None:
+        models = self.resolve_models()
+        if self.args.no_server and len(models) > 1:
+            raise SystemExit("--no-server attaches to one server, which serves one decision model; pass --models <tag>")
+        handle, writer = self.open_csv()
+        try:
+            for model in self.iter_models(models):
+                if model != self.needs_decision:   # each decision model needs its own server
+                    self.stop_server()
+                    self.needs_decision = model
+                    self.start_server(append_log=True)
+                print(f"\n--- Decision: {model} ---")
+                self._goldens(writer, model)
+                self._types(writer, model)
+                self._two_letter(writer, model)
+                self._invalid(writer, model)
+                self._too_long(writer, model)
+                self._image(writer, model)
+                self._docs(writer, model)
+        finally:
+            handle.close()
+        self.write_summary(models)
+
+
+if __name__ == "__main__":
+    raise SystemExit(run_task(DecisionSweep))

.github/workflows/model-sweep.yml

@@ -212,7 +212,7 @@ jobs:
       # others mid-sweep.
       fail-fast: false
       matrix:
-        # sweep_embedding.py and sweep_audio.py are ready and tested, but stay
+        # sweep_embedding.py, sweep_audio.py and sweep_decision.py are ready and tested, but stay
         # out of the matrix until llm and vision have proven themselves against
         # real hardware. Add them back here to enable them; nothing else needs
         # to change.
@@ -513,7 +513,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        # sweep_embedding.py and sweep_audio.py are ready and tested, but stay
+        # sweep_embedding.py, sweep_audio.py and sweep_decision.py are ready and tested, but stay
         # out of the matrix until llm and vision have proven themselves against
         # real hardware. Add them back here to enable them; nothing else needs
         # to change.

.github/workflows/ubuntu-build.yml

@@ -12,6 +12,19 @@ on:
   workflow_dispatch:
 
 jobs:
+  # The d3 System One contract (src/common/decision) against fixtures written by the upstream
+  # package: prompt bytes, request/question errors and answer objects. CPU only, no NPU.
+  d3-contract:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+      - name: Build and run the d3 contract test
+        run: |
+          cmake -S src/test/d3_contract -B build-d3-contract -DCMAKE_BUILD_TYPE=Release
+          cmake --build build-d3-contract -j"$(nproc)"
+          ctest --test-dir build-d3-contract --output-on-failure
+
   package-deb:
     runs-on: ubuntu-latest
     strategy:

docs/docs/benchmarks/d3_results.md

@@ -0,0 +1,46 @@
+---
+layout: docs
+title: Decision 3.0 (d3)
+parent: Benchmarks
+nav_order: 20
+---
+
+## ⚡ d3-lite on the NPU
+
+Accuracy and latency of `d3:lite` (vllm-sr/d3-lite v3.1.3, package D3-Lite-NPU2) served by FastFlowLM 1.0.10.
+
+> **Note:**
+> - AMD Ryzen AI MAX+ PRO 395 (Strix Halo, XDNA 2 NPU), 128 GB, Ubuntu 24.04, XRT, FLM's default NPU power mode (Performance).
+> - Latency is per request at the engine (no HTTP), median of 20 runs after warm-up; HTTP adds about a millisecond.
+
+### Accuracy against the upstream BF16 model
+
+| Set | Questions | Upstream BF16 | NPU (FLM) |
+|---|---:|---:|---:|
+| Decision Index public suite, stratified subset (gold accuracy) | 6,824 | 61.2% | 60.5% |
+| Same questions, top-1 agreement with BF16 | 7,226 | -- | 90.5% |
+| Hand-built contract set, top-1 agreement with BF16 | 311 | -- | 93.2% |
+
+Paired gold accuracy, upstream BF16 minus NPU: +0.69 pt (95% CI [+0.10, +1.28] pt).
+Prompts, token ids and answer objects are byte-identical to the upstream runtime.
+
+### Latency
+
+| Request | Median (ms) |
+|---|---:|
+| 1 question, 128 tokens | 109 |
+| 1 question, 330 tokens | 151 |
+| 1 question, 1,024 tokens | 430 |
+| 1 question, 2,048 tokens | 765 |
+| 1 question, 4,096 tokens | 1587 |
+| 1 question, 8,192 tokens | 3690 |
+| 1 question, 16,384 tokens | 8425 |
+| 1 question, 32,768 tokens | 22696 |
+| 4 questions, 330 tokens each | 536 |
+| 16 questions, 330 tokens each | 2067 |
+
+First request after `flm serve` is ready: 191 ms (steady state 168 ms over HTTP).
+
+Rows of 2,048 tokens and more: median of 15 runs (32,768: 3). From 1,536 tokens attention runs on the NPU (below
+that the CPU is faster); the first such request also loads the attention kernel (about 0.1 s) and from then on the
+decision model holds two NPU hardware contexts instead of one.

docs/docs/benchmarks/index.md

@@ -18,6 +18,7 @@ Browse detailed NPU benchmark results for each major model family supported by F
 - [Qwen3.6](qwen3.6_results/)
 - [Qwen3.8](qwen3.8_results/)
 - [gpt-oss](gpt-oss_results/)
+- [Decision 3.0 (d3)](d3_results/)
 - [LiquidAI/LFM2](lfm2_results/)
 - [Microsoft/Phi4](phi4_results/)
 - [Nanbeige4.1](nanbeige4.1_results/)