Voice AI Development:Inbound Call Agents Built for Production

# voice_agent.py — FastAPI WebSocket voice handler
import asyncio
from fastapi import WebSocket
import deepgram, openai, elevenlabs

async def voice_session(ws: WebSocket, session_id: str):
    await ws.accept()
    history = []  # swap for Redis at scale

    async for audio_chunk in ws.iter_bytes():
        # 1. Speech → Text (Deepgram nova-2)
        transcript = await deepgram.transcribe(
            audio_chunk, model="nova-2"
        )
        if not transcript.text or transcript.confidence < 0.65:
            continue  # drop low-confidence fragments

        # 2. LLM response (OpenAI, rolling 8-turn window)
        history.append({"role": "user", "content": transcript.text})
        response = await openai.chat.completions.create(
            model="gpt-4o",
            messages=[SYSTEM_PROMPT, *history[-8:]],
            max_tokens=120,
        )
        reply = response.choices[0].message.content
        history.append({"role": "assistant", "content": reply})

        # 3. Text → Speech (ElevenLabs Flash, streamed)
        audio_response = await elevenlabs.generate(
            text=reply,
            voice_id=VOICE_ID,
            model_id="eleven_flash_v2_5",
            stream=True,
        )
        async for chunk in audio_response:
            await ws.send_bytes(chunk)