voice-ai-development

name	voice-ai-development
description	Designs low-latency voice agents with OpenAI Realtime, Vapi, Deepgram STT, ElevenLabs TTS, and LiveKit. Use when the user mentions voice AI, speech-to-text, text-to-speech, or realtime voice. Not for text-only chat UIs (vercel-ai-sdk-expert) or local ComfyUI audio. Do not hardcode live API keys in examples.
version	1.0.1

When to Use

• User mentions or implies: voice ai, voice agent, speech to text, text to speech, realtime voice, vapi, deepgram, elevenlabs, livekit, openai realtime.
• Building low-latency, production-ready voice experiences.
• Need to choose the right combination of providers for each use case and optimize for perceived responsiveness.

Prerequisites

• Async programming (Python or Node.js)
• WebSocket basics
• Audio concepts (sample rate, codec)
• API keys for providers (OpenAI, Vapi, Deepgram, ElevenLabs, LiveKit)
• Audio handling knowledge

Procedure

1. OpenAI Realtime API (Native voice-to-voice with GPT-4o)

When to use: When you want integrated voice AI without separate STT/TTS.

import asyncio
import websockets
import json
import base64

OPENAI_API_KEY = "sk-..."

async def voice_session():
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(url, extra_headers=headers) as ws:
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": "alloy",  # alloy, echo, fable, onyx, nova, shimmer
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {
                    "model": "whisper-1"
                },
                "turn_detection": {
                    "type": "server_vad",  # Voice activity detection
                    "threshold": 0.5,
                    "prefix_padding_ms": 300,
                    "silence_duration_ms": 500
                },
                "tools": [
                    {
                        "type": "function",
                        "name": "get_weather",
                        "description": "Get weather for a location",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "location": {"type": "string"}
                            }
                        }
                    }
                ]
            }
        }))

        # Send audio (PCM16, 24kHz, mono)
        async def send_audio(audio_bytes):
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio_bytes).decode()
            }))

        # Receive events
        async for message in ws:
            event = json.loads(message)

            if event["type"] == "response.audio.delta":
                # Play audio chunk
                audio = base64.b64decode(event["delta"])
                play_audio(audio)

            elif event["type"] == "response.audio_transcript.done":
                print(f"Assistant said: {event['transcript']}")

            elif event["type"] == "input_audio_buffer.speech_started":
                print("User started speaking")

            elif event["type"] == "response.function_call_arguments.done":
                # Handle tool call
                name = event["name"]
                args = json.loads(event["arguments"])
                result = call_function(name, args)
                await ws.send(json.dumps({
                    "type": "conversation.item.create",
                    "item": {
                        "type": "function_call_output",
                        "call_id": event["call_id"],
                        "output": json.dumps(result)
                    }
                }))

2. Vapi Voice Agent (Phone-based agents, quick deployment)

When to use: Phone-based agents, quick deployment.

# Vapi provides hosted voice agents with webhooks
from flask import Flask, request, jsonify
import vapi

app = Flask(__name__)
client = vapi.Vapi(api_key="...")

# Create an assistant
assistant = client.assistants.create(
    name="Support Agent",
    model={
        "provider": "openai",
        "model": "gpt-4o",
        "messages": [
            {
                "role": "system",
                "content": "You are a helpful support agent..."
            }
        ]
    },
    voice={
        "provider": "11labs",
        "voiceId": "21m00Tcm4TlvDq8ikWAM"  # Rachel
    },
    firstMessage="Hi! How can I help you today?",
    transcriber={
        "provider": "deepgram",
        "model": "nova-2"
    }
)

# Webhook for conversation events
@app.route("/vapi/webhook", methods=["POST"])
def vapi_webhook():
    event = request.json

    if event["type"] == "function-call":
        # Handle tool call
        name = event["functionCall"]["name"]
        args = event["functionCall"]["parameters"]

        if name == "check_order":
            result = check_order(args["order_id"])
             jsonify({: result})

     event[] == :
        
        transcript = event[]
        save_transcript(event[][], transcript)

     jsonify({: })


call = client.calls.create(
    assistant_id=assistant.,
    customer={
        : 
    },
    phoneNumber={
        : 
    }
)


web_call = client.calls.create(
    assistant_id=assistant.,
    =
)

3. Deepgram STT + ElevenLabs TTS (Best-in-class transcription and synthesis)

When to use: High quality voice, custom pipeline.

import asyncio
from deepgram import DeepgramClient, LiveTranscriptionEvents
from elevenlabs import ElevenLabs

# Deepgram real-time transcription
deepgram = DeepgramClient(api_key="...")

async def transcribe_stream(audio_stream):
    connection = deepgram.listen.live.v("1")

    async def on_transcript(result):
        transcript = result.channel.alternatives[0].transcript
        if transcript:
            print(f"Heard: {transcript}")
            if result.is_final:
                # Process final transcript
                await handle_user_input(transcript)

    connection.on(LiveTranscriptionEvents.Transcript, on_transcript)

    await connection.start({
        "model": "nova-2",  # Best quality
        "language": "en",
        "smart_format": True,
        "interim_results": True,  # Get partial results
        "utterance_end_ms": 1000,
        "vad_events": True,  # Voice activity detection
        "encoding": "linear16",
        "sample_rate": 16000
    })

    # Stream audio
      chunk  audio_stream:
         connection.send(chunk)

     connection.finish()


eleven = ElevenLabs(api_key=)

 ():
    
    audio_stream = eleven.text_to_speech.convert_as_stream(
        voice_id=,  
        model_id=,  
        text=text,
        output_format=  
    )

     chunk  audio_stream:
         chunk


  ():
      eleven.text_to_speech.stream_async(
        voice_id=,
        model_id=
    )  tts:
          text_chunk  text_stream:
            audio =  tts.send(text_chunk)
             audio

        
        final_audio =  tts.flush()
         final_audio

4. LiveKit Real-time Infrastructure (WebRTC infrastructure for voice apps)

When to use: Building custom real-time voice apps.

from livekit import api, rtc
import asyncio

# Server-side: Create room and tokens
lk_api = api.LiveKitAPI(
    url="wss://your-livekit.livekit.cloud",
    api_key="...",
    api_secret="..."
)

async def create_room(room_name: str):
    room = await lk_api.room.create_room(
        api.CreateRoomRequest(name=room_name)
    )
    return room

def create_token(room_name: str, participant_name: str):
    token = api.AccessToken(
        api_key="...",
        api_secret="..."
    )
    token.with_identity(participant_name)
    token.with_grants(api.VideoGrants(
        room_join=True,
        room=room_name
    ))
    return token.to_jwt()

# Agent-side: Connect and process audio
async def voice_agent(room_name: str):
    room = rtc.Room()

    @room.on("track_subscribed")
    def on_track(track, publication, participant):
        if track.kind == rtc.TrackKind.KIND_AUDIO:
            # Process incoming audio
            audio_stream = rtc.AudioStream(track)
            asyncio.create_task(process_audio(audio_stream))

    token = create_token(room_name, "agent")
    await room.connect("wss://your-livekit.livekit.cloud", token)

    # Publish agent's audio
    source = rtc.AudioSource(sample_rate=, num_channels=)
    track = rtc.LocalAudioTrack.create_audio_track(, source)
     room.local_participant.publish_track(track)

    
      ():
         audio_chunk  text_to_speech(text):
             source.capture_frame(rtc.AudioFrame(
                data=audio_chunk,
                sample_rate=,
                num_channels=,
                samples_per_channel=(audio_chunk) // 
            ))

     room, speak


  ():
      frame  audio_stream:
        
         transcriber.send(frame.data)

5. Full Voice Agent Pipeline (Complete voice agent with all components)

When to use: Custom production voice agent.

import asyncio
from dataclasses import dataclass
from typing import AsyncIterator

@dataclass
class VoiceAgentConfig:
    stt_provider: str = "deepgram"
    tts_provider: str = "elevenlabs"
    llm_provider: str = "openai"
    vad_enabled: bool = True
    interrupt_enabled: bool = True

class VoiceAgent:
    def __init__(self, config: VoiceAgentConfig):
        self.config = config
        self.is_speaking = False
        self.conversation_history = []

    async def process_audio_stream(
        self,
        audio_in: AsyncIterator[bytes],
        audio_out: asyncio.Queue
    ):
        """Main audio processing loop."""

        # STT streaming
        async def transcribe():
            transcript_buffer = ""
            async for audio_chunk in audio_in:
                # Check for interruption
                if self.is_speaking and self.config.interrupt_enabled:
                      .detect_speech(audio_chunk):
                         .stop_speaking()

                result =  .stt.transcribe(audio_chunk)
                 result.is_final:
                     result.transcript

        
          user_text  transcribe():
              user_text.strip():
                

            .conversation_history.append({
                : ,
                : user_text
            })

            
            .is_speaking = 
              audio_chunk  .generate_response(user_text):
                 audio_out.put(audio_chunk)
            .is_speaking = 

      () -> AsyncIterator[]:
        

        
        llm_stream = .llm.stream_chat(.conversation_history)

        
        text_buffer = 
        full_response = 

          token  llm_stream:
            text_buffer += token
            full_response += token

            
             (text_buffer) >   token  :
                  audio  .tts.synthesize_stream(text_buffer):
                     audio
                text_buffer = 

        
         text_buffer:
              audio  .tts.synthesize_stream(text_buffer):
                 audio

        .conversation_history.append({
            : ,
            : full_response
        })

      () -> :
        
        
         .vad.is_speech(audio)

      ():
        
        .is_speaking = 
        
        

Pitfalls

• Non-Streaming TTS (HIGH): Non-streaming TTS adds significant latency. Fix: Use tts.synthesize_stream() or tts.convert_as_stream().
• Hardcoded Sample Rate (MEDIUM): Hardcoded sample rate may cause format mismatches. Fix: Define sample rates as constants, document expected formats.
• WebSocket Without Reconnection (HIGH): WebSocket connections need reconnection logic. Fix: Add retry loop with exponential backoff.
• Missing VAD Configuration (MEDIUM): VAD needs tuning for good user experience. Fix: Configure threshold and silence_duration_ms.
• Blocking Audio Processing (HIGH): Audio processing should be async to avoid blocking. Fix: Use async def and await for audio operations.
• Missing Interruption Handling (MEDIUM): Voice agents should handle user interruptions. Fix: Add barge-in detection and cancel current response.
• Audio Queue Without Clear (LOW): Audio queues should be clearable for interruptions. Fix: Add method to clear queue on interruption.
• WebSocket Without Error Handling (HIGH): WebSocket operations need error handling. Fix: Wrap in try/except for ConnectionClosed.

Verification

• Check WebSocket Connection: Ensure the WebSocket connection to the provider (e.g., OpenAI Realtime API) is established successfully. Look for session.update acceptance.
• Audio Stream Test: Verify that audio chunks are being received and decoded properly. Check response.audio.delta events.
• VAD Configuration: Confirm VAD events (input_audio_buffer.speech_started) are firing when expected.
• Tool Call Execution: Verify that function calls are received, executed, and results are sent back correctly.

Related Skills

• langgraph: Need complex agent logic behind voice.
• structured-output: Need to extract structured data from voice.
• langfuse: Need to monitor voice agent quality.
• twilio: Connect to Twilio for PSTN.
• nextjs-app-router: Need web interface for voice agent.