diff --git a/examples/online_serving/minicpmo/README.md b/examples/online_serving/minicpmo/README.md index 1212f0ff7..364c3a7a9 100644 --- a/examples/online_serving/minicpmo/README.md +++ b/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/) diff --git a/tests/benchmarks/duplex/test_omni_duplex_eval.py b/tests/benchmarks/duplex/test_omni_duplex_eval.py new file mode 100644 index 000000000..5f77aacc0 --- /dev/null +++ b/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 diff --git a/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py b/tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py new file mode 100644 index 000000000..192859de5 --- /dev/null +++ b/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"}] diff --git a/vllm_omni/benchmarks/duplex/__init__.py b/vllm_omni/benchmarks/duplex/__init__.py new file mode 100644 index 000000000..62d91dd86 --- /dev/null +++ b/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"] diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_clock.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_clock.py new file mode 100644 index 000000000..03997fc8e --- /dev/null +++ b/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") diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_dataset.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_dataset.py new file mode 100644 index 000000000..9d6013e80 --- /dev/null +++ b/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 diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_eval.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_eval.py new file mode 100644 index 000000000..a92376bd7 --- /dev/null +++ b/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 diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_judge.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_judge.py new file mode 100644 index 000000000..2fa4014d5 --- /dev/null +++ b/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) diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_media.py new file mode 100644 index 000000000..629d8e877 --- /dev/null +++ b/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 diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_metrics.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_metrics.py new file mode 100644 index 000000000..a373be58d --- /dev/null +++ b/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": "", + "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": , + "content_reasoning": "" +}} +""" + + +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": "" +}} +""" + + 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": "" +}} +""" diff --git a/vllm_omni/benchmarks/duplex/omni_duplex_eval_runner.py b/vllm_omni/benchmarks/duplex/omni_duplex_eval_runner.py new file mode 100644 index 000000000..d5a28aa06 --- /dev/null +++ b/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() diff --git a/vllm_omni/entrypoints/cli/__init__.py b/vllm_omni/entrypoints/cli/__init__.py index 2ffba6130..c13ac94be 100644 --- a/vllm_omni/entrypoints/cli/__init__.py +++ b/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"] diff --git a/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py b/vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py new file mode 100644 index 000000000..a43a5ec71 --- /dev/null +++ b/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) diff --git a/vllm_omni/experimental/fullduplex/client.py b/vllm_omni/experimental/fullduplex/client.py index 78ec2cd3d..77855bf6a 100644 --- a/vllm_omni/experimental/fullduplex/client.py +++ b/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)