[Frontend][Benchmark] Add judger and accuracy metrics for OmniInterac… (#7317)

Co-authored-by: Gao Han <hgaoaf@connect.ust.hk>
This commit is contained in:
ZacheryAUandGao Han authored and GitHub committed 2026-09-24 08:34:23 +00:00
1 parent 1d77e96d4d
commit 926ba72793
13 files changed
+4155 -20

No files matched your search

+70 -6
View File
@@ -302,6 +302,21 @@ vllm bench serve --omni \
selects all and oversize values use all available cases. Reference audio is required, and OmniInteract uses the
`/v1/realtime` endpoint.
To replay an existing sample set (for example `sampled_cases.jsonl` from a prior run), pass
`--omniinteract-video-list` instead of discovering cases. List order is preserved; `--num-prompts` takes the prefix of
the list (`0` or a value larger than the list length runs every row). Do not combine video-list mode with `--seed`,
`--dataset-path`, `--omniinteract-scenario-tags`, or `--omniinteract-scenario-focus`.
Use `--omniinteract-scenario-tags` to steer small runs toward paper-table scenarios (`realtime`, `proactive`,
`nested`, `interrupted`, `1qna`). By default the sampler covers each requested tag with at least one matching case
(when available under `--omniinteract-subsets`), then fills the remaining `--num-prompts` budget from other cases.
Add `--omniinteract-scenario-focus` to run only cases that match those tags. Example smoke coverage:
```bash
--num-prompts 5 \
--omniinteract-scenario-tags realtime proactive nested interrupted 1qna
```
Audio is replayed as 16 kHz PCM16 in 200 ms chunks and video at 1 FPS with real-time pacing. All selected media is decoded
before timing and remains in client memory for the run, so `--max-concurrency` does not limit media preparation memory; use
explicit `--num-prompts 0` only when the client has enough RAM for the full dataset. Media commands are bounded by
@@ -309,7 +324,9 @@ explicit `--num-prompts 0` only when the client has enough RAM for the full data
options apply. Use `--omniinteract-require-response` only for functional E2E cases; LISTEN is a valid benchmark result.
Each completed case writes `output.wav`, `wav_transcript.json`, `events.json`, `result.json`, and a final `.done` marker under
`--omniinteract-output-dir`. The root also contains `batch_summary.json` and `official_eval_manifest.jsonl`; failed cases write
`--omniinteract-output-dir`. The root also contains `batch_summary.json`, `official_eval_manifest.jsonl` (eligible cases only),
and `sampled_cases.jsonl` (every sampled case as absolute `video_path` / official `output_name` / `subset` rows for Lucky-Lance
OmniInteract `batch_inference_minicpmo.py --video_list`); failed cases write
`.failed.json`. Runs sharing one output root are serialized. Completion validates transport, response lifecycle, and artifacts,
not answer accuracy. Transcript timestamps are serialized playback-queue times. Playback ACKs report cumulative progress
incrementally along that serialized clock, like a live listener; the first ack for a response goes out as soon as its audio
@@ -318,6 +335,48 @@ arrives, checkpointing the response's history position so a later committed user
outputs are ineligible and omitted from the official manifest; `audio_clipped_bytes` records output beyond the rounded video
horizon.
Accuracy evaluation is opt-in and requires an already-running text judge with an OpenAI-compatible Chat Completions API. The
benchmark does not launch or stop the judge server. Add the following options to the command above:
```bash
vllm serve Qwen/Qwen2.5-7B-Instruct \
--served-model-name Qwen2.5-7B-Instruct \
--port 8001
```
Then add to the benchmark command:
```bash
--omniinteract-evaluate \
--omniinteract-judge-base-url http://127.0.0.1:8001 \
--omniinteract-judge-model Qwen2.5-7B-Instruct
```
`--omniinteract-judge-model` must match that judge process's `--served-model-name` (or its `--model`
string if `--served-model-name` is omitted). It is not a checkpoint path to load, and it is not the
Omni DUT `--served-model-name`. Point `--omniinteract-judge-base-url` at the judge, not at the Omni
realtime endpoint.
The early / core / interrupted-partial judge prompts follow the official English
templates in [Lucky-Lance/OmniInteract](https://github.com/Lucky-Lance/OmniInteract)
`eval/evaluation/llm_judge.py` (commit `de304cef35fd9a50a5caadb5090c34cfbf0dd868`).
They correspond to the OmniInteract paper appendix
([arXiv:2605.26485](https://arxiv.org/abs/2605.26485)) Listing A.1 (early-stage),
Listing A.2 (interrupted partial quality), and Listing A.3 (core-stage).
A local OpenAI-compatible judge is not the paper's GPT-4o judge, so reported
IA-QTF1 numbers are protocol-compatible rather than official paper-table scores.
After artifact publication, the evaluator builds `[start, t_a, end)` slots, judges early and core response text, writes
per-case details plus `evaluation/unified_eval_summary.json`, and prints IA-QTF1 columns
(1Q1A realtime / proactive / nested / Global, 1QnA, All Global), interruption diagnostics (NOR / PAQ / CSM), and nested
metrics (NCCS, inner / outer IA-QTF1, missed outer). Slices and sections with no slots in the evaluated sample are
omitted from the terminal report. The header lists how many evaluated cases carry each scenario tag
(`realtime`, `proactive`, `nested`, `interrupted`, `1qna`). Realtime and proactive
exclude nested inner/outer slots; 1Q1A Global recomputes F1 from those three TP/FP/FN aggregates; All Global includes every
scored slot plus unmatched-chunk false positives. Plain `wav_transcript.json` timestamps provide chunk-level approximate
timing. If transcript chunks include `aligned_words`, the evaluator splits boundary-crossing chunks and derives trigger
timing from word alignment.
Per-response TTFT and TTFP start when the server begins executing the native model-turn request that owns the response. RTF
continues to use client receipt of `response.created` through the last audio packet, divided by emitted audio duration. Global
TTFT, TTFP, and RTF cover the complete input-stream window. TPOT/ITL use engine stage-0 timing; ITL is emitted only when every
@@ -325,9 +384,12 @@ token interval is present within a continuous generation segment. Model-unit pac
excluded from TPOT/ITL. Raw request metrics retain `response_created_to_first_text_ms` and
`response_created_to_first_audio_ms` as client-envelope diagnostics.
The checked-in local performance configuration measures four deterministic cases from each OmniInteract subset (12 videos
total), with no benchmark warmups and a maximum concurrency of two. Each subset also sends one readiness request before its
measured cases. Run it from the repository root with:
The checked-in local performance configuration measures four deterministic cases from `1q1a`, `1q1a_math`, and
`1qna` (12 videos total), with no benchmark warmups and a maximum concurrency of two. Each subset also sends one readiness request before its
measured cases. The runner starts MiniCPM-o 4.5 on the first visible GPU and a text judge
(`Qwen/Qwen2.5-7B-Instruct`) on the second (`CUDA_VISIBLE_DEVICES=1` relative to the process). After artifacts land it
scores All Global IA-QTF1 through that local judge. This is protocol-compatible with the paper metric, not a GPT-4o
paper-table score. Run it from the repository root with two visible GPUs:
```bash
export HF_HOME=/path/to/persistent/huggingface-cache
@@ -338,12 +400,14 @@ bash tools/nightly/run_nightly_jobs.sh \
--label-substr minicpmo_4_5_omniinteract
```
The first run downloads the pinned OmniInteract archive into `HF_HOME`; later runs reuse that cache.
The first run downloads the pinned OmniInteract archive and the judge weights into `HF_HOME`; later runs reuse that cache.
It requires every case to commit its input, complete any emitted response lifecycles, and publish the expected WAV,
transcript, event, and result artifacts without errors. A valid LISTEN-only case may have no response audio or transcript
chunks. Official-manifest eligibility is reported separately because clipped or cancelled output is a benchmark-quality
signal, not a transport failure. This local performance test does not score answer accuracy.
signal, not a transport failure. Accuracy must finish with `status=ok` on every subset. After all three subsets finish,
All Global IA-QTF1 is recomputed from pooled `Global_TP` / `Global_FP` / `Global_FN` and must be at or above
`omniinteract_aggregate_min_ia_qtf1` (checked in as `0.2`).
### Video-MME Benchmark
+291
View File
@@ -249,6 +249,252 @@ def test_dataset_discovers_official_layouts_and_total_selection(tmp_path: Path):
assert [case.subset for case in selected] == ["1q1a", "1q1a_math"]
def _write_scenario_dataset(root: Path) -> Path:
data_root = root / "data"
one_q1a = data_root / "1q1a"
(one_q1a / "videos").mkdir(parents=True)
(one_q1a / "annotations").mkdir()
entries = []
specs = [
("realtime", "multi_turn", [{"question_time": 0.0, "answer_time": 1.0, "question_type": "realtime"}]),
("proactive", "multi_turn", [{"question_time": 0.0, "answer_time": 1.0, "question_type": "proactive"}]),
(
"nested",
"nested",
[
{
"question_time": 0.0,
"answer_time": 4.0,
"question_type": "proactive",
"question_text": "outer",
"answer_text": "outer",
},
{
"question_time": 1.0,
"answer_time": 2.0,
"question_type": "realtime",
"question_text": "inner",
"answer_text": "inner",
},
],
),
(
"interrupted",
"multi_turn",
[
{
"question_time": 0.0,
"answer_time": 1.0,
"question_type": "realtime",
"is_interrupted": True,
}
],
),
]
for name, scene, annotation in specs:
video = one_q1a / "videos" / f"{name}.mp4"
ann = one_q1a / "annotations" / f"{name}.json"
video.touch()
ann.write_text(json.dumps(annotation))
entries.append(
{
"video": f"videos/{name}.mp4",
"annotation": f"annotations/{name}.json",
"scene_type": scene,
}
)
(one_q1a / "video_json_map.json").write_text(json.dumps({"entries": entries}))
one_qna = data_root / "1qna"
(one_qna / "videos_bench").mkdir(parents=True)
(one_qna / "annotations").mkdir(parents=True)
(one_qna / "videos_bench" / "guide.mp4").touch()
(one_qna / "annotations" / "guide.json").write_text(
json.dumps(
{
"question_time": 0.0,
"question_text": "q",
"answers": [{"answer_time": 1.0, "answer_text": "a", "label": "step"}],
}
)
)
math_root = data_root / "1q1a_math"
(math_root / "videos").mkdir(parents=True)
(math_root / "annotations").mkdir()
(math_root / "videos" / "math.mp4").touch()
(math_root / "annotations" / "math.json").write_text(
json.dumps([{"question_time": 0.0, "answer_time": 1.0, "question_type": "realtime"}])
)
(math_root / "video_json_map.json").write_text(
json.dumps(
{
"entries": [
{
"video": "videos/math.mp4",
"annotation": "annotations/math.json",
"scene_type": "multi_turn",
}
]
}
)
)
return data_root
def test_scenario_cover_and_focus_sampling(tmp_path: Path) -> None:
root = _write_scenario_dataset(tmp_path)
covered = data.discover_omniinteract_cases(
root,
("1q1a", "1q1a_math", "1qna"),
num_prompts=5,
disable_shuffle=True,
scenario_tags=("realtime", "proactive", "nested", "interrupted", "1qna"),
)
assert len(covered) == 5
covered_tags = set().union(*(data.case_scenario_tags(case) for case in covered))
assert {"realtime", "proactive", "nested", "interrupted", "1qna"} <= covered_tags
focused = data.discover_omniinteract_cases(
root,
("1q1a", "1q1a_math", "1qna"),
num_prompts=0,
disable_shuffle=True,
scenario_tags=("nested", "1qna"),
scenario_focus=True,
)
assert {case.scene_type for case in focused} == {"nested", "1qna"}
def test_load_cases_from_video_list_preserves_order_and_caps(tmp_path: Path) -> None:
root = _write_dataset(tmp_path)
all_cases = data.discover_omniinteract_cases(root, data.OMNIINTERACT_SUBSETS, num_prompts=0, disable_shuffle=True)
ordered = list(reversed(all_cases))
video_list = tmp_path / "sampled_cases.jsonl"
video_list.write_text("".join(json.dumps(data.sampled_case_row(case)) + "\n" for case in ordered))
capped = data.load_omniinteract_cases_from_video_list(video_list, num_prompts=2)
assert [case.video_path for case in capped] == [case.video_path for case in ordered[:2]]
full = data.load_omniinteract_cases_from_video_list(video_list, num_prompts=0)
assert [case.video_path for case in full] == [case.video_path for case in ordered]
oversized = data.load_omniinteract_cases_from_video_list(video_list, num_prompts=99)
assert [case.video_path for case in oversized] == [case.video_path for case in ordered]
def test_load_cases_from_video_list_requires_existing_video(tmp_path: Path) -> None:
root = _write_dataset(tmp_path)
case = data.discover_omniinteract_cases(root, ("1q1a",), num_prompts=1, disable_shuffle=True)[0]
row = data.sampled_case_row(case)
row["video_path"] = str(tmp_path / "missing.mp4")
video_list = tmp_path / "bad.jsonl"
video_list.write_text(json.dumps(row) + "\n")
with pytest.raises(FileNotFoundError, match="video missing"):
data.load_omniinteract_cases_from_video_list(video_list, num_prompts=0)
def test_get_samples_from_video_list_without_dataset_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
root = _write_dataset(tmp_path)
cases = data.discover_omniinteract_cases(root, data.OMNIINTERACT_SUBSETS, num_prompts=0, disable_shuffle=True)
video_list = tmp_path / "sampled_cases.jsonl"
video_list.write_text("".join(json.dumps(data.sampled_case_row(case)) + "\n" for case in cases))
ref = tmp_path / "ref.wav"
ref.touch()
parser = TrackingArgumentParser()
OmniBenchmarkServingSubcommand.add_cli_args(parser)
args = parser.parse_args(
[
"--backend",
"openai-realtime-duplex",
"--dataset-name",
"omniinteract",
"--model",
MODEL,
"--endpoint",
"/v1/realtime",
"--num-prompts",
"2",
"--omniinteract-ref-audio",
str(ref),
"--omniinteract-video-list",
str(video_list),
]
)
preprocess_serve_args(args)
monkeypatch.setattr(
"vllm_omni.clients.duplex.reference_audio_data_url",
lambda _: "data:audio/wav;base64,ref",
)
monkeypatch.setattr(benchmark_patch, "prepare_media", lambda *a, **k: (1.0, b"pcm", ["frame"]))
samples = benchmark_patch.get_samples(args, None)
assert len(samples) == 2
assert [sample.omniinteract_case.video_path for sample in samples] == [case.video_path for case in cases[:2]]
def _write_overlapping_cover_dataset(root: Path) -> Path:
data_root = root / "data"
one_q1a = data_root / "1q1a"
(one_q1a / "videos").mkdir(parents=True)
(one_q1a / "annotations").mkdir()
entries = []
specs = [
(
"pair_a",
[
{
"question_time": 0.0,
"answer_time": 1.0,
"question_type": "realtime",
"is_interrupted": True,
}
],
),
(
"pair_b",
[
{
"question_time": 0.0,
"answer_time": 1.0,
"question_type": "realtime",
"is_interrupted": True,
}
],
),
(
"proactive_only",
[{"question_time": 0.0, "answer_time": 1.0, "question_type": "proactive"}],
),
]
for name, annotation in specs:
video = one_q1a / "videos" / f"{name}.mp4"
ann = one_q1a / "annotations" / f"{name}.json"
video.touch()
ann.write_text(json.dumps(annotation))
entries.append(
{
"video": f"videos/{name}.mp4",
"annotation": f"annotations/{name}.json",
"scene_type": "multi_turn",
}
)
(one_q1a / "video_json_map.json").write_text(json.dumps({"entries": entries}))
return data_root
def test_cover_sampling_skips_tags_already_covered(tmp_path: Path) -> None:
root = _write_overlapping_cover_dataset(tmp_path)
selected = data.discover_omniinteract_cases(
root,
("1q1a",),
num_prompts=2,
disable_shuffle=True,
scenario_tags=("realtime", "interrupted", "proactive"),
)
names = [case.video_path.stem for case in selected]
assert names == ["pair_a", "proactive_only"]
selected_tags = set().union(*(data.case_scenario_tags(case) for case in selected))
assert {"realtime", "interrupted", "proactive"} <= selected_tags
def _archive(path: Path, member: str) -> None:
with tarfile.open(path, "w") as archive:
content = b"{}"
@@ -542,6 +788,51 @@ def test_ineligible_outputs_are_excluded_from_manifest(tmp_path: Path, audio_tim
assert reason in result.official_eval_ineligible_reasons
oi.write_batch_artifacts(tmp_path / "out", [case], [result])
assert (tmp_path / "out" / "official_eval_manifest.jsonl").read_text() == ""
sampled = [
json.loads(line) for line in (tmp_path / "out" / "sampled_cases.jsonl").read_text().splitlines() if line.strip()
]
assert sampled == [
{
"video_path": str(case.video_path.resolve()),
"output_name": data.official_output_name(case),
"subset": case.subset,
}
]
def test_official_output_name_matches_minicpmo_batch_layout(tmp_path: Path):
one_q1a = _case(tmp_path, subset="1q1a", name="videos/clip_a.mp4")
one_qna = _case(tmp_path, subset="1qna", name="videos_bench/nested/guide.mp4")
assert data.official_output_name(one_q1a) == "1q1a/videos__clip_a"
assert data.official_output_name(one_qna) == "1qna/nested__guide"
row = data.sampled_case_row(one_q1a)
assert row["video_path"] == str(one_q1a.video_path.resolve())
assert row["output_name"] == "1q1a/videos__clip_a"
assert row["subset"] == "1q1a"
def test_sampled_cases_jsonl_lists_every_case_including_failures(tmp_path: Path):
ok_case = _case(tmp_path, name="videos/ok.mp4")
bad_case = _case(tmp_path, name="videos/bad.mp4")
ok_result = oi.OmniInteractCaseResult(
ok_case.subset, str(ok_case.video_path), str(oi._output_dir(tmp_path / "out", ok_case)), success=True
)
ok_result.eligible_for_official_eval = True
bad_result = oi.OmniInteractCaseResult(
bad_case.subset, str(bad_case.video_path), str(oi._output_dir(tmp_path / "out", bad_case)), success=False
)
oi.write_batch_artifacts(tmp_path / "out", [ok_case, bad_case], [ok_result, bad_result])
sampled = [
json.loads(line) for line in (tmp_path / "out" / "sampled_cases.jsonl").read_text().splitlines() if line.strip()
]
assert [row["output_name"] for row in sampled] == ["1q1a/videos__ok", "1q1a/videos__bad"]
manifest = [
json.loads(line)
for line in (tmp_path / "out" / "official_eval_manifest.jsonl").read_text().splitlines()
if line.strip()
]
assert len(manifest) == 1
assert manifest[0]["sample_id"].startswith("1q1a__")
def test_artifacts_require_complete_audio_and_transcript_response(tmp_path: Path):
+820
View File
@@ -0,0 +1,820 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
import json
from dataclasses import asdict
from pathlib import Path
import pytest
from vllm_omni.benchmarks.data_modules.omniinteract_dataset import (
OmniInteractCase,
OmniInteractEvaluationOptions,
)
from vllm_omni.benchmarks.omniinteract import OmniInteractCaseResult
from vllm_omni.benchmarks.omniinteract_eval import (
AlignedWord,
TranscriptChunk,
_summarize,
build_slots,
evaluate_batch,
evaluate_case,
evaluation_inputs_fingerprint,
match_slots,
)
from vllm_omni.benchmarks.omniinteract_judge import (
CoreJudgment,
EarlyJudgment,
OmniInteractJudge,
PartialJudgment,
)
pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.benchmark]
class _FixedJudge:
def judge_early(
self,
slot: dict[str, object],
full_context: str,
actual_text: str,
) -> EarlyJudgment:
del slot, full_context, actual_text
return EarlyJudgment("neutral", 1.0, "ok", "{}", "llm_json")
def judge_core(
self,
slot: dict[str, object],
full_context: str,
actual_text: str,
future_answers: str,
) -> CoreJudgment:
del slot, full_context, actual_text, future_answers
return CoreJudgment(1.0, "answer", False, "ok", "{}", "llm_json")
def judge_interrupted_partial(
self,
slot: dict[str, object],
actual_text: str,
) -> PartialJudgment:
del slot, actual_text
return PartialJudgment(1.0, False, "ok", "{}", "llm_json")
class _GeneratedJudge(OmniInteractJudge):
def __init__(self, responses: list[str]) -> None:
super().__init__("http://judge", "judge-model")
self.responses = iter(responses)
def _generate(self, system_prompt: str, user_prompt: str) -> str:
del system_prompt, user_prompt
return next(self.responses)
def test_build_slots_and_split_word_aligned_chunk() -> None:
slots = build_slots(
[
{
"question_time": 0.0,
"answer_time": 2.0,
"question_text": "question",
"answer_text": "answer",
"question_type": "realtime",
},
{
"question_time": 5.0,
"answer_time": 6.0,
"question_text": "next",
"answer_text": "next answer",
"question_type": "proactive",
},
],
"multi_turn",
)
chunk = TranscriptChunk(
source_id=0,
text="okay answer",
start=0.5,
end=2.5,
aligned_words=(
AlignedWord("okay", 0.5, 0.9),
AlignedWord("answer", 2.1, 2.5),
),
)
matched, unmatched = match_slots(slots, [chunk])
assert not unmatched
assert matched[0].early_chunks[0].text == "okay"
assert matched[0].core_chunks[0].text == "answer"
assert matched[0].core_chunks[0].start == pytest.approx(2.1)
def test_judge_parses_json_wrapped_in_model_text() -> None:
judge = _GeneratedJudge(
[
'result: {"flag":"FP_Hallucination","score":1,"rationale":"early"}',
'```json\n{"score":0.8,"trigger_phrase":"answer","spoiler":false,"rationale":"good"}\n```',
'{"score":0.7,"hallucination":false,"rationale":"partial"}',
]
)
slot = {"question_text": "q", "gt_answer": "answer"}
early = judge.judge_early(slot, "context", "guess")
core = judge.judge_core(slot, "context", "answer", "(none)")
partial = judge.judge_interrupted_partial(slot, "answer")
assert early.category == "hallucination"
assert early.score == 0.0
assert core.score == pytest.approx(0.8)
assert core.trigger_phrase == "answer"
assert partial.score == pytest.approx(0.7)
def test_evaluate_case_persists_accuracy_artifact(tmp_path: Path) -> None:
annotation = tmp_path / "annotation.json"
annotation.write_text(
json.dumps(
[
{
"question_time": 0.0,
"answer_time": 1.0,
"question_text": "question",
"answer_text": "answer",
"question_type": "realtime",
}
]
)
)
output_dir = tmp_path / "case-output"
output_dir.mkdir()
(output_dir / "wav_transcript.json").write_text(
json.dumps(
{
"chunks": [
{
"timestamp": [1.0, 1.5],
"text": "answer",
"aligned_words": [{"text": "answer", "start": 1.0, "end": 1.5}],
}
]
}
)
)
video = tmp_path / "video.mp4"
video.touch()
case = OmniInteractCase("1q1a", "video.mp4", video, annotation, "multi_turn")
result = OmniInteractCaseResult(
subset="1q1a",
video=str(video),
output_dir=str(output_dir),
success=True,
eligible_for_official_eval=True,
)
artifact = tmp_path / "evaluation.json"
evaluation = evaluate_case(case, result, _FixedJudge(), artifact)
assert evaluation["timing_precision"] == "word-aligned"
assert evaluation["summary"]["IA_QTF1"] == pytest.approx(1.0)
paper = evaluation["summary"]["paper_metrics"]["exp_f1"]
assert paper["all_global"]["IA_QTF1"] == pytest.approx(1.0)
assert paper["realtime"]["IA_QTF1"] == pytest.approx(1.0)
assert json.loads(artifact.read_text())["status"] == "ok"
fingerprint = evaluation["inputs_fingerprint"]
assert fingerprint["annotation_sha256"]
assert fingerprint["transcript_sha256"]
assert fingerprint["protocol_source"]
def _realtime_eval_case(tmp_path: Path, *, transcript_text: str) -> tuple[OmniInteractCase, OmniInteractCaseResult]:
annotation = tmp_path / "annotation.json"
annotation.write_text(
json.dumps(
[
{
"question_time": 0.0,
"answer_time": 1.0,
"question_text": "question",
"answer_text": "answer",
"question_type": "realtime",
}
]
)
)
output_dir = tmp_path / "case-output"
output_dir.mkdir(exist_ok=True)
chunks: list[dict[str, object]] = []
if transcript_text:
chunks.append({"timestamp": [1.0, 1.5], "text": transcript_text})
(output_dir / "wav_transcript.json").write_text(json.dumps({"chunks": chunks}))
video = tmp_path / "video.mp4"
video.touch()
case = OmniInteractCase("1q1a", "video.mp4", video, annotation, "multi_turn")
result = OmniInteractCaseResult(
subset="1q1a",
video=str(video),
output_dir=str(output_dir),
success=True,
eligible_for_official_eval=True,
)
return case, result
def _eval_options(tmp_path: Path, *, skip_existing: bool) -> OmniInteractEvaluationOptions:
return OmniInteractEvaluationOptions(
judge_base_url="http://127.0.0.1:9",
judge_model="cached-judge",
judge_api_key="EMPTY",
judge_timeout_s=60.0,
judge_max_tokens=32,
workers=1,
output_dir=tmp_path / "evaluation",
skip_existing=skip_existing,
)
class _ConfiguredFixedJudge(_FixedJudge):
def __init__(
self,
base_url: str,
model: str,
*,
api_key: str = "EMPTY",
timeout_s: float = 60.0,
max_tokens: int = 512,
) -> None:
del api_key, timeout_s
self.base_url = base_url
self.model = model
self.max_tokens = max_tokens
def test_skip_existing_reuses_matching_fingerprint(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
import vllm_omni.benchmarks.omniinteract_eval as eval_mod
case, result = _realtime_eval_case(tmp_path, transcript_text="answer")
options = _eval_options(tmp_path, skip_existing=False)
monkeypatch.setattr(eval_mod, "OmniInteractJudge", _ConfiguredFixedJudge)
first = evaluate_batch([case], [result], options)
assert first["summary"]["IA_QTF1"] == pytest.approx(1.0)
calls: list[str] = []
def _forbidden(
case: OmniInteractCase,
result: OmniInteractCaseResult,
judge: object,
output_path: Path,
config: object = None,
inputs_fingerprint: object = None,
) -> dict[str, object]:
del case, result, judge, output_path, config, inputs_fingerprint
calls.append("evaluate_case")
raise AssertionError("matching fingerprint must reuse the cached evaluation")
monkeypatch.setattr(eval_mod, "evaluate_case", _forbidden)
reused = evaluate_batch([case], [result], _eval_options(tmp_path, skip_existing=True))
assert calls == []
assert reused["summary"]["IA_QTF1"] == pytest.approx(1.0)
def test_skip_existing_rejects_stale_transcript(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
import vllm_omni.benchmarks.omniinteract_eval as eval_mod
case, result = _realtime_eval_case(tmp_path, transcript_text="answer")
options = _eval_options(tmp_path, skip_existing=True)
fingerprint = evaluation_inputs_fingerprint(
annotation_path=case.annotation_path,
transcript_path=Path(result.output_dir) / "wav_transcript.json",
judge_model=options.judge_model,
judge_base_url=options.judge_base_url,
judge_max_tokens=options.judge_max_tokens,
)
sample_id = f"{case.subset}__{Path(result.output_dir).name}"
cache_path = options.output_dir / f"{sample_id}.unified_eval.json"
cache_path.parent.mkdir(parents=True)
cache_path.write_text(
json.dumps(
{
"status": "ok",
"sample_id": sample_id,
"inputs_fingerprint": asdict(fingerprint),
"summary": {
"IA_QTF1": 1.0,
"num_slots": 1,
"num_unmatched_chunks": 0,
"Global_TP": 1,
"Global_FP": 0,
"Global_FN": 0,
},
"slots": [],
}
)
)
(Path(result.output_dir) / "wav_transcript.json").write_text(json.dumps({"chunks": []}))
monkeypatch.setattr(eval_mod, "OmniInteractJudge", _ConfiguredFixedJudge)
fresh = evaluate_batch([case], [result], options)
assert fresh["summary"]["IA_QTF1"] == pytest.approx(0.0)
class _MissingCoreScoreJudge(_ConfiguredFixedJudge):
def judge_core(
self,
slot: dict[str, object],
full_context: str,
actual_text: str,
future_answers: str,
) -> CoreJudgment:
del slot, full_context, actual_text, future_answers
return CoreJudgment(0.0, "", False, "Could not evaluate", "{}", "llm_parse_failed")
def test_evaluate_batch_reports_missing_score_as_parse_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
import vllm_omni.benchmarks.omniinteract_eval as eval_mod
case, result = _realtime_eval_case(tmp_path, transcript_text="answer")
options = _eval_options(tmp_path, skip_existing=False)
monkeypatch.setattr(eval_mod, "OmniInteractJudge", _MissingCoreScoreJudge)
evaluation = evaluate_batch([case], [result], options)
assert evaluation["status"] == "failed"
assert evaluation["evaluated"] == 0
assert evaluation["failed"] == 1
sample_id = f"{case.subset}__{Path(result.output_dir).name}"
cached = json.loads((options.output_dir / f"{sample_id}.unified_eval.json").read_text())
assert cached["status"] == "parse_failed"
class _CoreRationaleNumberJudge(OmniInteractJudge):
def __init__(
self,
base_url: str,
model: str,
*,
api_key: str = "EMPTY",
timeout_s: float = 60.0,
max_tokens: int = 512,
) -> None:
super().__init__(base_url, model, api_key=api_key, timeout_s=timeout_s, max_tokens=max_tokens)
def _generate(self, system_prompt: str, user_prompt: str) -> str:
del system_prompt, user_prompt
return '{"rationale":"Cannot score: 1 required reference is missing."}'
def test_evaluate_batch_rejects_core_rationale_number_as_parse_failure(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
import vllm_omni.benchmarks.omniinteract_eval as eval_mod
case, result = _realtime_eval_case(tmp_path, transcript_text="answer")
options = _eval_options(tmp_path, skip_existing=False)
monkeypatch.setattr(eval_mod, "OmniInteractJudge", _CoreRationaleNumberJudge)
evaluation = evaluate_batch([case], [result], options)
assert evaluation["status"] == "failed"
assert evaluation["evaluated"] == 0
assert evaluation["failed"] == 1
sample_id = f"{case.subset}__{Path(result.output_dir).name}"
cached = json.loads((options.output_dir / f"{sample_id}.unified_eval.json").read_text())
assert cached["status"] == "parse_failed"
assert cached["slots"][0]["stage_core"]["parse_source"] == "llm_parse_failed"
assert cached["slots"][0]["stage_core"]["S_core"] == 0.0
def test_skip_existing_rejects_parse_failed_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
import vllm_omni.benchmarks.omniinteract_eval as eval_mod
case, result = _realtime_eval_case(tmp_path, transcript_text="answer")
options = _eval_options(tmp_path, skip_existing=True)
fingerprint = evaluation_inputs_fingerprint(
annotation_path=case.annotation_path,
transcript_path=Path(result.output_dir) / "wav_transcript.json",
judge_model=options.judge_model,
judge_base_url=options.judge_base_url,
judge_max_tokens=options.judge_max_tokens,
)
sample_id = f"{case.subset}__{Path(result.output_dir).name}"
cache_path = options.output_dir / f"{sample_id}.unified_eval.json"
cache_path.parent.mkdir(parents=True)
cache_path.write_text(
json.dumps(
{
"status": "parse_failed",
"sample_id": sample_id,
"inputs_fingerprint": asdict(fingerprint),
"summary": {
"IA_QTF1": 0.0,
"num_slots": 1,
"num_unmatched_chunks": 0,
"Global_TP": 0,
"Global_FP": 0,
"Global_FN": 1,
},
"slots": [
{
"stage_core": {"parse_source": "llm_parse_failed", "S_core": 0.0},
"TP_n": 0.0,
"FP_delta": 0,
"FN_delta": 1,
}
],
}
)
)
monkeypatch.setattr(eval_mod, "OmniInteractJudge", _ConfiguredFixedJudge)
fresh = evaluate_batch([case], [result], options)
assert fresh["evaluated"] == 1
assert fresh["failed"] == 0
assert fresh["summary"]["IA_QTF1"] == pytest.approx(1.0)
assert json.loads(cache_path.read_text())["status"] == "ok"
def _slot_row(
*,
scene_type: str,
question_type: str,
tp_n: float,
fp: int = 0,
fn: int = 0,
nested_role: str | None = None,
nested_group_id: int | None = None,
sample_id: str = "s",
tp_core: float | None = None,
answer_start: float | None = 1.0,
) -> dict[str, object]:
core_tp = tp_n if tp_core is None else tp_core
return {
"sample_id": sample_id,
"scene_type": scene_type,
"question_type": question_type,
"nested_role": nested_role,
"nested_group_id": nested_group_id,
"TP_n": tp_n,
"TP_core": core_tp,
"FP_delta": fp,
"FN_delta": fn,
"is_interrupted": False,
"stage_core": {"answer_start": answer_start},
}
def test_paper_metrics_exclude_nested_from_realtime_and_proactive() -> None:
summary = _summarize(
[
_slot_row(scene_type="multi_turn", question_type="realtime", tp_n=1.0),
_slot_row(
scene_type="nested",
question_type="realtime",
tp_n=1.0,
nested_role="inner",
nested_group_id=1,
answer_start=0.5,
),
_slot_row(
scene_type="nested",
question_type="proactive",
tp_n=1.0,
nested_role="outer",
nested_group_id=1,
answer_start=1.5,
),
_slot_row(scene_type="multi_turn", question_type="proactive", tp_n=1.0),
_slot_row(scene_type="1QnA", question_type="step", tp_n=1.0, fn=0),
],
unmatched=0,
)
paper = summary["paper_metrics"]["exp_f1"]
assert paper["realtime"]["num_slots"] == 1
assert paper["proactive"]["num_slots"] == 1
assert paper["nested"]["num_slots"] == 2
assert paper["one_q1a_global"]["num_slots"] == 4
assert paper["one_qna"]["num_slots"] == 1
assert paper["all_global"]["num_slots"] == 5
# All Global includes 1QnA; 1Q1A Global does not.
assert paper["all_global"]["Global_TP"] == pytest.approx(5.0)
assert paper["one_q1a_global"]["Global_TP"] == pytest.approx(4.0)
assert summary["nested"]["NCCS"] == pytest.approx(1.0)
assert summary["nested"]["missing_q1_count"] == 0
assert summary["paper_metrics"]["exp_nested"]["missed_outer"] == 0
assert summary["paper_metrics"]["exp_nested"]["inner_IA_QTF1"] == pytest.approx(1.0)
assert summary["paper_metrics"]["exp_nested"]["outer_IA_QTF1"] == pytest.approx(1.0)
assert summary["scenario_case_counts"]["realtime"] == 1
assert summary["scenario_case_counts"]["proactive"] == 1
assert summary["scenario_case_counts"]["nested"] == 1
assert summary["scenario_case_counts"]["1qna"] == 1
def test_nccs_allows_missing_answer_start() -> None:
summary = _summarize(
[
_slot_row(
scene_type="nested",
question_type="realtime",
tp_n=0.8,
tp_core=0.8,
nested_role="inner",
nested_group_id=1,
answer_start=None,
),
_slot_row(
scene_type="nested",
question_type="proactive",
tp_n=0.8,
tp_core=0.8,
nested_role="outer",
nested_group_id=1,
answer_start=None,
),
],
unmatched=0,
)
assert summary["nested"]["success_pairs"] == 1
assert summary["nested"]["NCCS"] == pytest.approx(0.8)
def test_missed_outer_counts_zero_outer_core() -> None:
summary = _summarize(
[
_slot_row(
scene_type="nested",
question_type="realtime",
tp_n=1.0,
tp_core=1.0,
nested_role="inner",
nested_group_id=1,
answer_start=0.5,
),
_slot_row(
scene_type="nested",
question_type="proactive",
tp_n=0.0,
tp_core=0.0,
fn=1,
nested_role="outer",
nested_group_id=1,
answer_start=1.5,
),
],
unmatched=0,
)
nested = summary["paper_metrics"]["exp_nested"]
assert nested["missed_outer"] == 1
assert nested["NCCS"] == pytest.approx(0.0)
assert nested["inner_IA_QTF1"] == pytest.approx(1.0)
assert nested["outer_IA_QTF1"] == pytest.approx(0.0)
def test_cross_boundary_core_sets_effective_start_hint() -> None:
slots = build_slots(
[
{
"question_time": 0.0,
"answer_time": 2.0,
"question_text": "question",
"answer_text": "answer",
"question_type": "realtime",
}
],
"multi_turn",
)
chunk = TranscriptChunk(
source_id=0,
text="okay answer",
start=0.5,
end=2.5,
aligned_words=(
AlignedWord("okay", 0.5, 0.9),
AlignedWord("answer", 2.1, 2.5),
),
)
matched, unmatched = match_slots(slots, [chunk])
assert not unmatched
assert matched[0].core_chunks[0].effective_start_hint == pytest.approx(2.0)
class _NonNeutralEarlyJudge(_FixedJudge):
def judge_early(
self,
slot: dict[str, object],
full_context: str,
actual_text: str,
) -> EarlyJudgment:
del slot, full_context, actual_text
return EarlyJudgment("other", 1.0, "ignored", "{}", "llm_json")
def test_early_tp_ack_only_for_neutral(tmp_path: Path) -> None:
annotation = tmp_path / "annotation.json"
annotation.write_text(
json.dumps(
[
{
"question_time": 0.0,
"answer_time": 2.0,
"question_text": "question",
"answer_text": "answer",
"question_type": "realtime",
}
]
)
)
output_dir = tmp_path / "case-output"
output_dir.mkdir()
(output_dir / "wav_transcript.json").write_text(
json.dumps(
{
"chunks": [
{
"timestamp": [0.5, 0.9],
"text": "okay",
"aligned_words": [{"text": "okay", "start": 0.5, "end": 0.9}],
},
{
"timestamp": [2.1, 2.5],
"text": "answer",
"aligned_words": [{"text": "answer", "start": 2.1, "end": 2.5}],
},
]
}
)
)
video = tmp_path / "video.mp4"
video.touch()
case = OmniInteractCase("1q1a", "video.mp4", video, annotation, "multi_turn")
result = OmniInteractCaseResult(
subset="1q1a",
video=str(video),
output_dir=str(output_dir),
success=True,
eligible_for_official_eval=True,
)
evaluation = evaluate_case(case, result, _NonNeutralEarlyJudge(), tmp_path / "evaluation.json")
assert evaluation["slots"][0]["TP_ack"] == 0.0
assert evaluation["slots"][0]["stage_early"]["category"] == "other"
def test_paper_metrics_unknown_inner_does_not_underflow_realtime() -> None:
summary = _summarize(
[
_slot_row(scene_type="multi_turn", question_type="realtime", tp_n=0.5),
_slot_row(
scene_type="nested",
question_type="unknown",
tp_n=2.0,
nested_role="inner",
nested_group_id=1,
answer_start=0.5,
),
_slot_row(
scene_type="nested",
question_type="proactive",
tp_n=1.0,
nested_role="outer",
nested_group_id=1,
answer_start=1.5,
),
],
unmatched=0,
)
paper = summary["paper_metrics"]["exp_f1"]
assert paper["realtime"]["Global_TP"] == pytest.approx(0.5)
assert paper["realtime"]["num_slots"] == 1
assert paper["realtime"]["Precision"] >= 0.0
assert paper["one_q1a_global"]["Global_TP"] == pytest.approx(3.5)
def test_core_score_float_ignores_leading_counts() -> None:
from vllm_omni.benchmarks.omniinteract_judge import _score_float_from_text
assert _score_float_from_text("The answer covers 3 of the 5 key points, score: 0.6") == pytest.approx(0.6)
assert _score_float_from_text("score: 3") is None
assert _score_float_from_text("no score here 0.9") is None
judge = _GeneratedJudge(["The answer covers 3 of the 5 key points, score: 0.6"])
core = judge.judge_core({"question_text": "q", "gt_answer": "a"}, "ctx", "answer", "(none)")
assert core.score == pytest.approx(0.6)
assert core.parse_source == "llm_float_text"
def test_core_missing_json_score_does_not_use_rationale_number() -> None:
raw = '{"rationale":"Cannot score: 1 required reference is missing."}'
core = _GeneratedJudge([raw]).judge_core({"question_text": "q", "gt_answer": "a"}, "ctx", "answer", "(none)")
assert core.score == 0.0
assert core.parse_source == "llm_parse_failed"
def test_early_missing_score_and_unparsed_flag() -> None:
missing_score = _GeneratedJudge(['{"flag":"Neutral","rationale":"ok"}'])
early = missing_score.judge_early({"question_text": "q", "gt_answer": "a"}, "ctx", "hi")
assert early.category == "unparsed"
assert early.score == 0.0
assert early.parse_source == "llm_parse_failed"
garbage = _GeneratedJudge(["not json and not a known flag"])
unparsed = garbage.judge_early({"question_text": "q", "gt_answer": "a"}, "ctx", "hi")
assert unparsed.category == "unparsed"
assert unparsed.score == 0.0
assert unparsed.parse_source == "llm_parse_failed"
@pytest.mark.parametrize(
("path", "response"),
[
("early", '{"flag":"Neutral","rationale":"Could not evaluate this answer"}'),
("early", '{"flag":"Neutral","score":"invalid","rationale":"bad"}'),
("core", '{"rationale":"Could not evaluate this answer"}'),
("core", '{"rationale":"Cannot score: 1 required reference is missing."}'),
("core", '{"score":"invalid","trigger_phrase":"","spoiler":false,"rationale":"bad"}'),
("partial", '{"rationale":"Could not evaluate this answer"}'),
("partial", '{"score":"invalid","hallucination":false,"rationale":"bad"}'),
],
)
def test_missing_or_invalid_json_scores_are_parse_failures(path: str, response: str) -> None:
slot = {"question_text": "q", "gt_answer": "a"}
judge = _GeneratedJudge([response])
if path == "early":
judgment = judge.judge_early(slot, "ctx", "hi")
elif path == "core":
judgment = judge.judge_core(slot, "ctx", "answer", "(none)")
else:
judgment = judge.judge_interrupted_partial(slot, "answer")
assert judgment.score == 0.0
assert judgment.parse_source == "llm_parse_failed"
def test_explicit_zero_json_score_is_valid_llm_json() -> None:
slot = {"question_text": "q", "gt_answer": "a"}
early = _GeneratedJudge(['{"flag":"Neutral","score":0,"rationale":"zero"}']).judge_early(slot, "ctx", "hi")
core = _GeneratedJudge(['{"score":0.0,"trigger_phrase":"","spoiler":false,"rationale":"zero"}']).judge_core(
slot, "ctx", "answer", "(none)"
)
partial = _GeneratedJudge(['{"score":0,"hallucination":false,"rationale":"zero"}']).judge_interrupted_partial(
slot, "answer"
)
assert early.category == "neutral"
assert early.score == 0.0
assert early.parse_source == "llm_json"
assert core.score == 0.0
assert core.parse_source == "llm_json"
assert partial.score == 0.0
assert partial.parse_source == "llm_json"
@pytest.mark.parametrize("score_literal", ['"NaN"', '"Infinity"', '"-Infinity"', "NaN", "Infinity", "-Infinity"])
def test_nonfinite_json_scores_are_parse_failures(score_literal: str) -> None:
slot = {"question_text": "q", "gt_answer": "a"}
early = _GeneratedJudge([f'{{"flag":"Neutral","score":{score_literal},"rationale":"bad"}}']).judge_early(
slot, "ctx", "hi"
)
core = _GeneratedJudge(
[f'{{"score":{score_literal},"trigger_phrase":"answer","spoiler":false,"rationale":"bad"}}']
).judge_core(slot, "ctx", "answer", "(none)")
partial = _GeneratedJudge(
[f'{{"score":{score_literal},"hallucination":false,"rationale":"bad"}}']
).judge_interrupted_partial(slot, "answer")
assert early.score == 0.0
assert early.parse_source == "llm_parse_failed"
assert core.score == 0.0
assert core.parse_source == "llm_parse_failed"
assert core.trigger_phrase == ""
assert partial.score == 0.0
assert partial.parse_source == "llm_parse_failed"
def test_fingerprint_requires_evaluator_schema_version() -> None:
from vllm_omni.benchmarks.omniinteract_eval import EvaluationInputsFingerprint
payload = {
"annotation_sha256": "a" * 64,
"transcript_sha256": "b" * 64,
"judge_model": "m",
"judge_base_url": "http://127.0.0.1:9",
"judge_max_tokens": 32,
"protocol_source": "proto",
}
assert EvaluationInputsFingerprint.from_mapping(payload) is None
payload["evaluator_schema_version"] = 1
assert EvaluationInputsFingerprint.from_mapping(payload) is not None
payload["evaluator_schema_version"] = 999
matched = EvaluationInputsFingerprint.from_mapping(payload)
assert matched is not None
assert matched.evaluator_schema_version == 999
def test_summarize_counts_parse_sources() -> None:
rows = [
{
**_slot_row(scene_type="multi_turn", question_type="realtime", tp_n=1.0),
"stage_early": {"parse_source": "llm_json"},
"stage_core": {"parse_source": "llm_parse_failed", "trigger_fallback": True},
}
]
summary = _summarize(rows, unmatched=0)
assert summary["judge_parse"]["llm_json"] == 1
assert summary["judge_parse"]["llm_parse_failed"] == 1
assert summary["judge_parse"]["judge_calls"] == 2
assert summary["trigger_fallback_slots"] == 1
+47
View File
@@ -224,6 +224,53 @@ def test_preprocess_serve_args_applies_safe_omniinteract_prompt_default(
assert args.num_prompts == expected
@pytest.mark.parametrize(
("extra", "match"),
[
(["--seed", "7"], "cannot be combined with --seed"),
(["--dataset-path", "/tmp/omniinteract-data"], "cannot be combined with --dataset-path"),
(["--omniinteract-scenario-tags", "realtime"], "cannot be combined with --omniinteract-scenario-tags"),
(["--omniinteract-scenario-focus"], "cannot be combined with --omniinteract-scenario-focus"),
],
)
def test_preprocess_serve_args_rejects_omniinteract_video_list_conflicts(
tmp_path: Path,
extra: list[str],
match: str,
) -> None:
ref = tmp_path / "ref.wav"
ref.touch()
video_list = tmp_path / "list.jsonl"
video_list.write_text("{}\n")
parser = TrackingArgumentParser()
parser.add_argument("--dataset-name", default="sharegpt")
parser.add_argument("--backend", default="vllm")
parser.add_argument("--endpoint", default="/v1/completions")
parser.add_argument("--model", default="dummy")
parser.add_argument("--seed", type=int, default=0)
parser.add_argument("--dataset-path", default=None)
parser.add_argument("--num-prompts", type=int, default=1000)
parser.add_argument("--max-concurrency", type=int, default=None)
add_omni_args(parser)
args = parser.parse_args(
[
"--dataset-name",
"omniinteract",
"--backend",
"openai-realtime-duplex",
"--endpoint",
"/v1/realtime",
"--omniinteract-ref-audio",
str(ref),
"--omniinteract-video-list",
str(video_list),
*extra,
]
)
with pytest.raises(ValueError, match=match):
preprocess_serve_args(args)
@pytest.mark.parametrize(
("argv", "expected_extra_body", "expected_explicit"),
[
+216 -3
View File
@@ -100,6 +100,71 @@ class _SingleActiveContext:
stack.close()
def _config_for_test(test_name: str) -> dict[str, object] | None:
for config in BENCHMARK_CONFIGS:
if isinstance(config, dict) and config.get("test_name") == test_name:
return config
return None
def _judge_server_params_for_test(test_name: str) -> dict[str, object] | None:
config = _config_for_test(test_name)
if config is None:
return None
raw = config.get("judge_server_params")
if raw is None:
return None
if not isinstance(raw, dict):
raise TypeError(f"judge_server_params for {test_name} must be an object")
return raw
def _resolve_judge_cuda_visible_devices(configured: object) -> str:
"""Map a judge card index onto the process-visible CUDA device list.
``judge_server_params.cuda_visible_devices`` is an index into
``CUDA_VISIBLE_DEVICES`` when that env is set (``1`` with
``CUDA_VISIBLE_DEVICES=2,3`` selects physical GPU 3). Without the env it
is used as a raw device list.
"""
if not isinstance(configured, str) or not configured:
raise ValueError("judge_server_params.cuda_visible_devices must be a non-empty string")
parent = os.environ.get("CUDA_VISIBLE_DEVICES")
if parent is None or not parent.strip():
return configured
visible = [part.strip() for part in parent.split(",") if part.strip()]
if configured.isdigit():
index = int(configured)
if index < 0 or index >= len(visible):
raise ValueError(
f"judge_server_params.cuda_visible_devices={configured!r} is outside CUDA_VISIBLE_DEVICES={parent!r}"
)
return visible[index]
return configured
@contextmanager
def _start_judge_server(judge_params: dict[str, object]):
model = judge_params.get("model")
if not isinstance(model, str) or not model:
raise ValueError("judge_server_params.model must be a non-empty string")
extra = judge_params.get("extra_cli_args") or ()
if not isinstance(extra, list | tuple):
raise TypeError("judge_server_params.extra_cli_args must be a list")
visible = _resolve_judge_cuda_visible_devices(judge_params.get("cuda_visible_devices", "1"))
print(f"Starting OmniInteract judge with model: {model} on CUDA_VISIBLE_DEVICES={visible}")
with OmniServer(
model,
[str(item) for item in extra],
use_omni=bool(judge_params.get("use_omni", False)),
env_dict={"CUDA_VISIBLE_DEVICES": visible},
) as judge:
print(f"OmniInteract judge started on {judge.host}:{judge.port}")
yield judge
print("OmniInteract judge stopping...")
print("OmniInteract judge stopped")
# OmniServer defaults for flags not already present in JSON ``serve_args`` /
# ``extra_cli_args``. Add new (flag, value) pairs here rather than special-casing.
_OMNI_DEFAULT_SERVER_ARGS: tuple[tuple[str, str], ...] = (
@@ -200,9 +265,16 @@ def _start_omni_server(server_param):
server_args = ["--stage-overrides", stage_overrides] + server_args
if extra:
server_args = list(extra) + server_args
judge_params = _judge_server_params_for_test(test_name)
with OmniServer(model, server_args, use_omni=use_omni, env_dict=_omni_server_env()) as server:
server.test_name = test_name
server.judge_base_url = None
print("OmniServer started successfully")
if judge_params is None:
yield server
else:
with _start_judge_server(judge_params) as judge:
server.judge_base_url = f"http://{judge.host}:{judge.port}"
yield server
print("OmniServer stopping...")
@@ -260,6 +332,120 @@ def _resolve_num_warmups(params: dict[str, Any], *, default: int) -> int:
return value
def _is_finite_number(value: object) -> bool:
return isinstance(value, int | float) and not isinstance(value, bool) and math.isfinite(value)
_OMNIINTERACT_AGGREGATE_MIN_IA_QTF1 = "omniinteract_aggregate_min_ia_qtf1"
_OMNIINTERACT_AGGREGATE_SUBSETS = "omniinteract_aggregate_subsets"
_OMNIINTERACT_AGGREGATE_GROUP = "omniinteract_aggregate_group"
_OMNIINTERACT_AGGREGATE_COUNTS: dict[tuple[str, str], tuple[float, float, float]] = {}
_OMNIINTERACT_MIN_EVALUATED_CASES = 1
def _reset_omniinteract_aggregate_counts() -> None:
_OMNIINTERACT_AGGREGATE_COUNTS.clear()
def _ia_qtf1_from_counts(tp: float, fp: float, fn: float) -> float:
precision = tp / (tp + fp) if tp + fp > 0 else 0.0
recall = tp / (tp + fn) if tp + fn > 0 else 0.0
return (2.0 * precision * recall / (precision + recall)) if precision + recall > 0 else 0.0
def _assert_subset_evaluated(accuracy: dict[str, object]) -> None:
"""Reject a subset whose clipped or cancelled cases left nothing to score.
Individual ineligible cases may be skipped. A subset with fewer than
``_OMNIINTERACT_MIN_EVALUATED_CASES`` evaluated cases must not pass, or its
zero TP/FP/FN would be pooled as a successful accuracy result.
"""
evaluated = accuracy.get("evaluated")
if not isinstance(evaluated, int) or isinstance(evaluated, bool):
raise AssertionError("OmniInteract accuracy evaluated count is missing")
if evaluated < _OMNIINTERACT_MIN_EVALUATED_CASES:
raise AssertionError(
f"OmniInteract subset evaluated {evaluated} cases (skipped={accuracy.get('skipped')}); "
f"at least {_OMNIINTERACT_MIN_EVALUATED_CASES} evaluated case is required"
)
def _finite_accuracy_count(value: object, name: str) -> float:
assert _is_finite_number(value), f"OmniInteract accuracy summary {name} is missing"
return float(value)
def _aggregate_accuracy_params_for_test(test_name: str) -> dict[str, object]:
config = _config_for_test(test_name)
if config is None:
return {}
min_ia_qtf1 = config.get(_OMNIINTERACT_AGGREGATE_MIN_IA_QTF1)
if min_ia_qtf1 is None:
return {}
if not _is_finite_number(min_ia_qtf1):
raise ValueError("omniinteract_aggregate_min_ia_qtf1 must be a finite number")
subsets: list[str] = []
raw_params = config.get("benchmark_params") or []
if not isinstance(raw_params, list):
raise TypeError(f"benchmark_params for {test_name} must be a list")
for item in raw_params:
if not isinstance(item, dict) or not item.get("omniinteract_evaluate"):
continue
subset = item.get("omniinteract_subsets")
if not isinstance(subset, str) or not subset:
raise ValueError("omniinteract_aggregate_min_ia_qtf1 requires omniinteract_subsets on each evaluated case")
subsets.append(subset)
if not subsets:
raise ValueError("omniinteract_aggregate_min_ia_qtf1 requires at least one evaluated OmniInteract subset")
if len(set(subsets)) != len(subsets):
raise ValueError("omniinteract_aggregate_subsets must not contain duplicates")
return {
_OMNIINTERACT_AGGREGATE_MIN_IA_QTF1: float(min_ia_qtf1),
_OMNIINTERACT_AGGREGATE_SUBSETS: subsets,
_OMNIINTERACT_AGGREGATE_GROUP: test_name,
}
def _maybe_assert_aggregate_ia_qtf1(params: dict[str, object], acc_summary: dict[str, object]) -> None:
min_ia_qtf1 = params.get(_OMNIINTERACT_AGGREGATE_MIN_IA_QTF1)
if min_ia_qtf1 is None:
return
assert _is_finite_number(min_ia_qtf1), "omniinteract_aggregate_min_ia_qtf1 must be a finite number"
subsets = params.get(_OMNIINTERACT_AGGREGATE_SUBSETS)
if not isinstance(subsets, list | tuple) or not subsets:
raise ValueError("omniinteract_aggregate_min_ia_qtf1 requires omniinteract_aggregate_subsets")
if len(set(subsets)) != len(subsets):
raise ValueError("omniinteract_aggregate_subsets must not contain duplicates")
if not all(isinstance(item, str) and item for item in subsets):
raise ValueError("omniinteract_aggregate_subsets must be non-empty strings")
subset = params.get("omniinteract_subsets")
if not isinstance(subset, str) or not subset:
raise ValueError("omniinteract_aggregate_min_ia_qtf1 requires omniinteract_subsets")
if subset not in subsets:
raise ValueError(f"omniinteract_subsets {subset!r} is not in omniinteract_aggregate_subsets")
group = params.get(_OMNIINTERACT_AGGREGATE_GROUP)
if not isinstance(group, str) or not group:
raise ValueError("omniinteract_aggregate_min_ia_qtf1 requires omniinteract_aggregate_group")
tp = _finite_accuracy_count(acc_summary.get("Global_TP"), "Global_TP")
fp = _finite_accuracy_count(acc_summary.get("Global_FP"), "Global_FP")
fn = _finite_accuracy_count(acc_summary.get("Global_FN"), "Global_FN")
_OMNIINTERACT_AGGREGATE_COUNTS[(group, subset)] = (tp, fp, fn)
recorded = [_OMNIINTERACT_AGGREGATE_COUNTS.get((group, name)) for name in subsets]
present = [counts for counts in recorded if counts is not None]
if len(present) != len(subsets):
return
total_tp = sum(counts[0] for counts in present)
total_fp = sum(counts[1] for counts in present)
total_fn = sum(counts[2] for counts in present)
ia_qtf1 = _ia_qtf1_from_counts(total_tp, total_fp, total_fn)
print(
f"OmniInteract aggregate All Global IA-QTF1: {ia_qtf1:.6f} "
f"(TP={total_tp:.6f} / FP={total_fp:.6f} / FN={total_fn:.6f})"
)
assert ia_qtf1 >= float(min_ia_qtf1), f"OmniInteract aggregate All Global IA-QTF1 {ia_qtf1} is below {min_ia_qtf1}"
def assert_result(result, params, num_prompt) -> None:
assert result["completed"] == num_prompt, "Request failures exist"
if params.get("dataset_name") == "omniinteract":
@@ -271,6 +457,23 @@ def assert_result(result, params, num_prompt) -> None:
0,
), "OmniInteract requests did not all succeed"
assert summary.get("artifacts_complete") is True, "OmniInteract artifacts are incomplete"
if params.get("omniinteract_evaluate"):
accuracy = summary.get("accuracy")
assert isinstance(accuracy, dict), "OmniInteract accuracy is missing"
assert accuracy.get("status") == "ok", "OmniInteract accuracy did not complete"
assert accuracy.get("failed") == 0, "OmniInteract accuracy reported failed cases"
_assert_subset_evaluated(accuracy)
acc_summary = accuracy.get("summary")
assert isinstance(acc_summary, dict), "OmniInteract accuracy summary is missing"
ia_qtf1 = acc_summary.get("IA_QTF1")
assert _is_finite_number(ia_qtf1), "OmniInteract All Global IA-QTF1 is missing"
min_ia_qtf1 = params.get("omniinteract_min_ia_qtf1")
if min_ia_qtf1 is not None:
assert _is_finite_number(min_ia_qtf1), "omniinteract_min_ia_qtf1 must be a finite number"
assert float(ia_qtf1) >= float(min_ia_qtf1), (
f"OmniInteract All Global IA-QTF1 {ia_qtf1} is below {min_ia_qtf1}"
)
_maybe_assert_aggregate_ia_qtf1(params, acc_summary)
baseline = params.get("baseline")
hardware = result.get("Hardware")
hardware_baseline = baseline.get(hardware) if isinstance(baseline, dict) and isinstance(hardware, str) else None
@@ -304,8 +507,13 @@ def assert_result(result, params, num_prompt) -> None:
)
def test_performance_benchmark(omni_server, benchmark_params):
test_name = benchmark_params["test_name"]
params = benchmark_params["params"]
params = dict(benchmark_params["params"])
dataset_name = params.get("dataset_name", "")
if params.get("omniinteract_evaluate"):
judge_base_url = getattr(omni_server, "judge_base_url", None)
if not isinstance(judge_base_url, str) or not judge_base_url:
raise ValueError("omniinteract_evaluate requires top-level judge_server_params to start a judge")
params["omniinteract_judge_base_url"] = judge_base_url
host = omni_server.host
port = omni_server.port
@@ -349,6 +557,11 @@ def test_performance_benchmark(omni_server, benchmark_params):
"eval_phase",
"trust_remote_code",
"expected_duplex_audio_turns_per_session",
"omniinteract_min_ia_qtf1",
_OMNIINTERACT_AGGREGATE_MIN_IA_QTF1,
_OMNIINTERACT_AGGREGATE_SUBSETS,
_OMNIINTERACT_AGGREGATE_GROUP,
"judge_server_params",
"name",
"enable_negative_prompt",
"random_request_config",
@@ -403,7 +616,7 @@ def test_performance_benchmark(omni_server, benchmark_params):
num_warmups=_resolve_num_warmups(params, default=2),
benchmark_params_name=params.get("name") if isinstance(params.get("name"), str) else None,
)
assert_result(result, params, num_prompt)
assert_result(result, {**params, **_aggregate_accuracy_params_for_test(test_name)}, num_prompt)
# concurrency test
for sweep_index, (concurrency, num_prompt) in enumerate(zip(max_concurrency_list, num_prompt_list)):
@@ -422,4 +635,4 @@ def test_performance_benchmark(omni_server, benchmark_params):
num_warmups=_resolve_num_warmups(params, default=max(2, int(concurrency))),
benchmark_params_name=params.get("name") if isinstance(params.get("name"), str) else None,
)
assert_result(result, params, num_prompt)
assert_result(result, {**params, **_aggregate_accuracy_params_for_test(test_name)}, num_prompt)
@@ -7,7 +7,7 @@
"res": {
"cuda": "H100"
},
"num_cards": 1
"num_cards": 2
}
},
"local_model",
@@ -21,6 +21,16 @@
"--trust-remote-code"
]
},
"judge_server_params": {
"model": "Qwen/Qwen2.5-7B-Instruct",
"use_omni": false,
"cuda_visible_devices": "1",
"extra_cli_args": [
"--served-model-name",
"Qwen2.5-7B-Instruct"
]
},
"omniinteract_aggregate_min_ia_qtf1": 0.2,
"benchmark_params": [
{
"dataset_name": "omniinteract",
@@ -40,6 +50,8 @@
"omniinteract_ref_audio": "tests/assets/minicpmo_4_5/response_required_16k.wav",
"omniinteract_output_dir": "tests/dfx/perf/results/omniinteract/1q1a",
"omniinteract_max_video_duration_s": 600,
"omniinteract_evaluate": true,
"omniinteract_judge_model": "Qwen2.5-7B-Instruct",
"percentile-metrics": "ttft,tpot,itl,e2el,audio_ttfp,audio_rtf,audio_duration"
},
{
@@ -60,6 +72,8 @@
"omniinteract_ref_audio": "tests/assets/minicpmo_4_5/response_required_16k.wav",
"omniinteract_output_dir": "tests/dfx/perf/results/omniinteract/1q1a_math",
"omniinteract_max_video_duration_s": 600,
"omniinteract_evaluate": true,
"omniinteract_judge_model": "Qwen2.5-7B-Instruct",
"percentile-metrics": "ttft,tpot,itl,e2el,audio_ttfp,audio_rtf,audio_duration"
},
{
@@ -80,6 +94,8 @@
"omniinteract_ref_audio": "tests/assets/minicpmo_4_5/response_required_16k.wav",
"omniinteract_output_dir": "tests/dfx/perf/results/omniinteract/1qna",
"omniinteract_max_video_duration_s": 600,
"omniinteract_evaluate": true,
"omniinteract_judge_model": "Qwen2.5-7B-Instruct",
"percentile-metrics": "ttft,tpot,itl,e2el,audio_ttfp,audio_rtf,audio_duration"
}
]
@@ -694,6 +694,204 @@ def test_omniinteract_result_rejects_incomplete_artifacts():
)
def test_omniinteract_result_accepts_accuracy_at_floor():
from tests.dfx.perf.scripts.run_benchmark import assert_result
assert_result(
{
"completed": 4,
"omniinteract": {
"total": 4,
"success": 4,
"failed": 0,
"artifacts_complete": True,
"accuracy": {
"status": "ok",
"failed": 0,
"evaluated": 4,
"summary": {"IA_QTF1": 0.25},
},
},
},
{"dataset_name": "omniinteract", "omniinteract_evaluate": True, "omniinteract_min_ia_qtf1": 0.2},
4,
)
def test_omniinteract_result_rejects_missing_accuracy():
from tests.dfx.perf.scripts.run_benchmark import assert_result
with pytest.raises(AssertionError, match="accuracy is missing"):
assert_result(
{
"completed": 4,
"omniinteract": {"total": 4, "success": 4, "failed": 0, "artifacts_complete": True},
},
{"dataset_name": "omniinteract", "omniinteract_evaluate": True},
4,
)
def test_omniinteract_result_rejects_failed_accuracy():
from tests.dfx.perf.scripts.run_benchmark import assert_result
with pytest.raises(AssertionError, match="accuracy did not complete"):
assert_result(
{
"completed": 4,
"omniinteract": {
"total": 4,
"success": 4,
"failed": 0,
"artifacts_complete": True,
"accuracy": {"status": "failed", "failed": 1, "summary": {"IA_QTF1": 0.9}},
},
},
{"dataset_name": "omniinteract", "omniinteract_evaluate": True, "omniinteract_min_ia_qtf1": 0.0},
4,
)
def test_omniinteract_result_rejects_ia_qtf1_below_floor():
from tests.dfx.perf.scripts.run_benchmark import assert_result
with pytest.raises(AssertionError, match="IA-QTF1"):
assert_result(
{
"completed": 4,
"omniinteract": {
"total": 4,
"success": 4,
"failed": 0,
"artifacts_complete": True,
"accuracy": {
"status": "ok",
"failed": 0,
"evaluated": 4,
"summary": {"IA_QTF1": 0.1},
},
},
},
{"dataset_name": "omniinteract", "omniinteract_evaluate": True, "omniinteract_min_ia_qtf1": 0.2},
4,
)
def _omniinteract_accuracy_result(
*,
tp: float,
fp: float,
fn: float,
ia_qtf1: float = 0.5,
evaluated: int = 4,
skipped: int = 0,
) -> dict[str, object]:
return {
"completed": 4,
"omniinteract": {
"total": 4,
"success": 4,
"failed": 0,
"artifacts_complete": True,
"accuracy": {
"status": "ok",
"failed": 0,
"evaluated": evaluated,
"skipped": skipped,
"total": 4,
"summary": {"IA_QTF1": ia_qtf1, "Global_TP": tp, "Global_FP": fp, "Global_FN": fn},
},
},
}
def _omniinteract_aggregate_params(subset: str, *, group: str) -> dict[str, object]:
return {
"dataset_name": "omniinteract",
"omniinteract_evaluate": True,
"omniinteract_subsets": subset,
"omniinteract_aggregate_min_ia_qtf1": 0.2,
"omniinteract_aggregate_subsets": ["1q1a", "1q1a_math", "1qna"],
"omniinteract_aggregate_group": group,
}
def test_omniinteract_aggregate_waits_until_all_subsets():
from tests.dfx.perf.scripts.run_benchmark import _reset_omniinteract_aggregate_counts, assert_result
_reset_omniinteract_aggregate_counts()
assert_result(
_omniinteract_accuracy_result(tp=6.157502, fp=9, fn=8),
_omniinteract_aggregate_params("1q1a", group="wait"),
4,
)
assert_result(
_omniinteract_accuracy_result(tp=1.133541, fp=2, fn=2),
_omniinteract_aggregate_params("1q1a_math", group="wait"),
4,
)
def test_omniinteract_aggregate_accepts_pooled_ia_qtf1_at_floor():
from tests.dfx.perf.scripts.run_benchmark import _reset_omniinteract_aggregate_counts, assert_result
_reset_omniinteract_aggregate_counts()
assert_result(
_omniinteract_accuracy_result(tp=6.157502, fp=9, fn=8),
_omniinteract_aggregate_params("1q1a", group="pass"),
4,
)
assert_result(
_omniinteract_accuracy_result(tp=1.133541, fp=2, fn=2),
_omniinteract_aggregate_params("1q1a_math", group="pass"),
4,
)
assert_result(
_omniinteract_accuracy_result(tp=0.0, fp=11, fn=18, ia_qtf1=0.0),
_omniinteract_aggregate_params("1qna", group="pass"),
4,
)
def test_omniinteract_aggregate_rejects_pooled_ia_qtf1_below_floor():
from tests.dfx.perf.scripts.run_benchmark import _reset_omniinteract_aggregate_counts, assert_result
_reset_omniinteract_aggregate_counts()
assert_result(
_omniinteract_accuracy_result(tp=0.0, fp=1, fn=1, ia_qtf1=0.0),
_omniinteract_aggregate_params("1q1a", group="fail"),
4,
)
assert_result(
_omniinteract_accuracy_result(tp=0.0, fp=1, fn=1, ia_qtf1=0.0),
_omniinteract_aggregate_params("1q1a_math", group="fail"),
4,
)
with pytest.raises(AssertionError, match="aggregate All Global IA-QTF1"):
assert_result(
_omniinteract_accuracy_result(tp=0.0, fp=1, fn=1, ia_qtf1=0.0),
_omniinteract_aggregate_params("1qna", group="fail"),
4,
)
def test_omniinteract_aggregate_rejects_entirely_skipped_subset():
from tests.dfx.perf.scripts.run_benchmark import _reset_omniinteract_aggregate_counts, assert_result
_reset_omniinteract_aggregate_counts()
assert_result(
_omniinteract_accuracy_result(tp=4.0, fp=0.0, fn=0.0, ia_qtf1=1.0),
_omniinteract_aggregate_params("1q1a", group="skipped-subset"),
4,
)
with pytest.raises(AssertionError, match="evaluated 0 cases"):
assert_result(
_omniinteract_accuracy_result(tp=0.0, fp=0.0, fn=0.0, ia_qtf1=1.0, evaluated=0, skipped=4),
_omniinteract_aggregate_params("1q1a_math", group="skipped-subset"),
4,
)
def test_omni_tpot_baseline_accepts_measured_finite_sample():
from tests.dfx.perf.scripts.run_benchmark import assert_result
@@ -22,6 +22,7 @@ from vllm.tokenizers import TokenizerLike
from vllm.transformers_utils.repo_utils import hf_fs
OMNIINTERACT_SUBSETS = ("1q1a", "1q1a_math", "1qna")
OMNIINTERACT_SCENARIO_TAGS = ("realtime", "proactive", "nested", "interrupted", "1qna")
DEFAULT_OMNIINTERACT_REPO = "lucky-lance/OmniInteract"
DEFAULT_MAX_VIDEO_DURATION_S = 3600.0
logger = logging.getLogger(__name__)
@@ -36,6 +37,18 @@ class OmniInteractCase:
scene_type: str
@dataclass(frozen=True)
class OmniInteractEvaluationOptions:
judge_base_url: str
judge_model: str
judge_api_key: str
judge_timeout_s: float
judge_max_tokens: int
workers: int
output_dir: Path
skip_existing: bool = False
@dataclass(frozen=True)
class OmniInteractSessionOptions:
output_root: Path
@@ -44,6 +57,7 @@ class OmniInteractSessionOptions:
ref_audio: str
require_response: bool = False
max_video_duration_s: float = DEFAULT_MAX_VIDEO_DURATION_S
evaluation: OmniInteractEvaluationOptions | None = None
@dataclass(frozen=True)
@@ -70,14 +84,31 @@ class OmniInteractDataset(BenchmarkDataset):
subsets: Sequence[str] = OMNIINTERACT_SUBSETS,
random_seed: int = 0,
disable_shuffle: bool = False,
scenario_tags: Sequence[str] = (),
scenario_focus: bool = False,
video_list: str | Path | None = None,
) -> None:
super().__init__(
dataset_path=data_root or dataset_repo,
random_seed=random_seed,
disable_shuffle=disable_shuffle,
)
self.root = resolve_omniinteract_root(data_root, dataset_repo)
self.video_list = Path(video_list).expanduser().resolve() if video_list else None
self.subsets = tuple(subsets)
self.scenario_tags = normalize_scenario_tags(scenario_tags)
self.scenario_focus = bool(scenario_focus)
if self.video_list is not None:
if self.scenario_tags:
raise ValueError("OmniInteract video_list cannot be combined with scenario_tags")
if self.scenario_focus:
raise ValueError("OmniInteract video_list cannot be combined with scenario_focus")
if not self.video_list.is_file():
raise FileNotFoundError(f"--omniinteract-video-list does not exist: {self.video_list}")
self.root = None
else:
self.root = resolve_omniinteract_root(data_root, dataset_repo)
if self.scenario_focus and not self.scenario_tags:
raise ValueError("scenario_focus requires at least one scenario tag")
def sample(
self,
@@ -89,12 +120,18 @@ class OmniInteractDataset(BenchmarkDataset):
**_: Any,
) -> list[SampleRequest]:
del tokenizer
if self.video_list is not None:
cases = load_omniinteract_cases_from_video_list(self.video_list, num_prompts=num_requests)
else:
assert self.root is not None
cases = discover_omniinteract_cases(
self.root,
self.subsets,
num_prompts=num_requests,
seed=self.random_seed,
disable_shuffle=self.disable_shuffle,
scenario_tags=self.scenario_tags,
scenario_focus=self.scenario_focus,
)
return [
OmniInteractSampleRequest(
@@ -259,6 +296,182 @@ def _one_to_many_cases(root: Path) -> list[OmniInteractCase]:
return cases
def normalize_scenario_tags(tags: Sequence[str] | None) -> tuple[str, ...]:
"""Validate and dedupe scenario tags while preserving caller order."""
if not tags:
return ()
normalized: list[str] = []
seen: set[str] = set()
invalid: list[str] = []
for raw in tags:
tag = str(raw or "").strip().lower()
if not tag:
continue
if tag not in OMNIINTERACT_SCENARIO_TAGS:
invalid.append(tag)
continue
if tag not in seen:
seen.add(tag)
normalized.append(tag)
if invalid:
raise ValueError(
"Unsupported OmniInteract scenario tags: "
f"{sorted(set(invalid))}; choose from {list(OMNIINTERACT_SCENARIO_TAGS)}"
)
return tuple(normalized)
def _truthy(value: object) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, int | float):
return bool(value)
return str(value or "").strip().lower() in {"1", "true", "yes", "y", "t"}
def _annotation_rows(root: object) -> list[dict[str, object]]:
if isinstance(root, list):
return [row for row in root if isinstance(row, dict)]
if not isinstance(root, dict):
return []
rows: list[dict[str, object]] = []
items = root.get("items")
if isinstance(items, list):
rows.extend(row for row in items if isinstance(row, dict))
answers = root.get("answers")
if isinstance(answers, list):
rows.extend(row for row in answers if isinstance(row, dict))
conversations = root.get("conversations")
if isinstance(conversations, list):
rows.extend(row for row in conversations if isinstance(row, dict))
return rows
def case_scenario_tags(case: OmniInteractCase) -> frozenset[str]:
"""Infer paper-table scenario tags from a case's scene type and annotation."""
tags: set[str] = set()
scene = str(case.scene_type or "").strip().lower()
if scene == "1qna":
tags.add("1qna")
elif scene == "nested":
tags.add("nested")
try:
payload = json.loads(case.annotation_path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
raise ValueError(f"Invalid OmniInteract annotation: {case.annotation_path}") from exc
for row in _annotation_rows(payload):
question_type = str(row.get("question_type") or row.get("label") or "").strip().lower()
if scene == "multi_turn":
if question_type == "realtime":
tags.add("realtime")
elif question_type == "proactive":
tags.add("proactive")
if _truthy(row.get("is_interrupted")) or _truthy(row.get("interrupted")):
tags.add("interrupted")
return frozenset(tags)
def _select_with_scenario_tags(
cases: list[OmniInteractCase],
*,
num_prompts: int,
seed: int,
disable_shuffle: bool,
scenario_tags: Sequence[str],
scenario_focus: bool,
) -> list[OmniInteractCase]:
tags = normalize_scenario_tags(scenario_tags)
if not tags:
ordered = list(cases)
if not disable_shuffle:
random.Random(seed).shuffle(ordered)
if num_prompts:
if num_prompts > len(ordered):
logger.warning(
"Requested %d OmniInteract prompts but only %d are available; using all cases",
num_prompts,
len(ordered),
)
return ordered[:num_prompts]
return ordered
tag_sets = {case.video_path: case_scenario_tags(case) for case in cases}
if scenario_focus:
pool = [case for case in cases if tag_sets[case.video_path] & set(tags)]
missing = [tag for tag in tags if not any(tag in tag_sets[case.video_path] for case in pool)]
if missing:
raise ValueError(
f"No OmniInteract cases match scenario tags {missing} under the selected subsets (focus mode)"
)
if not pool:
raise ValueError(f"No OmniInteract cases match scenario tags {list(tags)}")
ordered = list(pool)
if not disable_shuffle:
random.Random(seed).shuffle(ordered)
if num_prompts:
if num_prompts > len(ordered):
logger.warning(
"Requested %d OmniInteract prompts but only %d match tags %s; using all matches",
num_prompts,
len(ordered),
list(tags),
)
return ordered[:num_prompts]
return ordered
rng = random.Random(seed)
buckets: dict[str, list[OmniInteractCase]] = {tag: [] for tag in tags}
for case in cases:
owned = tag_sets[case.video_path]
for tag in tags:
if tag in owned:
buckets[tag].append(case)
for tag, bucket in buckets.items():
if not bucket:
logger.warning(
"No OmniInteract cases available for scenario tag %r under the selected subsets",
tag,
)
elif not disable_shuffle:
rng.shuffle(bucket)
selected: list[OmniInteractCase] = []
selected_paths: set[Path] = set()
covered_tags: set[str] = set()
requested = set(tags)
for tag in tags:
if tag in covered_tags:
continue
for case in buckets[tag]:
if case.video_path in selected_paths:
continue
selected.append(case)
selected_paths.add(case.video_path)
covered_tags.update(tag_sets[case.video_path] & requested)
break
remaining = [case for case in cases if case.video_path not in selected_paths]
if not disable_shuffle:
rng.shuffle(remaining)
if num_prompts == 0:
return selected + remaining
if len(selected) > num_prompts:
logger.warning(
"OmniInteract cover tags need %d cases but --num-prompts=%d; truncating coverage set",
len(selected),
num_prompts,
)
return selected[:num_prompts]
return selected + remaining[: num_prompts - len(selected)]
def discover_omniinteract_cases(
root: Path,
subsets: Sequence[str],
@@ -266,6 +479,8 @@ def discover_omniinteract_cases(
num_prompts: int,
seed: int = 0,
disable_shuffle: bool = False,
scenario_tags: Sequence[str] = (),
scenario_focus: bool = False,
) -> list[OmniInteractCase]:
invalid = set(subsets) - set(OMNIINTERACT_SUBSETS)
if invalid:
@@ -276,6 +491,9 @@ def discover_omniinteract_cases(
raise ValueError("OmniInteract subsets must not contain duplicates")
if num_prompts < 0:
raise ValueError("num_prompts must be non-negative")
tags = normalize_scenario_tags(scenario_tags)
if scenario_focus and not tags:
raise ValueError("scenario_focus requires at least one scenario tag")
data_root, cases = _data_dir(root.resolve()), []
for subset in subsets:
subset_root = data_root / subset
@@ -285,15 +503,151 @@ def discover_omniinteract_cases(
cases.extend(selected)
if len({case.video_path for case in cases}) != len(cases):
raise ValueError("OmniInteract dataset contains duplicate video paths")
if not disable_shuffle:
random.Random(seed).shuffle(cases)
if num_prompts:
if num_prompts > len(cases):
logger.warning(
"Requested %d OmniInteract prompts but only %d are available; using all cases", num_prompts, len(cases)
return _select_with_scenario_tags(
cases,
num_prompts=num_prompts,
seed=seed,
disable_shuffle=disable_shuffle,
scenario_tags=tags,
scenario_focus=scenario_focus,
)
cases = cases[:num_prompts]
def _subset_root_for_video(video_path: Path, subset: str) -> Path:
"""Find the subset data root that owns an absolute video path."""
if subset not in OMNIINTERACT_SUBSETS:
raise ValueError(f"Unsupported OmniInteract subset: {subset!r}")
parts = video_path.resolve().parts
for index in range(len(parts) - 1, -1, -1):
if parts[index] != subset:
continue
candidate = Path(*parts[: index + 1])
if subset == "1qna":
if (candidate / "videos_bench").is_dir() and (candidate / "annotations").is_dir():
return candidate
elif (candidate / "video_json_map.json").is_file():
return candidate
raise ValueError(f"Could not locate OmniInteract subset root for {subset!r} from {video_path}")
def _case_from_video_list_row(row: dict[str, object], *, line_no: int) -> OmniInteractCase:
"""Rebuild an ``OmniInteractCase`` from one ``sampled_cases.jsonl`` row."""
subset = str(row.get("subset") or "")
video_text = str(row.get("video_path") or "")
if not subset or not video_text:
raise ValueError(f"OmniInteract video_list line {line_no} requires subset and video_path")
video_path = Path(video_text).expanduser().resolve()
if not video_path.is_file():
raise FileNotFoundError(f"OmniInteract video_list line {line_no} video missing: {video_path}")
subset_root = _subset_root_for_video(video_path, subset)
if subset == "1qna":
videos_root = (subset_root / "videos_bench").resolve()
annotations_root = (subset_root / "annotations").resolve()
if not video_path.is_relative_to(videos_root):
raise ValueError(f"OmniInteract video_list line {line_no} is outside videos_bench: {video_path}")
relative = video_path.relative_to(videos_root)
annotation = (subset_root / "annotations" / relative).with_suffix(".json").resolve()
if not annotation.is_relative_to(annotations_root):
raise ValueError(f"OmniInteract video_list line {line_no} unsafe annotation path: {relative}")
if not annotation.is_file():
raise FileNotFoundError(f"OmniInteract video_list line {line_no} annotation missing: {annotation}")
case = OmniInteractCase(
subset,
f"videos_bench/{relative.as_posix()}",
video_path,
annotation,
"1qna",
)
else:
by_path = {case.video_path.resolve(): case for case in _mapping_cases(subset_root, subset)}
case = by_path.get(video_path)
if case is None:
raise ValueError(f"OmniInteract video_list line {line_no} is not in {subset} mapping: {video_path}")
expected_name = str(row.get("output_name") or "")
if expected_name and expected_name != official_output_name(case):
raise ValueError(
f"OmniInteract video_list line {line_no} output_name mismatch: "
f"{expected_name!r} != {official_output_name(case)!r}"
)
return case
def load_omniinteract_cases_from_video_list(
video_list: str | Path,
*,
num_prompts: int,
) -> list[OmniInteractCase]:
"""Load cases from a ``sampled_cases.jsonl``-style list, preserving order.
``num_prompts`` caps the prefix of the list. ``0`` or a value larger than the
list length selects every row.
"""
if num_prompts < 0:
raise ValueError("num_prompts must be non-negative")
path = Path(video_list).expanduser().resolve()
if not path.is_file():
raise FileNotFoundError(f"--omniinteract-video-list does not exist: {path}")
cases: list[OmniInteractCase] = []
seen: set[Path] = set()
with path.open(encoding="utf-8") as handle:
for line_no, raw in enumerate(handle, start=1):
text = raw.strip()
if not text:
continue
try:
row = json.loads(text)
except json.JSONDecodeError as exc:
raise ValueError(f"OmniInteract video_list line {line_no} is not valid JSON") from exc
if not isinstance(row, dict):
raise ValueError(f"OmniInteract video_list line {line_no} must be a JSON object")
case = _case_from_video_list_row(row, line_no=line_no)
if case.video_path in seen:
raise ValueError(f"OmniInteract video_list contains duplicate video_path: {case.video_path}")
seen.add(case.video_path)
cases.append(case)
if not cases:
raise ValueError(f"OmniInteract video_list is empty: {path}")
if num_prompts == 0 or num_prompts >= len(cases):
return cases
return cases[:num_prompts]
def official_output_name(case: OmniInteractCase) -> str:
"""Match Lucky-Lance/OmniInteract ``collect_all_videos`` output_name layout.
``1q1a`` / ``1q1a_math`` use the mapping-relative video path with ``/`` →
``__``. ``1qna`` strips the ``videos_bench/`` prefix before the same rewrite.
"""
relative = case.video_rel.replace("\\", "/")
if case.subset == "1qna":
prefix = "videos_bench/"
if relative.startswith(prefix):
relative = relative[len(prefix) :]
stem = Path(relative).with_suffix("").as_posix().replace("/", "__")
return f"{case.subset}/{stem}"
def sampled_case_row(case: OmniInteractCase) -> dict[str, str]:
"""One official MiniCPM-o ``--video_list`` row for a sampled OmniInteract case.
``video_path`` reuses the absolute path already resolved from ``--dataset-path``
(local tree or Hub extract). Official
``batch_inference_minicpmo.py --video_list`` accepts these rows as-is.
"""
return {
"video_path": str(case.video_path.resolve()),
"output_name": official_output_name(case),
"subset": case.subset,
}
def case_manifest(case: OmniInteractCase, output_dir: Path) -> dict[str, Any]:
+9 -1
View File
@@ -32,6 +32,7 @@ from vllm_omni.benchmarks.data_modules.omniinteract_dataset import (
OmniInteractCase,
OmniInteractPreparedInput,
case_manifest,
sampled_case_row,
)
# The duplex client library (vllm_omni.clients) is imported lazily inside the
@@ -48,7 +49,7 @@ PCM16_SAMPLE_RATE = 16_000
PCM16_BYTES_PER_SAMPLE = 2
OUTPUT_SAMPLE_RATE = 24_000
SUCCESS_ARTIFACTS = (".done", "output.wav", "wav_transcript.json", "events.json", "result.json")
BATCH_ARTIFACTS = ("batch_summary.json", "official_eval_manifest.jsonl")
BATCH_ARTIFACTS = ("batch_summary.json", "official_eval_manifest.jsonl", "sampled_cases.jsonl")
ARTIFACT_LOCK_FILE = ".omniinteract.lock"
_INPUT_CHUNK_MS = 200
# Minimum new played audio between two incremental playback.ack sends for one
@@ -816,6 +817,7 @@ def write_batch_artifacts(
for case, result in zip(cases, results, strict=True)
if result.success and result.eligible_for_official_eval
]
sampled_rows = [sampled_case_row(case) for case in cases]
ineligible = [result for result in results if result.success and not result.eligible_for_official_eval]
if ineligible:
reasons = Counter(
@@ -831,6 +833,12 @@ def write_batch_artifacts(
root / "official_eval_manifest.jsonl",
"".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows),
)
# Every sampled case (including failed / ineligible) so official MiniCPM-o
# ``--video_list`` can replay the same video set without re-deriving ids.
_atomic_write_text(
root / "sampled_cases.jsonl",
"".join(json.dumps(row, ensure_ascii=False) + "\n" for row in sampled_rows),
)
def _websocket_url(config: OmniInteractBenchmarkConfig) -> str:
File diff suppressed because it is too large. Load diff
+495
View File
@@ -0,0 +1,495 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
"""Text-only LLM judge for OmniInteract unified evaluation.
The three prompt templates follow the official English judge protocol in
Lucky-Lance/OmniInteract ``eval/evaluation/llm_judge.py``
(commit de304cef35fd9a50a5caadb5090c34cfbf0dd868). They correspond to the
OmniInteract paper appendix (arXiv:2605.26485) Listing A.1 (early-stage),
Listing A.2 (interrupted partial quality), and Listing A.3 (core-stage).
A local OpenAI-compatible judge is not the paper's GPT-4o judge, so scores
are protocol-compatible rather than official paper-table numbers.
"""
from __future__ import annotations
import json
import math
import threading
import time
from collections.abc import Mapping
from dataclasses import dataclass
import regex as re
import requests
_JUDGE_MAX_ATTEMPTS = 3
_RETRYABLE_HTTP_STATUS = frozenset({502, 503, 504})
_SCORE_FLOAT_PATTERN = re.compile(
r"""score["']?\s*[:=]\s*([-+]?\d*\.?\d+(?:[eE][-+]?\d+)?)""",
re.IGNORECASE,
)
# Official English templates from OmniInteract llm_judge.py.
# Paper appendix: Listing A.1 early, A.2 interrupted-partial, A.3 core.
# Source: Lucky-Lance/OmniInteract@de304cef35fd9a50a5caadb5090c34cfbf0dd868
EARLY_SYSTEM_PROMPT = (
"You are a streaming voice assistant evaluation judge. "
"Judge only based on the given text. "
"Output must be parseable JSON with no other text."
)
EARLY_USER_TEMPLATE = """Determine whether the early output between start and t_a is an early hallucination.
[scene_type] {scene_type}
[slot] slot_id={slot_id},
turn_index={turn_index},
step_index={step_index},
boundary_type={boundary_type},
is_interrupted={is_interrupted}
[question] {question}
[current_gt_answer] {gt_answer}
[full_chunk_context] {full_context}
[early_actual_text] {actual_text}
Rules:
1. Greetings, confirmations, waiting, brief observations, and follow-up phrases -> Neutral.
2. If the model starts substantively answering, guessing unseen info, revealing
future steps, or making definitive factual claims -> FP.
3. For 1QnA first step, reciting the full procedure before acting -> FP.
4. score is interaction quality 0-1 when Neutral; 0 when hallucination.
Output JSON:
{{"flag":"Neutral|FP_Hallucination",
"score":float 0-1,
"rationale":"one sentence"}}"""
CORE_SYSTEM_PROMPT = (
"You are a strict streaming voice assistant core-answer evaluation judge. "
"Judge only based on the given text and reference answer. "
"Output must be parseable JSON with no other text."
)
CORE_USER_TEMPLATE = """Score the core output after t_a.
[scene_type] {scene_type}
[slot] slot_id={slot_id},
turn_index={turn_index},
step_index={step_index},
boundary_type={boundary_type},
is_interrupted={is_interrupted}
[question] {question}
[current_gt_answer] {gt_answer}
[future_gt_answers_or_steps]
{future_answers}
[full_chunk_context] {full_context}
[core_actual_text] {actual_text}
Rules:
1. score 0-1: correctness and coverage of core_actual_text vs gt_answer.
2. Off-topic, factual errors, or missing key answer -> low score.
3. 1QnA: reward only current-step info; penalize spoiling future steps or skipping the current step.
4. If score > 0, extract the earliest contiguous substring from core_actual_text
that establishes the answer as trigger_phrase.
5. trigger_phrase must be a verbatim substring; empty if score == 0.
Output JSON:
{{"score":float 0-1,
"trigger_phrase":"substring or empty",
"spoiler":true|false,
"rationale":"one sentence"}}"""
INTERRUPT_PARTIAL_SYSTEM_PROMPT = (
"You are a strict evaluator for interrupted voice-assistant answers. "
"Judge only from the provided text. "
"Return valid JSON only, with no extra text."
)
INTERRUPT_PARTIAL_USER_TEMPLATE = """Evaluate the quality of the assistant output
that was already spoken before or around an interruption.
[Task]
The assistant was answering, but the interaction was interrupted before
completion. The assistant was not required to complete the full original answer.
Score whether the content already spoken is relevant, correct, and useful for
the current ground-truth answer.
[Question] {question}
[Ground Truth Answer] {gt_answer}
[Assistant Output Already Spoken]
{actual_text}
Scoring Rules:
1. Score from 0 to 1.
2. Do not penalize incompleteness: a partial answer can receive a high score if the spoken part is correct and useful.
3. Score high when the spoken content overlaps with, paraphrases, or conveys useful parts of the ground truth.
4. Score low for acknowledgments or prefaces without substantive answer content.
5. Score low for wrong-question, irrelevant, or generic-filler output.
6. hallucination=true if the output contains clear incorrect facts, wrong target content, or unsupported content.
7. Ignore overflow duration when scoring quality; spill is measured separately.
Output JSON:
{{"score":float 0-1,
"hallucination":true|false,
"rationale":"one sentence"}}"""
class JudgeRequestError(RuntimeError):
"""Raised when the configured judge endpoint cannot produce a response."""
@dataclass(frozen=True)
class EarlyJudgment:
category: str
score: float
rationale: str
raw: str
parse_source: str
@dataclass(frozen=True)
class CoreJudgment:
score: float
trigger_phrase: str
spoiler: bool
rationale: str
raw: str
parse_source: str
@dataclass(frozen=True)
class PartialJudgment:
score: float
hallucination: bool
rationale: str
raw: str
parse_source: str
def _clamp_unit_interval(value: float) -> float:
return max(0.0, min(1.0, value))
def _json_unit_score(value: object) -> tuple[float | None, bool]:
"""Parse one JSON score into a finite unit-interval value.
The second flag is true when the value is a non-finite float (NaN or
infinity). Callers must treat that as a parse failure and award no credit.
A value that is not a float returns ``(None, False)`` so existing
fallbacks stay available.
"""
if isinstance(value, bool) or not isinstance(value, int | float | str):
return None, False
try:
parsed = float(value)
except ValueError:
return None, False
if not math.isfinite(parsed):
return None, True
return _clamp_unit_interval(parsed), False
def _required_json_unit_score(parsed: Mapping[str, object]) -> tuple[float | None, bool]:
"""Require an explicit finite ``score`` field on a parsed JSON object.
Returns ``(score, False)`` for a present, finite unit-interval value
(including an explicit ``0`` / ``0.0``). Returns ``(None, True)`` when
``score`` is missing, has an invalid type, or is non-finite — callers must
treat that as ``llm_parse_failed`` and must not invent a valid zero.
"""
if "score" not in parsed:
return None, True
score, non_finite = _json_unit_score(parsed["score"])
if non_finite or score is None:
return None, True
return score, False
def _text(value: object) -> str:
return str(value or "").strip()
def _bool(value: object) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, int | float):
return bool(value)
return _text(value).lower() in {"1", "true", "yes", "y", "t"}
def _first_json_object(text: str) -> dict[str, object] | None:
decoder = json.JSONDecoder()
position = text.find("{")
while position >= 0:
try:
value, _ = decoder.raw_decode(text[position:])
except json.JSONDecodeError:
position = text.find("{", position + 1)
continue
if isinstance(value, dict) and all(isinstance(key, str) for key in value):
return value
position = text.find("{", position + 1)
return None
def _score_float_from_text(text: str) -> float | None:
"""Parse a unit-interval score from text; ignore unrelated leading numbers."""
match = _SCORE_FLOAT_PATTERN.search(text)
if match is None:
return None
try:
value = float(match.group(1))
except ValueError:
return None
if 0.0 <= value <= 1.0:
return value
return None
def _chat_url(base_url: str) -> str:
base = base_url.rstrip("/")
if base.endswith("/chat/completions"):
return base
if base.endswith("/v1"):
return f"{base}/chat/completions"
return f"{base}/v1/chat/completions"
class OmniInteractJudge:
"""OpenAI-compatible client implementing the OmniInteract judge contract."""
def __init__(
self,
base_url: str,
model: str,
*,
api_key: str = "EMPTY",
timeout_s: float = 60.0,
max_tokens: int = 512,
) -> None:
if not base_url.strip():
raise ValueError("judge base URL is required")
if not model.strip():
raise ValueError("judge model is required")
self.base_url = base_url
self.model = model
self.api_key = api_key or "EMPTY"
self.timeout_s = timeout_s
self.max_tokens = max_tokens
self._thread_local = threading.local()
def _session(self) -> requests.Session:
session = getattr(self._thread_local, "session", None)
if session is None:
session = requests.Session()
self._thread_local.session = session
return session
def _generate(self, system_prompt: str, user_prompt: str) -> str:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
"temperature": 0.0,
"stream": False,
"max_tokens": self.max_tokens,
}
last_error: Exception | None = None
for attempt in range(_JUDGE_MAX_ATTEMPTS):
try:
response = self._session().post(
_chat_url(self.base_url),
headers=headers,
json=payload,
timeout=self.timeout_s,
)
except requests.Timeout as exc:
last_error = JudgeRequestError(f"judge request failed: {exc}")
if attempt + 1 < _JUDGE_MAX_ATTEMPTS:
time.sleep(0.5 * (2**attempt))
continue
raise last_error from exc
except requests.RequestException as exc:
raise JudgeRequestError(f"judge request failed: {exc}") from exc
if response.status_code in _RETRYABLE_HTTP_STATUS and attempt + 1 < _JUDGE_MAX_ATTEMPTS:
time.sleep(0.5 * (2**attempt))
continue
if not response.ok:
raise JudgeRequestError(f"judge returned HTTP {response.status_code}: {response.text[:500]}")
try:
body = response.json()
choices = body["choices"]
message = choices[0]["message"]
return str(message["content"])
except (KeyError, IndexError, TypeError, ValueError) as exc:
raise JudgeRequestError(f"unexpected judge response: {response.text[:500]}") from exc
assert last_error is not None
raise last_error
def judge_early(
self,
slot: Mapping[str, object],
full_context: str,
actual_text: str,
) -> EarlyJudgment:
raw = self._generate(
EARLY_SYSTEM_PROMPT,
EARLY_USER_TEMPLATE.format(
scene_type=_text(slot.get("scene_type")),
slot_id=slot.get("slot_id"),
turn_index=slot.get("turn_index"),
step_index=slot.get("step_index"),
boundary_type=_text(slot.get("boundary_type")),
is_interrupted=str(_bool(slot.get("is_interrupted"))).lower(),
question=_text(slot.get("question_text")),
gt_answer=_text(slot.get("gt_answer")),
full_context=full_context.strip(),
actual_text=actual_text.strip(),
),
)
parsed = _first_json_object(raw)
flag = _text(parsed.get("flag")) if parsed else raw.strip()
normalized = flag.lower().replace("-", "_").replace(" ", "")
if "hallucination" in normalized:
return EarlyJudgment(
category="hallucination",
score=0.0,
rationale=_text(parsed.get("rationale")) if parsed else "",
raw=raw,
parse_source="llm_json" if parsed else "llm_parse_failed",
)
if "neutral" in normalized:
if parsed is None:
return EarlyJudgment(
category="neutral",
score=0.0,
rationale="",
raw=raw,
parse_source="llm_parse_failed",
)
score, score_failed = _required_json_unit_score(parsed)
if score_failed:
return EarlyJudgment(
category="unparsed",
score=0.0,
rationale=_text(parsed.get("rationale")),
raw=raw,
parse_source="llm_parse_failed",
)
return EarlyJudgment(
category="neutral",
score=score if score is not None else 0.0,
rationale=_text(parsed.get("rationale")),
raw=raw,
parse_source="llm_json",
)
return EarlyJudgment(
category="unparsed",
score=0.0,
rationale=_text(parsed.get("rationale")) if parsed else "",
raw=raw,
parse_source="llm_parse_failed",
)
def judge_core(
self,
slot: Mapping[str, object],
full_context: str,
actual_text: str,
future_answers: str,
) -> CoreJudgment:
raw = self._generate(
CORE_SYSTEM_PROMPT,
CORE_USER_TEMPLATE.format(
scene_type=_text(slot.get("scene_type")),
slot_id=slot.get("slot_id"),
turn_index=slot.get("turn_index"),
step_index=slot.get("step_index"),
boundary_type=_text(slot.get("boundary_type")),
is_interrupted=str(_bool(slot.get("is_interrupted"))).lower(),
question=_text(slot.get("question_text")),
gt_answer=_text(slot.get("gt_answer")),
future_answers=future_answers.strip(),
full_context=full_context.strip(),
actual_text=actual_text.strip(),
),
)
parsed = _first_json_object(raw)
if parsed is not None:
score, json_score_failed = _required_json_unit_score(parsed)
if json_score_failed:
return CoreJudgment(
score=0.0,
trigger_phrase="",
spoiler=False,
rationale=_text(parsed.get("rationale")),
raw=raw,
parse_source="llm_parse_failed",
)
return CoreJudgment(
score=score if score is not None else 0.0,
trigger_phrase=_text(parsed.get("trigger_phrase")),
spoiler=_bool(parsed.get("spoiler")),
rationale=_text(parsed.get("rationale")),
raw=raw,
parse_source="llm_json",
)
fallback = _score_float_from_text(raw)
return CoreJudgment(
score=fallback if fallback is not None else 0.0,
trigger_phrase="",
spoiler=False,
rationale="",
raw=raw,
parse_source="llm_float_text" if fallback is not None else "llm_parse_failed",
)
def judge_interrupted_partial(
self,
slot: Mapping[str, object],
actual_text: str,
) -> PartialJudgment:
raw = self._generate(
INTERRUPT_PARTIAL_SYSTEM_PROMPT,
INTERRUPT_PARTIAL_USER_TEMPLATE.format(
question=_text(slot.get("question_text")),
gt_answer=_text(slot.get("gt_answer")),
actual_text=actual_text.strip(),
),
)
parsed = _first_json_object(raw)
if parsed is None:
return PartialJudgment(
score=0.0,
hallucination=False,
rationale="",
raw=raw,
parse_source="llm_parse_failed",
)
score, score_failed = _required_json_unit_score(parsed)
if score_failed:
return PartialJudgment(
score=0.0,
hallucination=False,
rationale=_text(parsed.get("rationale")),
raw=raw,
parse_source="llm_parse_failed",
)
return PartialJudgment(
score=score if score is not None else 0.0,
hallucination=_bool(parsed.get("hallucination")),
rationale=_text(parsed.get("rationale")),
raw=raw,
parse_source="llm_json",
)
+57 -1
View File
@@ -54,6 +54,7 @@ from vllm_omni.benchmarks.data_modules.daily_omni_dataset import (
from vllm_omni.benchmarks.data_modules.omniinteract_dataset import (
DEFAULT_OMNIINTERACT_REPO,
OmniInteractDataset,
OmniInteractEvaluationOptions,
OmniInteractPreparedInput,
OmniInteractSampleRequest,
OmniInteractSessionOptions,
@@ -92,6 +93,7 @@ from vllm_omni.benchmarks.omniinteract import (
from vllm_omni.benchmarks.omniinteract import (
write_batch_artifacts as write_omniinteract_batch_artifacts,
)
from vllm_omni.benchmarks.omniinteract_eval import evaluate_batch as evaluate_omniinteract_batch
from vllm_omni.metrics import definitions as defs
from vllm_omni.metrics.utils import coerce_bool, coerce_positive_float_scalar, coerce_positive_int_scalar
@@ -429,6 +431,39 @@ def _finalize_omniinteract_batch(
return compact_summary
async def _evaluate_omniinteract_batch(
input_requests: list[SampleRequest],
outputs: list[RequestFuncOutput],
) -> dict[str, object] | None:
rows = [
(sample, output)
for sample, output in zip(input_requests, outputs, strict=True)
if isinstance(sample, OmniInteractSampleRequest)
]
if not rows:
return None
options = rows[0][0].omniinteract_options
if not isinstance(options, OmniInteractSessionOptions) or options.evaluation is None:
return None
cases, results = [], []
for sample, output in rows:
result = getattr(output, "omniinteract_case_result", None)
if sample.omniinteract_case is None or not isinstance(result, OmniInteractCaseResult):
raise RuntimeError("OmniInteract benchmark output lost its dataset identity")
cases.append(sample.omniinteract_case)
results.append(result)
try:
return await asyncio.to_thread(
evaluate_omniinteract_batch,
cases,
results,
options.evaluation,
)
except Exception as exc: # noqa: BLE001 - post-hoc accuracy must not fail a finished benchmark
logger.exception("OmniInteract evaluation failed")
return {"status": "failed", "error": str(exc)}
def _prepare_omniinteract_batch(input_requests: list[SampleRequest]) -> None:
roots: set[Path] = set()
for sample in input_requests:
@@ -556,14 +591,32 @@ def get_samples(args, tokenizer, **kwargs):
subsets=tuple(getattr(args, "omniinteract_subsets")),
random_seed=args.seed,
disable_shuffle=getattr(args, "disable_shuffle", False),
scenario_tags=tuple(getattr(args, "omniinteract_scenario_tags", None) or ()),
scenario_focus=bool(getattr(args, "omniinteract_scenario_focus", False)),
video_list=getattr(args, "omniinteract_video_list", None),
)
output_root = Path(getattr(args, "omniinteract_output_dir"))
evaluation = None
if bool(getattr(args, "omniinteract_evaluate", False)):
evaluation_output = getattr(args, "omniinteract_eval_output_dir", None)
evaluation = OmniInteractEvaluationOptions(
judge_base_url=str(getattr(args, "omniinteract_judge_base_url", "http://127.0.0.1:8000")),
judge_model=str(getattr(args, "omniinteract_judge_model", "")),
judge_api_key=str(getattr(args, "omniinteract_judge_api_key", "EMPTY")),
judge_timeout_s=float(getattr(args, "omniinteract_judge_timeout_s", 60.0)),
judge_max_tokens=int(getattr(args, "omniinteract_judge_max_tokens", 512)),
workers=int(getattr(args, "omniinteract_eval_workers", 8)),
output_dir=Path(evaluation_output) if evaluation_output else output_root / "evaluation",
skip_existing=bool(getattr(args, "omniinteract_eval_skip_existing", False)),
)
options = OmniInteractSessionOptions(
output_root=Path(getattr(args, "omniinteract_output_dir")),
output_root=output_root,
timeout_s=float(getattr(args, "omniinteract_timeout_s")),
media_timeout_s=float(getattr(args, "omniinteract_media_timeout_s")),
ref_audio=str(getattr(args, "omniinteract_ref_audio")),
require_response=bool(getattr(args, "omniinteract_require_response")),
max_video_duration_s=float(getattr(args, "omniinteract_max_video_duration_s")),
evaluation=evaluation,
)
requests = dataset.sample(
tokenizer,
@@ -3478,6 +3531,9 @@ async def benchmark(
benchmark_duration = time.perf_counter() - benchmark_start_time
omniinteract_summary = _finalize_omniinteract_batch(input_requests, outputs)
omniinteract_evaluation = await _evaluate_omniinteract_batch(input_requests, outputs)
if omniinteract_summary is not None and omniinteract_evaluation is not None:
omniinteract_summary["accuracy"] = omniinteract_evaluation
metrics: Any
actual_output_lens: list[int] | int
@@ -32,6 +32,13 @@ def _positive_finite_float(value: str) -> float:
return parsed
def _positive_int(value: str) -> int:
parsed = int(value)
if parsed <= 0:
raise argparse.ArgumentTypeError(f"must be a positive integer, got {value!r}")
return parsed
def _existing_file(value: str) -> str:
path = Path(value).expanduser()
if not path.is_file():
@@ -42,6 +49,7 @@ def _existing_file(value: str) -> str:
def add_omniinteract_cli_args(parser: argparse.ArgumentParser) -> None:
from vllm_omni.benchmarks.data_modules.omniinteract_dataset import (
DEFAULT_MAX_VIDEO_DURATION_S,
OMNIINTERACT_SCENARIO_TAGS,
OMNIINTERACT_SUBSETS,
)
@@ -49,6 +57,35 @@ def add_omniinteract_cli_args(parser: argparse.ArgumentParser) -> None:
group.add_argument(
"--omniinteract-subsets", nargs="+", choices=OMNIINTERACT_SUBSETS, default=list(OMNIINTERACT_SUBSETS)
)
group.add_argument(
"--omniinteract-scenario-tags",
nargs="+",
choices=OMNIINTERACT_SCENARIO_TAGS,
default=None,
help=(
"Scenario tags used when sampling cases: realtime, proactive, nested, interrupted, 1qna. "
"Default behavior covers each requested tag with at least one case (when available), "
"then fills remaining --num-prompts from the rest of the selected subsets."
),
)
group.add_argument(
"--omniinteract-scenario-focus",
action="store_true",
help=(
"Only run cases matching --omniinteract-scenario-tags. "
"Without this flag, tags are used for coverage-first sampling."
),
)
group.add_argument(
"--omniinteract-video-list",
type=_existing_file,
help=(
"JSONL of sampled OmniInteract cases (video_path, output_name, subset), "
"typically sampled_cases.jsonl from a prior run. Preserves list order. "
"Cannot be combined with --seed, --dataset-path, or --omniinteract-scenario-tags. "
"With this flag, --num-prompts caps the prefix of the list (0 or oversized = all)."
),
)
group.add_argument(
"--omniinteract-timeout-s", type=_positive_finite_float, default=900.0, help="Complete session timeout."
)
@@ -76,6 +113,50 @@ def add_omniinteract_cli_args(parser: argparse.ArgumentParser) -> None:
default=Path("omniinteract-output"),
help="Directory for case and evaluator artifacts.",
)
group.add_argument(
"--omniinteract-evaluate",
action="store_true",
help="Run the text-only OmniInteract judge after generation and print an accuracy report.",
)
group.add_argument(
"--omniinteract-judge-base-url",
default="http://127.0.0.1:8000",
help="Base URL of an already-running OpenAI-compatible judge server.",
)
group.add_argument("--omniinteract-judge-model", help="Model name exposed by the judge server.")
group.add_argument(
"--omniinteract-judge-api-key",
default="EMPTY",
help="Bearer token for the judge server.",
)
group.add_argument(
"--omniinteract-judge-timeout-s",
type=_positive_finite_float,
default=60.0,
help="Timeout for each judge request.",
)
group.add_argument(
"--omniinteract-judge-max-tokens",
type=_positive_int,
default=512,
help="Maximum completion tokens for each judge request.",
)
group.add_argument(
"--omniinteract-eval-workers",
type=_positive_int,
default=8,
help="Maximum number of concurrent judge requests.",
)
group.add_argument(
"--omniinteract-eval-output-dir",
type=Path,
help="Accuracy artifact directory; defaults to OUTPUT_DIR/evaluation.",
)
group.add_argument(
"--omniinteract-eval-skip-existing",
action="store_true",
help="Reuse successful per-case evaluation artifacts.",
)
def add_multi_stage_cli_args(parser: argparse.ArgumentParser) -> None:
@@ -421,6 +502,22 @@ def preprocess_serve_args(args: argparse.Namespace) -> None:
raise ValueError("OmniInteract requires --endpoint /v1/realtime")
if not getattr(args, "omniinteract_ref_audio", None):
raise ValueError("OmniInteract requires --omniinteract-ref-audio")
if getattr(args, "omniinteract_evaluate", False) and not getattr(args, "omniinteract_judge_model", None):
raise ValueError("OmniInteract evaluation requires --omniinteract-judge-model")
if getattr(args, "omniinteract_video_list", None):
explicit = getattr(args, "explicit_keys", ())
if "seed" in explicit:
raise ValueError("--omniinteract-video-list cannot be combined with --seed")
if "dataset_path" in explicit:
raise ValueError("--omniinteract-video-list cannot be combined with --dataset-path")
if getattr(args, "omniinteract_scenario_tags", None):
raise ValueError("--omniinteract-video-list cannot be combined with --omniinteract-scenario-tags")
if getattr(args, "omniinteract_scenario_focus", False):
raise ValueError("--omniinteract-video-list cannot be combined with --omniinteract-scenario-focus")
if getattr(args, "omniinteract_scenario_focus", False) and not getattr(
args, "omniinteract_scenario_tags", None
):
raise ValueError("--omniinteract-scenario-focus requires --omniinteract-scenario-tags")
if getattr(args, "ignore_eos", False):
raise ValueError("OmniInteract does not support --ignore-eos")
if getattr(args, "profile", False):