[Python] Streaming audio/pcm bytes asynchronously in FastAPI

I am trying to build a speech demo where the user can speak through a React.js web interface and through a pipeline of AI models it receives back an audio response.

The problem I am facing is regarding two endpoints:

the first endpoint is the TTS (Text-To-Speech) service API endpoint:

@app.post('/synthesize_stream')
async def synthesize_stream(request: Request, background_tasks: BackgroundTasks):
"""
A streaming implementation that processes text chunks in parallel.
"""
session_id = request.headers.get("X-Session-ID")
if not session_id:
raise HTTPException(status_code=400, detail="Missing X-Session-ID header")

hook_url = "https://[my-ngrok-host]/speech/hook/audio"

tts_queue = asyncio.Queue()

async def process_text():
async for chunk_text in stream_arabic_sentence_chunker(async_token_stream_reader(request), token_limit=100):
if len(chunk_text.split()) < 3:
chunk_text = "." + chunk_text + " ,"
elif chunk_text[-1] not in ".?!؟":
chunk_text = "." + chunk_text.strip() + "."
chunk_text = chunk_text.replace("*", "")
chunk_text = re.sub(' +', ' ', chunk_text)
print(f"Adding chunk to queue: {chunk_text}")
await tts_queue.put(chunk_text)
await tts_queue.put(None)

async def process_tts():
while True:
chunk_text = await tts_queue.get()
if chunk_text is None:
break

print(f"Processing TTS for chunk: {chunk_text}")

loop = asyncio.get_running_loop()
try:
async with asyncio.Semaphore(2):
with inference_lock:
out = await loop.run_in_executor(
None,
lambda: model.inference(
chunk_text,
"ar",
gpt_cond_latent,
speaker_embedding,
temperature=0.8,
repetition_penalty=1000.0,
top_k=50,
top_p=0.8,
speed=1.0
)
)

audio_tensor = torch.tensor(out["wav"])
cleaned_chunk = await loop.run_in_executor(
None,
lambda: clean_audio_tensor(audio_tensor, original_sr=24000)
)

chunk_bytes = await loop.run_in_executor(
None,
lambda: pcm_bytes_from_tensor(cleaned_chunk, sample_rate=24000)
)

await send_to_hook(chunk_bytes, session_id, hook_url)
except Exception as e:
print(f"Error processing chunk: {e}")
finally:
tts_queue.task_done()

async def send_to_hook(audio_bytes, session_id, hook_url):
try:
await asyncio.to_thread(
requests.post,
hook_url,
headers={"Content-Type": "audio/pcm", "X-Session-ID": session_id},
data=audio_bytes
)
print(f"Sent audio chunk to hook")
except Exception as e:
print(f"Error sending to hook: {e}")

text_processor = asyncio.create_task(process_text())
tts_processor = asyncio.create_task(process_tts())
await asyncio.gather(text_processor, tts_processor)


return JSONResponse({"status": "accepted", "message": "Processing stream in background"})


It is a FastAPI endpoint which receives a stream of text chunks and then process it and eventually send the TTS of each processed chunk to an audio hook endpoint:

@router.post("/hook/audio")
async def audio_hook(request: Request):
"""
Hook endpoint that receives a stream of audio and forwards it to the
WebSocket connection associated with session_id.
"""
session_id = request.headers.get("X-Session-ID")
if not session_id:
raise HTTPException(status_code=400, detail="Missing X-Session-ID header")

if session_id not in audio_queues:
audio_queues[session_id] = asyncio.Queue()

try:
async for chunk in request.stream():
if chunk:
await audio_queues[session_id].put(chunk)
return {"status": "ok", "message": "Audio streamed successfully"}
except Exception as e:
logger.error(f"Error in audio hook stream: {e}")
raise HTTPException(status_code=500, detail="Error streaming audio")


Which is also a FastAPI endpoint and both services are running through uvicorn.

The problem that I am facing is that the sending of audio chunks to the hook only happens after finishing all processing of text and tts chunks which is not what was intended because I have a separate task to send the audio chunk to the audio hook immediately after it has finished processing.

I want each audio chunk when it becomes available it gets sent to the audio hook immediately and in parallel to other audio chunks being processed.

Thank you and any help will be appreciated

Continue reading...