Voice AI Agents: Building Real-Time Conversational AI

import asyncio
from typing import AsyncIterator

class SpeechToText:
    def __init__(self, provider: str = "deepgram"):
        self.provider = self._init_provider(provider)
        
    async def transcribe_stream(self, audio_stream: AsyncIterator[bytes]) -> AsyncIterator[str]:
        """Stream transcription with low latency"""
        async for chunk in self.provider.stream(audio_stream):
            if chunk.is_final:
                yield chunk.text
            else:
                # Interim results for real-time feedback
                yield f"[interim] {chunk.text}"

# Usage with streaming audio
async def handle_audio(audio_chunk: bytes):
    stt = SpeechToText(provider="deepgram")
    
    async for text in stt.transcribe_stream(audio_stream):
        if text.startswith("[interim]"):
            # Show interim (faster, less accurate)
            show_interim(text)
        else:
            # Final transcription
            await process_final(text)

class TextToSpeech:
    def __init__(self, provider: str = "elevenlabs"):
        self.provider = self._init_provider(provider)
        self.voice_id = "custom_voice_id"
        
    async def speak(self, text: str, stream: bool = True) -> AsyncIterator[bytes]:
        """Generate speech, optionally streaming"""
        if stream:
            async for audio_chunk in self.provider.stream(
                text=text,
                voice_id=self.voice_id,
                model="eleven_multilingual_v2"
            ):
                yield audio_chunk
        else:
            # Wait for full generation
            audio = await self.provider.generate(
                text=text,
                voice_id=self.voice_id
            )
            yield audio

# Usage
tts = TextToSpeech()
async for chunk in tts.speak("Hello! How can I help you today?"):
    play_audio(chunk)  # Stream to speaker immediately

class VoiceActivityDetector:
    def __init__(self):
        self.model = SileroVAD()
        
    async def detect_speech(self, audio_chunk: bytes) -> bool:
        """Detect if audio contains speech"""
        # Convert to proper format
        waveform = self.convert_audio(audio_chunk)
        
        # Get speech probabilities
        probabilities = await self.model.predict(waveform)
        
        # Return True if speech detected
        return max(probabilities) > 0.5
    
    async def wait_for_speech(self, audio_stream: AsyncIterator[bytes]) -> bytes:
        """Wait until speech is detected, then return audio"""
        buffer = []
        
        async for chunk in audio_stream:
            buffer.append(chunk)
            
            if await self.detect_speech(b"".join(buffer)):
                return b"".join(buffer)
            
            # Limit buffer size
            if len(buffer) > 100:
                buffer = buffer[-10:]

class VoiceAIAgent:
    def __init__(self, config: VoiceConfig):
        self.stt = SpeechToText(config.stt_provider)
        self.llm = LLM(config.llm_provider)
        self.tts = TextToSpeech(config.tts_provider)
        self.vad = VoiceActivityDetector()
        self.pipeline = AudioPipeline(config)
        
    async def handle_call(self, audio_stream: AsyncIterator[bytes]):
        """Main voice agent loop"""
        
        # Start with greeting
        await self.play_greeting()
        
        while True:
            # 1. Wait for user to start speaking
            user_audio = await self.vad.wait_for_speech(audio_stream)
            
            # 2. Transcribe speech to text
            user_text = ""
            async for transcript in self.stt.transcribe_stream(user_audio):
                if not transcript.startswith("[interim]"):
                    user_text = transcript
                    
            if not user_text:
                continue
                
            # 3. Check for exit conditions
            if self.is_exit(user_text):
                await self.say_goodbye()
                break
                
            # 4. Process with LLM
            response = await self.llm.chat(
                message=user_text,
                context=self.conversation_context
            )
            
            # 5. Generate and play response
            async for audio_chunk in self.tts.speak(response.text):
                await self.play_audio(audio_chunk)
                
            # 6. Update context
            self.conversation_context.add(user_text, response.text)
    
    async def play_greeting(self):
        greeting = "Hello! I'm your AI assistant. How can I help you today?"
        async for chunk in self.tts.speak(greeting):
            await self.play_audio(chunk)

class LowLatencyPipeline:
    """Optimized pipeline for minimal latency"""
    
    def __init__(self):
        self.buffer_ms = 100  # Buffer before processing
        self.chunk_size_ms = 100  # Audio chunk size
        
    async def process(self, audio: bytes) -> bytes:
        """
        Target latency breakdown:
        - Audio buffer: 100ms
        - STT: 150ms
        - LLM: 300ms (with caching)
        - TTS: 200ms (with prefetching)
        - Total: ~750ms (perceptible as real-time)
        """
        
        # Prefetch next TTS while current plays
        # Use streaming throughout
        # Cache common responses
        pass

# Deepgram STT
from deepgram import Deepgram

dg_client = Deepgram("YOUR_API_KEY")

async def transcribe_deepgram(audio_file: str):
    response = await dg_client.transcription.prerecorded(
        {"url": audio_file},
        {"punctuate": True, "utterances": True}
    )
    return response["results"]["channels"][0]["alternatives"][0]["transcript"]

# ElevenLabs TTS
import elevenlabs

async def speak_elevenlabs(text: str):
    audio = elevenlabs.generate(
        text=text,
        voice="Rachel",
        model="eleven_multilingual_v2"
    )
    return audio

# Vapi (Voice Agent Platform)
from vapi import Vapi

vapi = Vapi(token="YOUR_TOKEN")

# Start a call
call = await vapi.calls.create(
    assistant={
        "model": {"provider": "openai", "model": "gpt-4o"},
        "voice": {"provider": "elevenlabs", "voice_id": "Rachel"}
    },
    customer={"number": "+1234567890"}
)

class VoiceCloner:
    def __init__(self):
        self.elevenlabs = ElevenLabs()
        
    async def clone_voice(self, audio_samples: list[str], name: str):
        """Clone a voice from audio samples"""
        
        # Upload samples
        response = await self.elevenlabs.voices.add(
            name=name,
            files=audio_samples,
            description="Custom voice clone"
        )
        
        return response["voice_id"]
    
    async def generate_with_cloned_voice(self, text: str, voice_id: str):
        """Generate speech with cloned voice"""
        
        audio = await self.elevenlabs.generate(
            text=text,
            voice_id=voice_id
        )
        
        return audio

class EmotionalVoiceAI:
    def __init__(self):
        self.emotion_detector = EmotionDetector()
        self.emotional_tts = EmotionalTTS()
        
    async def detect_emotion(self, audio: bytes) -> str:
        """Detect emotion from user voice"""
        
        # Analyze audio for emotional markers
        emotion = await self.emotion_detector.analyze(audio)
        
        # Returns: "happy", "sad", "angry", "neutral", "surprised"
        return emotion
    
    async def respond_emotionally(self, text: str, user_emotion: str) -> bytes:
        """Generate emotionally appropriate response"""
        
        # Determine appropriate response emotion
        response_emotion = self.map_emotion(user_emotion)
        
        # Generate with emotion
        audio = await self.emotional_tts.speak(
            text=text,
            emotion=response_emotion,
            intensity=0.7
        )
        
        return audio
    
    def map_emotion(self, user_emotion: str) -> str:
        mapping = {
            "happy": "happy",
            "sad": "sympathetic",
            "angry": "calm",
            "neutral": "friendly",
            "surprised": "enthusiastic"
        }
        return mapping.get(user_emotion, "neutral")

class MultilingualVoiceAgent:
    def __init__(self):
        self.stt = MultilingualSTT()
        self.llm = MultilingualLLM()
        self.tts = MultilingualTTS()
        
    async def detect_language(self, audio: bytes) -> str:
        """Auto-detect spoken language"""
        return await self.stt.detect_language(audio)
    
    async def handle(self, audio: bytes):
        # Detect language
        lang = await self.detect_language(audio)
        
        # Transcribe in correct language
        text = await self.stt.transcribe(audio, language=lang)
        
        # Process in same language
        response = await self.llm.chat(text, language=lang)
        
        # Generate speech in same language
        audio_response = await self.tts.speak(response, language=lang)
        
        return audio_response

# AI customer service voice agent
async def customer_service_agent(call):
    agent = VoiceAIAgent(config)
    
    # Greet
    await agent.say("Thank you for calling. How can I help you today?")
    
    while True:
        # Get customer request
        audio = await agent.listen()
        request = await agent.transcribe(audio)
        
        # Classify intent
        intent = await classify_intent(request)
        
        if intent == "billing":
            response = await handle_billing(request)
        elif intent == "technical":
            response = await handle_technical(request)
        elif intent == "speak_to_human":
            await transfer_to_human(request)
            break
        else:
            response = await handle_general(request)
        
        # Respond
        await agent.respond(response)

# Appointment scheduling voice agent
async def book_appointment():
    agent = VoiceAIAgent(config)
    
    # Get appointment details
    await agent.respond("I'd be happy to help you book an appointment.")
    
    # Date
    await agent.respond("What date works for you?")
    date = await agent.get_date()
    
    # Time
    await agent.respond(f"Great, {date}. What time would you prefer?")
    time = await agent.get_time()
    
    # Service
    await agent.respond("What type of appointment?")
    service = await agent.get_service()
    
    # Confirm
    await agent.respond(
        f"Booking {service} on {date} at {time}. "
        "Shall I confirm this appointment?"
    )
    confirmed = await agent.confirm()
    
    if confirmed:
        await booking_system.create(date, time, service)
        await agent.respond("Your appointment is confirmed!")

# Modern AI-powered IVR
class AIVR:
    def __init__(self):
        self.agent = VoiceAIAgent(config)
        
    async def handle_call(self, audio_stream):
        # Natural language understanding - no menu trees!
        await self.agent.respond(
            "Hi, thanks for calling. How can I direct your call?"
        )
        
        audio = await self.agent.listen()
        request = await self.agent.transcribe(audio)
        
        # Understand intent naturally
        intent = await self.nlu.classify(request)
        
        # Route appropriately
        if intent.needs_human:
            await self.transfer(intent.department, intent.reason)
        else:
            await self.handle_automated(intent)

# Vapi - Voice AI platform
from vapi import Vapi

vapi = Vapi(token="YOUR_TOKEN")

# Create assistant
assistant = await vapi.assistants.create(
    name="Customer Service",
    model={
        "provider": "openai",
        "model": "gpt-4o",
        "system_prompt": "You are a helpful customer service agent."
    },
    voice={
        "provider": "elevenlabs",
        "voice_id": "Rachel"
    }
)

# Start outbound call
call = await vapi.calls.create(
    assistant_id=assistant.id,
    customer={"number": "+1234567890"}
)

# Bland AI - High-volume voice calls
import bland

client = bland.Client(api_key="YOUR_KEY")

# Create campaign
campaign = await client.campaigns.create(
    name="Appointment Reminder",
    llm_prompt="Call this number and remind about appointment...",
    voice_id="friendly_female"
)

# Make calls
await campaign.start(phone_numbers=["+1234567890"])

# Retell - Conversational voice AI
import retell

client = retell.Client(api_key="YOUR_KEY")

agent = await client.agents.create(
    name="Sales Agent",
    prompt="You are a sales representative...",
    voice="Mark"
)

# Register webhook for call events
webhook = await client.webhooks.register(
    url="https://your-server.com/webhook",
    events=["call_started", "call_ended", "transfer"]
)

# Good: Stream everything for minimum latency
async def handle_voice_input(audio_chunk):
    # Stream directly, don't wait for full audio
    async for text in stt.stream_transcribe(audio_chunk):
        if is_final(text):
            # Process immediately
            result = await llm.process(text)
            # Stream response back
            async for audio in tts.stream_speak(result):
                play_audio(audio)

# Bad: Wait for full transcription before processing
async def handle_voice_input(audio):
    # Wait for complete audio (BAD!)
    full_audio = await wait_for_complete_audio()
    
    # Then transcribe (slower)
    text = await stt.transcribe(full_audio)
    
    # Then process (even slower)
    result = await llm.process(text)
    
    # Then generate speech (slowest)
    audio = await tts.speak(result)
    play_audio(audio)
    # Total: 3-5 seconds (feels unnatural)

class InterruptHandler:
    def __init__(self):
        self.vad = VoiceActivityDetector()
        
    async def monitor_for_interrupt(self, audio_stream, current_response):
        """Watch for user interruption while AI is speaking"""
        
        while True:
            chunk = await audio_stream.receive()
            
            # Check if user started speaking
            if await self.vad.detect_speech(chunk):
                # User interrupted - stop current response
                await self.stop_speaking()
                
                # Return control to user
                return True
            
            # Check if response finished
            if await self.response_finished():
                return False
                
            await asyncio.sleep(0.1)

class CostOptimizer:
    def __init__(self):
        self.costs = {
            "stt": {"deepgram": "$0.004/min", "assemblyai": "$0.025/min"},
            "tts": {"elevenlabs": "$0.18/1k chars", "cartesia": "$0.03/min"}
        }
        
    def optimize(self, call_duration_minutes: float) -> dict:
        """Calculate optimal provider combination"""
        
        stt_cost = self.costs["stt"]["deepgram"] * call_duration_minutes
        tts_cost = self.costs["tts"]["cartesia"] * call_duration_minutes
        
        return {
            "stt_provider": "deepgram",
            "tts_provider": "cartesia",
            "total_cost_per_call": stt_cost + tts_cost
        }