From 802eaf2b297f038eeeedf121f303a8756c91457a Mon Sep 17 00:00:00 2001
From: "russell@unturf.com" <russell@unturf.com>
Date: Mon, 26 Jan 2026 19:32:51 -0500
Subject: [PATCH] Add sentence-by-sentence streaming for Qwen TTS

Split text into sentences and stream each as it's generated,
so first audio arrives much faster for long text.
---
 speech.py | 56 ++++++++++++++++++++++++++++++++++++++-----------------
 1 file changed, 39 insertions(+), 17 deletions(-)

diff --git a/speech.py b/speech.py
index 190e408..0274ea2 100755
--- a/speech.py
+++ b/speech.py
@@ -1107,17 +1107,12 @@ async def generate_speech(request: GenerateSpeechRequest):
         else:
             raise BadRequestError(f"Voice '{voice}' requires ref_audio and ref_text configuration", param='voice')
 
-        # Generate audio
-        audio_data = await asyncio.to_thread(
-            qwen_model.tts,
-            text=input_text,
-            language=language,
-            voice_prompt=voice_prompt
-        )
+        # Split text into sentences for streaming (first audio arrives faster)
+        sentences = simple_sentence_split(input_text, max_length=500)
+        logger.info(f"Split text into {len(sentences)} sentences for Qwen streaming")
 
-        # Qwen outputs float32 PCM at ~24kHz (sample rate from model)
-        sample_rate = str(qwen_model.sample_rate or 24000)
-        ffmpeg_args = build_ffmpeg_args(response_format, input_format="f32le", sample_rate=sample_rate)
+        # Qwen outputs float32 PCM at ~24kHz
+        ffmpeg_args = build_ffmpeg_args(response_format, input_format="f32le", sample_rate="24000")
 
         # Apply speed adjustment if needed
         if speed != 1.0:
@@ -1126,19 +1121,46 @@ async def generate_speech(request: GenerateSpeechRequest):
         ffmpeg_args.extend(["-"])
         ffmpeg_proc = subprocess.Popen(ffmpeg_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
 
-        # Use threading to avoid deadlock when audio is large
-        # (pipe buffers can fill causing ffmpeg to block on stdout write,
-        # which blocks our stdin write, causing deadlock)
-        def write_audio():
+        # Use queue + threading for sentence-by-sentence streaming
+        in_q = queue.Queue()
+
+        def generator():
+            """Process sentences sequentially and feed to queue"""
             try:
-                ffmpeg_proc.stdin.write(audio_data)
+                for idx, sentence in enumerate(sentences):
+                    logger.debug(f"Qwen processing sentence {idx+1}/{len(sentences)}: {len(sentence)} chars")
+                    audio_bytes = qwen_model.tts(
+                        text=sentence,
+                        language=language,
+                        voice_prompt=voice_prompt
+                    )
+                    in_q.put(audio_bytes)
+                    logger.debug(f"Qwen: queued sentence {idx+1}/{len(sentences)}")
+            except Exception as e:
+                logger.error(f"Qwen streaming error: {e}")
+            finally:
+                in_q.put(None)  # sentinel
+                logger.info(f"Qwen streaming complete: {len(sentences)} sentences processed")
+
+        def out_writer():
+            """Write audio from queue to ffmpeg stdin"""
+            try:
+                while True:
+                    chunk = in_q.get()
+                    if chunk is None:  # sentinel
+                        break
+                    ffmpeg_proc.stdin.write(chunk)
             except Exception as e:
                 logger.error(f"Qwen ffmpeg write error: {e}")
+                ffmpeg_proc.kill()
             finally:
                 ffmpeg_proc.stdin.close()
 
-        writer_thread = threading.Thread(target=write_audio, daemon=True)
-        writer_thread.start()
+        generator_worker = threading.Thread(target=generator, daemon=True)
+        generator_worker.start()
+
+        out_writer_worker = threading.Thread(target=out_writer, daemon=True)
+        out_writer_worker.start()
 
         return StreamingResponse(content=ffmpeg_proc.stdout, media_type=media_type)
     else: