Update app.py

app.py

@@ -350,54 +350,69 @@ async def stream_text_to_speech_cloning(
     output_format: str = Form("mp3", pattern="^(wav|mp3|flac)$"),
     reference_audio: UploadFile = File(...)):
     """
-    TRUE Real-Time Streaming - Fixed for immediate chunk delivery.
+    Sentence-by-Sentence Streaming using a high-performance, asyncio-native
+    producer-consumer pipeline.
     """
     if not hasattr(app.state, 'tts_wrapper'):
         raise HTTPException(status_code=503, detail="Service unavailable: Model not loaded")
 
     async def stream_generator():
         loop = asyncio.get_event_loop()
-        
-        try:
-            # 1. Setup: Convert audio and get reference encoding
-            converted_wav_buffer = await convert_to_wav_in_memory(reference_audio)
-            ref_audio_bytes = converted_wav_buffer.getvalue()
-            audio_hash = hashlib.sha256(ref_audio_bytes).hexdigest()
-            
-            ref_s = await loop.run_in_executor(
-                tts_executor, 
-                app.state.tts_wrapper._get_or_create_reference_encoding, 
-                audio_hash, 
-                ref_audio_bytes
-            )
-            
-            sentences = app.state.tts_wrapper._split_text_into_chunks(text)
-            logger.info(f"🚀 TRUE STREAMING: Processing {len(sentences)} chunks")
-            
-            def process_chunk(sentence_text):
-                with torch.no_grad():
-                    audio_chunk = app.state.tts_wrapper.tts_model.infer(sentence_text, ref_s, reference_text)
-                return app.state.tts_wrapper._convert_to_streamable_format(audio_chunk, output_format)
-
-            # 2. ✅ TRUE STREAMING: Process and yield chunks ONE BY ONE
-            for i, sentence in enumerate(sentences):
-                logger.info(f"🎯 Processing chunk {i+1}/{len(sentences)}: '{sentence[:50]}...'")
+        q = asyncio.Queue(maxsize=2)
+
+        async def producer():
+            try:
+                converted_wav_buffer = await convert_to_wav_in_memory(reference_audio)
+                ref_audio_bytes = converted_wav_buffer.getvalue()
+                audio_hash = hashlib.sha256(ref_audio_bytes).hexdigest()
                 
-                # Process this chunk and yield immediately
-                chunk_bytes = await loop.run_in_executor(tts_executor, process_chunk, sentence)
-                logger.info(f"✅ Yielded chunk {i+1}: {len(chunk_bytes)} bytes")
-                yield chunk_bytes
+                # ✅ Use LRU cache like blocking endpoint
+                ref_s = await loop.run_in_executor(
+                    tts_executor, 
+                    app.state.tts_wrapper._get_or_create_reference_encoding, 
+                    audio_hash, 
+                    ref_audio_bytes
+                )
                 
-            logger.info("🎉 Streaming complete - all chunks delivered")
+                sentences = app.state.tts_wrapper._split_text_into_chunks(text)
+                
+                def process_chunk(sentence_text):
+                    with torch.no_grad():
+                        audio_chunk = app.state.tts_wrapper.tts_model.infer(sentence_text, ref_s, reference_text)
+                    return app.state.tts_wrapper._convert_to_streamable_format(audio_chunk, output_format)
+
+                # Schedule all chunks to be processed in the background.
+                for sentence in sentences:
+                    task = loop.run_in_executor(tts_executor, process_chunk, sentence)
+                    await q.put(task) # Put the FUTURE, not the result, in the queue.
+
+            except Exception as e:
+                logger.error(f"Error in producer task: {e}")
+                await q.put(e)
+            finally:
+                await q.put(None) # Signal that all tasks have been scheduled.
+
+        producer_task = asyncio.create_task(producer())
+
+        # The CONSUMER's job is to wait for each result and yield it.
+        while True:
+            result = await q.get()
+            if result is None:
+                break
             
-        except Exception as e:
-            logger.error(f"❌ Stream error: {e}")
-            raise
+            if isinstance(result, Exception):
+                logger.error(f"Terminating stream due to producer error: {result}")
+                raise result
+            
+            # Await the result of the background task
+            chunk_bytes = await result
+            yield chunk_bytes
+            
+        await producer_task
 
     return StreamingResponse(
         stream_generator(),
-        media_type=f"audio/{'mpeg' if output_format == 'mp3' else output_format}",
-        headers={"Cache-Control": "no-cache"}
+        media_type=f"audio/{'mpeg' if output_format == 'mp3' else output_format}"
     )
 
 @app.get("/audio/{filename}")