[CI/Build][MiniCPM-o] Allow auto-response before input commit (#5567)

Signed-off-by: Sy03 <1370724210@qq.com>
This commit is contained in:
Sy03 authored and GitHub committed 2026-07-30 14:14:19 +00:00
1 parent f7e87e7cfd
commit 09c6ad4204
7 files changed
+104 -6

No files matched your search

@@ -265,14 +265,26 @@ class DemoState:
indices[event_type] = index
return indices
def event_order_ok(self, *, require_input_commit: bool = True) -> bool:
def event_order_ok(
self,
*,
require_input_commit: bool = True,
require_commit_before_response: bool = True,
) -> bool:
if not self.events or self.events[0].get("type") != "session.created":
return False
first_commit_index = self.first_index("input_audio_buffer.committed")
first_response_index = self.first_index("response.created")
if first_response_index is None:
return False
if require_input_commit and (first_commit_index is None or first_commit_index > first_response_index):
if require_input_commit and first_commit_index is None:
return False
if (
require_input_commit
and require_commit_before_response
and first_commit_index is not None
and first_commit_index > first_response_index
):
return False
indices_by_type = self.first_response_lifecycle_indices()
if not indices_by_type:
@@ -1409,7 +1421,10 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
require_input_commit=not continuous_input,
)
if validation_mode == "model-policy"
else state.event_order_ok(require_input_commit=not continuous_input)
else state.event_order_ok(
require_input_commit=not continuous_input,
require_commit_before_response=False,
)
)
input_transcription_ok = _input_transcription_ok(
state.input_transcription_count,
@@ -114,7 +114,7 @@ def test_duplex_single_session_response_required(omni_server, model_prefix: str,
)
)
)
assert result["ok"] is True
assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2)
assert result["audio_delta_count"] > 0
assert result["done_count"] == 1
assert result["error_count"] == 0
@@ -137,7 +137,7 @@ def test_duplex_two_sessions_resume_and_takeover(omni_server, model_prefix: str,
)
)
)
assert result["ok"] is True
assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2)
assert result["session_count"] == 2
assert result["resume"]["ok"] is True
assert result["takeover"]["ok"] is True
@@ -874,6 +874,39 @@ def test_realtime_duplex_demo_model_policy_accepts_continuous_input_without_comm
assert state.model_policy_event_order_ok(expected_turns=3, require_input_commit=False)
def test_realtime_duplex_demo_distinguishes_late_and_missing_commit():
demo = _load_demo_module()
state = demo.DemoState()
response_id = "resp-1"
item_id = f"item_{response_id}"
for event in (
{"type": "session.created"},
{"type": "response.created", "response_id": response_id},
{"type": "conversation.item.added", "item": {"id": item_id}},
{"type": "response.output_item.added", "response_id": response_id},
{"type": "response.content_part.added", "response_id": response_id},
{"type": "response.speak", "response_id": response_id},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA"},
{"type": "response.audio.done", "response_id": response_id},
{"type": "response.content_part.done", "response_id": response_id},
{"type": "response.output_item.done", "response_id": response_id},
{"type": "response.done", "response_id": response_id},
{"type": "input_audio_buffer.committed"},
):
state.add(event)
assert not state.event_order_ok(require_input_commit=True)
assert state.event_order_ok(
require_input_commit=True,
require_commit_before_response=False,
)
state.events.pop()
assert not state.event_order_ok(
require_input_commit=True,
require_commit_before_response=False,
)
def test_realtime_duplex_demo_accepts_overlap_unit_terminating_continuous_response():
demo = _load_demo_module()
state = demo.DemoState()
@@ -1,3 +1,4 @@
import concurrent.futures
import contextlib
import importlib
import os
@@ -7,6 +8,7 @@ import types
import pytest
from vllm_omni.diffusion.data import AttentionConfig, AttentionSpec
from vllm_omni.engine import async_omni_engine as async_omni_engine_module
from vllm_omni.engine.async_omni_engine import AsyncOmniEngine
from vllm_omni.engine.stage_init_utils import (
LogicalStageInitPlan,
@@ -19,6 +21,30 @@ from vllm_omni.engine.stage_runtime import StageRuntime
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
def test_orchestrator_startup_timeout_warns_how_to_raise_limits(monkeypatch):
engine = object.__new__(AsyncOmniEngine)
engine.orchestrator_thread = types.SimpleNamespace(is_alive=lambda: True)
monkeypatch.setattr(engine, "_try_shutdown", lambda _message: None)
ticks = iter((0.0, 1.0))
monkeypatch.setattr(async_omni_engine_module.time, "monotonic", lambda: next(ticks))
warnings: list[tuple[object, ...]] = []
monkeypatch.setattr(
async_omni_engine_module.logger,
"warning",
lambda *args: warnings.append(args),
)
with pytest.raises(TimeoutError, match="did not become ready within 1s"):
engine._wait_for_orchestrator_init(concurrent.futures.Future(), startup_timeout=1)
assert len(warnings) == 1
message = str(warnings[0][0])
assert "--init-timeout" in message
assert "--stage-init-timeout" in message
def _make_llm_metadata(
stage_id: int,
*,
@@ -25,6 +25,7 @@ from vllm.entrypoints.openai.engine.protocol import ErrorInfo, ErrorResponse
from vllm_omni.entrypoints.omni_base import OmniEngineDeadError
from vllm_omni.entrypoints.openai import api_server as api_server_module
from vllm_omni.entrypoints.openai import serving_speech as serving_speech_module
from vllm_omni.entrypoints.openai.audio_utils_mixin import AudioMixin
from vllm_omni.entrypoints.openai.protocol.audio import (
BatchSpeechRequest,
@@ -1390,6 +1391,20 @@ class TestTTSMethods:
# Verify speakers are normalized to lowercase
assert server.supported_speakers == {"ryan", "vivian", "aiden"}
def test_load_supported_speakers_skips_non_tts_omni_model(
self,
speech_server,
mocker: MockerFixture,
):
warning = mocker.patch.object(serving_speech_module.logger, "warning")
speech_server.engine_client.model_config = SimpleNamespace(
hf_config=SimpleNamespace(tts_config=SimpleNamespace())
)
assert speech_server._is_tts is False
assert speech_server._load_supported_speakers() == set()
warning.assert_not_called()
def test_load_supported_languages_from_config(self, speech_server):
"""Languages/dialects from codec_language_id are loaded title-cased; 'Auto' is added."""
speech_server._tts_model_type = "qwen3_tts"
+7
View File
@@ -465,6 +465,13 @@ class AsyncOmniEngine:
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
logger.warning(
"[AsyncOmniEngine] Orchestrator startup timed out after %ss. "
"Multi-stage deployments that initialize stages sequentially on one device "
"or load checkpoints from slow storage may need larger --init-timeout and "
"--stage-init-timeout values.",
startup_timeout,
)
self._try_shutdown("[AsyncOmniEngine] Failed to cleanup after orchestrator startup timeout")
raise TimeoutError(f"Orchestrator did not become ready within {startup_timeout}s")
try:
@@ -832,7 +832,9 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
config = self.engine_client.model_config.hf_config.audio_config
else:
# Default is qwen3_tts path
config = self.engine_client.model_config.hf_config.talker_config
config = getattr(self.engine_client.model_config.hf_config, "talker_config", None)
if config is None:
return set()
# Check for speakers in either spk_id or speaker_id
for attr_name in ["spk_id", "speaker_id"]: