From 856999559179001cbff4b715999ceb2dec329203 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Wed, 23 Sep 2026 20:52:43 +0800 Subject: [PATCH] Add bounded Omni Qwen3-TTS CPU Radeon whole-session stage --- .../experiments/probe_omni_crisp_tts.py | 180 +++++++ .../probe_omni_crisp_tts_entrypoint.py | 110 ++++ .../results/e2e_expansion_20260923/README.md | 3 +- .../evidence/qwen_tts_omni_hybrid/README.md | 28 + ...p_tts_pinned_profile20_20260923.driver.log | 495 ++++++++++++++++++ ...i_crisp_tts_pinned_profile20_20260923.json | 491 +++++++++++++++++ ...ni_crisp_tts_pinned_profile20_20260923.log | 167 ++++++ ...isp_tts_pinned_profile20_asr_20260923.json | 16 + ...risp_tts_pinned_profile20_asr_20260923.log | 17 + ...risp_tts_pinned_public_20260923.driver.log | 52 ++ ...omni_crisp_tts_pinned_public_20260923.json | 26 + .../omni_crisp_tts_pinned_public_20260923.log | 79 +++ ...s_pinned_reject_device_20260923.driver.log | 100 ++++ ...isp_tts_pinned_reject_device_20260923.json | 26 + ...s_pinned_reject_memory_20260923.driver.log | 100 ++++ ...isp_tts_pinned_reject_memory_20260923.json | 26 + .../qwen_tts_radeon890m_joint/README.md | 2 +- .../evidence/summary/README.md | 6 +- docs/design/omni_edge_backends.md | 9 +- docs/design/omni_edge_e2e_roadmap.md | 2 + vllm_omni/config/pipeline_registry.py | 7 +- vllm_omni/engine/backends/__init__.py | 4 + vllm_omni/engine/backends/crisp_tts.py | 352 +++++++++++++ .../models/qwen3_tts/pipeline.py | 17 + 24 files changed, 2308 insertions(+), 7 deletions(-) create mode 100644 benchmarks/edge_harness/experiments/probe_omni_crisp_tts.py create mode 100644 benchmarks/edge_harness/experiments/probe_omni_crisp_tts_entrypoint.py create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/README.md create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_profile20_20260923.driver.log create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_profile20_20260923.json create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_profile20_20260923.log create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_profile20_asr_20260923.json create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_profile20_asr_20260923.log create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_public_20260923.driver.log create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_public_20260923.json create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_public_20260923.log create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_device_20260923.driver.log create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_device_20260923.json create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_memory_20260923.driver.log create mode 100644 benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_memory_20260923.json create mode 100644 vllm_omni/engine/backends/crisp_tts.py diff --git a/benchmarks/edge_harness/experiments/probe_omni_crisp_tts.py b/benchmarks/edge_harness/experiments/probe_omni_crisp_tts.py new file mode 100644 index 000000000..ee55089f5 --- /dev/null +++ b/benchmarks/edge_harness/experiments/probe_omni_crisp_tts.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +"""Profile pinned Qwen3-TTS CustomVoice through the Omni CrispASR stage.""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import math +import time +import wave +from pathlib import Path +from types import SimpleNamespace + +import numpy as np + + +def _rank(values: list[float], fraction: float) -> float: + return sorted(values)[math.ceil(len(values) * fraction) - 1] + + +def _save_wav(path: Path, audio) -> None: + pcm = np.rint(audio.numpy() * 32768).astype(" None: + parser = argparse.ArgumentParser(description=__doc__) + for name in ( + "server_bin", "server_sha256", "talker_file", "talker_sha256", "codec_file", + "codec_sha256", "server_log", "output_report", "output_wav", + "punc_file", "punc_sha256", + ): + parser.add_argument("--" + name.replace("_", "-"), required=True) + parser.add_argument("--expected-gpu-name", default="AMD Radeon(TM) 890M Graphics") + parser.add_argument("--warmups", type=int, default=1) + parser.add_argument("--repeats", type=int, default=20) + parser.add_argument("--abort-check", action="store_true") + parser.add_argument("--capacity-gib", type=int, default=16) + parser.add_argument("--reserve-gib", type=int, default=8) + args = parser.parse_args() + if args.warmups < 0 or args.repeats <= 0 or args.reserve_gib <= 0 or args.capacity_gib < args.reserve_gib: + parser.error("invalid warmup/repeat count or memory budget") + + from vllm_omni.config.stage_config import DeployConfig, StageDeployConfig, merge_pipeline_deploy + from vllm_omni.engine.stage_runtime import StageRuntime + from vllm_omni.model_executor.models.qwen3_tts.pipeline import QWEN3_TTS_CRISP_HYBRID_PIPELINE + + backend = { + "name": "external.crisp.tts.v1", + "server_bin": args.server_bin, + "server_sha256": args.server_sha256, + "talker_file": args.talker_file, + "talker_sha256": args.talker_sha256, + "codec_file": args.codec_file, + "codec_sha256": args.codec_sha256, + "punc_file": args.punc_file, + "punc_sha256": args.punc_sha256, + "log_file": args.server_log, + "expected_gpu_name": args.expected_gpu_name, + "ggml_vk_visible_devices": "1", + "memory_overhead_bytes": 5 << 30, + "max_text_bytes": 4096, + "max_wav_bytes": 4 << 20, + "max_audio_s": 10, + "request_timeout_s": 180, + "start_timeout_s": 180, + "voice": "ryan", + "seed": 42, + } + deploy = DeployConfig( + async_chunk=False, + stages=[StageDeployConfig( + stage_id=0, + backend=backend, + resource_budget={ + "capacities": {"host_ram": args.capacity_gib << 30}, + "demands": {"host_ram": args.reserve_gib << 30}, + }, + )], + ) + pipeline = QWEN3_TTS_CRISP_HYBRID_PIPELINE + configs = [stage.to_omegaconf() for stage in merge_pipeline_deploy(pipeline, deploy)] + runtime = StageRuntime(configs, "local-qwen-tts-crisp", "", stage_init_timeout=180, async_chunk=False) + report = { + "scope": "complete resident Qwen3-TTS CustomVoice text-to-WAV through Omni StageRuntime/StagePool", + "placement": "Radeon Vulkan talker/codec + CPU FP32 code predictor", + "warmup_count": args.warmups, + "measured_count": args.repeats, + "memory_budget": deploy.stages[0].resource_budget, + } + try: + started = time.perf_counter() + runtime.initialize() + report["startup_s"] = time.perf_counter() - started + pool = runtime.stage_pools[0] + report["execution_plan"] = pool.stage_client.execution_plan + state = SimpleNamespace(sampling_params_list=[None]) + + async def request_one(text: str, *, save: bool = False) -> dict: + request_id = f"tts-{time.monotonic_ns()}" + started = time.perf_counter() + await pool.submit_initial(request_id, state, {"text": text, "voice": "ryan", "seed": 42}) + deadline = time.monotonic() + 180 + while True: + output = pool.poll_graph_output(0) + if output is not None: + break + if time.monotonic() >= deadline: + raise TimeoutError("Omni CrispASR stage request timed out") + await asyncio.sleep(0.002) + wall_s = time.perf_counter() - started + try: + if output.error: + raise RuntimeError(output.error) + payload = output.multimodal_output + audio, sample_rate = payload["audio"], payload["sr"] + metadata = output.custom_output["audio_metadata"] + assert sample_rate == 24000 and audio.ndim == 1 + assert len(audio) == metadata["pcm_frames"] and len(audio) > 0 + assert float(audio.square().mean().sqrt()) > 1e-5 + if save: + _save_wav(Path(args.output_wav), audio) + return { + "text": text, + "wall_s": wall_s, + "frames": len(audio), + "duration_s": metadata["duration_s"], + "pcm_sha256": metadata["pcm_sha256"], + "stage_event": output.custom_output["stage_event"], + } + finally: + output.release_stage_buffers() + await asyncio.sleep(0) + + first = "Hello from the local computer." + second = "The blue car is parked beside the library." + report["checks"] = [await request_one(first), await request_one(second)] + report["warmups"] = [await request_one(first) for _ in range(args.warmups)] + report["measured"] = [await request_one(first, save=i == 0) for i in range(args.repeats)] + walls = [row["wall_s"] for row in report["measured"]] + report["nearest_rank_p50_wall_s"] = _rank(walls, 0.5) + report["nearest_rank_p95_wall_s"] = _rank(walls, 0.95) + report["all_same_pcm_sha256"] = len({row["pcm_sha256"] for row in report["measured"]}) == 1 + if args.abort_check: + request_id = f"tts-abort-{time.monotonic_ns()}" + await pool.submit_initial(request_id, state, {"text": second, "voice": "ryan", "seed": 42}) + await asyncio.sleep(0.01) + await pool.abort_requests([request_id]) + await asyncio.sleep(0.1) + report["abort_check"] = { + "stale_output": pool.poll_graph_output(0) is not None, + "worker_exited": pool.stage_client._proc.poll() is not None, + "ledger_after_abort": runtime.resource_ledger.snapshot(), + } + if report["abort_check"]["stale_output"] or not report["abort_check"]["worker_exited"]: + raise RuntimeError("CrispASR cancellation left stale output or live worker") + report["status"] = "passed" + except BaseException as exc: + report["status"] = "failed" + report["error"] = f"{type(exc).__name__}: {exc}" + raise + finally: + runtime.shutdown() + if runtime.resource_ledger is not None: + report["ledger_after_shutdown"] = runtime.resource_ledger.snapshot() + destination = Path(args.output_report) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + print(json.dumps(report, indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/benchmarks/edge_harness/experiments/probe_omni_crisp_tts_entrypoint.py b/benchmarks/edge_harness/experiments/probe_omni_crisp_tts_entrypoint.py new file mode 100644 index 000000000..80db36b83 --- /dev/null +++ b/benchmarks/edge_harness/experiments/probe_omni_crisp_tts_entrypoint.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +"""Verify public AsyncOmni audio delivery from the pinned hybrid TTS stage.""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import tempfile +import time +from pathlib import Path + +import numpy as np +import yaml + + +async def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + for name in ( + "model_config_dir", "server_bin", "server_sha256", "talker_file", "talker_sha256", + "codec_file", "codec_sha256", "punc_file", "punc_sha256", "server_log", "output_report", + ): + parser.add_argument("--" + name.replace("_", "-"), required=True) + args = parser.parse_args() + + from vllm_omni.entrypoints.async_omni import AsyncOmni + from vllm_omni.model_executor.models.qwen3_tts.pipeline import QWEN3_TTS_CRISP_HYBRID_PIPELINE + + backend = { + "name": "external.crisp.tts.v1", + "server_bin": args.server_bin, + "server_sha256": args.server_sha256, + "talker_file": args.talker_file, + "talker_sha256": args.talker_sha256, + "codec_file": args.codec_file, + "codec_sha256": args.codec_sha256, + "punc_file": args.punc_file, + "punc_sha256": args.punc_sha256, + "log_file": args.server_log, + "expected_gpu_name": "AMD Radeon(TM) 890M Graphics", + "ggml_vk_visible_devices": "1", + "memory_overhead_bytes": 5 << 30, + "max_text_bytes": 4096, + "max_wav_bytes": 4 << 20, + "max_audio_s": 10, + "voice": "ryan", + "seed": 42, + } + report = {"entrypoint": "AsyncOmni.generate", "pipeline": QWEN3_TTS_CRISP_HYBRID_PIPELINE.model_type} + engine = None + try: + with tempfile.TemporaryDirectory(prefix="omni-crisp-tts-") as temporary: + deployment = Path(temporary) / "deploy.yaml" + deployment.write_text(yaml.safe_dump({ + "pipeline": QWEN3_TTS_CRISP_HYBRID_PIPELINE.model_type, + "async_chunk": False, + "stages": [{ + "stage_id": 0, + "backend": backend, + "resource_budget": { + "capacities": {"host_ram": 16 << 30}, + "demands": {"host_ram": 8 << 30}, + }, + }], + }), encoding="utf-8") + started = time.perf_counter() + engine = AsyncOmni( + model=str(Path(args.model_config_dir).resolve(strict=True)), + deploy_config=str(deployment), stage_init_timeout=180, init_timeout=240, + ) + report["startup_s"] = time.perf_counter() - started + outputs = [] + started = time.perf_counter() + async for output in engine.generate( + {"text": "Hello from the local computer.", "voice": "ryan", "seed": 42}, + request_id="public-tts-1", + ): + if output.error: + raise RuntimeError(output.error) + audio = output.multimodal_output["audio"] + pcm = np.rint(audio.numpy() * 32768).astype(" + asyncio.run(main()) + File "C:\Users\zhout\AppData\Local\Programs\Python\Python312\Lib\asyncio\runners.py", line 195, in run + return runner.run(main) + ^^^^^^^^^^^^^^^^ + File "C:\Users\zhout\AppData\Local\Programs\Python\Python312\Lib\asyncio\runners.py", line 118, in run + return self._loop.run_until_complete(task) + ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + File "C:\Users\zhout\AppData\Local\Programs\Python\Python312\Lib\asyncio\base_events.py", line 691, in run_until_complete + return future.result() + ^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\benchmarks\edge_harness\experiments\probe_omni_crisp_tts.py", line 100, in main + runtime.initialize() + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 278, in initialize + raise exc + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 252, in initialize + initialized_clients = self._initialize_stage_replicas(stage_plans, self._stage_init_timeout) + ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 589, in _initialize_stage_replicas + raise primary_exc + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 550, in _init_group + client = self._initialize_replica( + ^^^^^^^^^^^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 817, in _initialize_replica + return create_graph_client( + ^^^^^^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\backends\__init__.py", line 19, in create_graph_client + return CrispTTSStageClient(metadata, backend, ledger, reservation) + ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\backends\crisp_tts.py", line 139, in __init__ + raise ValueError("this shared-RAM hybrid route is qualified only for the HX370 Radeon 890M") +ValueError: this shared-RAM hybrid route is qualified only for the HX370 Radeon 890M diff --git a/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_device_20260923.json b/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_device_20260923.json new file mode 100644 index 000000000..d6608dadc --- /dev/null +++ b/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_device_20260923.json @@ -0,0 +1,26 @@ +{ + "scope": "complete resident Qwen3-TTS CustomVoice text-to-WAV through Omni StageRuntime/StagePool", + "placement": "Radeon Vulkan talker/codec + CPU FP32 code predictor", + "warmup_count": 0, + "measured_count": 1, + "memory_budget": { + "capacities": { + "host_ram": 17179869184 + }, + "demands": { + "host_ram": 8589934592 + } + }, + "status": "failed", + "error": "ValueError: this shared-RAM hybrid route is qualified only for the HX370 Radeon 890M", + "ledger_after_shutdown": { + "capacities": { + "host_ram": 17179869184 + }, + "reserved": { + "host_ram": 0 + }, + "owners": [], + "quarantined": [] + } +} diff --git a/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_memory_20260923.driver.log b/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_memory_20260923.driver.log new file mode 100644 index 000000000..2113b9c7d --- /dev/null +++ b/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_memory_20260923.driver.log @@ -0,0 +1,100 @@ +INFO 09-23 20:50:44 [patch.py:259] NVFP4 W4A4 weight_scale NaN-clamp: installed. +INFO 09-23 20:50:44 [patch.py:510] inductor factorable-divisibility patch: installed. +INFO 09-23 20:50:48 [stage_init_utils.py:768] [stage_init] Set VLLM_WORKER_MULTIPROC_METHOD=spawn +WARNING 09-23 20:50:48 [stage_init_utils.py:1887] [stage_init] Failed to load transfer config: [Errno 13] Permission denied: 'C:\\Users\\zhout\\w2' +ERROR 09-23 20:50:50 [stage_runtime.py:265] [StageRuntime] Stage initialization failed; shutting down 0 initialized client(s) + +ERROR 09-23 20:50:50 [stage_runtime.py:265] Traceback (most recent call last): + +ERROR 09-23 20:50:50 [stage_runtime.py:265] File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 252, in initialize + +ERROR 09-23 20:50:50 [stage_runtime.py:265] initialized_clients = self._initialize_stage_replicas(stage_plans, self._stage_init_timeout) + +ERROR 09-23 20:50:50 [stage_runtime.py:265] ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +ERROR 09-23 20:50:50 [stage_runtime.py:265] File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 589, in _initialize_stage_replicas + +ERROR 09-23 20:50:50 [stage_runtime.py:265] raise primary_exc + +ERROR 09-23 20:50:50 [stage_runtime.py:265] File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 550, in _init_group + +ERROR 09-23 20:50:50 [stage_runtime.py:265] client = self._initialize_replica( + +ERROR 09-23 20:50:50 [stage_runtime.py:265] ^^^^^^^^^^^^^^^^^^^^^^^^^ + +ERROR 09-23 20:50:50 [stage_runtime.py:265] File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 817, in _initialize_replica + +ERROR 09-23 20:50:50 [stage_runtime.py:265] return create_graph_client( + +ERROR 09-23 20:50:50 [stage_runtime.py:265] ^^^^^^^^^^^^^^^^^^^^ + +ERROR 09-23 20:50:50 [stage_runtime.py:265] File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\backends\__init__.py", line 19, in create_graph_client + +ERROR 09-23 20:50:50 [stage_runtime.py:265] return CrispTTSStageClient(metadata, backend, ledger, reservation) + +ERROR 09-23 20:50:50 [stage_runtime.py:265] ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +ERROR 09-23 20:50:50 [stage_runtime.py:265] File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\backends\crisp_tts.py", line 132, in __init__ + +ERROR 09-23 20:50:50 [stage_runtime.py:265] raise ResourceUnavailable("TTS weights plus declared state/workspace/headroom exceed reservation") + +ERROR 09-23 20:50:50 [stage_runtime.py:265] vllm_omni.engine.resource_ledger.ResourceUnavailable: TTS weights plus declared state/workspace/headroom exceed reservation +{ + "scope": "complete resident Qwen3-TTS CustomVoice text-to-WAV through Omni StageRuntime/StagePool", + "placement": "Radeon Vulkan talker/codec + CPU FP32 code predictor", + "warmup_count": 0, + "measured_count": 1, + "memory_budget": { + "capacities": { + "host_ram": 17179869184 + }, + "demands": { + "host_ram": 4294967296 + } + }, + "status": "failed", + "error": "ResourceUnavailable: TTS weights plus declared state/workspace/headroom exceed reservation", + "ledger_after_shutdown": { + "capacities": { + "host_ram": 17179869184 + }, + "reserved": { + "host_ram": 0 + }, + "owners": [], + "quarantined": [] + } +} +Traceback (most recent call last): + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\benchmarks\edge_harness\experiments\probe_omni_crisp_tts.py", line 180, in + asyncio.run(main()) + File "C:\Users\zhout\AppData\Local\Programs\Python\Python312\Lib\asyncio\runners.py", line 195, in run + return runner.run(main) + ^^^^^^^^^^^^^^^^ + File "C:\Users\zhout\AppData\Local\Programs\Python\Python312\Lib\asyncio\runners.py", line 118, in run + return self._loop.run_until_complete(task) + ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + File "C:\Users\zhout\AppData\Local\Programs\Python\Python312\Lib\asyncio\base_events.py", line 691, in run_until_complete + return future.result() + ^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\benchmarks\edge_harness\experiments\probe_omni_crisp_tts.py", line 100, in main + runtime.initialize() + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 278, in initialize + raise exc + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 252, in initialize + initialized_clients = self._initialize_stage_replicas(stage_plans, self._stage_init_timeout) + ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 589, in _initialize_stage_replicas + raise primary_exc + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 550, in _init_group + client = self._initialize_replica( + ^^^^^^^^^^^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\stage_runtime.py", line 817, in _initialize_replica + return create_graph_client( + ^^^^^^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\backends\__init__.py", line 19, in create_graph_client + return CrispTTSStageClient(metadata, backend, ledger, reservation) + ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + File "\\wsl.localhost\Ubuntu\home\zhout\project\edge_infer\vllm-omni-edge\vllm_omni\engine\backends\crisp_tts.py", line 132, in __init__ + raise ResourceUnavailable("TTS weights plus declared state/workspace/headroom exceed reservation") +vllm_omni.engine.resource_ledger.ResourceUnavailable: TTS weights plus declared state/workspace/headroom exceed reservation diff --git a/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_memory_20260923.json b/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_memory_20260923.json new file mode 100644 index 000000000..b31d70312 --- /dev/null +++ b/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/omni_crisp_tts_pinned_reject_memory_20260923.json @@ -0,0 +1,26 @@ +{ + "scope": "complete resident Qwen3-TTS CustomVoice text-to-WAV through Omni StageRuntime/StagePool", + "placement": "Radeon Vulkan talker/codec + CPU FP32 code predictor", + "warmup_count": 0, + "measured_count": 1, + "memory_budget": { + "capacities": { + "host_ram": 17179869184 + }, + "demands": { + "host_ram": 4294967296 + } + }, + "status": "failed", + "error": "ResourceUnavailable: TTS weights plus declared state/workspace/headroom exceed reservation", + "ledger_after_shutdown": { + "capacities": { + "host_ram": 17179869184 + }, + "reserved": { + "host_ram": 0 + }, + "owners": [], + "quarantined": [] + } +} diff --git a/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_radeon890m_joint/README.md b/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_radeon890m_joint/README.md index 84a8bcd43..158a90271 100644 --- a/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_radeon890m_joint/README.md +++ b/benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_radeon890m_joint/README.md @@ -1,6 +1,6 @@ # Native Windows CPU + Radeon 890M Qwen3-TTS scoped whole-model path -The 0.6B CustomVoice model completed two named English **text-to-WAV** requests in [CrispASR v0.8.35](https://github.com/CrispStrobe/CrispASR/releases/tag/v0.8.35), outside Omni. This is a **joint CPU+iGPU result**: the talker and codec were routed to Radeon 890M Vulkan, while the iterative code predictor was explicitly pinned to CPU FP32. It is not an all-Radeon pass or a CPU+iGPU+NPU pass. The [DirectML route failed quality](../qwen_tts_radeon890m_directml/README.md). Neither route establishes Omni streaming or release qualification. +The 0.6B CustomVoice model completed two named English **text-to-WAV** requests in [CrispASR v0.8.35](https://github.com/CrispStrobe/CrispASR/releases/tag/v0.8.35), outside Omni. A subsequent [bounded Omni whole-session stage](../qwen_tts_omni_hybrid/README.md) now executes the same hybrid route with exact PCM parity on both inputs. This is a **joint CPU+iGPU result**: the talker and codec were routed to Radeon 890M Vulkan, while the iterative code predictor was explicitly pinned to CPU FP32. It is not an all-Radeon pass or a CPU+iGPU+NPU pass. The [DirectML route failed quality](../qwen_tts_radeon890m_directml/README.md). Neither route establishes playable Omni streaming or release qualification. The host was Windows 11 build 26200 on Ryzen AI 9 HX 370 with Radeon 890M driver `32.0.22018.6001`; an RTX 5090 Laptop was also present. `GGML_VK_VISIBLE_DEVICES=1` exposed only the Radeon adapter, enumerated as `Vulkan0`; `-dev 1` selected the original physical adapter. The [first placement log](crisp_tts_radeon_cp_cpu_20260923.log) records `AMD Radeon(TM) 890M Graphics`, `using preferred GPU backend: Vulkan0`, `code_pred CPU-pinned (72 tensors, 432 MB, copy=f32)`, and `codec: GPU default - loading weights onto Vulkan0`. The [runtime's routing source](https://github.com/CrispStrobe/CrispASR/blob/v0.8.35/src/qwen3_tts.cpp) routes the talker off Vulkan unless `CRISPASR_QWEN3_TTS_VULKAN_NATIVE=1`; that override was set for the passing runs, and the CPU-talkback message was absent. This is placement evidence for the named runtime configuration, not a per-operator profiler trace. diff --git a/benchmarks/edge_harness/results/e2e_profiling_20260922/evidence/summary/README.md b/benchmarks/edge_harness/results/e2e_profiling_20260922/evidence/summary/README.md index 3d8b5082a..81c13067c 100644 --- a/benchmarks/edge_harness/results/e2e_profiling_20260922/evidence/summary/README.md +++ b/benchmarks/edge_harness/results/e2e_profiling_20260922/evidence/summary/README.md @@ -1,6 +1,6 @@ # E2E check and profiling -All 60 pairings have a disposition. The timing tables below remain the 2026-09-22 measurements. The configuration matrix and affected per-cell reasons incorporate the [2026-09-23 scoped recovery runs](../../../e2e_recovery_20260923/README.md) and [WSL CPU/native Windows expansion](../../../e2e_expansion_20260923/README.md). The expansion adds separate 20-request serial MiniCPM-o text-to-speech profiles on WSL CPU and native Windows GPU, a Spark output-head component profile on the HX370 AMD NPU, complete Spark text-request profiles through standalone llama.cpp and a bounded Omni whole-session stage on native Windows CPU and Radeon 890M, a standalone Qwen3-TTS text-to-WAV profile on native Windows CPU, and a separate hybrid CPU+Radeon 890M Qwen3-TTS profile. Profiling completion does not imply full release qualification. +All 60 pairings have a disposition. The timing tables below remain the 2026-09-22 measurements. The configuration matrix and affected per-cell reasons incorporate the [2026-09-23 scoped recovery runs](../../../e2e_recovery_20260923/README.md) and [WSL CPU/native Windows expansion](../../../e2e_expansion_20260923/README.md). The expansion adds separate 20-request serial MiniCPM-o text-to-speech profiles on WSL CPU and native Windows GPU, a Spark output-head component profile on the HX370 AMD NPU, complete Spark text-request profiles through standalone llama.cpp and a bounded Omni whole-session stage on native Windows CPU and Radeon 890M, a standalone Qwen3-TTS text-to-WAV profile on native Windows CPU, and a hybrid CPU+Radeon 890M Qwen3-TTS profile now repeated through a bounded Omni whole-session stage. Profiling completion does not imply full release qualification. ## Configuration × model @@ -12,7 +12,7 @@ NOT E2E means no complete device/model workload was verified. “Scoped E2E” i | PC HX370: CPU / native Windows | scoped E2E: Omni Spark text | scoped E2E: standalone text → WAV | NOT E2E | NOT E2E | synthetic policy pass | | PC HX370 + RTX 5090 Laptop / WSL | profile completed | profile completed | scoped E2E: text + image | scoped E2E: text/image/audio → speech | synthetic policy pass | | PC HX370 + RTX 5090 Laptop / Windows | profile completed | profile completed | NOT E2E | scoped E2E: text/image/audio → speech | synthetic policy pass | -| PC HX370 + Radeon 890M / Windows worker | scoped E2E: Omni Spark text | scoped E2E: Radeon talker/codec + CPU predictor | NOT E2E | NOT E2E | NOT E2E | +| PC HX370 + Radeon 890M / Windows worker | scoped E2E: Omni Spark text | scoped E2E: Omni Radeon talker/codec + CPU predictor | NOT E2E | NOT E2E | NOT E2E | | PC HX370 + AMD NPU / Windows worker | NOT E2E (output-head component passed) | NOT E2E | NOT E2E | NOT E2E | NOT E2E | | PC CPU+iGPU+NPU +/- discrete GPU, joint model execution | NOT E2E | NOT E2E | NOT E2E | NOT E2E | NOT E2E | | PC Snapdragon X Elite CRD / AI Hub | NOT E2E | NOT E2E | NOT E2E | NOT E2E | NOT E2E | @@ -46,7 +46,7 @@ NOT E2E means no complete device/model workload was verified. “Scoped E2E” i | PC HX370 + RTX 5090 Laptop / Windows | MiniCPM-o 4.5 | SCOPED_E2E | The verified checkpoint completed native Windows three-stage BF16 text, one synthetic image, and one synthetic audio input to text+speech in separate constrained plans with 5 GiB thinker host offload and explicit 2,048/512-token limits. Text and all 32 thinker token IDs matched WSL CUDA; a 5.2 s WAV was nonzero. The red-square image produced the same description and 27 thinker token IDs as WSL plus a nonzero 5.2 s WAV in 17.15 s. The 2 s/440 Hz tone produced the same whistling description and nine IDs as WSL plus a nonzero 2.16 s WAV in 8.26 s. The initial text run needed 900/600 s startup limits after a 300 s timeout; a separate 1-warmup/20-measured serial text run had request wall p50 18.38 s/p95 18.73 s, with sampled host available RAM as low as 0.78 GB and pagefile use up to 11.17 GB. These synthetic inputs do not establish general image/audio or speech quality. | Check real image/audio suites, video input, speech intelligibility/alignment, concurrency, interruption, loading-peak admission and sustained thermal/power behavior before broader qualification; [native evidence](../../../e2e_expansion_20260923/README.md). | | PC HX370 + RTX 5090 Laptop / Windows | InternVLA-A1 | SYNTHETIC_POLICY_PASS | The Place_Markpen checkpoint strict-loaded from a UNC path and produced finite `[1, 50, 32]` actions on synthetic zero observations/noise. Policy/output CUDA placement was checked; one native forward took 1.12 s, with 6.68 GB peak allocated. Relative L2 difference from WSL CUDA was 0.83% (no task tolerance). | Validate real observations/reference actions, units/timestamps, memory and repeated latency before task qualification; [native evidence](../../../e2e_expansion_20260923/README.md). | | PC HX370 + Radeon 890M / Windows worker | Spark-X2.5 | SCOPED_E2E_OMNI_COMPLETE_REQUEST | The same pinned Spark-X2.5-1.7B Q4_K_M GGUF completed text prefill/decode through a bounded Omni whole-session stage with llama.cpp Vulkan1. The log identifies Radeon 890M and 29/29 layers offloaded there, rather than RTX 5090; an intentionally wrong device name was refused. `Paris` and `1511` passed; one warmup plus 20 serial complete requests were 20/20 correct, nearest-rank wall p50/p95 2.10/2.14 s. A public AsyncOmni request passed; an oversized context was rejected with a later correct request; in-flight cancellation produced no stale output and released the ledger. The prior standalone run was 2.18/2.20 s. No streaming, restart-after-cancel, GPU-memory peak or sustained power/thermal check is claimed. | Qualify public configuration, broader text/token quality, streaming and cancellation recovery, shared-RAM/GPU peak and sustained use; [Omni raw evidence](../../../e2e_expansion_20260923/evidence/spark_omni_llamacpp/README.md), [standalone evidence](../../../e2e_expansion_20260923/evidence/spark_radeon890m_llamacpp/README.md). | -| PC HX370 + Radeon 890M / Windows worker | Qwen3-TTS 0.6B CustomVoice | SCOPED_E2E_STANDALONE_HYBRID | The Q8_0 CustomVoice GGUF and F16 codec GGUF completed two named text-to-WAV requests in standalone CrispASR on native Windows with Vulkan0 placed on Radeon 890M for the talker/codec and the 432 MB code predictor explicitly pinned to CPU FP32. Whisper tiny.en transcribed both sentences exactly (WER 0 as an intelligibility proxy). One warmup plus 20 serial resident-server requests for one repeated prompt had nearest-rank client wall p50/p95 3.310/3.345 s for identical 2.56 s PCM (RTF 1.29/1.31). The default Vulkan configuration actually fell back to CPU; forcing all stages to Vulkan ran away to the 360-frame cap. A separate DirectML route failed waveform quality. No Omni binding, playable streaming, loading peak, broad quality or sustained power/thermal run was verified. | Bind the whole hybrid session with explicit CPU+iGPU placement/admission and cancellation; validate quality, real-time playback and full-chain benefit before broader qualification; [joint backend evidence](../../../e2e_expansion_20260923/evidence/qwen_tts_radeon890m_joint/README.md), [failed DirectML route](../../../e2e_expansion_20260923/evidence/qwen_tts_radeon890m_directml/README.md). | +| PC HX370 + Radeon 890M / Windows worker | Qwen3-TTS 0.6B CustomVoice | SCOPED_E2E_OMNI_COMPLETE_REQUEST_HYBRID | The pinned Q8_0 CustomVoice talker, F16 codec and punctuation GGUFs completed two named text-to-WAV requests through an Omni whole-session CrispASR stage on native Windows. The runtime log identifies Radeon 890M Vulkan0 for talker/codec and the 432 MB code predictor pinned to CPU FP32; this is not all-Radeon or NPU execution. Both PCM hashes exactly match the earlier standalone outputs. Whisper tiny.en transcribed the first Omni WAV and both standalone WAVs exactly (WER 0 as an intelligibility proxy). One warmup plus 20 serial complete Omni requests for 2.56 s of audio had identical PCM and nearest-rank wall p50/p95 3.298/3.336 s (RTF 1.29/1.30). A public AsyncOmni audio request passed; cancellation drained the worker without stale output or remaining reservation. The earlier standalone profile was 3.310/3.345 s. The default Vulkan configuration previously fell back to CPU, all-Vulkan ran away to a 360-frame cap, and DirectML failed waveform quality. No playable stream, post-cancel restart, loading peak, broad speech quality or sustained power/thermal run is claimed. | Validate playable streaming, restart/recovery, broader voice quality, shared-RAM/GPU peak, concurrency and sustained real-time behavior before wider qualification; [Omni raw evidence](../../../e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/README.md), [standalone evidence](../../../e2e_expansion_20260923/evidence/qwen_tts_radeon890m_joint/README.md), [failed DirectML route](../../../e2e_expansion_20260923/evidence/qwen_tts_radeon890m_directml/README.md). | | PC HX370 + Radeon 890M / Windows worker | Qwen3.8-27B | NOT_E2E_PROFILEABLE | FP32 448x448 vision tower: DirectML 85/113 nodes; normalized max error 9.095e-06 versus torch CPU. Historical component path, not full 27B or current graph-stage image-to-text integration. | Recover the hashed tower bundle and qualify its current Omni graph handoff with the language stage. | | PC HX370 + Radeon 890M / Windows worker | MiniCPM-o 4.5 | NOT_E2E_PROFILEABLE | No placement- and quality-verified whole-model graph bundle/stateful adapter for this model on this AMD route. Small backend graphs do not establish model support. | Supply and validate the missing matching artifact/backend, then run the complete specified workload. | | PC HX370 + Radeon 890M / Windows worker | InternVLA-A1 | NOT_E2E_PROFILEABLE | No placement- and quality-verified whole-model graph bundle/stateful adapter for this model on this AMD route. Small backend graphs do not establish model support. | Supply and validate the missing matching artifact/backend, then run the complete specified workload. | diff --git a/docs/design/omni_edge_backends.md b/docs/design/omni_edge_backends.md index cb08163d6..aa611aef3 100644 --- a/docs/design/omni_edge_backends.md +++ b/docs/design/omni_edge_backends.md @@ -3,12 +3,17 @@ The first backend is `external.graph.v1`: one complete, stateless graph operation per request, with a fixed input bucket and copied host tensors. A second backend, `external.llamacpp.text.v1`, owns one pinned GGUF whole-model server process and -returns complete text requests. Both use the normal `Omni` / `AsyncOmni` APIs, +returns complete text requests. `external.crisp.tts.v1` owns a pinned +CustomVoice GGUF talker, codec and local hybrid CPU+Radeon process, returning +complete PCM audio. All use the normal `Omni` / `AsyncOmni` APIs, `PipelineConfig`, `StageRuntime`, `StagePool` and orchestrator. Native vLLM stages retain their model execution, KV cache, sampling and batching. No accelerator is mandatory. The llama.cpp backend currently uses the complete-request graph control path for admission and acknowledgement; it is not an ONNX graph and does not yet support incremental token events or restart after active cancellation. +The CrispASR route likewise lacks playable incremental PCM and post-cancel +restart; its current placement gate is specific to the measured HX370 Radeon +890M shared-RAM configuration. ## Boundaries @@ -57,6 +62,8 @@ memory-accounting and unavailable-device gaps remain documented repair items. This section preserves the 2026-09-22 audit. For the latest per-cell disposition and 2026-09-23 follow-up evidence, use the [current 60-cell E2E matrix](../../benchmarks/edge_harness/results/e2e_profiling_20260922/evidence/summary/README.md). The HX370 AMD NPU now has a placement- and quality-verified **Spark 1.7B output-head component** through Omni, while complete Spark generation on that route remains unverified; see the [component record](../../benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/spark_amd_npu_output_head/README.md). Native Windows CPU and Radeon 890M completed [scoped whole-model Spark text requests](../../benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/spark_radeon890m_llamacpp/README.md) with the same Q4_K_M GGUF in standalone llama.cpp, then [complete requests through a bounded Omni stage](../../benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/spark_omni_llamacpp/README.md). The Omni route verifies pinned artifacts, actual placement, one-slot memory admission, terminal events and cancellation drain. Incremental streaming, restart after cancellation, memory peaks and broader quality remain open. These results advance the M3 hardware/backend evidence without claiming a joint multi-accelerator plan. +The [Qwen3-TTS CPU+Radeon whole-session path](../../benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/README.md) also completed two named text-to-audio inputs and 20 serial Omni requests with exact PCM parity to its standalone record. It verifies pinned executable and weights, logged hybrid placement, one-slot shared-RAM admission, terminal audio and cancellation drain. Playable streaming, post-cancel restart, loading/iGPU peaks, sustained use and broader speech quality remain open; RTF still exceeds 1. No joint CPU+iGPU+NPU claim follows from this result. + **All 60 named pairings were checked on 2026-09-22** through actual runs, raw-record review or artifact/runtime preflight. This is not 60 successful executions. Each cell links to its evidence, finding and next required step. diff --git a/docs/design/omni_edge_e2e_roadmap.md b/docs/design/omni_edge_e2e_roadmap.md index 46a137e26..fd44ca377 100644 --- a/docs/design/omni_edge_e2e_roadmap.md +++ b/docs/design/omni_edge_e2e_roadmap.md @@ -133,6 +133,8 @@ outputs need reference analysis; timing alone cannot identify their cause. The 2026-09-23 [Spark GGUF Omni follow-up](../../benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/spark_omni_llamacpp/README.md) adds a separate native Windows CPU/Radeon 890M whole-session option. Its pinned 1.7B Q4_K_M artifact passed two named complete text requests and 20 serial inventory requests per device through `StageRuntime`/`StagePool`; public `AsyncOmni` short requests passed on both devices. It verifies CPU or Vulkan1 placement, explicit 4 GiB host reservation, terminal event/acknowledgement and cancellation drain. This does not change the vLLM M0 acceptance or establish M1 mobile generation. Its next gates are incremental output, post-cancel restart, broader token/quality parity, loading/device-memory peaks, concurrency and sustained power/thermal behavior. +The [Qwen3-TTS Omni hybrid follow-up](../../benchmarks/edge_harness/results/e2e_expansion_20260923/evidence/qwen_tts_omni_hybrid/README.md) adds one complete-request stage for the native Windows HX370 CPU+Radeon 890M route. A pinned Q8_0 talker and F16 codec produced two named PCM outputs exactly matching the prior standalone backend; one warmup plus 20 serial requests measured 3.298/3.336 s p50/p95 for 2.56 s of audio. Public `AsyncOmni` audio, cancellation drain and explicit 8 GiB host admission passed. This is progress toward M2 integration, not M2 acceptance: RTF remains above 1, and playable chunk streaming, post-cancel restart, loading/iGPU peaks, broader voice quality and 30-minute thermal behavior remain open. No NPU participated. + VLA qualification here ends at action data. Physical robot control, collision avoidance and actuator safety belong to the external controller. diff --git a/vllm_omni/config/pipeline_registry.py b/vllm_omni/config/pipeline_registry.py index 1f92d4a7d..b27cc8f50 100644 --- a/vllm_omni/config/pipeline_registry.py +++ b/vllm_omni/config/pipeline_registry.py @@ -111,7 +111,11 @@ from vllm_omni.model_executor.models.qwen3_omni.pipeline import ( QWEN3_OMNI_THINKER_ONLY_PIPELINE, resolve_qwen3_omni_pipeline, ) -from vllm_omni.model_executor.models.qwen3_tts.pipeline import QWEN3_TTS_PIPELINE, QWEN3_TTS_TALKER_ONLY_PIPELINE +from vllm_omni.model_executor.models.qwen3_tts.pipeline import ( + QWEN3_TTS_CRISP_HYBRID_PIPELINE, + QWEN3_TTS_PIPELINE, + QWEN3_TTS_TALKER_ONLY_PIPELINE, +) from vllm_omni.model_executor.models.spark2_5.pipeline import ( SPARK2_5_GGUF_TEXT_PIPELINE, SPARK2_5_PIPELINE, @@ -144,6 +148,7 @@ OMNI_PIPELINES: dict[str, PipelineConfig | PipelineResolverFunc] = { "qwen3_omni_moe_thinker_only": QWEN3_OMNI_THINKER_ONLY_PIPELINE, "qwen3_tts": QWEN3_TTS_PIPELINE, "qwen3_tts_talker_only": QWEN3_TTS_TALKER_ONLY_PIPELINE, + "qwen3_tts_crisp_hybrid": QWEN3_TTS_CRISP_HYBRID_PIPELINE, # Single-stage text: an AR decoder with no companion stage. It lives in # the same registry as the multi-stage pipelines so that one code path # resolves every model (see the M0 local text mode). diff --git a/vllm_omni/engine/backends/__init__.py b/vllm_omni/engine/backends/__init__.py index fb69290b9..34cadbf96 100644 --- a/vllm_omni/engine/backends/__init__.py +++ b/vllm_omni/engine/backends/__init__.py @@ -13,4 +13,8 @@ def create_graph_client(metadata, config, ledger, reservation): from .llamacpp import LlamaCppTextStageClient return LlamaCppTextStageClient(metadata, backend, ledger, reservation) + if name == "external.crisp.tts.v1": + from .crisp_tts import CrispTTSStageClient + + return CrispTTSStageClient(metadata, backend, ledger, reservation) raise ValueError(f"unsupported complete-request stage backend: {name!r}") diff --git a/vllm_omni/engine/backends/crisp_tts.py b/vllm_omni/engine/backends/crisp_tts.py new file mode 100644 index 000000000..e0eeab3e1 --- /dev/null +++ b/vllm_omni/engine/backends/crisp_tts.py @@ -0,0 +1,352 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Bounded complete-request Qwen3-TTS stage backed by local CrispASR. + +CrispASR owns the talker, code predictor and codec state. Omni owns the +reservation, process, request identity, terminal audio and output acknowledgement. +This backend does not advertise playable incremental audio or post-cancel restart. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +import hashlib +import io +import json +import os +import re +import socket +import subprocess +import time +import urllib.error +import urllib.request +import uuid +import wave +from pathlib import Path +from typing import Any + +import numpy as np +import torch +from omni_stage_contracts import StageEvent, StageRequest +from vllm.outputs import CompletionOutput + +from vllm_omni.engine.resource_ledger import ResourceUnavailable +from vllm_omni.engine.stage_client import StageClientBase +from vllm_omni.outputs import OmniRequestOutput + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(8 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _local_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +def _post_wav(url: str, text: str, seed: int, *, timeout: float, max_bytes: int) -> bytes: + payload = json.dumps({"input": text, "response_format": "wav", "seed": seed}).encode() + request = urllib.request.Request( + url, data=payload, headers={"Content-Type": "application/json"}, method="POST" + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + if response.status != 200: + raise RuntimeError(f"CrispASR returned HTTP {response.status}") + wav_bytes = response.read(max_bytes + 1) + if len(wav_bytes) > max_bytes: + raise ResourceUnavailable("CrispASR WAV exceeds admitted output bound") + return wav_bytes + + +def _decode_wav(wav_bytes: bytes, max_audio_s: float) -> tuple[torch.Tensor, bytes, int]: + with wave.open(io.BytesIO(wav_bytes), "rb") as wav: + channels, width, sample_rate, frames = ( + wav.getnchannels(), wav.getsampwidth(), wav.getframerate(), wav.getnframes() + ) + if (channels, width, sample_rate) != (1, 2, 24000) or not 0 < frames <= max_audio_s * sample_rate: + raise ValueError("CrispASR WAV violates 24 kHz mono PCM16 or duration contract") + pcm = wav.readframes(frames) + if len(pcm) != frames * 2: + raise ValueError("CrispASR returned truncated PCM") + audio = torch.from_numpy(np.frombuffer(pcm, dtype=" None: + for name, value in vars(metadata).items(): + setattr(self, name, value) + self.stage_type = "graph" # Omni's complete-request control path. + self._ledger, self._reservation = ledger, reservation + self._generation = uuid.uuid4().hex + self._proc: subprocess.Popen | None = None + self._log_stream = None + self._closed = False + self._active: str | None = None + self._task: asyncio.Task | None = None + self._output: OmniRequestOutput | None = None + self._epoch = 0 + self._memory_pool = str(config.get("memory_pool", "host_ram")) + self._max_text_bytes = int(config.get("max_text_bytes", 4096)) + self._max_wav_bytes = int(config.get("max_wav_bytes", 4 << 20)) + self._max_audio_s = float(config.get("max_audio_s", 10)) + self._request_timeout_s = float(config.get("request_timeout_s", 180)) + self._voice = str(config.get("voice", "ryan")) + self._seed = int(config.get("seed", 42)) + self._port = int(config.get("port") or _local_port()) + try: + if ( + self._max_text_bytes <= 0 or self._max_wav_bytes <= 0 or self._max_audio_s <= 0 + or self._request_timeout_s <= 0 or not 0 < self._port < 65536 + or self._seed < 1 or not re.fullmatch(r"[a-z_]+", self._voice) + ): + raise ValueError("invalid CrispASR text, WAV, voice, seed, timeout or port bound") + if self._memory_pool not in reservation.demands: + raise ResourceUnavailable("CrispASR memory pool absent from stage reservation") + demand = reservation.demands[self._memory_pool] + if self._max_wav_bytes > demand: + raise ResourceUnavailable("CrispASR WAV bound exceeds stage reservation") + self._binary = Path(config["server_bin"]).resolve(strict=True) + self._talker = Path(config["talker_file"]).resolve(strict=True) + self._codec = Path(config["codec_file"]).resolve(strict=True) + self._punc = Path(config["punc_file"]).resolve(strict=True) + expected = { + "server_sha256": (self._binary, str(config["server_sha256"]).lower()), + "talker_sha256": (self._talker, str(config["talker_sha256"]).lower()), + "codec_sha256": (self._codec, str(config["codec_sha256"]).lower()), + "punc_sha256": (self._punc, str(config["punc_sha256"]).lower()), + } + for label, (path, digest) in expected.items(): + if _sha256(path) != digest: + raise ValueError(f"CrispASR {label} differs from the declared artifact hash") + overhead = int(config["memory_overhead_bytes"]) + size = self._talker.stat().st_size + self._codec.stat().st_size + self._punc.stat().st_size + if overhead <= 0 or size + overhead > demand: + raise ResourceUnavailable("TTS weights plus declared state/workspace/headroom exceed reservation") + self._expected_gpu_name = str(config["expected_gpu_name"]) + if not self._expected_gpu_name: + raise ValueError("hybrid TTS stage requires expected_gpu_name") + if self._expected_gpu_name != "AMD Radeon(TM) 890M Graphics" or str( + config.get("ggml_vk_visible_devices", "1") + ) != "1": + raise ValueError("this shared-RAM hybrid route is qualified only for the HX370 Radeon 890M") + self._log_path = Path(config["log_file"]).resolve() + self._log_path.parent.mkdir(parents=True, exist_ok=True) + self._log_stream = self._log_path.open("wb") + env = os.environ.copy() + env["GGML_VK_VISIBLE_DEVICES"] = str(config.get("ggml_vk_visible_devices", "1")) + env["CRISPASR_QWEN3_TTS_VULKAN_NATIVE"] = "1" + env["CRISPASR_QWEN3_TTS_CP_BACKEND"] = "cpu-f32" + command = [ + str(self._binary), "--server", "--host", "127.0.0.1", "--port", str(self._port), + "--backend", "qwen3-tts-customvoice", "-m", str(self._talker), + "--codec-model", str(self._codec), "--voice", self._voice, + "--punc-model", str(self._punc), + "--gpu-backend", "vulkan", "-dev", "1", "--no-flash-attn", + "-n", "96", "--seed", str(self._seed), "--verbose", + ] + flags = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0 + self._proc = subprocess.Popen( + command, stdout=self._log_stream, stderr=subprocess.STDOUT, + env=env, creationflags=flags, + ) + base = f"http://127.0.0.1:{self._port}" + deadline = time.monotonic() + float(config.get("start_timeout_s", 180)) + while True: + if self._proc.poll() is not None: + raise RuntimeError(f"CrispASR exited during load; see {self._log_path}") + try: + with urllib.request.urlopen(base + "/health", timeout=1) as response: + if response.status == 200: + break + except (urllib.error.URLError, TimeoutError, OSError): + pass + if time.monotonic() >= deadline: + raise TimeoutError(f"CrispASR did not become ready; see {self._log_path}") + time.sleep(0.2) + log_text = self._log_path.read_text(encoding="utf-8", errors="replace") + placement = ( + f"0 = {self._expected_gpu_name}" in log_text + and "using preferred GPU backend: Vulkan0" in log_text + and "code_pred CPU-pinned" in log_text + and "codec: GPU default - loading weights onto Vulkan0" in log_text + and "qwen3_tts: loaded " in log_text + and f"loaded punctuation model '{self._punc}'" in log_text + and "falling back to CPU" not in log_text + ) + if not placement: + raise RuntimeError("REFUSE_DEVICE_PLACEMENT: CrispASR hybrid route differs from declared CPU+Radeon") + import psutil + + self._loaded_rss_bytes = int(psutil.Process(self._proc.pid).memory_info().rss) + if self._loaded_rss_bytes > demand: + raise ResourceUnavailable("loaded CrispASR RSS exceeds stage reservation") + self._base_url = base + self.execution_plan = { + "backend": "external.crisp.tts.v1", + "stage_id": self.stage_id, + "worker_generation": self._generation, + "worker_pid": self._proc.pid, + "artifact_sha256": {label: digest for label, (_, digest) in expected.items()}, + "placement": "Radeon Vulkan0 talker/codec + CPU FP32 code predictor", + "expected_gpu_name": self._expected_gpu_name, + "ggml_vk_visible_devices": env["GGML_VK_VISIBLE_DEVICES"], + "loaded_rss_bytes": self._loaded_rss_bytes, + "reserved_bytes": dict(reservation.demands), + "memory_overhead_bytes": overhead, + "max_text_bytes": self._max_text_bytes, + "max_wav_bytes": self._max_wav_bytes, + "max_audio_s": self._max_audio_s, + "voice": self._voice, + "seed": self._seed, + "request_capacity": 1, + "stateful_session": "CrispASR-owned; complete WAV per request", + "evidence": "B", + } + except BaseException: + self.shutdown() + raise + + async def add_request_async(self, request_id: str, prompt: Any, params: Any = None) -> None: + self.check_health() + if self._active is not None: + raise ResourceUnavailable("CrispASR stage has an unacknowledged request; capacity is one") + if not isinstance(prompt, dict) or not isinstance(prompt.get("text"), str) or not prompt["text"]: + raise ValueError("TTS prompt must contain nonempty text") + text = prompt["text"] + if len(text.encode("utf-8")) > self._max_text_bytes: + raise ResourceUnavailable("TTS text exceeds admitted request bound") + if prompt.get("voice", self._voice) != self._voice or int(prompt.get("seed", self._seed)) != self._seed: + raise ValueError("TTS voice and seed must match the pinned stage configuration") + self._epoch += 1 + request = StageRequest(request_id, self.stage_id, self._epoch, self._generation) + self._active = request_id + self._task = asyncio.create_task(self._run(request, text), name=f"crisp-tts-{request_id}") + + async def _run(self, request: StageRequest, text: str) -> None: + try: + started = time.perf_counter() + wav = await asyncio.to_thread( + _post_wav, self._base_url + "/v1/audio/speech", text, self._seed, + timeout=self._request_timeout_s, max_bytes=self._max_wav_bytes, + ) + audio, pcm, frames = _decode_wav(wav, self._max_audio_s) + wall_s = time.perf_counter() - started + completion = CompletionOutput(0, "", [], None, None, finish_reason="stop") + completion.multimodal_output = {"audio": audio, "sr": 24000} + event = StageEvent( + request.request_id, self.stage_id, request.epoch, 1, "audio", + self._generation, terminal=True, + ) + output = OmniRequestOutput( + request_id=request.request_id, + prompt=text, + stage_id=self.stage_id, + final_output_type="audio", + outputs=[completion], + _custom_output={ + "stage_event": dataclasses.asdict(event), + "audio_metadata": { + "sample_rate_hz": 24000, + "channels": 1, + "pcm_frames": frames, + "duration_s": frames / 24000, + "pcm_sha256": hashlib.sha256(pcm).hexdigest(), + }, + }, + metrics={"crisp_tts_wall_s": wall_s}, + ) + except asyncio.CancelledError: + raise + except Exception as exc: + output = OmniRequestOutput.from_error(request.request_id, str(exc)) + output.stage_id = self.stage_id + if not self._closed and self._epoch == request.epoch: + loop = asyncio.get_running_loop() + output._stage_release = lambda: loop.call_soon_threadsafe( + self.acknowledge, request.request_id, request.epoch, request.worker_generation + ) + self._output = output + + def get_graph_output_nowait(self) -> OmniRequestOutput | None: + output, self._output = self._output, None + return output + + def acknowledge(self, request_id: str, epoch: int, generation: str) -> None: + if ( + self._active == request_id and epoch == self._epoch + and generation == self._generation + and (self._task is None or self._task.done()) + ): + self._active = None + self._task = None + + async def abort_requests_async(self, request_ids: list[str]) -> None: + if self._active not in request_ids: + return + self._epoch += 1 + self._output = None + if self._task is not None and not self._task.done(): + drained = await asyncio.to_thread(self._terminate) + self._closed = True + done, _ = await asyncio.wait({self._task}, timeout=5) + if done: + self._ledger.release(self._reservation, drained=drained) + else: + self._ledger.release(self._reservation, drained=False) + self._task.add_done_callback( + lambda task: self._ledger.release( + self._reservation, drained=drained and not task.cancelled() + ) + ) + self._active = None + + def _terminate(self) -> bool: + proc = self._proc + try: + if proc is not None and proc.poll() is None: + try: + proc.terminate() + except ProcessLookupError: + pass + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + except OSError: + return False + finally: + if self._log_stream is not None: + self._log_stream.close() + self._log_stream = None + return proc is None or proc.poll() is not None + + def check_health(self) -> None: + if self._closed or self._proc is None or self._proc.poll() is not None: + from vllm.v1.engine.exceptions import EngineDeadError + + raise EngineDeadError() + + async def collective_rpc_async(self, method, timeout=None, args=(), kwargs=None): + raise NotImplementedError(f"CrispASR backend does not implement collective RPC {method}") + + def shutdown(self) -> None: + self._closed = True + self._output = None + drained = self._terminate() + task = self._task + if task is not None and not task.done(): + self._ledger.release(self._reservation, drained=False) + task.add_done_callback( + lambda done: self._ledger.release(self._reservation, drained=drained and not done.cancelled()) + ) + else: + self._ledger.release(self._reservation, drained=drained) diff --git a/vllm_omni/model_executor/models/qwen3_tts/pipeline.py b/vllm_omni/model_executor/models/qwen3_tts/pipeline.py index 3b25dabb6..6dce12d50 100644 --- a/vllm_omni/model_executor/models/qwen3_tts/pipeline.py +++ b/vllm_omni/model_executor/models/qwen3_tts/pipeline.py @@ -83,3 +83,20 @@ QWEN3_TTS_TALKER_ONLY_PIPELINE = PipelineConfig( ), ), ) + + +# Whole-session hybrid fallback for the pinned CustomVoice GGUF artifact. +# CrispASR owns talker, iterative code predictor and codec within one backend; +# graph is only the Omni complete-request admission and acknowledgement path. +QWEN3_TTS_CRISP_HYBRID_PIPELINE = PipelineConfig( + model_type="qwen3_tts_crisp_hybrid", + stages=( + StagePipelineConfig( + stage_id=0, + model_stage="audio", + execution_type=StageExecutionType.GRAPH, + final_output=True, + final_output_type="audio", + ), + ), +)