Voice AI &
Speech Synthesis APIs

import asyncio
import os
from cartesia import AsyncCartesia

# Production Low-Latency Interruption Handler & Voice Streamer
async def stream_voice_response(text_stream, audio_ws):
    cartesia = AsyncCartesia(api_key=os.getenv("CARTESIA_API_KEY"))
    
    # Initialize low-latency sonic context
    voice_ctx = cartesia.tts.websocket()
    
    try:
        async for chunk in text_stream:
            # Send incremental text chunks as the LLM generates them
            await voice_ctx.send(
                model_id="sonic-english",
                transcript=chunk,
                voice_id="694f12bc-c4db-490f-b1e4-a42c0d52f952",
                continue_=True
            )
            
            # Forward binary audio frames directly to phone/browser
            async for audio_chunk in voice_ctx.receive():
                await audio_ws.send_bytes(audio_chunk["audio"])
                
    except asyncio.CancelledError:
        # User interrupted: cancel remote generation immediately
        print("User barge-in detected! Halting TTS synthesis stream...")
        await voice_ctx.cancel()
        raise

Frequently Asked Questions

Commonly evaluated trade-offs, contractual pitfalls, and latency optimization rules.