From e7e05a55cf09772378cc24b0df3127a4294d97ba Mon Sep 17 00:00:00 2001 From: dsinghvi Date: Tue, 24 Mar 2026 14:52:01 +0000 Subject: [PATCH] [Feature] Speech batch entrypoint (#1701) Signed-off-by: dsinghvi Signed-off-by: Divyansh Singhvi --- docs/serving/speech_api.md | 162 ++++++- .../qwen3_tts/batch_speech_client.py | 161 +++++++ .../online_serving/test_qwen3_tts_batch.py | 433 ++++++++++++++++++ .../openai_api/test_serving_speech.py | 147 +++++- vllm_omni/entrypoints/omni_base.py | 1 + vllm_omni/entrypoints/openai/api_server.py | 38 +- .../entrypoints/openai/protocol/audio.py | 58 ++- .../entrypoints/openai/serving_speech.py | 102 ++++- 8 files changed, 1094 insertions(+), 8 deletions(-) create mode 100644 examples/online_serving/qwen3_tts/batch_speech_client.py create mode 100644 tests/e2e/online_serving/test_qwen3_tts_batch.py diff --git a/docs/serving/speech_api.md b/docs/serving/speech_api.md index 7969ca441..6d033bc20 100644 --- a/docs/serving/speech_api.md +++ b/docs/serving/speech_api.md @@ -317,7 +317,7 @@ curl -X POST http://localhost:8091/v1/audio/speech \ }' --output cloned.wav ``` -upload voice +### Upload Voice ```bash curl -X POST http://localhost:8091/v1/audio/voices \ -F "audio_sample=@/path/to/voice_sample.wav" \ @@ -325,7 +325,7 @@ curl -X POST http://localhost:8091/v1/audio/voices \ -F "name=custom_voice_1" ``` -use upload voice +### Use Uploaded Voice ```bash curl -X POST http://localhost:8091/v1/audio/speech \ -H "Content-Type: application/json" \ @@ -336,6 +336,164 @@ curl -X POST http://localhost:8091/v1/audio/speech \ }' --output cloned.wav ``` +## Batch Speech Generation + +The batch endpoint synthesizes multiple texts in a single request, returning all results as JSON with base64-encoded audio. + +### Endpoint + +``` +POST /v1/audio/speech/batch +Content-Type: application/json +``` + +### Request Parameters + +| Parameter | Type | Default | Description | +|-----------|------|---------|-------------| +| `items` | array | **required** | List of items to synthesize (1–32) | +| `model` | string | server's model | Model to use | +| `voice` | string | null | Default voice for all items | +| `response_format` | string | "wav" | Default audio format for all items | +| `speed` | float | 1.0 | Default playback speed (0.25–4.0) | +| `task_type` | string | null | Default TTS task type | +| `language` | string | null | Default language | +| `instructions` | string | null | Default voice style instructions | +| `ref_audio` | string | null | Default reference audio (Base task) | +| `ref_text` | string | null | Default reference transcript (Base task) | +| `max_new_tokens` | integer | null | Default max tokens | + +Each item in the `items` array requires only `input` (the text). All other fields are optional and override the batch-level defaults when set: + +| Field | Type | Description | +|-------|------|-------------| +| `input` | string | **required** — text to synthesize | +| `voice` | string | Override voice for this item | +| `response_format` | string | Override format for this item | +| `speed` | float | Override speed for this item | +| `task_type` | string | Override task type | +| `language` | string | Override language | +| `instructions` | string | Override instructions | +| `ref_audio` | string | Override reference audio | +| `ref_text` | string | Override reference transcript | +| `max_new_tokens` | integer | Override max tokens | + +### Response Format + +```json +{ + "id": "speech-batch-abc123", + "results": [ + { + "index": 0, + "status": "success", + "audio_data": "", + "media_type": "audio/wav" + }, + { + "index": 1, + "status": "error", + "error": "Input text cannot be empty" + } + ], + "total": 2, + "succeeded": 1, + "failed": 1 +} +``` + +### Examples + +**Basic batch with shared defaults:** + +```bash +curl -X POST http://localhost:8091/v1/audio/speech/batch \ + -H "Content-Type: application/json" \ + -d '{ + "items": [ + {"input": "Hello, how are you?"}, + {"input": "Goodbye, see you later!"} + ], + "voice": "vivian", + "language": "English" + }' +``` + +**Per-item overrides (different voices and formats):** + +```bash +curl -X POST http://localhost:8091/v1/audio/speech/batch \ + -H "Content-Type: application/json" \ + -d '{ + "items": [ + {"input": "Hello!", "voice": "vivian", "response_format": "mp3"}, + {"input": "你好！", "voice": "ryan", "language": "Chinese"} + ], + "response_format": "wav" + }' +``` + +**Voice cloning with shared reference audio (Base task):** + +```bash +curl -X POST http://localhost:8091/v1/audio/speech/batch \ + -H "Content-Type: application/json" \ + -d '{ + "items": [ + {"input": "First sentence in the cloned voice."}, + {"input": "Second sentence in the cloned voice."} + ], + "task_type": "Base", + "ref_audio": "https://example.com/reference.wav", + "ref_text": "Transcript of the reference audio" + }' +``` + +Setting `ref_audio` at the batch level applies it to all items, avoiding the need to repeat it per item. + +**Decoding the response in Python:** + +```python +import base64 +import httpx + +response = httpx.post( + "http://localhost:8091/v1/audio/speech/batch", + json={ + "items": [ + {"input": "First sentence."}, + {"input": "Second sentence."}, + ], + "voice": "vivian", + }, + timeout=300.0, +) + +for result in response.json()["results"]: + if result["status"] == "success": + audio_bytes = base64.b64decode(result["audio_data"]) + with open(f"output_{result['index']}.wav", "wb") as f: + f.write(audio_bytes) +``` + +### Configuration + +| Parameter | Source | Default | Description | +|-----------|--------|---------|-------------| +| `tts_batch_max_items` | engine kwarg | 32 | Maximum number of items per batch request | + +All items are fanned out to `generate()` concurrently. The engine's stage worker automatically batches them up to the configured `max_batch_size` and queues the rest — no client-side throttling needed. + +For best throughput, use a batch-optimized stage config with `max_batch_size > 1`: + +```bash +vllm serve Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice \ + --stage-configs-path vllm_omni/model_executor/stage_configs/qwen3_tts_batch.yaml \ + --omni --port 8091 --trust-remote-code --enforce-eager +``` + +The default `qwen3_tts.yaml` uses `max_batch_size: 1` (single request). The `qwen3_tts_batch.yaml` config sets `max_batch_size: 4` for ~4x throughput. + ## Supported Models ### Qwen3-TTS diff --git a/examples/online_serving/qwen3_tts/batch_speech_client.py b/examples/online_serving/qwen3_tts/batch_speech_client.py new file mode 100644 index 000000000..7d48e650f --- /dev/null +++ b/examples/online_serving/qwen3_tts/batch_speech_client.py @@ -0,0 +1,161 @@ +"""Batch speech client for Qwen3-TTS via /v1/audio/speech/batch endpoint. + +This script demonstrates how to synthesize multiple texts in a single request. +A particularly useful scenario is voice cloning: set ref_audio once at the +batch level and generate many utterances in the cloned voice without repeating +the reference for each item. + +Start the server (with batch-optimized config for best throughput): + + vllm serve Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice \ + --stage-configs-path vllm_omni/model_executor/stage_configs/qwen3_tts_batch.yaml \ + --trust-remote-code + +Examples: + # Batch with a predefined voice + python batch_speech_client.py \ + --texts "Hello, how are you?" "Goodbye, see you later!" + + # Voice cloning: one ref_audio, many outputs + python batch_speech_client.py \ + --task-type Base \ + --ref-audio /path/to/reference.wav \ + --ref-text "Transcript of the reference audio" \ + --texts "First cloned sentence." "Second cloned sentence." \ + "Third cloned sentence." +""" + +import argparse +import base64 +import os + +import httpx + +DEFAULT_API_BASE = "http://localhost:8091" +DEFAULT_API_KEY = "EMPTY" + + +def encode_audio_to_base64(audio_path: str) -> str: + """Encode a local audio file to a base64 data URL.""" + if not os.path.exists(audio_path): + raise FileNotFoundError(f"Audio file not found: {audio_path}") + + ext = os.path.splitext(audio_path)[1].lower() + mime_map = {".wav": "audio/wav", ".mp3": "audio/mpeg", ".flac": "audio/flac", ".ogg": "audio/ogg"} + mime_type = mime_map.get(ext, "audio/wav") + + with open(audio_path, "rb") as f: + audio_b64 = base64.b64encode(f.read()).decode("utf-8") + return f"data:{mime_type};base64,{audio_b64}" + + +def run_batch(args) -> None: + """Send a batch TTS request and save each result to a file.""" + items = [{"input": text} for text in args.texts] + + payload: dict = { + "items": items, + "response_format": args.response_format, + } + if args.voice: + payload["voice"] = args.voice + if args.language: + payload["language"] = args.language + if args.task_type: + payload["task_type"] = args.task_type + if args.instructions: + payload["instructions"] = args.instructions + if args.max_new_tokens: + payload["max_new_tokens"] = args.max_new_tokens + + # Voice cloning parameters (shared across all items) + if args.ref_audio: + if args.ref_audio.startswith(("http://", "https://")): + payload["ref_audio"] = args.ref_audio + else: + payload["ref_audio"] = encode_audio_to_base64(args.ref_audio) + if args.ref_text: + payload["ref_text"] = args.ref_text + + print(f"Sending batch of {len(items)} item(s) to {args.api_base}") + if args.ref_audio: + print("Voice cloning mode — ref_audio applied to all items") + + url = f"{args.api_base}/v1/audio/speech/batch" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {args.api_key}", + } + + with httpx.Client(timeout=300.0) as client: + response = client.post(url, json=payload, headers=headers) + + if response.status_code != 200: + print(f"Error {response.status_code}: {response.text}") + return + + data = response.json() + print(f"Total: {data['total']} Succeeded: {data['succeeded']} Failed: {data['failed']}") + + os.makedirs(args.output_dir, exist_ok=True) + for result in data["results"]: + idx = result["index"] + if result["status"] == "success": + audio_bytes = base64.b64decode(result["audio_data"]) + out_path = os.path.join(args.output_dir, f"batch_{idx}.{args.response_format}") + with open(out_path, "wb") as f: + f.write(audio_bytes) + print(f" [{idx}] saved {len(audio_bytes)} bytes -> {out_path}") + else: + print(f" [{idx}] FAILED: {result['error']}") + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Batch speech client for /v1/audio/speech/batch", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + + parser.add_argument("--api-base", default=DEFAULT_API_BASE, help="API base URL") + parser.add_argument("--api-key", default=DEFAULT_API_KEY, help="API key") + + # Texts to synthesize + parser.add_argument( + "--texts", + nargs="+", + required=True, + help="One or more texts to synthesize", + ) + + # Shared voice settings + parser.add_argument("--voice", default="vivian", help="Speaker name (default: vivian)") + parser.add_argument("--language", default=None, help="Language: Auto, Chinese, English, etc.") + parser.add_argument("--instructions", default=None, help="Voice style/emotion instructions") + parser.add_argument( + "--task-type", + default=None, + choices=["CustomVoice", "VoiceDesign", "Base"], + help="TTS task type (default: CustomVoice)", + ) + + # Voice cloning (Base task) + parser.add_argument("--ref-audio", default=None, help="Reference audio path or URL for voice cloning") + parser.add_argument("--ref-text", default=None, help="Reference audio transcript for voice cloning") + + # Generation + parser.add_argument("--max-new-tokens", type=int, default=None, help="Max new tokens per item") + parser.add_argument( + "--response-format", + default="wav", + choices=["wav", "mp3", "flac", "pcm", "aac", "opus"], + help="Audio format (default: wav)", + ) + parser.add_argument("--output-dir", "-o", default="batch_output", help="Output directory (default: batch_output)") + + return parser.parse_args() + + +if __name__ == "__main__": + args = parse_args() + run_batch(args) diff --git a/tests/e2e/online_serving/test_qwen3_tts_batch.py b/tests/e2e/online_serving/test_qwen3_tts_batch.py new file mode 100644 index 000000000..d0d633661 --- /dev/null +++ b/tests/e2e/online_serving/test_qwen3_tts_batch.py @@ -0,0 +1,433 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +""" +E2E tests for the /v1/audio/speech/batch endpoint. + +Validates bulk synthesis via the batch API with actual model inference. +""" + +import os + +os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn" +os.environ["VLLM_TEST_CLEAN_GPU_MEMORY"] = "0" + +import base64 +import shutil +import struct +import tempfile +import time +from pathlib import Path + +import httpx +import pytest +import yaml + +from tests.conftest import ( + OmniServer, + convert_audio_file_to_text, + cosine_similarity_text, +) +from tests.utils import hardware_test + +MODEL = "Qwen/Qwen3-TTS-12Hz-0.6B-CustomVoice" + + +def get_stage_config(name: str = "qwen3_tts.yaml"): + return str(Path(__file__).parent.parent.parent.parent / "vllm_omni" / "model_executor" / "stage_configs" / name) + + +@pytest.fixture(scope="module") +def omni_server(): + """Start vLLM-Omni server with CustomVoice model.""" + stage_config_path = get_stage_config() + + with OmniServer( + MODEL, + [ + "--stage-configs-path", + stage_config_path, + "--stage-init-timeout", + "120", + "--trust-remote-code", + "--enforce-eager", + "--disable-log-stats", + ], + ) as server: + yield server + + +def make_batch_request( + host: str, + port: int, + items: list[dict], + voice: str | None = "vivian", + language: str | None = "English", + response_format: str = "wav", + timeout: float = 300.0, + **kwargs, +) -> httpx.Response: + """Make a request to the /v1/audio/speech/batch endpoint.""" + url = f"http://{host}:{port}/v1/audio/speech/batch" + payload: dict = { + "items": items, + "response_format": response_format, + } + if voice is not None: + payload["voice"] = voice + if language is not None: + payload["language"] = language + payload.update(kwargs) + + with httpx.Client(timeout=timeout) as client: + return client.post(url, json=payload) + + +def verify_wav_bytes(audio_bytes: bytes) -> bool: + """Verify decoded audio bytes form a valid WAV.""" + if len(audio_bytes) < 44: + return False + return audio_bytes[:4] == b"RIFF" and audio_bytes[8:12] == b"WAVE" + + +def assert_not_silence(pcm_bytes: bytes): + """Assert PCM16 samples are not all identical.""" + samples = struct.unpack(f"<{len(pcm_bytes) // 2}h", pcm_bytes) + unique = set(samples) + assert len(unique) > 1, f"All-silence: {len(samples)} samples, unique={unique}" + + +MIN_AUDIO_BYTES = 10000 + + +class TestSpeechBatchE2E: + """E2E tests for /v1/audio/speech/batch endpoint.""" + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch_basic_two_items(self, omni_server) -> None: + """Batch with two items returns two successful base64-encoded results.""" + items = [ + {"input": "Hello, how are you today?"}, + {"input": "The weather is nice outside."}, + ] + response = make_batch_request( + host=omni_server.host, + port=omni_server.port, + items=items, + ) + + assert response.status_code == 200, f"Request failed: {response.text}" + data = response.json() + + assert data["total"] == 2 + assert data["succeeded"] == 2 + assert data["failed"] == 0 + assert data["id"].startswith("speech-batch-") + assert len(data["results"]) == 2 + + for i, result in enumerate(data["results"]): + assert result["index"] == i + assert result["status"] == "success" + assert result["audio_data"] is not None + assert result["error"] is None + audio_bytes = base64.b64decode(result["audio_data"]) + assert verify_wav_bytes(audio_bytes), f"Item {i}: invalid WAV" + assert len(audio_bytes) > MIN_AUDIO_BYTES, f"Item {i}: audio too small ({len(audio_bytes)} bytes)" + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch_single_item(self, omni_server) -> None: + """Batch with a single item works correctly.""" + items = [{"input": "Single item batch test."}] + response = make_batch_request( + host=omni_server.host, + port=omni_server.port, + items=items, + ) + + assert response.status_code == 200, f"Request failed: {response.text}" + data = response.json() + + assert data["total"] == 1 + assert data["succeeded"] == 1 + assert data["failed"] == 0 + assert len(data["results"]) == 1 + assert data["results"][0]["status"] == "success" + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch_per_item_voice_override(self, omni_server) -> None: + """Per-item voice overrides the batch-level default.""" + items = [ + {"input": "First item with default voice."}, + {"input": "Second item with ryan voice.", "voice": "ryan"}, + ] + response = make_batch_request( + host=omni_server.host, + port=omni_server.port, + items=items, + voice="vivian", + ) + + assert response.status_code == 200, f"Request failed: {response.text}" + data = response.json() + + assert data["total"] == 2 + assert data["succeeded"] == 2 + # Both items should produce valid audio + for result in data["results"]: + assert result["status"] == "success" + audio_bytes = base64.b64decode(result["audio_data"]) + assert verify_wav_bytes(audio_bytes) + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch_multiple_languages(self, omni_server) -> None: + """Batch items with different languages per item.""" + items = [ + {"input": "Hello, nice to meet you.", "language": "English"}, + {"input": "你好，很高兴认识你。", "language": "Chinese"}, + ] + response = make_batch_request( + host=omni_server.host, + port=omni_server.port, + items=items, + ) + + assert response.status_code == 200, f"Request failed: {response.text}" + data = response.json() + + assert data["succeeded"] == 2 + for result in data["results"]: + assert result["status"] == "success" + audio_bytes = base64.b64decode(result["audio_data"]) + assert len(audio_bytes) > MIN_AUDIO_BYTES + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch_whisper_transcription(self, omni_server) -> None: + """Whisper transcription of batch output matches input text.""" + input_text = "Good morning, welcome to the speech synthesis test." + items = [{"input": input_text}] + response = make_batch_request( + host=omni_server.host, + port=omni_server.port, + items=items, + ) + + assert response.status_code == 200, f"Request failed: {response.text}" + data = response.json() + assert data["succeeded"] == 1 + + audio_bytes = base64.b64decode(data["results"][0]["audio_data"]) + assert verify_wav_bytes(audio_bytes) + + with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: + f.write(audio_bytes) + wav_path = f.name + + try: + transcript = convert_audio_file_to_text(wav_path) + print(f"Whisper transcript: {transcript}") + assert len(transcript.strip()) > 0, "Empty transcript — likely silence" + similarity = cosine_similarity_text(transcript.lower(), input_text.lower()) + print(f"Cosine similarity: {similarity:.3f}") + assert similarity > 0.7, f"Transcript mismatch: similarity={similarity:.2f}, transcript='{transcript}'" + finally: + os.unlink(wav_path) + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch_result_indices_ordered(self, omni_server) -> None: + """Result indices match the input item order.""" + items = [ + {"input": "First sentence."}, + {"input": "Second sentence."}, + {"input": "Third sentence."}, + ] + response = make_batch_request( + host=omni_server.host, + port=omni_server.port, + items=items, + ) + + assert response.status_code == 200, f"Request failed: {response.text}" + data = response.json() + + indices = [r["index"] for r in data["results"]] + assert indices == [0, 1, 2] + + +class TestSpeechBatchValidation: + """Validation / error-handling tests for the batch endpoint.""" + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch_empty_items_rejected(self, omni_server) -> None: + """Empty items list returns a 4xx error.""" + url = f"http://{omni_server.host}:{omni_server.port}/v1/audio/speech/batch" + payload = {"items": [], "voice": "vivian"} + + with httpx.Client(timeout=30.0) as client: + response = client.post(url, json=payload) + + assert response.status_code in (400, 422) + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch_exceeds_max_items(self, omni_server) -> None: + """Batch exceeding 32 items returns 400 error.""" + items = [{"input": f"Item {i}"} for i in range(33)] + response = make_batch_request( + host=omni_server.host, + port=omni_server.port, + items=items, + ) + + assert response.status_code == 400 + + +def _make_batch2_stage_config() -> str: + """Create a temporary stage config with max_batch_size=2 for both stages.""" + src = Path(get_stage_config("qwen3_tts_batch.yaml")) + with open(src) as f: + cfg = yaml.safe_load(f) + + for stage in cfg["stage_args"]: + stage["runtime"]["max_batch_size"] = 2 + + tmp_dir = tempfile.mkdtemp(prefix="tts_batch2_") + dst = Path(tmp_dir) / "qwen3_tts_batch2.yaml" + with open(dst, "w") as f: + yaml.dump(cfg, f) + return str(dst), tmp_dir + + +def make_single_request( + host: str, + port: int, + text: str, + voice: str = "vivian", + language: str = "English", + timeout: float = 300.0, +) -> httpx.Response: + """Make a request to the single /v1/audio/speech endpoint.""" + url = f"http://{host}:{port}/v1/audio/speech" + payload = {"input": text, "voice": voice, "language": language} + with httpx.Client(timeout=timeout) as client: + return client.post(url, json=payload) + + +@pytest.fixture(scope="module") +def omni_server_batch2(): + """Start vLLM-Omni server with max_batch_size=2 config.""" + config_path, tmp_dir = _make_batch2_stage_config() + + with OmniServer( + MODEL, + [ + "--stage-configs-path", + config_path, + "--stage-init-timeout", + "120", + "--trust-remote-code", + "--enforce-eager", + "--disable-log-stats", + ], + ) as server: + yield server + + shutil.rmtree(tmp_dir, ignore_errors=True) + + +class TestSpeechBatchSize2: + """E2E tests with max_batch_size=2 to verify true batched inference.""" + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch2_produces_valid_audio(self, omni_server_batch2) -> None: + """Batch of 2 items with batched engine produces valid audio.""" + items = [ + {"input": "Hello, this is the first sentence."}, + {"input": "And this is the second sentence."}, + ] + response = make_batch_request( + host=omni_server_batch2.host, + port=omni_server_batch2.port, + items=items, + ) + + assert response.status_code == 200, f"Request failed: {response.text}" + data = response.json() + + assert data["total"] == 2 + assert data["succeeded"] == 2 + assert data["failed"] == 0 + + # Save audio files for inspection + output_dir = Path(tempfile.mkdtemp(prefix="tts_batch2_output_")) + for result in data["results"]: + assert result["status"] == "success" + audio_bytes = base64.b64decode(result["audio_data"]) + assert verify_wav_bytes(audio_bytes), f"Item {result['index']}: invalid WAV" + assert len(audio_bytes) > MIN_AUDIO_BYTES + + wav_path = output_dir / f"batch_item_{result['index']}.wav" + wav_path.write_bytes(audio_bytes) + print(f" Item {result['index']}: {len(audio_bytes)} bytes -> {wav_path}") + + print(f"\nBatch audio saved to: {output_dir}") + + @pytest.mark.core_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100"}, num_cards=1) + def test_batch2_vs_sequential_timing(self, omni_server_batch2) -> None: + """Compare batch-of-2 vs 2 sequential single requests. + + With max_batch_size=2 the engine can process both items + concurrently, so the batch call should be faster than two + sequential single calls (or at least comparable). + """ + text_a = "The quick brown fox jumps over the lazy dog." + text_b = "A journey of a thousand miles begins with a single step." + + host = omni_server_batch2.host + port = omni_server_batch2.port + + # --- Sequential: 2 single /v1/audio/speech calls --- + t0 = time.perf_counter() + resp_a = make_single_request(host, port, text_a) + resp_b = make_single_request(host, port, text_b) + sequential_time = time.perf_counter() - t0 + + assert resp_a.status_code == 200, f"Single A failed: {resp_a.text}" + assert resp_b.status_code == 200, f"Single B failed: {resp_b.text}" + + # --- Batch: 1 /v1/audio/speech/batch call with 2 items --- + items = [{"input": text_a}, {"input": text_b}] + t0 = time.perf_counter() + resp_batch = make_batch_request(host, port, items) + batch_time = time.perf_counter() - t0 + + assert resp_batch.status_code == 200, f"Batch failed: {resp_batch.text}" + batch_data = resp_batch.json() + assert batch_data["succeeded"] == 2 + + # Verify both produced valid audio + for result in batch_data["results"]: + audio_bytes = base64.b64decode(result["audio_data"]) + assert verify_wav_bytes(audio_bytes) + assert len(audio_bytes) > MIN_AUDIO_BYTES + + print(f"\n Sequential (2 singles): {sequential_time:.2f}s") + print(f" Batch (2 items): {batch_time:.2f}s") + print(f" Speedup: {sequential_time / batch_time:.2f}x") diff --git a/tests/entrypoints/openai_api/test_serving_speech.py b/tests/entrypoints/openai_api/test_serving_speech.py index 67abd7617..ce32eb8b8 100644 --- a/tests/entrypoints/openai_api/test_serving_speech.py +++ b/tests/entrypoints/openai_api/test_serving_speech.py @@ -20,7 +20,12 @@ from vllm.entrypoints.openai.engine.protocol import ErrorInfo, ErrorResponse from vllm_omni.entrypoints.openai import api_server as api_server_module from vllm_omni.entrypoints.openai.audio_utils_mixin import AudioMixin -from vllm_omni.entrypoints.openai.protocol.audio import CreateAudio, OpenAICreateSpeechRequest +from vllm_omni.entrypoints.openai.protocol.audio import ( + BatchSpeechRequest, + CreateAudio, + OpenAICreateSpeechRequest, + SpeechBatchItem, +) from vllm_omni.entrypoints.openai.serving_speech import ( OmniOpenAIServingSpeech, _create_wav_header, @@ -181,6 +186,7 @@ def test_app(mocker: MockerFixture): mock_engine_client.generate = mocker.MagicMock(side_effect=mock_generate_fn) mock_engine_client.default_sampling_params_list = [{}] + mock_engine_client.tts_batch_max_items = 32 # Mock models to have an is_base_model method mock_models = mocker.MagicMock() @@ -194,6 +200,9 @@ def test_app(mocker: MockerFixture): request_logger=mock_request_logger, ) + # Skip TTS validation in tests (mock doesn't set up supported_speakers) + speech_server._validate_tts_request = mocker.MagicMock(return_value=None) + # Patch the signature of create_speech to remove 'raw_request' for FastAPI route introspection original_create_speech = speech_server.create_speech _ = mocker.MagicMock(side_effect=original_create_speech) @@ -231,6 +240,7 @@ def test_app(mocker: MockerFixture): return {"voices": speakers, "uploaded_voices": uploaded_voices} app.add_api_route("/v1/audio/voices", list_voices, methods=["GET"]) + app.add_api_route("/v1/audio/speech/batch", speech_server.create_speech_batch, methods=["POST"]) # Add upload_voice endpoint async def upload_voice(audio_sample: UploadFile = File(...), consent: str = Form(...), name: str = Form(...)): @@ -923,7 +933,7 @@ class TestStreamingProtocolValidation: def test_stream_validation_errors(self): """stream=True requires response_format not in ('pcm', 'wav') and speed=1.0.""" - with pytest.raises(ValidationError, match="requires response_format not in \\('pcm', 'wav'\\)"): + with pytest.raises(ValidationError, match="requires response_format='pcm' or 'wav'"): OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="mp3") with pytest.raises(ValidationError, match="Speed adjustment is not supported"): OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="pcm", speed=2.0) @@ -1028,6 +1038,139 @@ class TestStreamingResponse: assert "audio/wav" in response.headers["content-type"] +class TestSpeechBatchAPI: + """Tests for the /v1/audio/speech/batch endpoint.""" + + def test_batch_success(self, client): + """Batch with two items should return two successful results with base64 audio.""" + payload = { + "items": [ + {"input": "Hello world"}, + {"input": "Goodbye world"}, + ], + "response_format": "wav", + } + response = client.post("/v1/audio/speech/batch", json=payload) + assert response.status_code == 200 + body = response.json() + assert body["total"] == 2 + assert body["succeeded"] == 2 + assert body["failed"] == 0 + assert all(r["status"] == "success" for r in body["results"]) + assert all(r["audio_data"] is not None for r in body["results"]) + # Verify audio_data is valid base64 + import base64 + + for r in body["results"]: + decoded = base64.b64decode(r["audio_data"]) + assert len(decoded) > 0 + + def test_batch_single_item(self, client): + """Batch with a single item should work.""" + payload = {"items": [{"input": "Solo"}]} + response = client.post("/v1/audio/speech/batch", json=payload) + assert response.status_code == 200 + body = response.json() + assert body["total"] == 1 + assert body["succeeded"] == 1 + + def test_batch_empty_items_rejected(self, client): + """Empty items list should be rejected by Pydantic validation.""" + response = client.post("/v1/audio/speech/batch", json={"items": []}) + assert response.status_code == 422 + + def test_batch_too_many_items(self, client): + """Exceeding the batch max items limit (default 32) should be rejected.""" + payload = {"items": [{"input": f"text {i}"} for i in range(33)]} + with pytest.raises(ValueError, match="exceeding the maximum"): + client.post("/v1/audio/speech/batch", json=payload) + + def test_batch_max_items_allowed(self, client): + """Exactly 32 items should be accepted.""" + payload = {"items": [{"input": f"text {i}"} for i in range(32)]} + response = client.post("/v1/audio/speech/batch", json=payload) + assert response.status_code == 200 + body = response.json() + assert body["total"] == 32 + assert body["succeeded"] == 32 + + def test_batch_results_have_correct_indices(self, client): + """Each result should have an index matching its position.""" + payload = {"items": [{"input": f"text {i}"} for i in range(3)]} + response = client.post("/v1/audio/speech/batch", json=payload) + body = response.json() + indices = [r["index"] for r in body["results"]] + assert indices == [0, 1, 2] + + def test_batch_response_has_id(self, client): + """Batch response should have a unique id starting with 'speech-batch-'.""" + payload = {"items": [{"input": "Hello"}]} + response = client.post("/v1/audio/speech/batch", json=payload) + body = response.json() + assert body["id"].startswith("speech-batch-") + + +class TestMergeBatchItem: + """Tests for the _merge_batch_item static method.""" + + def test_item_override_wins(self): + """Per-item voice should override batch-level voice.""" + batch = BatchSpeechRequest( + items=[SpeechBatchItem(input="hi", voice="Ryan")], + voice="Vivian", + ) + merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0]) + assert merged.voice == "Ryan" + + def test_batch_default_used(self): + """Batch-level voice should be used when item doesn't specify one.""" + batch = BatchSpeechRequest( + items=[SpeechBatchItem(input="hi")], + voice="Vivian", + ) + merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0]) + assert merged.voice == "Vivian" + + def test_response_format_override(self): + """Per-item response_format should override batch default.""" + batch = BatchSpeechRequest( + items=[SpeechBatchItem(input="hi", response_format="mp3")], + response_format="wav", + ) + merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0]) + assert merged.response_format == "mp3" + + def test_stream_always_false(self): + """Merged requests should always have stream=False.""" + batch = BatchSpeechRequest(items=[SpeechBatchItem(input="hi")]) + merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0]) + assert merged.stream is False + + def test_all_fields_merge(self): + """All overridable fields should merge correctly.""" + batch = BatchSpeechRequest( + items=[ + SpeechBatchItem( + input="hello", + voice="Ryan", + language="English", + speed=1.5, + task_type="CustomVoice", + max_new_tokens=512, + ) + ], + voice="Vivian", + language="Chinese", + speed=1.0, + ) + merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0]) + assert merged.voice == "Ryan" + assert merged.language == "English" + assert merged.speed == 1.5 + assert merged.task_type == "CustomVoice" + assert merged.max_new_tokens == 512 + + class TestAsyncOmniSupportedTasks: """Test that AsyncOmni reports correct supported tasks based on output modalities.""" diff --git a/vllm_omni/entrypoints/omni_base.py b/vllm_omni/entrypoints/omni_base.py index 082be95d8..96df0591e 100644 --- a/vllm_omni/entrypoints/omni_base.py +++ b/vllm_omni/entrypoints/omni_base.py @@ -88,6 +88,7 @@ class OmniBase: self.log_stats = log_stats self.async_chunk = async_chunk self.output_modalities = output_modalities or [] + self.tts_batch_max_items: int = kwargs.pop("tts_batch_max_items", 32) logger.info("[%s] Initializing with model %s", self.__class__.__name__, model) st = time.time() diff --git a/vllm_omni/entrypoints/openai/api_server.py b/vllm_omni/entrypoints/openai/api_server.py index 3e14bc2e5..454cb261e 100644 --- a/vllm_omni/entrypoints/openai/api_server.py +++ b/vllm_omni/entrypoints/openai/api_server.py @@ -87,7 +87,7 @@ from vllm_omni.entrypoints.openai.image_api_utils import ( encode_image_base64, parse_size, ) -from vllm_omni.entrypoints.openai.protocol.audio import OpenAICreateSpeechRequest +from vllm_omni.entrypoints.openai.protocol.audio import BatchSpeechRequest, OpenAICreateSpeechRequest from vllm_omni.entrypoints.openai.protocol.images import ( ImageData, ImageGenerationRequest, @@ -950,6 +950,42 @@ async def create_speech(request: OpenAICreateSpeechRequest, raw_request: Request raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)) from e +@router.post( + "/v1/audio/speech/batch", + dependencies=[Depends(validate_json_request)], + responses={ + HTTPStatus.OK.value: {"model": dict}, + HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse}, + HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse}, + HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse}, + }, +) +@with_cancellation +@load_aware_call +async def create_speech_batch(request: BatchSpeechRequest, raw_request: Request): + handler = Omnispeech(raw_request) + if handler is None: + base_server = getattr(raw_request.app.state, "openai_serving_tokenization", None) + if base_server is None: + raise HTTPException( + status_code=HTTPStatus.NOT_FOUND.value, + detail="The model does not support Speech API", + ) + return base_server.create_error_response(message="The model does not support Speech API") + try: + result = await handler.create_speech_batch(request) + if isinstance(result, ErrorResponse): + return JSONResponse( + content=result.model_dump(), + status_code=result.error.code if result.error else 400, + ) + return JSONResponse(content=result.model_dump()) + except ValueError as e: + raise HTTPException(status_code=HTTPStatus.BAD_REQUEST.value, detail=str(e)) from e + except Exception as e: + raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)) from e + + @router.get( "/v1/audio/voices", responses={ diff --git a/vllm_omni/entrypoints/openai/protocol/audio.py b/vllm_omni/entrypoints/openai/protocol/audio.py index a47976663..e8b4f65d5 100644 --- a/vllm_omni/entrypoints/openai/protocol/audio.py +++ b/vllm_omni/entrypoints/openai/protocol/audio.py @@ -73,7 +73,7 @@ class OpenAICreateSpeechRequest(BaseModel): if self.stream: if self.response_format not in ("pcm", "wav"): raise ValueError( - "Streaming (stream=true) requires response_format not in ('pcm', 'wav'). " + "Streaming (stream=true) requires response_format='pcm' or 'wav'. " f"Got response_format='{self.response_format}'." ) if self.speed is None: @@ -102,6 +102,62 @@ class AudioResponse(BaseModel): media_type: str +# --- Batch Speech Models --- + + +class SpeechBatchItem(BaseModel): + """Per-item input for batch speech. Only `input` is required; + all other fields override the batch-level defaults when set.""" + + input: str + voice: str | None = None + instructions: str | None = None + response_format: Literal["wav", "pcm", "flac", "mp3", "aac", "opus"] | None = None + speed: float | None = Field(default=None, ge=0.25, le=4.0) + task_type: Literal["CustomVoice", "VoiceDesign", "Base"] | None = None + language: str | None = None + ref_audio: str | None = None + ref_text: str | None = None + x_vector_only_mode: bool | None = None + max_new_tokens: int | None = None + initial_codec_chunk_frames: int | None = Field(default=None, ge=0) + + +class BatchSpeechRequest(BaseModel): + """Top-level request for batch speech generation. + Fields here act as shared defaults; per-item overrides win.""" + + model: str | None = None + items: list[SpeechBatchItem] = Field(..., min_length=1) + voice: str | None = None + instructions: str | None = None + response_format: Literal["wav", "pcm", "flac", "mp3", "aac", "opus"] = "wav" + speed: float | None = Field(default=1.0, ge=0.25, le=4.0) + task_type: Literal["CustomVoice", "VoiceDesign", "Base"] | None = None + language: str | None = None + ref_audio: str | None = None + ref_text: str | None = None + x_vector_only_mode: bool | None = None + max_new_tokens: int | None = None + initial_codec_chunk_frames: int | None = Field(default=None, ge=0) + + +class SpeechBatchItemResult(BaseModel): + index: int + status: Literal["success", "error"] + audio_data: str | None = None + media_type: str | None = None + error: str | None = None + + +class BatchSpeechResponse(BaseModel): + id: str + results: list[SpeechBatchItemResult] + total: int + succeeded: int + failed: int + + class StreamingSpeechSessionConfig(BaseModel): """Configuration sent as the first WebSocket message for streaming TTS.""" diff --git a/vllm_omni/entrypoints/openai/serving_speech.py b/vllm_omni/entrypoints/openai/serving_speech.py index 3c8278f14..d23b80b31 100644 --- a/vllm_omni/entrypoints/openai/serving_speech.py +++ b/vllm_omni/entrypoints/openai/serving_speech.py @@ -13,6 +13,7 @@ import numpy as np from fastapi import Request, UploadFile from fastapi.responses import Response, StreamingResponse from transformers.utils.hub import cached_file +from vllm.entrypoints.openai.engine.protocol import ErrorResponse from vllm.entrypoints.openai.engine.serving import OpenAIServing from vllm.logger import init_logger from vllm.multimodal.media import MediaConnector @@ -22,8 +23,12 @@ from vllm_omni.entrypoints.openai.audio_utils_mixin import AudioMixin from vllm_omni.entrypoints.openai.metadata_manager import MetadataManager from vllm_omni.entrypoints.openai.protocol.audio import ( AudioResponse, + BatchSpeechRequest, + BatchSpeechResponse, CreateAudio, OpenAICreateSpeechRequest, + SpeechBatchItem, + SpeechBatchItemResult, ) from vllm_omni.outputs import OmniRequestOutput @@ -167,6 +172,9 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin): logger.info(f"Loaded {len(self.supported_speakers)} supported speakers: {sorted(self.supported_speakers)}") logger.info(f"Loaded {len(self.uploaded_speakers)} uploaded speakers") + # Batch configuration + self._batch_max_items: int = getattr(self.engine_client, "tts_batch_max_items", 32) + # Load speech tokenizer codec parameters for prompt length estimation self._codec_frame_rate: float | None = self._load_codec_frame_rate() @@ -1023,7 +1031,8 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin): async def _generate_audio_bytes( self, request: OpenAICreateSpeechRequest, - ) -> tuple[bytes, str]: + base64_encode: bool = False, + ) -> tuple[bytes | str, str]: request_id, generator, _ = await self._prepare_speech_generation(request) final_output: OmniRequestOutput | None = None @@ -1058,7 +1067,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin): response_format=request.response_format or "wav", speed=request.speed or 1.0, stream_format=request.stream_format, - base64_encode=False, + base64_encode=base64_encode, ) audio_response: AudioResponse = self.create_audio(audio_obj) return audio_response.audio_data, audio_response.media_type @@ -1129,3 +1138,92 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin): except Exception as e: logger.exception("Speech generation failed: %s", e) return self.create_error_response(f"Speech generation failed: {e}") + + @staticmethod + def _merge_batch_item( + batch: BatchSpeechRequest, + item: SpeechBatchItem, + ) -> OpenAICreateSpeechRequest: + """Merge batch-level defaults with per-item overrides into a full request.""" + + def _pick(field: str): + """Return item-level value if set, else batch-level value.""" + item_val = getattr(item, field, None) + return item_val if item_val is not None else getattr(batch, field, None) + + picked_speed = _pick("speed") + return OpenAICreateSpeechRequest( + input=item.input, + model=batch.model, + voice=_pick("voice"), + instructions=_pick("instructions"), + response_format=_pick("response_format") or "wav", + speed=picked_speed if picked_speed is not None else 1.0, + stream=False, + task_type=_pick("task_type"), + language=_pick("language"), + ref_audio=_pick("ref_audio"), + ref_text=_pick("ref_text"), + x_vector_only_mode=_pick("x_vector_only_mode"), + max_new_tokens=_pick("max_new_tokens"), + initial_codec_chunk_frames=_pick("initial_codec_chunk_frames"), + ) + + async def create_speech_batch( + self, + batch_request: BatchSpeechRequest, + ) -> BatchSpeechResponse | ErrorResponse: + """Generate speech for multiple items concurrently.""" + if len(batch_request.items) > self._batch_max_items: + raise ValueError( + f"Batch contains {len(batch_request.items)} items, exceeding the maximum of {self._batch_max_items}." + ) + + error_check_ret = await self._check_model(batch_request) + if error_check_ret is not None: + return error_check_ret + + if self.engine_client.errored: + raise self.engine_client.dead_error + + batch_id = f"speech-batch-{random_uuid()}" + + merged_requests = [self._merge_batch_item(batch_request, item) for item in batch_request.items] + + async def _run_item(idx: int, req: OpenAICreateSpeechRequest) -> SpeechBatchItemResult: + validation_error = self._validate_tts_request(req) + if validation_error is not None: + return SpeechBatchItemResult(index=idx, status="error", error=validation_error) + try: + audio_data, media_type = await self._generate_audio_bytes(req, base64_encode=True) + except Exception as e: + logger.exception("Batch item %d failed: %s", idx, e) + return SpeechBatchItemResult(index=idx, status="error", error=str(e)) + return SpeechBatchItemResult( + index=idx, + status="success", + audio_data=audio_data, + media_type=media_type, + ) + + results = await asyncio.gather( + *[_run_item(i, req) for i, req in enumerate(merged_requests)], + return_exceptions=True, + ) + + final_results: list[SpeechBatchItemResult] = [] + for i, r in enumerate(results): + if isinstance(r, BaseException): + logger.exception("Batch item %d raised unexpected exception: %s", i, r) + final_results.append(SpeechBatchItemResult(index=i, status="error", error=str(r))) + else: + final_results.append(r) + + succeeded = sum(1 for r in final_results if r.status == "success") + return BatchSpeechResponse( + id=batch_id, + results=final_results, + total=len(final_results), + succeeded=succeeded, + failed=len(final_results) - succeeded, + )