diff --git a/docs/serving/speech_api.md b/docs/serving/speech_api.md index d4871ce15fa..1db6263cdfe 100644 --- a/docs/serving/speech_api.md +++ b/docs/serving/speech_api.md @@ -279,10 +279,15 @@ curl -X POST http://localhost:8091/v1/audio/voices \ ## Streaming Text Input (WebSocket) -The `/v1/audio/speech/stream` WebSocket endpoint accepts text incrementally and generates audio per sentence as boundaries are detected. +The `/v1/audio/speech/stream` WebSocket endpoint accepts text incrementally. +By default (`split_granularity=none`) it buffers until `input.done` and +synthesizes that flush as **one** TTS request, which keeps long-form timbre +stable. Set `split_granularity` to `sentence` or `clause` to emit a request +at each detected boundary (lower time-to-first-audio for STT/LLM pipelines). -> Note: text input is always streamed incrementally. Audio output remains sentence-scoped: -> use `stream_audio=false` for one binary frame per sentence, or `stream_audio=true` for one or more PCM chunks per sentence. +> Note: `stream_audio` only changes how **audio bytes** are framed (one WAV/PCM +> payload vs chunked PCM). Text segmentation is controlled separately by +> `split_granularity`. ### WebSocket Protocol @@ -314,14 +319,16 @@ upstream LLM) pays the WebSocket handshake once instead of once per utterance. - The session config is sticky. Send `input.text` again straight after `session.done` to reuse it, or send another `session.config` first to change - voice, format, or reference audio. A `session.config` sent while text is - still buffered is rejected so no pending input is silently dropped. -- An utterance is the flush unit, not a linguistic one: it is whatever text was - buffered when `input.done` arrived, of any length, synthesized as one request. - `utterance_index` counts those flushes across the connection, so it tells you - which `input.done` a frame belongs to. `sentence_index` counts within one - flush and so pairs with `total_sentences`, which means every utterance reports - `sentence_index: 0` of `total_sentences: 1` (or `0` for an empty buffer). + voice, format, or reference audio. A `session.config` sent in the middle of + an utterance is rejected so no pending input is silently dropped and so a + split utterance cannot end up half in one voice and half in another. +- An utterance is the flush unit, not a linguistic one: it is one `input.done` + cycle. `utterance_index` counts those flushes across the connection, so it + tells you which `input.done` a frame belongs to. `sentence_index` counts the + TTS requests inside one flush and so pairs with `total_sentences`: with the + default `split_granularity=none` that is always `sentence_index: 0` of + `total_sentences: 1` (or `0` for an empty buffer), while `sentence` or + `clause` counts the linguistic units actually synthesized. - End the connection with `session.close`, or by closing the socket. An idle connection is still closed after the server's idle timeout, which now also applies to the gap between utterances. @@ -332,7 +339,15 @@ All REST API parameters are supported, plus: | Parameter | Type | Default | Description | | ----------- | ------ | --------- | ------------- | -| `stream_audio` | bool | false | Stream one or more PCM chunks for the buffered input over WebSocket | +| `stream_audio` | bool | false | Stream one or more PCM chunks for each TTS request over WebSocket | +| `split_granularity` | string | `"none"` | `"none"`: one request per `input.done`. `"sentence"`: split on `.!?` plus CJK `。！？…`, Indic danda `।॥`, and Arabic `؟`. `"clause"`: also split on `,;，；،؛`. | +| `seed` | integer | null | Forwarded to the speech engine for this session | + +ASCII punctuation only ends a unit when whitespace or `input.done` follows it, +and decimals (`3.14`), thousands separators (`1,000`), abbreviations (`Dr.`, +`e.g.`) and initials (`J. R.`) are not treated as boundaries. A punctuation run +and any closing quote or bracket stay with the unit they close, so `Wait...` +and `He said "Hello."` are one request each. ```bash DELETE /v1/audio/voices/{name} diff --git a/examples/online_serving/text_to_speech/README.md b/examples/online_serving/text_to_speech/README.md index 500e52773d8..50f0e58a526 100644 --- a/examples/online_serving/text_to_speech/README.md +++ b/examples/online_serving/text_to_speech/README.md @@ -629,13 +629,21 @@ Raw PCM streaming requires `stream_format="audio"`, `response_format="pcm"`, and ### Streaming WebSocket -The `/v1/audio/speech/stream` endpoint accepts text incrementally and synthesizes the buffered text as one continuous request on `input.done`: +The `/v1/audio/speech/stream` endpoint accepts text incrementally and, by default, synthesizes the buffered text as one continuous request on `input.done`: ```bash python qwen3_tts/streaming_speech_client.py --text "Hello world. How are you? I am fine." python qwen3_tts/streaming_speech_client.py --text "..." --simulate-stt --stt-delay 0.1 ``` +For per-sentence audio (including Indic danda `।`) before `input.done`: + +```bash +python qwen3_tts/streaming_speech_client.py \ + --text "नमस्ते। कैसे हो?" \ + --split-granularity sentence +``` + `input.done` flushes without closing, so repeating `--text` synthesizes several utterances over one connection: ```bash diff --git a/examples/online_serving/text_to_speech/qwen3_tts/streaming_speech_client.py b/examples/online_serving/text_to_speech/qwen3_tts/streaming_speech_client.py index e141a4c485f..793a5a6d53c 100644 --- a/examples/online_serving/text_to_speech/qwen3_tts/streaming_speech_client.py +++ b/examples/online_serving/text_to_speech/qwen3_tts/streaming_speech_client.py @@ -24,6 +24,11 @@ --text "Hello world. How are you? I am fine." \ --simulate-stt --stt-delay 0.1 + # Opt into per-sentence synthesis (Indic danda, CJK, Latin) + python streaming_speech_client.py \ + --text "नमस्ते। कैसे हो?" \ + --split-granularity sentence + # Receive JSON sidecar chunks with word-level timestamps python streaming_speech_client.py \ --text "Hello world. How are you?" \ @@ -73,8 +78,9 @@ def frame_basename(msg: dict) -> str: """Name a file after the utterance and sentence a frame belongs to. - Every utterance is sentence 0, so the utterance index is what keeps the - files of one connection from overwriting each other. + Every utterance uses `sentence_index` for units inside that flush. + With default `split_granularity=none` that index stays 0; sentence mode + uses 0, 1, ... so files from one connection do not overwrite each other. """ return f"utterance_{msg['utterance_index']:03d}_sentence_{msg['sentence_index']:03d}" @@ -368,6 +374,13 @@ def main(): ) parser.add_argument("--speed", type=float, default=1.0, help="Playback speed (0.25-4.0)") parser.add_argument("--max-new-tokens", type=int, default=None, help="Max tokens") + parser.add_argument("--seed", type=int, default=None, help="Sampling seed forwarded to the server") + parser.add_argument( + "--split-granularity", + default=None, + choices=["none", "sentence", "clause"], + help="Text split mode (default: server none = one request per input.done)", + ) # Base task options parser.add_argument("--ref-audio", default=None, help="Reference audio") @@ -409,6 +422,8 @@ def main(): "max_new_tokens", "ref_audio", "ref_text", + "seed", + "split_granularity", ]: val = getattr(args, key.replace("-", "_"), None) if val is not None: diff --git a/tests/entrypoints/openai_api/test_serving_speech_stream.py b/tests/entrypoints/openai_api/test_serving_speech_stream.py index 7c553af8f56..6414ab4ff8b 100644 --- a/tests/entrypoints/openai_api/test_serving_speech_stream.py +++ b/tests/entrypoints/openai_api/test_serving_speech_stream.py @@ -197,7 +197,7 @@ def test_session_config_rejected_while_input_is_buffered(self, mocker: MockerFix error = ws.receive_json() assert error["type"] == "error" - assert "while input is buffered" in error["message"] + assert "while an utterance is in progress" in error["message"] # The buffered text survives the rejected reconfiguration. ws.send_json({"type": "input.done"}) @@ -742,6 +742,8 @@ async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_ra config.speaker_embedding = None config.stream_audio = True config.word_timestamps = False + config.seed = None + config.non_streaming_mode = None with pytest.raises(WebSocketDisconnect): asyncio.run( @@ -758,6 +760,163 @@ async def mock_generate_pcm_chunks(_generator, _request_id, *, include_sample_ra assert websocket.send_json.await_count == 2 +class TestWebSocketSentenceSplitting: + def test_sentence_granularity_emits_one_request_per_sentence(self, mocker: MockerFixture): + app, speech_service = _build_test_app(mocker=mocker) + + with TestClient(app) as client: + with client.websocket_connect("/v1/audio/speech/stream") as ws: + ws.send_json( + { + "type": "session.config", + "voice": "Vivian", + "split_granularity": "sentence", + } + ) + ws.send_json({"type": "input.text", "text": "Hello world. How are you? "}) + ws.send_json({"type": "input.done"}) + + first = ws.receive_json() + assert first["sentence_index"] == 0 + assert first["sentence_text"] == "Hello world." + ws.receive_bytes() + assert ws.receive_json()["type"] == "audio.done" + + second = ws.receive_json() + assert second["sentence_index"] == 1 + assert second["sentence_text"] == "How are you?" + ws.receive_bytes() + assert ws.receive_json()["type"] == "audio.done" + + assert ws.receive_json() == { + "type": "session.done", + "utterance_index": 0, + "total_sentences": 2, + } + + assert speech_service._generate_audio_bytes.await_count == 2 + assert [call.args[0].input for call in speech_service._generate_audio_bytes.await_args_list] == [ + "Hello world.", + "How are you?", + ] + + def test_sentence_granularity_emits_before_input_done(self, mocker: MockerFixture): + app, speech_service = _build_test_app(mocker=mocker) + + with TestClient(app) as client: + with client.websocket_connect("/v1/audio/speech/stream") as ws: + ws.send_json( + { + "type": "session.config", + "voice": "Vivian", + "split_granularity": "sentence", + } + ) + ws.send_json({"type": "input.text", "text": "First sentence. "}) + + start = ws.receive_json() + assert start["sentence_text"] == "First sentence." + ws.receive_bytes() + assert ws.receive_json()["type"] == "audio.done" + assert speech_service._generate_audio_bytes.await_count == 1 + + ws.send_json({"type": "input.done"}) + assert ws.receive_json() == { + "type": "session.done", + "utterance_index": 0, + "total_sentences": 1, + } + + def test_indic_danda_splits_without_latin_period(self, mocker: MockerFixture): + app, speech_service = _build_test_app(mocker=mocker) + + with TestClient(app) as client: + with client.websocket_connect("/v1/audio/speech/stream") as ws: + ws.send_json( + { + "type": "session.config", + "voice": "Vivian", + "language": "Auto", + "split_granularity": "sentence", + } + ) + ws.send_json({"type": "input.text", "text": "नमस्ते। कैसे हो?"}) + ws.send_json({"type": "input.done"}) + + first = ws.receive_json() + assert first["sentence_text"] == "नमस्ते।" + ws.receive_bytes() + assert ws.receive_json()["type"] == "audio.done" + + second = ws.receive_json() + assert second["sentence_text"] == "कैसे हो?" + ws.receive_bytes() + assert ws.receive_json()["type"] == "audio.done" + assert ws.receive_json()["total_sentences"] == 2 + + def test_session_config_rejected_after_a_split_unit_was_emitted(self, mocker: MockerFixture): + """The splitter buffer is empty here, but the utterance is still open.""" + app, speech_service = _build_test_app(mocker=mocker) + + with TestClient(app) as client: + with client.websocket_connect("/v1/audio/speech/stream") as ws: + ws.send_json( + { + "type": "session.config", + "voice": "Vivian", + "split_granularity": "sentence", + } + ) + ws.send_json({"type": "input.text", "text": "First sentence. "}) + ws.receive_json() + ws.receive_bytes() + assert ws.receive_json()["type"] == "audio.done" + + ws.send_json({"type": "session.config", "voice": "Serena"}) + error = ws.receive_json() + assert error["type"] == "error" + assert "while an utterance is in progress" in error["message"] + + ws.send_json({"type": "input.text", "text": "Second sentence. "}) + assert ws.receive_json()["sentence_index"] == 1 + ws.receive_bytes() + assert ws.receive_json()["type"] == "audio.done" + + ws.send_json({"type": "input.done"}) + assert ws.receive_json() == { + "type": "session.done", + "utterance_index": 0, + "total_sentences": 2, + } + + # Reconfiguration is allowed again once the utterance closed. + ws.send_json({"type": "session.config", "voice": "Serena"}) + ws.send_json({"type": "input.text", "text": "Third."}) + ws.send_json({"type": "input.done"}) + ws.receive_json() + ws.receive_bytes() + ws.receive_json() + ws.receive_json() + + voices = [call.args[0].voice for call in speech_service._generate_audio_bytes.await_args_list] + assert voices == ["Vivian", "Vivian", "Serena"] + + def test_seed_is_forwarded_to_speech_request(self, mocker: MockerFixture): + app, speech_service = _build_test_app(mocker=mocker) + + with TestClient(app) as client: + with client.websocket_connect("/v1/audio/speech/stream") as ws: + ws.send_json({"type": "session.config", "voice": "Vivian", "seed": 42}) + ws.send_json({"type": "input.text", "text": "Hello."}) + ws.send_json({"type": "input.done"}) + ws.receive_json() + ws.receive_bytes() + ws.receive_json() + ws.receive_json() + + assert speech_service._generate_audio_bytes.await_args_list[0].args[0].seed == 42 + + class TestGeneratePcmChunksContract: """Guard: _generate_pcm_chunks must exist on OmniOpenAIServingSpeech. diff --git a/tests/entrypoints/openai_api/test_speech_text_splitter.py b/tests/entrypoints/openai_api/test_speech_text_splitter.py new file mode 100644 index 00000000000..ca2eb76d87e --- /dev/null +++ b/tests/entrypoints/openai_api/test_speech_text_splitter.py @@ -0,0 +1,127 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +import pytest + +from vllm_omni.entrypoints.openai import speech_text_splitter +from vllm_omni.entrypoints.openai.speech_text_splitter import SpeechTextSplitter, extract_complete_units + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +class TestExtractCompleteUnits: + def test_ascii_sentence_needs_whitespace_or_flush(self): + units, remainder, _ = extract_complete_units("Hello world.", frozenset(".!?。"), flush=False) + assert units == [] + assert remainder == "Hello world." + + units, remainder, _ = extract_complete_units("Hello world.", frozenset(".!?。"), flush=True) + assert units == ["Hello world."] + assert remainder == "" + + def test_ascii_sentence_with_trailing_space(self): + units, remainder, _ = extract_complete_units("Hello world. More", frozenset(".!?"), flush=False) + assert units == ["Hello world."] + assert remainder == "More" + + def test_decimal_is_not_a_sentence_boundary(self): + units, remainder, _ = extract_complete_units("It costs 3.14 dollars. ", frozenset(".!?"), flush=False) + assert units == ["It costs 3.14 dollars."] + assert remainder == "" + + def test_thousands_separator_is_not_a_clause_boundary(self): + units, _, _ = extract_complete_units("It is 1,000 rupees, sir. ", frozenset(".!?,;"), flush=False) + assert units == ["It is 1,000 rupees,", "sir."] + + def test_punctuation_run_stays_one_unit(self): + units, _, _ = extract_complete_units("Wait... Really?! Yes.", frozenset(".!?"), flush=True) + assert units == ["Wait...", "Really?!", "Yes."] + + def test_closing_quote_stays_with_its_sentence(self): + units, _, _ = extract_complete_units('He said "Hello." Bye.', frozenset(".!?"), flush=True) + assert units == ['He said "Hello."', "Bye."] + + def test_abbreviations_and_initials_do_not_split(self): + units, _, _ = extract_complete_units("Dr. Smith, e.g. J. R. Tolkien. Next.", frozenset(".!?"), flush=True) + assert units == ["Dr. Smith, e.g. J. R. Tolkien.", "Next."] + + def test_long_token_before_period_is_not_an_abbreviation(self): + units, _, _ = extract_complete_units("We landed in Washington. Next.", frozenset(".!?"), flush=True) + assert units == ["We landed in Washington.", "Next."] + + def test_abbreviation_lookback_is_bounded(self, monkeypatch): + # One WebSocket frame can reach ~128 KiB; an unbounded lookback walked + # the whole prefix at every `.`, quadratic in the frame size. + checks = 0 + is_token_char = speech_text_splitter._is_token_char + + def counting_is_token_char(ch): + nonlocal checks + checks += 1 + return is_token_char(ch) + + monkeypatch.setattr(speech_text_splitter, "_is_token_char", counting_is_token_char) + text = "a." * 20000 + units, _, _ = extract_complete_units(text, frozenset(".!?"), flush=True) + assert units == [text] + assert checks <= len(text) * (speech_text_splitter._MAX_ABBREVIATION_LEN + 1) + + def test_cjk_period_completes_once_the_next_char_is_known(self): + # The trailing 。 waits: a closing delimiter could still follow it. + units, remainder, _ = extract_complete_units("你好。世界。", frozenset(".!?。！？"), flush=False) + assert units == ["你好。"] + assert remainder == "世界。" + + units, remainder, _ = extract_complete_units("你好。世界。", frozenset(".!?。！？"), flush=True) + assert units == ["你好。", "世界。"] + assert remainder == "" + + def test_indic_danda_and_question_mark(self): + units, remainder, _ = extract_complete_units("नमस्ते। कैसे हो?", frozenset(".!?।॥؟"), flush=True) + assert units == ["नमस्ते।", "कैसे हो?"] + assert remainder == "" + + def test_scan_cursor_skips_already_examined_text(self): + units, remainder, scan = extract_complete_units("no terminator yet", frozenset(".!?"), flush=False) + assert units == [] + assert remainder == "no terminator yet" + assert scan == len(remainder) + + units, remainder, scan = extract_complete_units( + "no terminator yet. Next", frozenset(".!?"), flush=False, scan_from=scan + ) + assert units == ["no terminator yet."] + assert remainder == "Next" + + +class TestSpeechTextSplitter: + def test_none_buffers_until_flush(self): + splitter = SpeechTextSplitter("none") + assert splitter.feed("Hello world. ") == [] + assert splitter.flush() == ["Hello world."] + assert splitter.flush() == [] + + def test_sentence_mode_incremental_words(self): + splitter = SpeechTextSplitter("sentence") + assert splitter.feed("Hello ") == [] + assert splitter.feed("world. ") == ["Hello world."] + assert splitter.feed("How are you?") == [] + assert splitter.flush() == ["How are you?"] + + def test_sentence_mode_resumes_scan_across_feeds(self): + splitter = SpeechTextSplitter("sentence") + assert splitter.feed("It costs 3") == [] + assert splitter.feed(".") == [] + assert splitter.feed("14 dollars. ") == ["It costs 3.14 dollars."] + + def test_clause_mode_splits_on_comma(self): + splitter = SpeechTextSplitter("clause") + assert splitter.feed("Hello, world. ") == ["Hello,", "world."] + + def test_clause_mode_splits_on_arabic_semicolon(self): + splitter = SpeechTextSplitter("clause") + assert splitter.feed("مرحبا؛ كيف حالك") == ["مرحبا؛"] + + def test_empty_flush(self): + splitter = SpeechTextSplitter("sentence") + assert splitter.flush() == [] diff --git a/vllm_omni/entrypoints/openai/protocol/audio.py b/vllm_omni/entrypoints/openai/protocol/audio.py index f204ae209ae..2446114df17 100644 --- a/vllm_omni/entrypoints/openai/protocol/audio.py +++ b/vllm_omni/entrypoints/openai/protocol/audio.py @@ -578,6 +578,22 @@ class StreamingSpeechSessionConfig(BaseModel): "frames (existing behavior)." ), ) + seed: int | None = Field( + default=None, + ge=_INT64_MIN, + le=_INT64_MAX, + description="Random seed forwarded to /v1/audio/speech for this session.", + ) + split_granularity: Literal["none", "sentence", "clause"] = Field( + default="none", + description=( + "How incoming input.text is segmented before TTS. 'none' (default) " + "buffers until input.done and runs one request, matching the " + "long-form timbre-continuity path. 'sentence' emits a request at " + "each sentence boundary (Latin .!? plus CJK/Indic/Arabic marks). " + "'clause' also splits on commas/semicolons for lower TTFA." + ), + ) @model_validator(mode="after") def validate_streaming_constraints(self) -> "StreamingSpeechSessionConfig": diff --git a/vllm_omni/entrypoints/openai/serving_speech_stream.py b/vllm_omni/entrypoints/openai/serving_speech_stream.py index 0f883f17920..32e9fa40d01 100644 --- a/vllm_omni/entrypoints/openai/serving_speech_stream.py +++ b/vllm_omni/entrypoints/openai/serving_speech_stream.py @@ -3,8 +3,10 @@ """WebSocket handler for streaming text input TTS. -Accepts text incrementally via WebSocket, buffers it until input.done, and -generates audio once for the buffered input using the existing TTS pipeline. +Accepts text incrementally via WebSocket. By default (split_granularity=none) +it buffers until input.done and generates audio once for the buffered input. +Opting into split_granularity=sentence or clause emits a TTS request at each +detected boundary so incremental STT/LLM clients can hear audio before flush. input.done is a flush, not a close: it ends the current utterance and the connection stays open, so the next utterance reuses the same connection @@ -13,9 +15,10 @@ The session config is sticky across flushes and can be replaced by sending another session.config between utterances. -"Utterance" here names the flush unit rather than any linguistic unit: it is -whatever text the client had buffered when it sent input.done, from a word to -several paragraphs, synthesized as a single request. +"Utterance" here names the flush unit. With split_granularity=none it is +whatever text was buffered when input.done arrived, synthesized as one +request. With sentence/clause splitting it is still one input.done cycle, +but sentence_index counts linguistic units inside that flush. Protocol: Client -> Server: @@ -32,11 +35,10 @@ {"type": "audio.done", "utterance_index": 0, "sentence_index": 0} {"type": "session.done", "utterance_index": 0, "total_sentences": N} {"type": "error", "message": "..."} - # session.done ends the flushed utterance, not the connection. An - # utterance is just the flush unit: whatever text was buffered when - # input.done arrived, of any length. utterance_index counts those - # flushes across the connection, while sentence_index counts within - # one of them and so pairs with total_sentences. + # session.done ends the flushed utterance, not the connection. + # utterance_index counts input.done flushes. sentence_index counts + # TTS requests inside one flush (always 0 of 1 when split_granularity + # is none; 0..N-1 when sentence/clause splitting is enabled). Server -> Client (when word_timestamps=true): {"type": "audio.start", "utterance_index": 0, "sentence_index": 0, @@ -64,6 +66,7 @@ StreamingSpeechSessionConfig, ) from vllm_omni.entrypoints.openai.serving_speech import OmniOpenAIServingSpeech +from vllm_omni.entrypoints.openai.speech_text_splitter import SpeechTextSplitter from vllm_omni.utils.forced_aligner import extract_word_timestamps logger = init_logger(__name__) @@ -79,11 +82,10 @@ class OmniStreamingSpeechHandler: """Handles WebSocket sessions for streaming text-input TTS. - A connection carries one or more utterances. Text arrives incrementally, - is buffered until input.done, and audio is generated once for the - buffered input using the existing OmniOpenAIServingSpeech pipeline. The - connection outlives each utterance so a client can keep synthesizing on - it without reconnecting. + A connection carries one or more utterances. Text arrives incrementally. + With split_granularity=none it is buffered until input.done and synthesized + as one request. sentence/clause modes emit a request at each boundary, + including before input.done. The connection outlives each utterance. Args: speech_service: The existing TTS serving instance (reused for @@ -105,17 +107,20 @@ def __init__( async def handle_session(self, websocket: WebSocket) -> None: """Main loop for a single WebSocket connection. - Serves any number of utterances: text is buffered until input.done, - which flushes it as one TTS request and then leaves the connection - open for the next one. Rejecting a message is only fatal before the - first valid session.config; afterwards the error is reported and the - connection survives. + Serves any number of utterances. input.done always ends a flush; + split_granularity controls whether that flush is one TTS request or + several sentence/clause requests. """ await websocket.accept() config: StreamingSpeechSessionConfig | None = None - text_parts: list[str] = [] + splitter = SpeechTextSplitter("none") utterance_index = 0 + sentence_index = 0 + # An utterance is in progress from the first input.text until the next + # input.done. Buffered text is not enough to detect it: sentence/clause + # splitting can leave the buffer empty after emitting a unit. + utterance_open = False try: while True: @@ -140,10 +145,10 @@ async def handle_session(self, websocket: WebSocket) -> None: msg_type = msg.get("type") if msg_type == "session.config": - if text_parts: + if utterance_open: await self._send_error( websocket, - "session.config cannot be applied while input is buffered; send input.done first", + "session.config cannot be applied while an utterance is in progress; send input.done first", ) continue new_config = await self._build_config(websocket, msg) @@ -152,6 +157,7 @@ async def handle_session(self, websocket: WebSocket) -> None: return # Error already sent, connection closing continue # Keep serving with the previous config config = new_config + splitter = SpeechTextSplitter(config.split_granularity) elif config is None: await self._send_error( @@ -165,32 +171,39 @@ async def handle_session(self, websocket: WebSocket) -> None: if not isinstance(text, str): await self._send_error(websocket, "input.text requires a string value") continue - text_parts.append(text) + if text: + utterance_open = True + for unit in splitter.feed(text): + await self._generate_and_send( + websocket, + config, + unit, + utterance_index=utterance_index, + sentence_index=sentence_index, + ) + sentence_index += 1 elif msg_type == "input.done": - full_text = "".join(text_parts).strip() - text_parts.clear() - total_sentences = 0 - if full_text: - # However long the buffered text is, the pipeline takes - # it as one request, so every flush is sentence 0 of 1. + for unit in splitter.flush(): await self._generate_and_send( websocket, config, - full_text, + unit, utterance_index=utterance_index, - sentence_index=0, + sentence_index=sentence_index, ) - total_sentences = 1 + sentence_index += 1 await websocket.send_json( { "type": "session.done", "utterance_index": utterance_index, - "total_sentences": total_sentences, + "total_sentences": sentence_index, } ) utterance_index += 1 + sentence_index = 0 + utterance_open = False elif msg_type == "session.close": await websocket.close() @@ -310,6 +323,7 @@ async def _generate_and_send( speaker_embedding=config.speaker_embedding, stream=config.stream_audio, word_timestamps=config.word_timestamps, + seed=config.seed, ) start_payload = { diff --git a/vllm_omni/entrypoints/openai/speech_text_splitter.py b/vllm_omni/entrypoints/openai/speech_text_splitter.py new file mode 100644 index 00000000000..1f25a8dba2e --- /dev/null +++ b/vllm_omni/entrypoints/openai/speech_text_splitter.py @@ -0,0 +1,204 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Incremental text splitting for WebSocket TTS input. + +Default serving still synthesizes one request per ``input.done`` (see +``split_granularity='none'``). Opting into ``sentence`` or ``clause`` restores +per-boundary requests so incremental STT/LLM clients can start audio before +the full utterance arrives. + +Terminators cover Latin, CJK, Indic danda, and Arabic question/semicolon marks +so Hindi and similar scripts are not stuck waiting for ``.`` / ``?``. +""" + +from __future__ import annotations + +# ASCII punctuation is ambiguous (abbreviations, decimals, thousands +# separators), so it only closes a unit when whitespace or a flush follows. +# Script-specific marks below are unambiguous and close a unit as soon as the +# character after them is known. +_ASCII_SENTENCE_TERMINATORS = frozenset(".!?") +_ASCII_CLAUSE_TERMINATORS = _ASCII_SENTENCE_TERMINATORS | frozenset(",;") +_SCRIPT_SENTENCE_TERMINATORS = frozenset( + { + "。", + "！", + "？", + "…", + "।", # Devanagari danda + "॥", # Devanagari double danda + "؟", # Arabic question mark + } +) +_SCRIPT_CLAUSE_TERMINATORS = _SCRIPT_SENTENCE_TERMINATORS | { + "，", + "；", + "،", # Arabic comma + "؛", # Arabic semicolon +} + +_SENTENCE_TERMINATORS = _ASCII_SENTENCE_TERMINATORS | _SCRIPT_SENTENCE_TERMINATORS +_CLAUSE_TERMINATORS = _ASCII_CLAUSE_TERMINATORS | _SCRIPT_CLAUSE_TERMINATORS + +# Trailing characters that belong to the unit they close, so `Wait...` and +# `He said "Hello."` stay one TTS request instead of several. +_CLOSING_DELIMITERS = frozenset("\"'”’»)]}』」）】") +_RUN_CHARS = _CLAUSE_TERMINATORS | _CLOSING_DELIMITERS + +# Abbreviations that end in `.` mid-sentence. Single letters are covered +# separately so initials such as `J. R. R.` do not split either. +_ABBREVIATIONS = frozenset( + { + "dr", + "mr", + "mrs", + "ms", + "prof", + "sr", + "jr", + "st", + "mt", + "fig", + "no", + "vs", + "etc", + "approx", + "dept", + "inc", + "ltd", + "co", + "al", + "e.g", + "i.e", + "u.s", + "u.k", + "a.m", + "p.m", + } +) +_MAX_ABBREVIATION_LEN = max(len(abbrev) for abbrev in _ABBREVIATIONS) + + +def _is_token_char(ch: str) -> bool: + return ch.isalnum() or ch == "." + + +def _is_numeric_separator(buffer: str, index: int) -> bool: + """True for the `.`/`,` inside `3.14` or `1,000`.""" + if buffer[index] not in ".,": + return False + if index == 0 or not buffer[index - 1].isdigit(): + return False + return index + 1 < len(buffer) and buffer[index + 1].isdigit() + + +def _is_abbreviation(buffer: str, index: int) -> bool: + """True for the `.` of `Dr.`, `e.g.`, or an initial like `J.`.""" + if buffer[index] != ".": + return False + # Bounded so a frame of `a.a.a...` cannot walk the whole prefix at every + # `.`. Every abbreviation fits inside the limit, so hitting it can only + # rule a token out, never produce a false positive. + limit = max(0, index - _MAX_ABBREVIATION_LEN) + start = index + while start > limit and _is_token_char(buffer[start - 1]): + start -= 1 + if start > 0 and _is_token_char(buffer[start - 1]): + return False + token = buffer[start:index] + if not token: + return False + if len(token) == 1 and token.isalpha(): + return True + return token.lower().strip(".") in _ABBREVIATIONS + + +def extract_complete_units( + buffer: str, + terminators: frozenset[str], + *, + flush: bool, + scan_from: int = 0, +) -> tuple[list[str], str, int]: + """Split ``buffer`` into complete units. + + Returns ``(units, remainder, next_scan)``. ``next_scan`` is an offset into + ``remainder``: everything before it has been examined already, so a caller + appending more text can pass it back as ``scan_from`` instead of rescanning + the whole buffer on every ``input.text`` message. + """ + units: list[str] = [] + last_split = 0 + i = max(scan_from, 0) + length = len(buffer) + while i < length: + ch = buffer[i] + if ch not in terminators or _is_numeric_separator(buffer, i) or _is_abbreviation(buffer, i): + i += 1 + continue + + # Absorb the punctuation/closing-delimiter run so `Wait...` emits once. + run_end = i + 1 + while run_end < length and buffer[run_end] in _RUN_CHARS: + run_end += 1 + + if run_end >= length and not flush: + # The run may still grow; re-examine this terminator next feed. + return units, buffer[last_split:], i - last_split + + ascii_terminator = ch in _ASCII_CLAUSE_TERMINATORS + if run_end < length and ascii_terminator and not buffer[run_end].isspace(): + i = run_end + continue + + piece = buffer[last_split:run_end].strip() + if piece: + units.append(piece) + i = run_end + while i < length and buffer[i].isspace(): + i += 1 + last_split = i + + remainder = buffer[last_split:] + if flush: + tail = remainder.strip() + if tail: + units.append(tail) + return units, "", 0 + return units, remainder, len(remainder) + + +class SpeechTextSplitter: + """Stateful splitter used by one WebSocket utterance.""" + + def __init__(self, granularity: str = "none") -> None: + if granularity not in ("none", "sentence", "clause"): + raise ValueError(f"Unsupported split_granularity: {granularity!r}") + self.granularity = granularity + self._buf = "" + self._scan = 0 + + def _terminators(self) -> frozenset[str] | None: + if self.granularity == "none": + return None + if self.granularity == "clause": + return _CLAUSE_TERMINATORS + return _SENTENCE_TERMINATORS + + def feed(self, text: str) -> list[str]: + self._buf += text + terminators = self._terminators() + if terminators is None: + return [] + units, self._buf, self._scan = extract_complete_units(self._buf, terminators, flush=False, scan_from=self._scan) + return units + + def flush(self) -> list[str]: + terminators = self._terminators() + if terminators is None: + piece = self._buf.strip() + self._buf = "" + self._scan = 0 + return [piece] if piece else [] + units, self._buf, self._scan = extract_complete_units(self._buf, terminators, flush=True, scan_from=self._scan) + return units