[Feature] Speech batch entrypoint (#1701)

Signed-off-by: dsinghvi <divyanshsinghvi@gmail.com>
Signed-off-by: Divyansh Singhvi <divyanshsinghvi@gmail.com>
This commit is contained in:
dsinghvi authored and GitHub committed 2026-03-24 10:52:01 -04:00
1 parent 7f112041da
commit e7e05a55cf
8 files changed
+1094 -8

No files matched your search

+160 -2
View File
@@ -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": "<base64-encoded audio>",
"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
@@ -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)
@@ -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")
@@ -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."""
+1
View File
@@ -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()
+37 -1
View File
@@ -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={
+57 -1
View File
@@ -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."""
+100 -2
View File
@@ -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,
)