Voice AI Call Center Agents: Architecture Blueprint & Production Stack

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio
import json
import httpx

app = FastAPI(title="Voice AI WebSocket Orchestrator")

@app.websocket("/ws/call/{call_id}")
async def handle_voice_call_websocket(websocket: WebSocket, call_id: str):
    await websocket.accept()
    print(f"Voice Call Connected: {call_id}")
    
    # Interrupt Event Handle for Customer Barge-in
    interrupt_event = asyncio.Event()
    
    async def receive_telephony_audio():
        """Listens for inbound audio chunks and VAD speech events from Twilio/SIP."""
        try:
            while True:
                data = await websocket.receive_text()
                event = json.loads(data)
                
                # Check for VAD Speech-Started Event (Customer Barge-in)
                if event.get("event") == "speech_started":
                    print("Barge-in detected! Halting active TTS playback stream.")
                    interrupt_event.set()
                elif event.get("event") == "media":
                    # Pass raw base64 PCM audio chunk to Deepgram STT stream
                    pass
        except WebSocketDisconnect:
            print(f"Call disconnected: {call_id}")

    async def stream_agent_voice_response(user_transcript: str):
        """Streams LLM tokens into Cartesia Sonic TTS and sends audio frames to caller."""
        interrupt_event.clear()
        
        # 1. Stream tokens from internal vLLM model endpoint
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST", 
                "http://vllm-voice-node.internal:8000/v1/chat/completions",
                json={
                    "model": "Meta-Llama-3.1-8B-Instruct",
                    "messages": [{"role": "user", "content": user_transcript}],
                    "stream": True
                }
            ) as response:
                async for chunk in response.aiter_text():
                    if interrupt_event.is_set():
                        print("Execution interrupted mid-sentence.")
                        break
                    
                    # Convert token chunk to Cartesia Sonic TTS audio bytes & send
                    # await websocket.send_json({"event": "media", "media": {"payload": base64_pcm}})

    # Spawn concurrent receive and dialogue tasks
    receive_task = asyncio.create_task(receive_telephony_audio())
    try:
        await receive_task
    except Exception as err:
        print(f"Error handling voice websocket: {str(err)}")