[Frontend] Opt-in WebSocket TTS split_granularity and session seed (#7046)

Signed-off-by: rk9595 <rakesh.kariya@somaiya.edu>
Signed-off-by: Rakesh Kariya <rakesh.kariya@somaiya.edu>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

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}

examples/online_serving/text_to_speech/README.md

@@ -796,13 +796,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

examples/online_serving/text_to_speech/qwen3_tts/streaming_speech_client.py

@@ -24,6 +24,11 @@ Usage:
         --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 @@ except ImportError:
 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:

tests/entrypoints/openai_api/test_serving_speech_stream.py

@@ -197,7 +197,7 @@ class TestStreamingSpeechWebSocket:
 
                 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 @@ class TestStreamingSpeechWebSocket:
         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 @@ class TestStreamingSpeechWebSocket:
         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.
 

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() == []

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":

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 @@ session.close, on the idle timeout, or when the client closes the socket.
 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 @@ Protocol:
         {"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 @@ from vllm_omni.entrypoints.openai.protocol.audio import (
     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 @@ _MAX_INPUT_TEXT_MESSAGE_SIZE = 128 * 1024
 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 @@ class OmniStreamingSpeechHandler:
     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 @@ class OmniStreamingSpeechHandler:
                 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 @@ class OmniStreamingSpeechHandler:
                             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 @@ class OmniStreamingSpeechHandler:
                     if not isinstance(text, str):
                         await self._send_error(websocket, "input.text requires a string value")
                         continue
-                    text_parts.append(text)
-
-                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.
+                    if text:
+                        utterance_open = True
+                    for unit in splitter.feed(text):
                         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
+
+                elif msg_type == "input.done":
+                    for unit in splitter.flush():
+                        await self._generate_and_send(
+                            websocket,
+                            config,
+                            unit,
+                            utterance_index=utterance_index,
+                            sentence_index=sentence_index,
+                        )
+                        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 @@ class OmniStreamingSpeechHandler:
             speaker_embedding=config.speaker_embedding,
             stream=config.stream_audio,
             word_timestamps=config.word_timestamps,
+            seed=config.seed,
         )
 
         start_payload = {

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