feat(benchmark): add point-cloud VQA pipeline

dimos/benchmark/evaluation/registry.py

@@ -31,6 +31,7 @@
 LOCAL_NAME_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
 BUILTIN_EVALUATIONS = {
     "frozen-integer-qa": ("dimos.benchmark.short_horizon_qa.evaluation:frozen_integer_qa"),
+    "point-cloud-vqa": ("dimos.benchmark.vqa.evaluation:point_cloud_vqa"),
 }
 
 

dimos/benchmark/vqa/evaluation.py

@@ -0,0 +1,144 @@
+# Copyright 2026 Dimensional Inc.
+"""Multiple-choice VQA evaluation plugin using public image artifacts only."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+import json
+from pathlib import Path
+import re
+
+import cv2
+from pydantic import BaseModel, ConfigDict, Field
+
+from dimos.benchmark.evaluation.models import (
+    ArtifactNativeResult,
+    ArtifactReference,
+    EvaluationReport,
+    SummaryItem,
+)
+from dimos.benchmark.evaluation.protocol import EvaluationContext
+from dimos.models.vl.openai import OpenAIVlModel
+from dimos.msgs.sensor_msgs.Image import Image
+
+
+class VqaEvaluationConfig(BaseModel):
+    """Location and vision model for a generated VQA evaluation dataset."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True, strict=True)
+
+    dataset: str = Field(min_length=1)
+    model: str = Field(default="gpt-4o-mini", min_length=1)
+
+
+class MultipleChoiceVqaEvaluation:
+    """Score a vision model against generated image-question-choice VQA cases."""
+
+    name = "point-cloud-vqa"
+    config_model: type[BaseModel] = VqaEvaluationConfig
+
+    def __init__(self, vision_factory: Callable[[str], OpenAIVlModel] | None = None) -> None:
+        self._vision_factory = vision_factory or (lambda model: OpenAIVlModel(model_name=model))
+
+    def run(self, config: BaseModel, context: EvaluationContext) -> EvaluationReport:
+        if not isinstance(config, VqaEvaluationConfig):
+            raise TypeError("point-cloud-vqa received the wrong configuration type")
+        dataset = Path(config.dataset).expanduser()
+        if not dataset.is_absolute():
+            dataset = context.spec_dir / dataset
+        dataset = dataset.resolve()
+        cases = _load_jsonl(dataset / "cases.jsonl")
+        labels = {item["id"]: item["answer"] for item in _load_jsonl(dataset / "labels.jsonl")}
+        model = self._vision_factory(config.model)
+        results = [_evaluate_case(dataset, model, case, labels) for case in cases]
+        artifact = context.workspace / "vqa-results.json"
+        artifact.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8")
+        passed = sum(item["passed"] for item in results)
+        total = len(results)
+        return EvaluationReport(
+            summary=(
+                SummaryItem(key="cases", label="Cases", value=total),
+                SummaryItem(key="passed", label="Passed", value=passed),
+                SummaryItem(
+                    key="accuracy", label="Accuracy", value=passed / total if total else 0.0
+                ),
+            ),
+            native_result=ArtifactNativeResult(
+                artifact=ArtifactReference(
+                    path=artifact.relative_to(context.workspace).as_posix(),
+                    label="VQA case results",
+                    media_type="application/json",
+                )
+            ),
+            artifacts=(
+                ArtifactReference(
+                    path=artifact.relative_to(context.workspace).as_posix(),
+                    label="VQA case results",
+                    media_type="application/json",
+                ),
+            ),
+        )
+
+
+def _evaluate_case(
+    dataset: Path, model: OpenAIVlModel, case: dict[str, object], labels: dict[str, str]
+) -> dict[str, object]:
+    case_id = _required_string(case, "id")
+    choices = _choices(case)
+    expected = labels.get(case_id)
+    if expected is None:
+        raise ValueError(f"missing private label for case {case_id}")
+    if expected not in choices:
+        raise ValueError(f"private label for {case_id} is not an allowed choice")
+    image = cv2.imread(str(dataset / _required_string(case, "image")))
+    if image is None:
+        raise ValueError(f"unable to load public image for case {case_id}")
+    prompt = (
+        f"{_required_string(case, 'question')}\n\n"
+        f"Choices: {', '.join(choices)}.\n"
+        "Use only the supplied image. End with exactly `ANSWER: <choice>`."
+    )
+    response = model.query(Image.from_numpy(image), prompt)
+    answer = _parse_choice(response, choices)
+    return {
+        "id": case_id,
+        "expected": expected,
+        "answer": answer,
+        "passed": answer == expected,
+        "raw_response": response,
+    }
+
+
+def _load_jsonl(path: Path) -> list[dict[str, object]]:
+    if not path.is_file():
+        raise ValueError(f"missing VQA dataset file: {path}")
+    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]
+
+
+def _required_string(item: dict[str, object], key: str) -> str:
+    value = item.get(key)
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"VQA case requires non-empty {key}")
+    return value
+
+
+def _choices(case: dict[str, object]) -> tuple[str, ...]:
+    value = case.get("choices")
+    if (
+        not isinstance(value, list)
+        or len(value) < 2
+        or not all(isinstance(item, str) for item in value)
+    ):
+        raise ValueError("VQA case requires at least two string choices")
+    return tuple(value)
+
+
+def _parse_choice(response: str, choices: tuple[str, ...]) -> str | None:
+    match = re.search(r"^ANSWER:\s*(.+?)\s*$", response, re.MULTILINE)
+    if match is None:
+        return None
+    answer = match.group(1)
+    return answer if answer in choices else None
+
+
+point_cloud_vqa = MultipleChoiceVqaEvaluation()

dimos/benchmark/vqa/generation/adapters.py

@@ -0,0 +1,51 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""MoonDream and EdgeTAM adapters for the single-frame VQA pipeline."""
+
+from __future__ import annotations
+
+from dimos.models.segmentation.edge_tam import EdgeTAMImageSegmenter
+from dimos.models.vl.moondream import MoondreamVlModel
+from dimos.msgs.sensor_msgs.Image import Image
+from dimos.perception.detection.type.detection2d.imageDetections2D import ImageDetections2D
+
+
+class MoondreamObjectDetector:
+    """Adapt MoonDream's query detection API to the grounding interface."""
+
+    def __init__(self, model: MoondreamVlModel) -> None:
+        self._model = model
+
+    def detect(self, image: Image, query: str) -> ImageDetections2D:
+        return self._model.query_detections(image, query)
+
+    def locate(self, image: Image, query: str) -> ImageDetections2D:
+        points = self._model.query_points(image, f"center of the {query}")
+        for point in points:
+            point.name = query
+        return points
+
+
+class EdgeTamObjectSegmenter:
+    """Adapt EdgeTAM single-image segmentation to the grounding interface."""
+
+    def __init__(self, segmenter: EdgeTAMImageSegmenter) -> None:
+        self._segmenter = segmenter
+
+    def segment(self, detections: ImageDetections2D) -> ImageDetections2D:
+        return self._segmenter.segment(detections)
+
+    def segment_points(self, points: ImageDetections2D) -> ImageDetections2D:
+        return self._segmenter.segment_points(points)

dimos/benchmark/vqa/generation/dataset.py

@@ -0,0 +1,149 @@
+# Copyright 2026 Dimensional Inc.
+"""Persist VQA generation evidence and a simple multiple-choice evaluation export."""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+import json
+from pathlib import Path
+from typing import Any
+
+import cv2
+
+from dimos.benchmark.vqa.models import (
+    AcceptedOracleResult,
+    BooleanAnswerContract,
+    CalibratedFrame,
+    GroundTruthResult,
+    QuestionIntent,
+    QuestionProposal,
+    RejectedOracleResult,
+)
+
+
+def write_frame_record(
+    output: Path,
+    frame: CalibratedFrame,
+    recording: str,
+    frame_index: int,
+    intents: list[QuestionIntent | QuestionProposal],
+    results: list[GroundTruthResult | AcceptedOracleResult | RejectedOracleResult],
+    metadata: dict[str, Any],
+) -> None:
+    """Write one frame's public cases alongside its private generation audit record."""
+    output.mkdir(parents=True, exist_ok=False)
+    image_path = output / "image.jpg"
+    if not cv2.imwrite(str(image_path), frame.image.data):
+        raise RuntimeError(f"failed to write {image_path}")
+    accepted = [result for result in results if _is_accepted(result)]
+    cases, labels = _evaluation_rows(frame.id, accepted)
+    _write_json(
+        output / "frame.json",
+        {
+            "schema_version": "1.0",
+            "frame_id": frame.id,
+            "recording": recording,
+            "frame_index": frame_index,
+            "image": image_path.name,
+            "question_count": len(intents),
+            "accepted_question_count": len(accepted),
+            "rejected_question_count": len(results) - len(accepted),
+            **metadata,
+        },
+    )
+    _write_json(output / "ground_truth.json", [_private_result(item) for item in results])
+    _write_json(output / "cases.json", cases)
+    _write_json(output / "labels.json", labels)
+
+
+def write_dataset_manifest(output: Path) -> dict[str, int]:
+    """Build aggregate public cases and private labels from completed frame records."""
+    frames = sorted(path for path in output.glob("frame-*") if (path / "frame.json").is_file())
+    case_rows: list[dict[str, Any]] = []
+    label_rows: list[dict[str, Any]] = []
+    accepted = 0
+    rejected = 0
+    for path in frames:
+        frame = json.loads((path / "frame.json").read_text())
+        case_rows.extend(
+            {**case, "image": f"{path.name}/{case['image']}"}
+            for case in json.loads((path / "cases.json").read_text())
+        )
+        label_rows.extend(json.loads((path / "labels.json").read_text()))
+        accepted += frame["accepted_question_count"]
+        rejected += frame["rejected_question_count"]
+    _write_jsonl(output / "cases.jsonl", case_rows)
+    _write_jsonl(output / "labels.jsonl", label_rows)
+    return {
+        "frame_count": len(frames),
+        "accepted_question_count": accepted,
+        "rejected_question_count": rejected,
+    }
+
+
+def _is_accepted(result: GroundTruthResult | AcceptedOracleResult | RejectedOracleResult) -> bool:
+    return isinstance(result, AcceptedOracleResult) or (
+        isinstance(result, GroundTruthResult) and result.status == "answered"
+    )
+
+
+def _evaluation_rows(
+    frame_id: str, results: list[GroundTruthResult | AcceptedOracleResult | RejectedOracleResult]
+) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
+    cases: list[dict[str, Any]] = []
+    labels: list[dict[str, str]] = []
+    for result in results:
+        if isinstance(result, RejectedOracleResult):
+            continue
+        if isinstance(result, AcceptedOracleResult):
+            contract = result.answer_contract
+            choices = (
+                ("yes", "no") if isinstance(contract, BooleanAnswerContract) else contract.choices
+            )
+            case_id = f"{frame_id}-{result.proposal.id}"
+            question = result.proposal.question
+            answer = result.answer
+        else:
+            case_id = result.question.id
+            question = result.question.question
+            choices = result.question.allowed_answers
+            answer = result.answer
+        if answer is None or answer not in choices:
+            raise ValueError(f"accepted VQA case {case_id} must have a choice answer")
+        cases.append(
+            {"id": case_id, "image": "image.jpg", "question": question, "choices": choices}
+        )
+        labels.append({"id": case_id, "answer": answer})
+    return cases, labels
+
+
+def _private_result(
+    result: GroundTruthResult | AcceptedOracleResult | RejectedOracleResult,
+) -> dict[str, Any]:
+    if isinstance(result, AcceptedOracleResult):
+        return {
+            "status": "answered",
+            "answer": result.answer,
+            "proposal": asdict(result.proposal),
+            "answer_contract": asdict(result.answer_contract),
+            "evidence_ids": result.evidence_ids,
+            "tool_results": [asdict(item) for item in result.tool_results],
+            "trace": [asdict(item) for item in result.trace],
+        }
+    if isinstance(result, RejectedOracleResult):
+        return {
+            "status": "rejected",
+            "reason": result.reason,
+            "proposal": asdict(result.proposal),
+            "tool_results": [asdict(item) for item in result.tool_results],
+            "trace": [asdict(item) for item in result.trace],
+        }
+    return asdict(result)
+
+
+def _write_json(path: Path, payload: Any) -> None:
+    path.write_text(json.dumps(payload, indent=2) + "\n")
+
+
+def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
+    path.write_text("".join(f"{json.dumps(row, sort_keys=True)}\n" for row in rows))