[Benchmark][MiniCPM-o] Add Omni-DuplexEval support (#6634)

Signed-off-by: Zhanshenzhui <99891002+zyforsure@users.noreply.github.com>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>

examples/online_serving/minicpmo/README.md

@@ -192,6 +192,64 @@ stronger `response-required` mode is diagnostic: it requires a purpose-built
 two-response WAV, its `--input-sha256`, and an
 `--expect-second-response-substring` value.
 
+## Run Omni-DuplexEval
+
+Omni-DuplexEval generation uses the vLLM-Omni native MiniCPM-o duplex
+endpoint. Evaluation uses a separate OpenAI-compatible multimodal judge
+served from the same vLLM-Omni environment. This validates the vLLM-Omni
+duplex client and local judge integration; it does not reproduce the paper's
+original MiniCPM-o inference implementation.
+
+Start the duplex generation server:
+
+```bash
+vllm serve openbmb/MiniCPM-o-4_5 --omni --trust-remote-code \
+    --deploy-config vllm_omni/deploy/minicpmo_4_5_duplex.yaml \
+    --served-model-name openbmb/MiniCPM-o-4_5 \
+    --host 0.0.0.0 --port 8099
+```
+
+Start a separate multimodal judge. The allowed path must contain any local
+videos passed with `--judge-video-mode video_url`:
+
+```bash
+vllm serve Qwen/Qwen2.5-VL-7B-Instruct \
+    --served-model-name Qwen/Qwen2.5-VL-7B-Instruct \
+    --allowed-local-media-path /data/omni-duplex-eval \
+    --host 0.0.0.0 --port 8000
+```
+
+Generate, evaluate, and summarize:
+
+```bash
+vllm bench omni-duplex-eval --omni generate \
+    --url ws://127.0.0.1:8099/v1/realtime?duplex=1 \
+    --model openbmb/MiniCPM-o-4_5 \
+    --ref-audio /data/ref.wav \
+    --dataset Hothan/Omni-DuplexEval \
+    --concurrency 2 \
+    --response-root /data/omni-duplex-eval/responses
+
+vllm bench omni-duplex-eval --omni evaluate \
+    --dataset Hothan/Omni-DuplexEval \
+    --response-root /data/omni-duplex-eval/responses \
+    --score-root /data/omni-duplex-eval/scores \
+    --judge-base-url http://127.0.0.1:8000 \
+    --judge-model Qwen/Qwen2.5-VL-7B-Instruct \
+    --judge-video-mode video_url \
+    --eval-workers 4
+
+vllm bench omni-duplex-eval --omni summarize \
+    --score-root /data/omni-duplex-eval/scores
+```
+
+The dataset names such as `RTD_OCR` and `PR_correction` are Hugging Face
+splits, not dataset configurations. Pass one with `--split`, or omit it to
+run all nine splits. `--limit 1` is useful for a smoke test. Realtime
+generation records `clock=media`; artifacts generated with
+`--pace as-fast-as-possible` record `clock=invalid` and evaluation rejects
+them unless `--allow-invalid-clock` is explicit.
+
 ## Related examples
 
 - [Offline MiniCPM-o inference](../../offline_inference/minicpmo/)

tests/benchmarks/duplex/test_omni_duplex_eval.py

@@ -0,0 +1,143 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+from vllm_omni.benchmarks.duplex import omni_duplex_eval_eval as eval_module
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_clock import (
+    extract_timed_sentences,
+    normalize_response_items,
+    split_text,
+    validate_clock,
+)
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import (
+    DuplexSample,
+    canonical_task_type,
+    family_for_split,
+    load_samples,
+    task_type_for_split,
+)
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_metrics import (
+    PROTOCOL_PIN,
+    build_content_prompt,
+    build_reminder_prompt,
+    build_temporal_prompt,
+    parse_judge_json,
+    reminder_window,
+    summarize_pr_results,
+    temporal_window,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+def test_dataset_mapping_and_manifest(tmp_path):
+    manifest = tmp_path / "samples.json"
+    manifest.write_text(json.dumps([{"id": "a", "split": "RTD_OCR", "video": "clip.mp4"}]), encoding="utf-8")
+    sample = load_samples(manifest, media_root=tmp_path)[0]
+    assert sample.family == "rtd"
+    assert sample.video == str(tmp_path / "clip.mp4")
+    assert family_for_split("PR_event_reminder") == "pr"
+    assert canonical_task_type("post-event-reminder") == "post_event_reminder"
+    assert task_type_for_split("PR_correction") == "correction"
+    assert load_samples([{"id": "pr", "split": "PR_correction"}])[0].task_type == "correction"
+
+
+def test_hugging_face_subset_names_are_splits(monkeypatch):
+    calls = []
+
+    def fake_load_dataset(name, **kwargs):
+        calls.append((name, kwargs))
+        return [{"id": "pr", "question_text": "Correct this."}]
+
+    monkeypatch.setitem(sys.modules, "datasets", SimpleNamespace(load_dataset=fake_load_dataset))
+    sample = load_samples("Hothan/Omni-DuplexEval", split="PR_correction")[0]
+    assert calls == [("Hothan/Omni-DuplexEval", {"split": "PR_correction"})]
+    assert sample.split == "PR_correction"
+    assert sample.task_type == "correction"
+
+
+def test_response_aliases_and_clock_guard():
+    assert split_text("One. Two!") == ["One.", "Two!"]
+    assert normalize_response_items({"chunks": [{"text": "x", "current_time": 800}]}) == [
+        {"sentence": "x", "start": 800.0, "end": 800.0}
+    ]
+    with pytest.raises(ValueError, match="clock=invalid"):
+        validate_clock({"clock": "invalid"})
+    validate_clock({"clock": "invalid"}, allow_invalid=True)
+    timed = extract_timed_sentences([{"type": "response.output_text.delta", "delta": "Done.", "_media_clock_ms": 800}])
+    assert [item.as_dict() for item in timed] == [{"sentence": "Done.", "start": 0.8, "end": 0.8}]
+    with pytest.raises(ValueError, match="only clock=media"):
+        extract_timed_sentences([], clock="wall")
+
+
+def test_protocol_windows_and_parsing():
+    assert PROTOCOL_PIN == "ca3c122b4d4bf67afd6b18ea5e724b4561bdde48"
+    assert temporal_window(4, 5, 10) == (2.0, 3.0)
+    assert temporal_window(1, 1.2, 10) == (0.0, 0.5)
+    assert reminder_window(5) == (5.0, 15.0)
+    assert parse_judge_json('noise {"content_score": 2.345, "is_relevant": 1}') == {
+        "content_score": 2.35,
+        "is_relevant": 1,
+    }
+    assert parse_judge_json('noise "success_score": 1') == {"success_score": 1}
+    assert summarize_pr_results([{"task_type": "correction", "all_success": 1}])["mean_all_success"] == 1.0
+
+
+def test_protocol_prompt_wording_is_pinned():
+    prompts = [
+        build_temporal_prompt(1, 2, "response", "question"),
+        build_content_prompt("response", "question", ["one", "two"]),
+        build_reminder_prompt("instruction", "response", "proactive_reminder"),
+        build_reminder_prompt("instruction", "response", "post_event_reminder"),
+        build_reminder_prompt("instruction", "response", "correction", "answer"),
+    ]
+    assert [hashlib.sha256(prompt.encode()).hexdigest() for prompt in prompts] == [
+        "5f7f5fcb4a37e4d2eea8ba23beabf74630e6fc2fb5b2661ceb94403254bb04d5",
+        "1fccbed6c1ed8ea1c76d5564d7289ed6adca3678f6a0da2cc735942123d64457",
+        "d0a0e3a9d873d2f368c5c99b9fe46681f683cc36a9fe0e09b99d6cbca1d6dd0c",
+        "08f9951243c859d271f82ee53da22cd0bdab1fe93c6ca04f4870d3b959c0803a",
+        "3de131a7e36cb4b8ac369bb05e87ede518be20221e214df3644211c179131a73",
+    ]
+
+
+def test_frame_sample_content_passes_frames(tmp_path, monkeypatch):
+    response = tmp_path / "response.json"
+    response.write_text(json.dumps([{"sentence": "A person moves.", "start": 1, "end": 2}]), encoding="utf-8")
+    response.with_name("response.meta.json").write_text(json.dumps({"clock": "media"}), encoding="utf-8")
+    video = tmp_path / "video.mp4"
+    video.write_bytes(b"video")
+    monkeypatch.setattr(eval_module, "extract_jpeg", lambda *args, **kwargs: b"\xff\xd8jpeg")
+
+    class Judge:
+        def temporal(self, *args, **kwargs):
+            return '{"temporal_score": 3, "is_relevant": 1}'
+
+        def content(self, prompt, video, frames=None, *, mode="video_url"):
+            assert mode == "frame-sample"
+            assert frames == [b"\xff\xd8jpeg", b"\xff\xd8jpeg"]
+            return '{"content_score": 3}'
+
+    sample = DuplexSample(
+        id="rtd",
+        split="RTD_OCR",
+        family="rtd",
+        task_type=None,
+        video=video,
+        video_duration=4,
+    )
+    score = eval_module.evaluate_sample(
+        sample,
+        response,
+        tmp_path / "score.json",
+        Judge(),
+        judge_video_mode="frame-sample",
+    )
+    assert score["content"]["frame_count"] == 2

tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py

@@ -0,0 +1,194 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+import pytest
+
+from vllm_omni.benchmarks.duplex import omni_duplex_eval_runner as runner
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DuplexSample
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge
+from vllm_omni.entrypoints.cli.benchmark import omni_duplex_eval as cli
+from vllm_omni.entrypoints.cli.benchmark.omni_duplex_eval import OmniDuplexEvalSubcommand
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.benchmark]
+
+
+def test_cli_generate_evaluate_summarize_flow(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys):
+    manifest = tmp_path / "manifest.json"
+    manifest.write_text(
+        json.dumps(
+            [
+                {
+                    "id": "sample-1",
+                    "split": "PR_correction",
+                    "question_text": "What changed?",
+                    "answer1": "The object moved.",
+                }
+            ]
+        ),
+        encoding="utf-8",
+    )
+    response_root = tmp_path / "responses"
+    score_root = tmp_path / "scores"
+
+    async def fake_generate(sample, *, output_root, **kwargs):
+        output = Path(output_root) / sample.split / f"{sample.id}.json"
+        output.parent.mkdir(parents=True, exist_ok=True)
+        output.write_text(json.dumps([{"sentence": "It moved.", "start": 0, "end": 1}]), encoding="utf-8")
+        output.with_name(output.stem + ".meta.json").write_text(json.dumps({"clock": "media"}), encoding="utf-8")
+        return output
+
+    class FakeJudge:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def chat(self, *args, **kwargs):
+            return '{"success_score": 1, "is_relevant": 1}'
+
+    monkeypatch.setattr(cli, "generate_sample", fake_generate)
+    monkeypatch.setattr(cli, "DuplexJudge", FakeJudge)
+
+    parser = argparse.ArgumentParser()
+    OmniDuplexEvalSubcommand.add_cli_args(parser)
+
+    common = ["--dataset", str(manifest), "--family", "pr"]
+    generate = parser.parse_args(
+        [
+            "generate",
+            *common,
+            "--model",
+            "mock",
+            "--ref-audio",
+            str(manifest),
+            "--response-root",
+            str(response_root),
+            "--concurrency",
+            "2",
+        ]
+    )
+    OmniDuplexEvalSubcommand.cmd(generate)
+    evaluate = parser.parse_args(
+        [
+            "evaluate",
+            *common,
+            "--response-root",
+            str(response_root),
+            "--score-root",
+            str(score_root),
+            "--judge-model",
+            "mock-judge",
+            "--eval-workers",
+            "2",
+        ]
+    )
+    OmniDuplexEvalSubcommand.cmd(evaluate)
+    OmniDuplexEvalSubcommand.cmd(parser.parse_args(["summarize", "--score-root", str(score_root)]))
+
+    summary = json.loads(capsys.readouterr().out)
+    assert summary["samples"] == 1
+    assert summary["pr"]["mean_all_success"] == 1.0
+
+
+@pytest.mark.asyncio
+async def test_generate_exercises_realtime_socket_and_media_clock(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    import websockets
+
+    received = []
+
+    async def handler(websocket):
+        response_started = False
+        async for raw in websocket:
+            event = json.loads(raw)
+            received.append(event)
+            if event["type"] == "session.update":
+                await websocket.send(json.dumps({"type": "session.created"}))
+            elif event["type"] == "input_audio_buffer.append" and not response_started:
+                response_started = True
+                await websocket.send(json.dumps({"type": "response.created", "response": {"id": "r1"}}))
+                await websocket.send(
+                    json.dumps({"type": "response.output_text.delta", "response_id": "r1", "delta": "Done."})
+                )
+                await websocket.send(
+                    json.dumps(
+                        {
+                            "type": "response.audio.delta",
+                            "response_id": "r1",
+                            "delta": base64.b64encode(b"\0\0" * 240).decode(),
+                            "sample_rate_hz": 24_000,
+                        }
+                    )
+                )
+            elif event["type"] == "input_audio_buffer.commit":
+                await websocket.send(json.dumps({"type": "response.done", "response": {"id": "r1"}}))
+            elif event["type"] == "session.close":
+                await websocket.send(json.dumps({"type": "session.closed"}))
+                return
+
+    monkeypatch.setattr(runner, "read_audio_pcm16", lambda path: b"\0\0" * (16_000 * 8 // 10))
+    monkeypatch.setattr(runner, "video_duration", lambda path: 0.8)
+    monkeypatch.setattr(runner, "iter_jpegs", lambda *args, **kwargs: iter([(0.0, b"jpeg")]))
+    audio = tmp_path / "question.wav"
+    video = tmp_path / "video.mp4"
+    ref = tmp_path / "ref.wav"
+    for path in (audio, video, ref):
+        path.write_bytes(b"data")
+    sample = DuplexSample("sample", "PR_correction", "pr", "correction", video, audio)
+
+    async with websockets.serve(handler, "127.0.0.1", 0) as server:
+        port = server.sockets[0].getsockname()[1]
+        output = await runner.generate_sample(
+            sample,
+            url=f"ws://127.0.0.1:{port}/v1/realtime?duplex=1",
+            model="mock",
+            ref_audio=ref,
+            output_root=tmp_path / "responses",
+        )
+
+    assert json.loads(output.read_text(encoding="utf-8")) == [{"sentence": "Done.", "start": 0.8, "end": 0.8}]
+    meta = json.loads(output.with_name("sample.meta.json").read_text(encoding="utf-8"))
+    assert meta["response_done"] is True
+    assert meta["drain_timeout"] is None
+    assert any(event["type"] == "playback.ack" for event in received)
+
+
+def test_judge_exercises_openai_http_schema():
+    requests = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_POST(self):
+            length = int(self.headers["Content-Length"])
+            requests.append((self.path, self.headers["Authorization"], json.loads(self.rfile.read(length))))
+            payload = json.dumps({"choices": [{"message": {"content": '{"success_score": 1}'}}]}).encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        judge = DuplexJudge(f"http://127.0.0.1:{server.server_port}", "judge", api_key="token")
+        assert judge.chat("prompt") == '{"success_score": 1}'
+    finally:
+        server.shutdown()
+        thread.join()
+        server.server_close()
+
+    path, authorization, payload = requests[0]
+    assert path == "/v1/chat/completions"
+    assert authorization == "Bearer token"
+    assert payload["model"] == "judge"
+    assert payload["messages"] == [{"role": "user", "content": "prompt"}]

vllm_omni/benchmarks/duplex/__init__.py

@@ -0,0 +1,9 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""CPU-friendly Omni-DuplexEval generation and scoring helpers."""
+
+from .omni_duplex_eval_dataset import DuplexSample, load_samples
+from .omni_duplex_eval_metrics import PROTOCOL_PIN
+
+__all__ = ["DuplexSample", "PROTOCOL_PIN", "load_samples"]

vllm_omni/benchmarks/duplex/omni_duplex_eval_clock.py

@@ -0,0 +1,106 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Media-clock sentence extraction and response JSON normalization."""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from typing import Any
+
+import regex as re
+
+_TERMINAL = re.compile(r"(.+?[.!?。！？]+)(?:\s+|$)", re.S)
+
+
+@dataclass
+class TimedSentence:
+    sentence: str
+    start: float
+    end: float
+
+    def as_dict(self) -> dict[str, Any]:
+        return {"sentence": self.sentence, "start": round(self.start, 3), "end": round(self.end, 3)}
+
+
+def split_text(text: str) -> list[str]:
+    text = str(text or "").strip()
+    if not text:
+        return []
+    parts: list[str] = []
+    position = 0
+    for match in _TERMINAL.finditer(text):
+        parts.append(match.group(1).strip())
+        position = match.end()
+    tail = text[position:].strip()
+    if tail:
+        parts.append(tail)
+    return [part for part in parts if part]
+
+
+def extract_timed_sentences(events: Iterable[dict[str, Any]], *, clock: str = "media") -> list[TimedSentence]:
+    """Split output deltas while anchoring text to input ``audio_end_ms``."""
+    if clock != "media":
+        raise ValueError("only clock=media is supported")
+    sentences: list[TimedSentence] = []
+    pending = ""
+    start: float | None = None
+    last: float | None = None
+    for event in events:
+        event_type = event.get("type")
+        if event_type not in {"response.output_text.delta", "response.audio_transcript.delta", "response.text.delta"}:
+            continue
+        delta = event.get("delta")
+        if not isinstance(delta, str) or not delta.strip():
+            continue
+        timestamp = event.get("audio_end_ms")
+        if timestamp is None:
+            timestamp = event.get("_media_clock_ms", event.get("current_time_ms", 0.0))
+        try:
+            # Every accepted event field is explicitly millisecond-valued.
+            now = float(timestamp) / 1000.0
+        except (TypeError, ValueError):
+            now = last or 0.0
+        pending += delta
+        start = now if start is None else start
+        last = now
+        pieces = split_text(pending)
+        complete = pieces if pieces and re.search(r"[.!?。！？]$", pending.strip()) else pieces[:-1]
+        if complete:
+            for piece in complete:
+                sentences.append(TimedSentence(piece, start, max(start, now)))
+            consumed = " ".join(complete)
+            pending = pending[len(consumed) :].strip()
+            start = now if pending else None
+    if pending.strip():
+        sentences.append(TimedSentence(pending.strip(), start or last or 0.0, last or start or 0.0))
+    return sentences
+
+
+def normalize_response_items(raw: Any) -> list[dict[str, Any]]:
+    if isinstance(raw, dict):
+        items = raw.get("sentences", raw.get("chunks"))
+    else:
+        items = raw
+    if not isinstance(items, list):
+        raise ValueError("response JSON must be a list or contain sentences/chunks")
+    normalized = []
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        text = item.get("sentence", item.get("text", item.get("content", "")))
+        if not isinstance(text, str) or not text.strip():
+            continue
+        try:
+            start = float(item.get("start", item.get("current_time", item.get("time", 0.0))) or 0.0)
+            end = float(item.get("end", item.get("current_time", item.get("time", start))) or start)
+        except (TypeError, ValueError):
+            continue
+        normalized.append({"sentence": text.strip(), "start": start, "end": max(start, end)})
+    return normalized
+
+
+def validate_clock(meta: dict[str, Any], *, allow_invalid: bool = False) -> None:
+    if meta.get("clock") == "invalid" and not allow_invalid:
+        raise ValueError("generate artifact has clock=invalid; pass --allow-invalid-clock to score it")

vllm_omni/benchmarks/duplex/omni_duplex_eval_dataset.py

@@ -0,0 +1,193 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Dataset and split normalization for Omni-DuplexEval.
+
+The loader accepts a Hugging Face dataset id, a JSON/JSONL manifest, or an
+already materialized iterable.  Media is deliberately resolved at use time so
+the benchmark remains usable in air-gapped environments.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+DEFAULT_DATASET = "Hothan/Omni-DuplexEval"
+RTD_SPLITS = frozenset(
+    {
+        "RTD_world_knowledge",
+        "RTD_counting",
+        "RTD_fine_grained_movement",
+        "RTD_interaction_relation",
+        "RTD_OCR",
+        "RTD_Omni",
+    }
+)
+PR_SPLITS = frozenset({"PR_correction", "PR_event_reminder", "PR_post_event_reminder"})
+_TASK_ALIASES = {
+    "correction": "correction",
+    "pr_correction": "correction",
+    "event_reminder": "proactive_reminder",
+    "pr_event_reminder": "proactive_reminder",
+    "proactive_reminder": "proactive_reminder",
+    "post_event_reminder": "post_event_reminder",
+    "pr_post_event_reminder": "post_event_reminder",
+}
+
+
+def canonical_task_type(value: str) -> str:
+    key = str(value).strip().lower().replace("-", "_").replace(" ", "_")
+    if key not in _TASK_ALIASES:
+        raise ValueError(f"unsupported Omni-DuplexEval task type: {value!r}")
+    return _TASK_ALIASES[key]
+
+
+def task_type_for_split(split: str) -> str:
+    """Match the task routing used by the official HF batch evaluator."""
+    key = str(split).strip().lower()
+    if "correction" in key:
+        return "correction"
+    if "post_event" in key:
+        return "post_event_reminder"
+    if key.startswith("pr"):
+        return "proactive_reminder"
+    raise ValueError(f"cannot infer proactive-reminder task from split {split!r}")
+
+
+def family_for_split(split: str, task_type: str | None = None) -> str:
+    if split in RTD_SPLITS or str(split).upper().startswith("RTD"):
+        return "rtd"
+    if split in PR_SPLITS or str(split).upper().startswith("PR"):
+        return "pr"
+    if task_type:
+        return "rtd" if str(task_type).lower().startswith("rtd") else "pr"
+    raise ValueError(f"cannot infer benchmark family from split {split!r}")
+
+
+def _value(row: dict[str, Any], *keys: str, default: Any = None) -> Any:
+    for key in keys:
+        if key in row and row[key] is not None:
+            return row[key]
+    return default
+
+
+def _float(value: Any, default: float | None = None) -> float | None:
+    try:
+        return None if value is None else float(value)
+    except (TypeError, ValueError):
+        return default
+
+
+@dataclass(frozen=True)
+class DuplexSample:
+    id: str
+    split: str
+    family: str
+    task_type: str | None
+    video: Any
+    question_audio: Any = None
+    question_text: str = ""
+    answer1: str = ""
+    answer2: str = ""
+    reminder1: Any = None
+    reminder2: Any = None
+    video_duration: float | None = None
+    video_type: str = ""
+    raw: dict[str, Any] | None = None
+
+    @classmethod
+    def from_row(cls, row: dict[str, Any], *, split: str | None = None, media_root: Path | None = None) -> DuplexSample:
+        chosen_split = str(split or _value(row, "split", "subset", "config", default=""))
+        task_value = _value(row, "task_type", "task", "type")
+        family = str(_value(row, "family", default="") or family_for_split(chosen_split, task_value))
+        task = None
+        if family == "pr":
+            task = canonical_task_type(str(task_value)) if task_value else task_type_for_split(chosen_split)
+        sample_id = str(_value(row, "id", "sample_id", "uid", "name", default=""))
+        video = _value(row, "video", "video_path", "video_file")
+        audio = _value(row, "question_audio", "audio", "question_wav")
+        if media_root:
+
+            def resolve(value: Any) -> Any:
+                if not isinstance(value, str):
+                    return value
+                candidate = Path(value).expanduser()
+                return str(candidate if candidate.is_absolute() else media_root / candidate)
+
+            video, audio = resolve(video), resolve(audio)
+        return cls(
+            id=sample_id,
+            split=chosen_split,
+            family=family,
+            task_type=task,
+            video=video,
+            question_audio=audio,
+            question_text=str(_value(row, "question_text", "question", "instruction", default="") or ""),
+            answer1=str(_value(row, "answer1", "answer", "ground_answer", default="") or ""),
+            answer2=str(_value(row, "answer2", default="") or ""),
+            reminder1=_value(row, "reminder1", "reminder_1"),
+            reminder2=_value(row, "reminder2", "reminder_2"),
+            video_duration=_float(_value(row, "video_duration", "duration")),
+            video_type=str(_value(row, "video_type", default="") or ""),
+            raw=dict(row),
+        )
+
+
+def _read_manifest(path: Path) -> list[dict[str, Any]]:
+    if path.suffix.lower() == ".jsonl":
+        return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if isinstance(payload, dict):
+        payload = payload.get("data", payload.get("samples", [payload]))
+    if not isinstance(payload, list):
+        raise ValueError("manifest must contain a JSON list")
+    return payload
+
+
+def load_samples(
+    dataset: str | Path | Iterable[dict[str, Any]] = DEFAULT_DATASET,
+    *,
+    split: str | None = None,
+    family: str = "all",
+    media_root: str | Path | None = None,
+    limit: int | None = None,
+    ids: Iterable[str] | None = None,
+) -> list[DuplexSample]:
+    if isinstance(dataset, (str, Path)) and Path(str(dataset)).exists():
+        rows = _read_manifest(Path(str(dataset)))
+    elif not isinstance(dataset, (str, Path)):
+        rows = list(dataset)
+    else:
+        try:
+            from datasets import load_dataset
+        except ImportError as exc:
+            raise RuntimeError("datasets is required for Hugging Face loading; use a local manifest instead") from exc
+        kwargs: dict[str, Any] = {}
+        if split and split != "all":
+            kwargs["split"] = split
+        loaded = load_dataset(str(dataset), **kwargs)
+        if isinstance(loaded, dict):
+            rows = []
+            for name, table in loaded.items():
+                rows.extend({**dict(row), "split": name} for row in table)
+        else:
+            rows = [dict(row) for row in loaded]
+    wanted = set(str(item) for item in ids) if ids else None
+    root = Path(media_root).expanduser() if media_root else None
+    result = []
+    for row in rows:
+        sample = DuplexSample.from_row(row, split=None if split in (None, "all") else split, media_root=root)
+        if split and split != "all" and sample.split != split:
+            continue
+        if family != "all" and sample.family != family:
+            continue
+        if wanted is not None and sample.id not in wanted:
+            continue
+        result.append(sample)
+    if limit is not None:
+        result = result[: max(0, limit)]
+    return result

vllm_omni/benchmarks/duplex/omni_duplex_eval_eval.py

@@ -0,0 +1,245 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Evaluate and summarize generated Omni-DuplexEval artifacts."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+from typing import Any
+
+from .omni_duplex_eval_clock import normalize_response_items, validate_clock
+from .omni_duplex_eval_dataset import DuplexSample
+from .omni_duplex_eval_judge import DuplexJudge
+from .omni_duplex_eval_media import extract_jpeg, materialize_media, video_duration
+from .omni_duplex_eval_metrics import (
+    PROTOCOL_PIN,
+    build_content_prompt,
+    build_reminder_prompt,
+    build_temporal_prompt,
+    parse_judge_json,
+    reminder_window,
+    summarize_pr_results,
+    summarize_temporal_results,
+    temporal_window,
+)
+
+
+def _read(path: Path) -> Any:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _write(path: Path, value: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def _text(items: list[dict[str, Any]]) -> str:
+    return " ".join(str(item.get("sentence", item.get("text", ""))).strip() for item in items).strip()
+
+
+def evaluate_sample(
+    sample: DuplexSample,
+    response_path: Path,
+    score_path: Path,
+    judge: DuplexJudge,
+    *,
+    judge_fps: int = 2,
+    judge_video_mode: str = "video_url",
+    window_size: float = 10.0,
+    allow_invalid_clock: bool = False,
+) -> dict[str, Any]:
+    raw = _read(response_path)
+    meta = (
+        _read(response_path.with_name(response_path.stem + ".meta.json"))
+        if response_path.with_name(response_path.stem + ".meta.json").exists()
+        else {}
+    )
+    validate_clock(meta, allow_invalid=allow_invalid_clock)
+    items = normalize_response_items(raw)
+    result: dict[str, Any] = {
+        "id": sample.id,
+        "family": sample.family,
+        "task_type": sample.task_type,
+        "protocol_pin": PROTOCOL_PIN,
+        "response_meta": meta,
+        "judge_model": getattr(judge, "model", None),
+        "judge_video_mode": judge_video_mode,
+    }
+    if sample.family == "rtd":
+        video_path = materialize_media(sample.video, score_path.parent / ".media", sample.id, ".mp4")
+        duration = sample.video_duration or video_duration(video_path)
+        temporal_rows = []
+        for item in items:
+            window = temporal_window(item["start"], item["end"], duration)
+            if not window:
+                temporal_rows.append(
+                    {
+                        **item,
+                        "sentence_start": item["start"],
+                        "sentence_end": item["end"],
+                        "sentence_duration": max(0.0, item["end"] - item["start"]),
+                        "error": "No valid temporal window",
+                        "temporal_score": 0,
+                        "is_relevant": 0,
+                    }
+                )
+                continue
+            frames = _extract_frames(video_path, _times(*window, fps=judge_fps))
+            parsed = parse_judge_json(
+                judge.temporal(build_temporal_prompt(*window, item["sentence"], sample.question_text), frames)
+            )
+            temporal_rows.append(
+                {
+                    **item,
+                    "sentence_start": item["start"],
+                    "sentence_end": item["end"],
+                    "sentence_duration": max(0.0, item["end"] - item["start"]),
+                    "window_start": window[0],
+                    "window_end": window[1],
+                    "error": None,
+                    **parsed,
+                }
+            )
+        content_frames = None
+        if judge_video_mode == "frame-sample":
+            content_frames = _extract_frames(video_path, _content_frame_times(duration))
+        content = parse_judge_json(
+            judge.content(
+                build_content_prompt(_text(items), sample.question_text, [sample.answer1, sample.answer2]),
+                video_path,
+                frames=content_frames,
+                mode=judge_video_mode,
+            )
+        )
+        if content_frames is not None:
+            content["frame_count"] = len(content_frames)
+        result.update(
+            {
+                "temporal": {"sentences": temporal_rows, "summary": summarize_temporal_results(temporal_rows)},
+                "content": content,
+            }
+        )
+    else:
+        if sample.task_type is None:
+            raise ValueError(f"missing proactive-reminder task type for split {sample.split!r}")
+        task = sample.task_type
+        times = _reminder_times(sample)
+        event_rows = []
+        if task == "correction" or not times:
+            response = _text(items)
+            parsed = parse_judge_json(
+                judge.chat(build_reminder_prompt(sample.question_text, response, task, sample.answer1))
+            )
+            event_rows.append({"response_segment": response, **parsed})
+        else:
+            for start in times:
+                lo, hi = reminder_window(start, window_size)
+                response = " ".join(item["sentence"] for item in items if lo <= item["start"] <= hi)
+                if not response:
+                    event_rows.append({"start_time": start, "response_segment": "", "success_score": 0})
+                else:
+                    parsed = parse_judge_json(
+                        judge.chat(build_reminder_prompt(sample.question_text, response, task, sample.answer1))
+                    )
+                    event_rows.append({"start_time": start, "response_segment": response, **parsed})
+        scores = [int(row.get("success_score", 0)) for row in event_rows]
+        result["pr"] = {
+            "events": event_rows,
+            "all_success": bool(scores) and all(score == 1 for score in scores),
+            "total_score": int(bool(scores) and all(score == 1 for score in scores)),
+        }
+    _write(score_path, result)
+    return result
+
+
+def _times(start: float, end: float, *, fps: int) -> list[float]:
+    count = max(1, int((end - start) * fps))
+    return (
+        [start + (end - start) / 2] if count == 1 else [start + i * (end - start) / (count - 1) for i in range(count)]
+    )
+
+
+def _extract_frames(path: str | Path, timestamps: list[float]) -> list[bytes]:
+    """Extract only decodable JPEG frames; clips can end on non-keyframes."""
+    frames = []
+    for timestamp in timestamps:
+        try:
+            frame = extract_jpeg(path, timestamp=timestamp)
+        except (OSError, subprocess.CalledProcessError, ValueError):
+            continue
+        if frame.startswith(b"\xff\xd8"):
+            frames.append(frame)
+    return frames
+
+
+def _content_frame_times(duration: float) -> list[float]:
+    """Sample the full video at approximately one frame every three seconds."""
+    if duration <= 0:
+        return []
+    times = []
+    timestamp = 0.0
+    while timestamp < duration:
+        times.append(min(timestamp, max(0.0, duration - 0.01)))
+        timestamp += 3.0
+    # Qwen2.5-VL has a finite per-request image-token budget. Preserve
+    # coverage while bounding long clips to a safely portable frame count.
+    if len(times) > 16:
+        stride = (len(times) - 1) / 15
+        times = [times[round(i * stride)] for i in range(16)]
+    return times
+
+
+def _reminder_times(sample: DuplexSample) -> list[float]:
+    values = []
+    for value in (sample.reminder1, sample.reminder2):
+        if isinstance(value, dict):
+            value = value.get("start", value.get("time"))
+        try:
+            if value is not None:
+                values.append(float(value))
+        except (TypeError, ValueError):
+            pass
+    return values
+
+
+def summarize_scores(score_root: str | Path) -> dict[str, Any]:
+    rows = []
+    for path in sorted(Path(score_root).rglob("*.json")):
+        if not path.name.endswith("_summary.json"):
+            row = _read(path)
+            row.setdefault("split", path.parent.name)
+            rows.append(row)
+    rtd_temporal = [
+        row["temporal"]["summary"] | {"content_score": row.get("content", {}).get("content_score", 0.0)}
+        for row in rows
+        if "temporal" in row
+    ]
+    pr = [{"task_type": row.get("task_type"), **row.get("pr", {})} for row in rows if "pr" in row]
+    result = {"protocol_pin": PROTOCOL_PIN, "samples": len(rows)}
+    if rtd_temporal:
+        by_task: dict[str, list[dict[str, float]]] = {}
+        for row in rows:
+            if "temporal" in row:
+                by_task.setdefault(str(row.get("split", row.get("task_type", "unknown"))), []).append(
+                    {
+                        "content": float(row.get("content", {}).get("content_score", 0.0)),
+                        "temporal": float(row["temporal"]["summary"].get("avg_temporal_score", 0.0)),
+                    }
+                )
+        result["rtd"] = {
+            "mean_content_score": sum(item["content_score"] for item in rtd_temporal) / len(rtd_temporal),
+            "mean_avg_temporal_score": sum(item["avg_temporal_score"] for item in rtd_temporal) / len(rtd_temporal),
+            "by_task": {
+                task: {
+                    "mean_content_score": sum(item["content"] for item in values) / len(values),
+                    "mean_avg_temporal_score": sum(item["temporal"] for item in values) / len(values),
+                }
+                for task, values in by_task.items()
+            },
+        }
+    if pr:
+        result["pr"] = summarize_pr_results(pr)
+    return result

vllm_omni/benchmarks/duplex/omni_duplex_eval_judge.py

@@ -0,0 +1,75 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""OpenAI-compatible local judge client."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pybase64 as base64
+import requests
+
+
+def _build_openai_url(base_url: str, api_path: str) -> str:
+    base = base_url.rstrip("/")
+    normalized_path = api_path if api_path.startswith("/") else f"/{api_path}"
+    if base.endswith(normalized_path):
+        return base
+    if base.endswith("/v1"):
+        return f"{base}{normalized_path}"
+    return f"{base}/v1{normalized_path}"
+
+
+class DuplexJudge:
+    def __init__(self, base_url: str, model: str, *, api_key: str = "EMPTY", timeout: int = 600) -> None:
+        self.base_url, self.model, self.api_key, self.timeout = base_url, model, api_key, timeout
+
+    def chat(self, content: Any, *, system: str | None = None, max_tokens: int = 1200) -> str:
+        messages = ([{"role": "system", "content": system}] if system else []) + [{"role": "user", "content": content}]
+        response = requests.post(
+            _build_openai_url(self.base_url, "/chat/completions"),
+            json={
+                "model": self.model,
+                "messages": messages,
+                "temperature": 0.1,
+                "max_tokens": max_tokens,
+            },
+            headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
+            timeout=self.timeout,
+        )
+        if not response.ok:
+            raise requests.HTTPError(
+                f"{response.status_code} response from judge: {response.text[:500]}", response=response
+            )
+        return str(response.json()["choices"][0]["message"]["content"])
+
+    def temporal(self, prompt: str, frames: list[bytes]) -> str:
+        content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]
+        content.extend(
+            {"type": "image_url", "image_url": {"url": "data:image/jpeg;base64," + base64.b64encode(frame).decode()}}
+            for frame in frames
+        )
+        return self.chat(content, system="You are a careful evaluator for real-time multimodal systems.")
+
+    def content(
+        self,
+        prompt: str,
+        video: str | Path | None = None,
+        frames: list[bytes] | None = None,
+        *,
+        mode: str = "video_url",
+    ) -> str:
+        content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]
+        if mode == "video_url" and video is not None:
+            content.append({"type": "video_url", "video_url": {"url": "file://" + str(Path(video).resolve())}})
+        else:
+            content.extend(
+                {
+                    "type": "image_url",
+                    "image_url": {"url": "data:image/jpeg;base64," + base64.b64encode(frame).decode()},
+                }
+                for frame in (frames or [])
+            )
+        return self.chat(content)

vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py

@@ -0,0 +1,113 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Small ffmpeg-backed media helpers used by the duplex runner and judge."""
+
+from __future__ import annotations
+
+import subprocess
+import wave
+from collections.abc import Iterator
+from pathlib import Path
+from typing import Any
+
+from vllm_omni.experimental.fullduplex.client import PCM16_BYTES_PER_SAMPLE, PCM16_SAMPLE_RATE
+
+
+def _run_ffmpeg(args: list[str]) -> bytes:
+    result = subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", *args], capture_output=True, check=True)
+    return result.stdout
+
+
+def materialize_media(value: Any, output_dir: str | Path, stem: str, suffix: str) -> Path:
+    """Resolve common Hugging Face media values to a local file."""
+    if isinstance(value, str | Path):
+        return Path(value)
+    path = value.get("path") if isinstance(value, dict) else getattr(value, "path", None)
+    if path:
+        return Path(path)
+    payload = value.get("bytes") if isinstance(value, dict) else None
+    if payload is None and isinstance(value, bytes | bytearray):
+        payload = value
+    if payload is None:
+        raise ValueError(f"cannot materialize media value for {stem!r}")
+    destination = Path(output_dir) / f"{stem}{suffix}"
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    destination.write_bytes(bytes(payload))
+    return destination
+
+
+def video_duration(path: str | Path) -> float:
+    result = subprocess.run(
+        ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(path)],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return max(0.0, float(result.stdout.strip()))
+
+
+def extract_jpeg(path: str | Path, *, timestamp: float, quality: int = 3) -> bytes:
+    return _run_ffmpeg(
+        [
+            "-ss",
+            f"{max(0.0, timestamp):.3f}",
+            "-i",
+            str(path),
+            "-frames:v",
+            "1",
+            "-q:v",
+            str(quality),
+            "-f",
+            "image2",
+            "pipe:1",
+        ]
+    )
+
+
+def iter_jpegs(
+    path: str | Path, *, fps: float = 1.0, duration: float | None = None, quality: int = 3
+) -> Iterator[tuple[float, bytes]]:
+    end = duration if duration is not None else video_duration(path)
+    step = 1.0 / fps
+    timestamp = 0.0
+    while timestamp < end:
+        try:
+            frame = extract_jpeg(path, timestamp=timestamp, quality=quality)
+        except (OSError, subprocess.CalledProcessError):
+            break
+        if frame:
+            yield timestamp, frame
+        timestamp += step
+
+
+def read_audio_pcm16(path: str | Path) -> bytes:
+    source = Path(path)
+    if source.suffix.lower() == ".wav":
+        try:
+            with wave.open(str(source), "rb") as wav_file:
+                if (
+                    wav_file.getnchannels() == 1
+                    and wav_file.getsampwidth() == 2
+                    and wav_file.getframerate() == PCM16_SAMPLE_RATE
+                    and wav_file.getcomptype() == "NONE"
+                ):
+                    return wav_file.readframes(wav_file.getnframes())
+        except wave.Error:
+            pass
+    return _run_ffmpeg(["-i", str(source), "-f", "s16le", "-ac", "1", "-ar", str(PCM16_SAMPLE_RATE), "pipe:1"])
+
+
+def iter_av_units(
+    audio_pcm16: bytes, frames: Iterator[tuple[float, bytes]], *, unit_ms: int = 1000
+) -> Iterator[tuple[bytes, bytes | None]]:
+    unit_bytes = PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * unit_ms // 1000
+    frame_iter = iter(frames)
+    next_frame = next(frame_iter, None)
+    for offset in range(0, len(audio_pcm16), unit_bytes):
+        timestamp = offset / (PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE)
+        frame = None
+        if next_frame is not None and next_frame[0] <= timestamp + unit_ms / 1000:
+            frame = next_frame[1]
+            next_frame = next(frame_iter, None)
+        yield audio_pcm16[offset : offset + unit_bytes], frame

vllm_omni/benchmarks/duplex/omni_duplex_eval_metrics.py

@@ -0,0 +1,260 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Protocol prompts, windows, parsing, and aggregation.
+
+Prompt wording and window math are copied verbatim from the MIT-licensed
+OpenBMB/Omni-DuplexEval evaluator at :data:`PROTOCOL_PIN`.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterable
+from typing import Any
+
+import regex as re
+
+# Copied from OpenBMB/Omni-DuplexEval@PROTOCOL_PIN (MIT).
+# Do not change prompt wording without bumping the pin and recording a score delta.
+PROTOCOL_PIN = "ca3c122b4d4bf67afd6b18ea5e724b4561bdde48"
+
+
+def temporal_window(sentence_start: float, sentence_end: float, video_duration: float) -> tuple[float, float] | None:
+    start = max(0.0, float(sentence_start) - 2.0)
+    end = max(start + 0.5, float(sentence_end) - 2.0)
+    return (start, end) if end <= float(video_duration) and end - start >= 0.5 else None
+
+
+def reminder_window(start_time: float, window_size: float = 10.0) -> tuple[float, float]:
+    return float(start_time), float(start_time) + float(window_size)
+
+
+def parse_judge_json(text: str) -> dict[str, Any]:
+    for match in re.findall(r"\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}", str(text), re.S):
+        try:
+            value = json.loads(re.sub(r",\s*([}\]])", r"\1", match))
+        except (TypeError, ValueError, json.JSONDecodeError):
+            continue
+        if isinstance(value, dict):
+            break
+    else:
+        value = {}
+        patterns = [
+            (r'"temporal_score"\s*:\s*(\d+)', "temporal_score", int),
+            (r'"is_relevant"\s*:\s*(\d+)', "is_relevant", int),
+            (r'"success_score"\s*:\s*([01])', "success_score", int),
+            (r'"content_score"\s*:\s*(\d+(?:\.\d+)?)', "content_score", float),
+        ]
+        for pattern, key, caster in patterns:
+            if match := re.search(pattern, str(text)):
+                parsed = caster(match.group(1))
+                value[key] = round(parsed, 2) if caster is float else parsed
+        for pattern, key in [
+            (r'"temporal_reasoning"\s*:\s*"([^"]*)"', "temporal_reasoning"),
+            (r'"content_reasoning"\s*:\s*"([^"]*)"', "content_reasoning"),
+            (r'"reasoning"\s*:\s*"([^"]*)"', "reasoning"),
+        ]:
+            if match := re.search(pattern, str(text), re.S):
+                value[key] = match.group(1).strip()
+                break
+    for key in ("temporal_score", "is_relevant", "success_score"):
+        if key in value:
+            try:
+                value[key] = int(float(value[key]))
+            except (TypeError, ValueError):
+                value[key] = 0
+    if "content_score" in value:
+        try:
+            value["content_score"] = round(float(value["content_score"]), 2)
+        except (TypeError, ValueError):
+            value["content_score"] = 0.0
+    return value
+
+
+def summarize_temporal_results(results: Iterable[dict[str, Any]]) -> dict[str, Any]:
+    rows = list(results)
+    valid = [row for row in rows if not row.get("error")]
+    relevant = [row for row in valid if int(row.get("is_relevant", 0)) == 1]
+    irrelevant = [row for row in valid if int(row.get("is_relevant", 0)) == 0]
+    scores = [int(row.get("temporal_score", 0)) for row in relevant]
+    total_valid_duration = sum(float(row.get("sentence_duration", 0.0)) for row in valid)
+    irrelevant_duration = sum(float(row.get("sentence_duration", 0.0)) for row in irrelevant)
+    return {
+        "avg_temporal_score": round(sum(scores) / len(scores), 4) if scores else 0.0,
+        "total_sentences": len(rows),
+        "evaluated_sentences": len(valid),
+        "error_count": len(rows) - len(valid),
+        "relevant_sentences_count": len(relevant),
+        "irrelevant_sentences_count": len(irrelevant),
+        "score_distribution": {
+            "3_points": scores.count(3),
+            "2_points": scores.count(2),
+            "1_point": scores.count(1),
+            "0_points": scores.count(0),
+        },
+        "relevance_stats": {
+            "first_relevant_time": min((row.get("sentence_start") for row in relevant), default=None),
+            "total_relevant_duration": round(sum(float(row.get("sentence_duration", 0.0)) for row in relevant), 4),
+            "total_irrelevant_duration": round(irrelevant_duration, 4),
+            "irrelevant_duration_ratio": round(
+                irrelevant_duration / total_valid_duration if total_valid_duration > 0 else 0.0, 4
+            ),
+        },
+        "excluded_irrelevant_sentences": [
+            {
+                "sentence": row.get("sentence", ""),
+                "start": row.get("sentence_start"),
+                "end": row.get("sentence_end"),
+                "temporal_score": row.get("temporal_score", 0),
+            }
+            for row in irrelevant
+        ],
+    }
+
+
+def summarize_pr_results(results: Iterable[dict[str, Any]]) -> dict[str, Any]:
+    rows = list(results)
+    successes = [int(row.get("all_success", row.get("total_score", 0))) for row in rows]
+    by_task: dict[str, list[int]] = {}
+    for row, score in zip(rows, successes, strict=False):
+        by_task.setdefault(str(row.get("task_type", "unknown")), []).append(score)
+    return {
+        "samples": len(rows),
+        "mean_all_success": sum(successes) / len(successes) if successes else 0.0,
+        "by_task": {task: sum(values) / len(values) for task, values in by_task.items()},
+    }
+
+
+def build_temporal_prompt(start_time: float, end_time: float, response_text: str, question: str) -> str:
+    return f"""
+You are evaluating a real-time video description system.
+
+Basic information:
+- Video segment: {start_time:.2f}s to {end_time:.2f}s.
+- User instruction: {question}
+- Model response sentence: "{response_text}"
+
+Evaluation principles:
+- Judge only the current video segment.
+- Ignore response content that clearly belongs to earlier or later segments unless it contradicts the current segment.
+- Language does not matter; evaluate semantic correctness and temporal alignment only.
+- Decide whether the sentence is a substantive response or only a filler/polite phrase.
+
+Temporal sensitivity score:
+- 3: Excellent temporal alignment. The sentence accurately describes the current segment and follows the instruction.
+- 2: Mostly aligned. Minor inaccuracies, omissions, or harmless references to nearby segments are allowed.
+- 1: Poor alignment. Major inaccuracies, wrong timing, or mostly irrelevant content.
+- 0: No meaningful alignment with the current segment.
+
+Relevance label:
+- 1: The sentence contains substantive task- or video-related information.
+- 0: The sentence is only filler, acknowledgement, meta-commentary, or generic text without useful content.
+
+Return exactly this JSON object:
+{{
+  "temporal_score": <0, 1, 2, or 3>,
+  "temporal_reasoning": "<brief explanation>",
+  "is_relevant": <0 or 1>
+}}
+"""
+
+
+def build_content_prompt(response_text: str, question: str, references: list[str] | None = None) -> str:
+    reference_block = ""
+    if references:
+        reference_block = "Reference annotations, for reference only:\n"
+        for index, reference in enumerate(references, 1):
+            reference_block += f"{index}. {reference}\n"
+
+    return f"""
+You are a precise content-accuracy evaluator for video description.
+
+Basic information:
+- User instruction: {question}
+- Model response: "{response_text}"
+{reference_block}
+
+Evaluation goal:
+Judge whether the model response is factually accurate for the whole video and aligned with the instruction.
+Consider object/action/color/count/spatial/event errors, hallucinations, omissions, and irrelevant content.
+Reference annotations are optional guidance; the primary evidence is the video itself.
+
+Scoring:
+- Use a decimal score from 0.00 to 3.00.
+- Start from 3.00 and deduct for each error.
+- Use 0.00 only when the response is empty, completely irrelevant, or contains no correct video facts.
+- Output exactly two decimal places.
+
+Return exactly this JSON object:
+{{
+  "content_score": <decimal from 0.00 to 3.00>,
+  "content_reasoning": "<brief explanation with main errors and final score>"
+}}
+"""
+
+
+def build_reminder_prompt(instruction: str, response: str, task_type: str, ground_answer: str = "") -> str:
+    if task_type == "correction":
+        return f"""
+You are judging whether a model successfully completed a correction task.
+
+Task description:
+The user instruction contains an incorrect statement about the video.
+The model should identify the incorrect part and provide the correct information.
+
+Input:
+- User instruction: {instruction}
+- Reference correction: {ground_answer}
+- Model response: "{response}"
+
+Success criteria:
+1. Identify the error implied by the user instruction and the reference correction.
+2. Check whether the model corrected all required error points.
+3. The corrected content must preserve the correct context, including subject, object, action, and attributes.
+4. Ignore extra information that is unrelated to both the instruction and the reference, unless it contradicts them.
+
+Scoring:
+- 1 = all required error points are corrected and the context is consistent.
+- 0 = at least one required correction is missing, wrong, or contextually inconsistent.
+
+Return exactly this JSON object:
+{{
+  "success_score": <0 or 1>,
+  "reasoning": "<brief explanation covering the required correction points>"
+}}
+"""
+
+    if task_type == "post_event_reminder":
+        timing_note = "The response is evaluated immediately after the target event has occurred."
+    else:
+        timing_note = "The response is evaluated at the target event time or immediately after it."
+
+    return f"""
+You are judging whether a model successfully completed an event-reminder task.
+
+Task description:
+The user gave an instruction asking the system to remind them when a specific event happens.
+{timing_note}
+The provided model text is the response segment generated in the evaluation window.
+
+Input:
+- User instruction: {instruction}
+- Model response segment: "{response}"
+
+Success criteria:
+1. The response clearly refers to the target event in the instruction.
+2. The response communicates a reminder, notification, or confirmation that the event has happened.
+3. Vague narration, unrelated descriptions, or wrong-event reminders are failures.
+4. The response does not need to match the instruction wording exactly if the meaning is clear.
+
+Scoring:
+- 1 = successful reminder.
+- 0 = unsuccessful reminder.
+
+Return exactly this JSON object:
+{{
+  "success_score": <0 or 1>,
+  "reasoning": "<brief explanation>"
+}}
+"""

vllm_omni/benchmarks/duplex/omni_duplex_eval_runner.py

@@ -0,0 +1,111 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Single-sample MiniCPM-o native-duplex generation."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+import pybase64 as base64
+
+from vllm_omni.experimental.fullduplex.client import RealtimeDuplexClient, build_realtime_url, wait_for
+
+from .omni_duplex_eval_clock import extract_timed_sentences
+from .omni_duplex_eval_media import iter_av_units, iter_jpegs, materialize_media, read_audio_pcm16, video_duration
+
+
+def _ref_audio(path: str | Path) -> str:
+    value = Path(path).expanduser().read_bytes()
+    return "data:audio/wav;base64," + base64.b64encode(value).decode("ascii")
+
+
+async def generate_sample(
+    sample: Any,
+    *,
+    url: str,
+    model: str,
+    ref_audio: str | Path,
+    output_root: str | Path,
+    fps: float = 1.0,
+    mix: str = "question",
+    pace: str = "realtime",
+    clock: str = "media",
+    overwrite: bool = False,
+    unit_ms: int = 1000,
+) -> Path:
+    output = Path(output_root) / sample.split / f"{sample.id}.json"
+    meta_path = output.with_name(output.stem + ".meta.json")
+    if output.exists() and not overwrite:
+        return output
+    if mix != "question":
+        raise NotImplementedError("v1 supports mix=question; soundtrack mixing is reserved for P1")
+    media_dir = output.parent / ".media"
+    audio_path = materialize_media(sample.question_audio, media_dir, f"{sample.id}_question", ".wav")
+    video_path = materialize_media(sample.video, media_dir, sample.id, ".mp4")
+    pcm = read_audio_pcm16(audio_path)
+    duration = sample.video_duration or video_duration(video_path)
+    frames = iter_jpegs(video_path, fps=fps, duration=duration)
+    realtime = pace == "realtime"
+    if pace not in {"realtime", "as-fast-as-possible"}:
+        raise ValueError("pace must be realtime or as-fast-as-possible")
+    client = RealtimeDuplexClient(build_realtime_url(url, model, autostart=False))
+    response_done = False
+    drain_timeout = None
+    close_timeout = None
+    async with client:
+        await client.configure(model, ref_audio=_ref_audio(ref_audio), instructions="Streaming Omni Conversation.")
+        ack_task = asyncio.create_task(_ack_playback(client))
+        try:
+            await client.stream_av_units(iter_av_units(pcm, frames, unit_ms=unit_ms), realtime=realtime)
+            await client.commit()
+            try:
+                await wait_for(
+                    lambda: client.events.count("response.done") > 0,
+                    timeout_s=max(20.0, duration + 20.0),
+                    label="response.done",
+                )
+                response_done = True
+            except TimeoutError as exc:
+                drain_timeout = str(exc)
+        finally:
+            ack_task.cancel()
+            try:
+                await ack_task
+            except asyncio.CancelledError:
+                pass
+        await client.acknowledge_playback()
+        try:
+            await client.close_session(timeout_s=20.0)
+        except TimeoutError as exc:
+            close_timeout = str(exc)
+        events = list(client.events.events)
+    timed = [sentence.as_dict() for sentence in extract_timed_sentences(events, clock=clock)]
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(timed, ensure_ascii=False, indent=2), encoding="utf-8")
+    meta = {
+        "id": sample.id,
+        "split": sample.split,
+        "clock": clock if realtime else "invalid",
+        "pace": pace,
+        "mix": mix,
+        "fps": fps,
+        "unit_ms": unit_ms,
+        "model": model,
+        "response_done": response_done,
+        "drain_timeout": drain_timeout,
+        "close_timeout": close_timeout,
+        "ref_audio_sha256": hashlib.sha256(Path(ref_audio).read_bytes()).hexdigest(),
+    }
+    meta_path.write_text(json.dumps(meta, indent=2), encoding="utf-8")
+    return output
+
+
+async def _ack_playback(client: RealtimeDuplexClient) -> None:
+    while True:
+        await asyncio.sleep(0.25)
+        await client.acknowledge_playback()

vllm_omni/entrypoints/cli/__init__.py

@@ -1,3 +1,6 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
 """CLI helpers for vLLM-Omni entrypoints."""
 
 # To ensure patch imports work properly, disable unused import checks
@@ -6,8 +9,9 @@
 from vllm_omni.benchmarks.patch import patch
 # isort: on
 
+from vllm_omni.entrypoints.cli.benchmark.omni_duplex_eval import OmniDuplexEvalSubcommand
 from vllm_omni.entrypoints.cli.benchmark.serve import OmniBenchmarkServingSubcommand
 
 from .serve import OmniServeCommand
 
-__all__ = ["OmniServeCommand", "OmniBenchmarkServingSubcommand"]
+__all__ = ["OmniServeCommand", "OmniBenchmarkServingSubcommand", "OmniDuplexEvalSubcommand"]

vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py

@@ -0,0 +1,134 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+import argparse
+import asyncio
+import concurrent.futures
+import json
+from pathlib import Path
+
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_dataset import DEFAULT_DATASET, load_samples
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_eval import evaluate_sample, summarize_scores
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_judge import DuplexJudge
+from vllm_omni.benchmarks.duplex.omni_duplex_eval_runner import generate_sample
+from vllm_omni.entrypoints.cli.benchmark.base import OmniBenchmarkSubcommandBase
+
+
+def _common(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument("--dataset", default=DEFAULT_DATASET)
+    parser.add_argument("--split", default="all")
+    parser.add_argument("--family", choices=("all", "rtd", "pr"), default="all")
+    parser.add_argument("--media-root")
+    parser.add_argument("--limit", type=int)
+    parser.add_argument("--ids", nargs="*")
+
+
+def add_cli_args(parser: argparse.ArgumentParser) -> None:
+    actions = parser.add_subparsers(dest="action", required=True)
+    generate = actions.add_parser("generate")
+    _common(generate)
+    generate.add_argument("--url", default="ws://localhost:8099/v1/realtime?duplex=1")
+    generate.add_argument("--model", required=True)
+    generate.add_argument("--ref-audio", required=True)
+    generate.add_argument("--response-root", required=True)
+    generate.add_argument("--fps", type=float, default=1.0)
+    generate.add_argument("--mix", choices=("question",), default="question")
+    generate.add_argument("--pace", choices=("realtime", "as-fast-as-possible"), default="realtime")
+    generate.add_argument("--clock", choices=("media",), default="media")
+    generate.add_argument("--concurrency", type=int, default=1)
+    generate.add_argument("--overwrite", action="store_true")
+    evaluate = actions.add_parser("evaluate")
+    _common(evaluate)
+    evaluate.add_argument("--response-root", required=True)
+    evaluate.add_argument("--score-root", required=True)
+    evaluate.add_argument("--judge-base-url", default="http://127.0.0.1:8000")
+    evaluate.add_argument("--judge-model", required=True)
+    evaluate.add_argument("--judge-api-key", default="EMPTY")
+    evaluate.add_argument("--judge-video-mode", choices=("video_url", "frame-sample"), default="video_url")
+    evaluate.add_argument("--judge-fps", type=int, default=2)
+    evaluate.add_argument("--window-size", type=float, default=10.0)
+    evaluate.add_argument("--allow-invalid-clock", action="store_true")
+    evaluate.add_argument("--eval-workers", type=int, default=1)
+    evaluate.add_argument("--overwrite", action="store_true")
+    summarize = actions.add_parser("summarize")
+    summarize.add_argument("--score-root", required=True)
+
+
+def run(args: argparse.Namespace) -> int:
+    if args.action == "summarize":
+        print(json.dumps(summarize_scores(args.score_root), ensure_ascii=False, indent=2))
+        return 0
+
+    samples = load_samples(
+        args.dataset,
+        split=args.split,
+        family=args.family,
+        media_root=args.media_root,
+        limit=args.limit,
+        ids=args.ids,
+    )
+    if args.action == "generate":
+        if args.concurrency < 1:
+            raise ValueError("--concurrency must be at least 1")
+
+        async def generate() -> None:
+            semaphore = asyncio.Semaphore(args.concurrency)
+
+            async def generate_one(sample) -> None:
+                async with semaphore:
+                    await generate_sample(
+                        sample,
+                        url=args.url,
+                        model=args.model,
+                        ref_audio=args.ref_audio,
+                        output_root=args.response_root,
+                        fps=args.fps,
+                        mix=args.mix,
+                        pace=args.pace,
+                        clock=args.clock,
+                        overwrite=args.overwrite,
+                    )
+
+            await asyncio.gather(*(generate_one(sample) for sample in samples))
+
+        asyncio.run(generate())
+        return 0
+
+    if args.eval_workers < 1:
+        raise ValueError("--eval-workers must be at least 1")
+    judge = DuplexJudge(args.judge_base_url, args.judge_model, api_key=args.judge_api_key)
+
+    def evaluate_one(sample) -> None:
+        response_path = Path(args.response_root) / sample.split / f"{sample.id}.json"
+        score_path = Path(args.score_root) / sample.split / f"{sample.id}.json"
+        if score_path.exists() and not args.overwrite:
+            return
+        evaluate_sample(
+            sample,
+            response_path,
+            score_path,
+            judge,
+            judge_fps=args.judge_fps,
+            judge_video_mode=args.judge_video_mode,
+            window_size=args.window_size,
+            allow_invalid_clock=args.allow_invalid_clock,
+        )
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=args.eval_workers) as executor:
+        list(executor.map(evaluate_one, samples))
+    return 0
+
+
+class OmniDuplexEvalSubcommand(OmniBenchmarkSubcommandBase):
+    """Run the Omni-DuplexEval generation or scoring workflow."""
+
+    name = "omni-duplex-eval"
+    help = "Generate, evaluate, or summarize Omni-DuplexEval artifacts."
+
+    @classmethod
+    def add_cli_args(cls, parser: argparse.ArgumentParser) -> None:
+        add_cli_args(parser)
+
+    @staticmethod
+    def cmd(args: argparse.Namespace) -> None:
+        run(args)

vllm_omni/experimental/fullduplex/client.py

@@ -435,6 +435,7 @@ class RealtimeDuplexClient:
         self.events = RealtimeEventCollector()
         self._ws: Any = None
         self._reader_task: asyncio.Task[None] | None = None
+        self._media_clock_ms = 0
 
     async def __aenter__(self) -> RealtimeDuplexClient:
         self._ws = await websockets.connect(
@@ -463,6 +464,7 @@ class RealtimeDuplexClient:
                     continue
                 event = json.loads(raw)
                 if isinstance(event, dict):
+                    event.setdefault("_media_clock_ms", self._media_clock_ms)
                     self.events.add(event)
         except ConnectionClosed:
             return
@@ -559,21 +561,36 @@ class RealtimeDuplexClient:
             PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE * chunk_ms // 1000,
             PCM16_BYTES_PER_SAMPLE,
         )
+        await self.stream_av_units(
+            ((pcm16[offset : offset + chunk_bytes], None) for offset in range(0, len(pcm16), chunk_bytes)),
+            realtime=realtime,
+        )
+
+    async def stream_av_units(
+        self,
+        units: Any,
+        *,
+        realtime: bool = True,
+    ) -> None:
+        """Stream PCM16 units, optionally attaching a JPEG to each unit."""
         audio_end_ms = 0
-        for offset in range(0, len(pcm16), chunk_bytes):
-            chunk = pcm16[offset : offset + chunk_bytes]
+        for chunk, frame in units:
+            if not chunk:
+                continue
             duration_ms = len(chunk) * 1000 // (PCM16_SAMPLE_RATE * PCM16_BYTES_PER_SAMPLE)
             audio_end_ms += duration_ms
-            await self.send(
-                {
-                    "type": "input_audio_buffer.append",
-                    "audio": base64.b64encode(chunk).decode("ascii"),
-                    "input_audio_format": "pcm16",
-                    "sample_rate_hz": PCM16_SAMPLE_RATE,
-                    "duration_ms": duration_ms,
-                    "audio_end_ms": audio_end_ms,
-                }
-            )
+            self._media_clock_ms = audio_end_ms
+            event: dict[str, object] = {
+                "type": "input_audio_buffer.append",
+                "audio": base64.b64encode(chunk).decode("ascii"),
+                "input_audio_format": "pcm16",
+                "sample_rate_hz": PCM16_SAMPLE_RATE,
+                "duration_ms": duration_ms,
+                "audio_end_ms": audio_end_ms,
+            }
+            if frame is not None:
+                event["video_frames"] = [base64.b64encode(frame).decode("ascii") if isinstance(frame, bytes) else frame]
+            await self.send(event)
             if realtime:
                 await asyncio.sleep(duration_ms / 1000)