AI Voice Agents Complete Guide 2026: Building Conversational AI Systems

import asyncio
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncIterator, Optional, List, Dict
import numpy as np

@dataclass
class TranscriptionResult:
    """Result from speech recognition."""
    text: str
    confidence: float
    is_final: bool
    start_time: float
    end_time: float
    words: List[Dict] = None
    
    @property
    def duration(self) -> float:
        return self.end_time - self.start_time

class ASREngine(ABC):
    """Abstract base class for ASR engines."""
    
    @abstractmethod
    async def initialize(self) -> None:
        """Initialize the ASR engine."""
        pass
    
    @abstractmethod
    async def recognize(
        self, 
        audio_chunk: bytes,
        sample_rate: int = 16000
    ) -> TranscriptionResult:
        """Recognize speech from audio chunk."""
        pass
    
    @abstractmethod
    async def recognize_streaming(
        self, 
        audio_stream: AsyncIterator[bytes]
    ) -> AsyncIterator[TranscriptionResult]:
        """Process streaming audio."""
        pass

class WhisperASR(ASREngine):
    """OpenAI Whisper-based ASR implementation."""
    
    def __init__(
        self,
        model_name: str = "base",
        language: str = "en",
        device: str = "cuda"
    ):
        self.model_name = model_name
        self.language = language
        self.device = device
        self.model = None
        self.processor = None
    
    async def initialize(self) -> None:
        """Load Whisper model."""
        # In production, load model in executor to avoid blocking
        loop = asyncio.get_event_loop()
        
        import whisper
        self.model = await loop.run_in_executor(
            None,
            lambda: whisper.load_model(self.model_name, device=self.device)
        )
        
        # Load processor
        self.processor = whisper.WhisperProcessor.from_pretrained(
            f"openai/whisper-{self.model_name}"
        )
    
    async def recognize(
        self, 
        audio_chunk: bytes,
        sample_rate: int = 16000
    ) -> TranscriptionResult:
        """Recognize speech from audio chunk."""
        
        # Convert bytes to numpy array
        audio_np = np.frombuffer(audio_chunk, dtype=np.float32)
        
        # Run recognition in executor
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(
            None,
            lambda: self.model.transcribe(
                audio_np,
                language=self.language,
                fp16=self.device == "cuda"
            )
        )
        
        return TranscriptionResult(
            text=result["text"].strip(),
            confidence=result.get("avg_logprob", -1.0),
            is_final=True,
            start_time=0.0,
            end_time=result.get("duration", 0.0),
            words=result.get("words", [])
        )
    
    async def recognize_streaming(
        self, 
        audio_stream: AsyncIterator[bytes]
    ) -> AsyncIterator[TranscriptionResult]:
        """Process streaming audio with VAD."""
        
        buffer = []
        vad = VoiceActivityDetector()
        
        async for audio_chunk in audio_stream:
            # Add to buffer
            buffer.append(audio_chunk)
            
            # Check for speech activity
            if vad.is_speaking(audio_chunk):
                # Continue accumulating
                continue
            
            # Silence detected - process accumulated audio
            if buffer:
                combined_audio = b''.join(buffer)
                
                # Process with 5-second window limit
                if len(combined_audio) > 5 * 16000 * 4:  # 5 seconds
                    result = await self.recognize(combined_audio)
                    yield result
                
                buffer = []

class StreamingASRWithInterim:
    """ASR with interim results for real-time feedback."""
    
    def __init__(self, asr_engine: ASREngine):
        self.asr_engine = asr_engine
        self.audio_buffer = []
        self.interim_threshold = 0.5  # seconds of silence
    
    async def process_audio(
        self,
        audio_chunk: bytes,
        sample_rate: int = 16000
    ) -> List[TranscriptionResult]:
        """Process audio and return both interim and final results."""
        
        results = []
        self.audio_buffer.append(audio_chunk)
        
        # Check for silence to determine if utterance is complete
        if self._is_silent(audio_chunk, sample_rate):
            if self.audio_buffer:
                # Process complete utterance
                full_audio = b''.join(self.audio_buffer)
                final_result = await self.asr_engine.recognize(full_audio, sample_rate)
                results.append(final_result)
                self.audio_buffer = []
        else:
            # Generate interim result while speaking
            current_audio = b''.join(self.audio_buffer)
            
            # Only generate interim results periodically to avoid overhead
            if len(self.audio_buffer) % 5 == 0:
                interim_result = await self.asr_engine.recognize(current_audio, sample_rate)
                interim_result.is_final = False
                results.append(interim_result)
        
        return results
    
    def _is_silent(self, audio_chunk: bytes, sample_rate: int) -> bool:
        """Detect if audio chunk is silence."""
        audio_np = np.frombuffer(audio_chunk, dtype=np.float32)
        rms = np.sqrt(np.mean(audio_np ** 2))
        return rms < 0.01  # Threshold for silence

class VoiceInputNormalizer:
    """Normalizes speech input for better NLU processing."""
    
    def __init__(self):
        self.common_corrections = {
            "um            "uh":": "",
 "",
            "like": "",  # filler words
            "you know": "",
            "actually": "",
            "basically": "",
            "literally": ""
        }
    
    def normalize(self, text: str) -> str:
        """Normalize speech transcript."""
        
        # Remove filler words
        for filler, replacement in self.common_corrections.items():
            text = text.replace(filler, replacement)
        
        # Fix common speech-to-text errors
        text = self._fix_common_errors(text)
        
        # Add punctuation based on context
        text = self._add_punctuation(text)
        
        # Clean up whitespace
        text = ' '.join(text.split())
        
        return text
    
    def _fix_common_errors(self, text: str) -> str:
        """Fix common ASR transcription errors."""
        
        corrections = {
            "to too": "to",
            "two too": "to",
            "their there": "there",
            "your you're": "you're",
            "its it's": "it's"
        }
        
        for wrong, correct in corrections.items():
            parts = wrong.split()
            if len(parts) == 2:
                # Replace pattern only if it appears as separate words
                text = text.replace(wrong, correct)
        
        return text
    
    def _add_punctuation(self, text: str) -> str:
        """Add punctuation to unpunctuated speech text."""
        
        # Simple heuristics for punctuation
        if not text.endswith(('.', '?', '!')):
            text += '.'
        
        # Add question marks for question patterns
        question_words = ['what', 'how', 'why', 'when', 'where', 'who', 'which']
        if any(text.lower().startswith(q) for q in question_words):
            text = text.rstrip('.') + '?'
        
        return text

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from enum import Enum

class ConversationMode(Enum):
    """Voice conversation modes."""
    COMMAND = "command"  # Direct commands
    NAVIGATION = "navigation"  # Menu traversal
    INFORMATION = "information"  # Q&A
    TRANSACTION = "transaction"  # Multi-step transactions

@dataclass
class TurnContext:
    """Context for a conversation turn."""
    session_id: str
    user_id: str
    conversation_mode: ConversationMode = ConversationMode.INFORMATION
    current_intent: Optional[str] = None
    entities: Dict[str, Any] = field(default_factory=dict)
    slot_values: Dict[str, str] = field(default_factory=dict)
    sentiment: str = "neutral"
    confidence: float = 1.0
    raw_text: str = ""
    asr_confidence: float = 1.0

class VoiceNLU:
    """NLU optimized for voice input."""
    
    def __init__(self, intent_classifier, entity_extractor):
        self.intent_classifier = intent_classifier
        self.entity_extractor = entity_extractor
        self.context_history: Dict[str, List[TurnContext]] = {}
        self.max_history = 10
    
    async def process_input(
        self,
        text: str,
        context: TurnContext,
        asr_confidence: float = 1.0
    ) -> TurnContext:
        """Process voice input and update context."""
        
        # Normalize speech input
        normalizer = VoiceInputNormalizer()
        normalized_text = normalizer.normalize(text)
        
        # Update raw text
        context.raw_text = text
        context.asr_confidence = asr_confidence
        
        # Low ASR confidence - may need confirmation
        if asr_confidence < 0.7:
            context.confidence = 0.5
            context.entities["_low_confidence"] = True
        
        # Classify intent
        intent_result = await self.intent_classifier.classify(
            normalized_text,
            context=context
        )
        context.current_intent = intent_result.intent
        context.confidence *= intent_result.confidence
        
        # Extract entities
        entities = await self.entity_extractor.extract(
            normalized_text,
            context=context
        )
        
        # Merge entities with context
        context.entities.update(entities)
        
        # Resolve pronouns and references
        context = await self._resolve_references(context, normalized_text)
        
        # Update sentiment
        context.sentiment = await self._detect_sentiment(normalized_text)
        
        # Store in history
        self._add_to_history(context.session_id, context)
        
        return context
    
    async def _resolve_references(
        self, 
        context: TurnContext,
        text: str
    ) -> TurnContext:
        """Resolve pronouns and implicit references."""
        
        # Check for implicit references
        if any(word in text.lower() for word in ['it', 'that', 'this', 'there']):
            # Try to resolve from previous context
            history = self.context_history.get(context.session_id, [])
            
            if history and history[-1].entities:
                # Copy relevant entities from previous turn
                last_entities = history[-1].entities
                
                # Propagate entity if mentioned implicitly
                if 'location' in last_entities and 'location' not in context.entities:
                    context.entities['location'] = last_entities['location']
        
        # Resolve time references
        context = await self._resolve_time_references(context, text)
        
        return context
    
    async def _resolve_time_references(
        self,
        context: TurnContext,
        text: str
    ) -> TurnContext:
        """Resolve relative time references."""
        
        import re
        from datetime import datetime, timedelta
        
        text_lower = text.lower()
        
        # Simple relative time patterns
        time_patterns = {
            r'\btoday\b': 0,
            r'\btomorrow\b': 1,
            r'\bnext week\b': 7,
            r'\bnext month\b': 30,
        }
        
        for pattern, days in time_patterns.items():
            if re.search(pattern, text_lower):
                target_date = datetime.now() + timedelta(days=days)
                context.entities['resolved_date'] = target_date.isoformat()
                break
        
        return context
    
    def _add_to_history(self, session_id: str, context: TurnContext) -> None:
        """Add turn to conversation history."""
        
        if session_id not in self.context_history:
            self.context_history[session_id] = []
        
        self.context_history[session_id].append(context)
        
        # Limit history size
        if len(self.context_history[session_id]) > self.max_history:
            self.context_history[session_id] = \
                self.context_history[session_id][-self.max_history:]

from enum import Enum
from typing import Callable, Dict, List, Optional, Any
import asyncio

class DialogueAct(Enum):
    """Speech acts in conversation."""
    GREETING = "greeting"
    QUESTION = "question"
    ANSWER = "answer"
    CONFIRMATION = "confirmation"
    REJECTION = "rejection"
    COMMAND = "command"
    APOLOGY = "apology"
    CLOSING = "closing"

@dataclass
class DialogueState:
    """Current state of the dialogue."""
    session_id: str
    current_node: str
    collected_slots: Dict[str, Any] = field(default_factory=dict)
    required_slots: List[str] = field(default_factory=list)
    confirmed_slots: Dict[str, bool] = field(default_factory=dict)
    intent: Optional[str] = None
    topic: str = "general"
    subtopic: Optional[str] = None
    is Escalated: bool = False
    human_handoff: bool = False
    conversation_turns: int = 0

class DialogueNode:
    """A node in the dialogue flow."""
    
    def __init__(
        self,
        node_id: str,
        prompts: List[str],
        expected_slots: List[str] = None,
        intents: List[str] = None,
        next_node_map: Dict[str, str] = None,
        actions: List[Callable] = None,
        condition: Callable = None
    ):
        self.node_id = node_id
        self.prompts = prompts
        self.expected_slots = expected_slots or []
        self.intents = intents or []
        self.next_node_map = next_node_map or {}
        self.actions = actions or []
        self.condition = condition
    
    def get_prompt(self, context: TurnContext) -> str:
        """Get appropriate prompt for context."""
        # Simple rotation through prompts
        turn = context.conversation_turns % len(self.prompts)
        return self.prompts[turn]

class DialogueManager:
    """Manages dialogue flow and state."""
    
    def __init__(self, nlu: VoiceNLU, tts: "TTSEngine"):
        self.nlu = nlu
        self.tts = tts
        self.dialogue_flows: Dict[str, DialogueFlow] = {}
        self.active_sessions: Dict[str, DialogueState] = {}
        self.default_flow = "general"
    
    async def process_turn(
        self,
        session_id: str,
        user_input: str,
        asr_confidence: float = 1.0
    ) -> Dict[str, Any]:
        """Process a conversation turn."""
        
        # Get or create session state
        state = self._get_session_state(session_id)
        
        # Get current node
        flow = self.dialogue_flows.get(
            state.topic, 
            self.dialogue_flows[self.default_flow]
        )
        node = flow.get_node(state.current_node)
        
        # Process NLU
        context = TurnContext(
            session_id=session_id,
            user_id="unknown",  # Would come from auth
            raw_text=user_input
        )
        
        context = await self.nlu.process_input(
            user_input, 
            context, 
            asr_confidence
        )
        
        # Update state
        state.conversation_turns += 1
        state.current_intent = context.current_intent
        
        # Execute node actions
        for action in node.actions:
            await action(context, state)
        
        # Determine next node
        next_node_id = self._determine_next_node(
            node, context, state
        )
        
        state.current_node = next_node_id
        next_node = flow.get_node(next_node_id)
        
        # Generate response
        response_text = next_node.get_prompt(context)
        
        # Check for slot filling
        missing_slots = self._get_missing_slots(
            next_node.expected_slots, 
            state
        )
        
        if missing_slots:
            # Ask for missing information
            response_text = self._generate_slot_prompt(
                missing_slots, 
                context
            )
        
        # Generate speech
        audio = await self.tts.synthesize(response_text)
        
        return {
            "text": response_text,
            "audio": audio,
            "state": state,
            "should_confirm": len(missing_slots) == 0 and len(next_node.expected_slots) > 0,
            "is_complete": next_node.is_terminal
        }
    
    def _determine_next_node(
        self,
        current_node: DialogueNode,
        context: TurnContext,
        state: DialogueState
    ) -> str:
        """Determine the next dialogue node based on intent."""
        
        # Check intent-based transitions
        if context.current_intent in current_node.next_node_map:
            return current_node.next_node_map[context.current_intent]
        
        # Check condition-based transitions
        if current_node.condition:
            for intent, next_node in current_node.next_node_map.items():
                test_context = TurnContext(
                    session_id=context.session_id,
                    user_id=context.user_id,
                    current_intent=intent
                )
                if current_node.condition(test_context):
                    return next_node
        
        # Default fallback
        return current_node.next_node_map.get("default", current_node.node_id)
    
    def _get_missing_slots(
        self,
        expected_slots: List[str],
        state: DialogueState
    ) -> List[str]:
        """Get list of unfilled required slots."""
        
        return [
            slot for slot in expected_slots 
            if slot not in state.collected_slots
        ]
    
    def _generate_slot_prompt(
        self,
        missing_slots: List[str],
        context: TurnContext
    ) -> str:
        """Generate prompt for missing slot information."""
        
        slot_prompts = {
            "name": "What is your name?",
            "date": "What date would you like?",
            "time": "What time works for you?",
            "location": "Where would you like this?",
            "phone": "May I have your phone number?",
            "email": "What is your email address?",
            "people": "How many people will be attending?"
        }
        
        prompts = [slot_prompts.get(s, f"Could you provide your {s}?") 
                   for s in missing_slots]
        
        return " ".join(prompts)

import asyncio
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, List, Dict

@dataclass
class TTSResult:
    """Result from text-to-speech synthesis."""
    audio_data: bytes
    duration: float
    sample_rate: int
    format: str = "wav"

class TTSEngine(ABC):
    """Abstract base class for TTS engines."""
    
    @abstractmethod
    async def synthesize(
        self,
        text: str,
        voice_id: str = None,
        **kwargs
    ) -> TTSResult:
        """Synthesize speech from text."""
        pass
    
    @abstractmethod
    async def synthesize_streaming(
        self,
        text: str,
        voice_id: str = None
    ) -> AsyncIterator[bytes]:
        """Synthesize speech with streaming output."""
        pass

class OpenAITTS(TTSEngine):
    """OpenAI TTS implementation."""
    
    def __init__(
        self,
        api_key: str,
        model: str = "tts-1",
        voice: str = "alloy"
    ):
        self.api_key = api_key
        self.model = model
        self.voice = voice
    
    async def synthesize(
        self,
        text: str,
        voice_id: str = None,
        **kwargs
    ) -> TTSResult:
        """Synthesize speech using OpenAI TTS."""
        
        import requests
        
        voice = voice_id or self.voice
        
        response = requests.post(
            "https://api.openai.com/v1/audio/speech",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": self.model,
                "voice": voice,
                "input": text,
                "response_format": "wav"
            }
        )
        
        audio_data = response.content
        
        # Calculate duration (approximate)
        # In production, use audio library for accurate duration
        duration = len(audio_data) / (24000 * 2)  # Assuming 24kHz, 16-bit
        
        return TTSResult(
            audio_data=audio_data,
            duration=duration,
            sample_rate=24000,
            format="wav"
        )
    
    async def synthesize_streaming(
        self,
        text: str,
        voice_id: str = None
    ) -> AsyncIterator[bytes]:
        """Stream synthesis for lower latency."""
        # OpenAI TTS-1 doesn't support true streaming
        # This would use a different provider for production streaming
        result = await self.synthesize(text, voice_id)
        yield result.audio_data

class CoquiTTS(ASREngine):
    """Open-source TTS using Coqui."""
    
    def __init__(
        self,
        model_path: str,
        config_path: str = None,
        device: str = "cuda"
    ):
        self.model_path = model_path
        self.config_path = config_path
        self.device = device
        self.model = None
    
    async def synthesize(
        self,
        text: str,
        voice_id: str = None,
        **kwargs
    ) -> TTSResult:
        """Synthesize speech using Coqui TTS."""
        
        loop = asyncio.get_event_loop()
        
        # Load model if needed
        if self.model is None:
            from TTS.api import TTS
            self.model = await loop.run_in_executor(
                None,
                lambda: TTS(model_path=self.model_path)
            )
        
        # Generate speech
        wav = await loop.run_in_executor(
            None,
            lambda: self.model.tts(text)
        )
        
        # Convert to bytes
        import numpy as np
        audio_np = np.array(wav)
        
        # Convert to 16-bit PCM
        audio_int16 = (audio_np * 32767).astype(np.int16)
        audio_bytes = audio_int16.tobytes()
        
        return TTSResult(
            audio_data=audio_bytes,
            duration=len(wav) / 24000,
            sample_rate=24000,
            format="raw"
        )

@dataclass
class VoiceProfile:
    """Defines a voice persona for the agent."""
    voice_id: str
    name: str
    provider: str
    characteristics: Dict[str, Any]
    use_cases: List[str]
    languages: List[str]
    emotional_range: Dict[str, float]  #悲伤,高兴,激动等

class VoiceManager:
    """Manages voice selection and customization."""
    
    def __init__(self):
        self.voices: Dict[str, VoiceProfile] = {}
        self.default_voice: Optional[str] = None
    
    def register_voice(self, profile: VoiceProfile) -> None:
        """Register a new voice profile."""
        self.voices[profile.voice_id] = profile
    
    def select_voice(
        self,
        context: TurnContext,
        content_type: str = "general"
    ) -> str:
        """Select appropriate voice based on context."""
        
        # Get available voices for content type
        candidates = [
            v for v in self.voices.values()
            if content_type in v.use_cases
        ]
        
        if not candidates:
            return self.default_voice
        
        # Adjust based on sentiment
        if context.sentiment == "negative":
            # More empathetic voice for complaints/issues
            candidates = [
                v for v in candidates 
                if v.emotional_range.get("empathetic", 0) > 0.5
            ]
        
        # Select first matching voice
        return candidates[0].voice_id if candidates else self.default_voice
    
    def adjust_for_context(
        self,
        text: str,
        voice_id: str,
        context: TurnContext
    ) -> str:
        """Modify text for voice characteristics."""
        
        profile = self.voices.get(voice_id)
        if not profile:
            return text
        
        # Adjust vocabulary based on voice characteristics
        formality = profile.characteristics.get("formality", 0.5)
        
        if formality > 0.7:
            # More formal language
            text = text.replace("gonna", "going to")
            text = text.replace("wanna", "want to")
            text = text.replace("yeah", "yes")
        
        return text

class LatencyOptimizer:
    """Optimizes various latency components."""
    
    def __init__(self):
        self.target_latencies = {
            "asr": 0.3,  # seconds
            "nlu": 0.5,
            "generation": 1.0,
            "tts": 0.5,
            "total": 2.0
        }
    
    async def measure_and_optimize(
        self,
        session_id: str,
        audio_data: bytes
    ) -> Dict[str, float]:
        """Measure and optimize latency."""
        
        timings = {}
        
        # ASR timing
        start = asyncio.get_event_loop().time()
        asr_result = await self.asr.recognize(audio_data)
        timings["asr"] = asyncio.get_event_loop().time() - start
        
        # NLU timing
        start = asyncio.get_event_loop().time()
        context = await self.nlu.process_input(asr_result.text, context)
        timings["nlu"] = asyncio.get_event_loop().time() - start
        
        # Generation timing
        start = asyncio.get_event_loop().time()
        response = await self.generate_response(context)
        timings["generation"] = asyncio.get_event_loop().time() - start
        
        # TTS timing
        start = asyncio.get_event_loop().time()
        audio = await self.tts.synthesize(response)
        timings["tts"] = asyncio.get_event_loop().time() - start
        
        timings["total"] = sum(timings.values())
        
        # Log for optimization
        await self._log_timings(session_id, timings)
        
        # Trigger optimizations if needed
        if timings["total"] > self.target_latencies["total"]:
            await self._optimize_pipeline(timings)
        
        return timings
    
    async def _optimize_pipeline(self, timings: Dict[str, float]) -> None:
        """Apply optimizations based on timing analysis."""
        
        # If ASR is slow, consider:
        # - Smaller model
        # - Caching common phrases
        # - Streaming with VAD
        
        # If generation is slow, consider:
        # - Caching frequent responses
        # - Smaller LLM
        # - Template-based fallbacks
        
        # If TTS is slow, consider:
        # - Pre-synthesizing common responses
        # - Streaming synthesis
        # - Caching voice segments
        pass

class VoiceAgentFallbacks:
    """Fallback strategies for various failure modes."""
    
    def __init__(self):
        self.fallback_tts = None
        self.fallback_asr = None
    
    async def handle_asr_failure(
        self,
        audio_data: bytes,
        error: Exception
    ) -> str:
        """Handle ASR failure with fallbacks."""
        
        # Try fallback ASR if available
        if self.fallback_asr:
            try:
                result = await self.fallback_asr.recognize(audio_data)
                return result.text
            except Exception:
                pass
        
        # Ask user to repeat
        return "__RETRY__"
    
    async def handle_tts_failure(
        self,
        text: str,
        error: Exception
    ) -> bytes:
        """Handle TTS failure with fallbacks."""
        
        # Try fallback TTS
        if self.fallback_tts:
            try:
                result = await self.fallback_tts.synthesize(text)
                return result.audio_data
            except Exception:
                pass
        
        # Return empty audio and suggest alternative
        return b"__FALLBACK__"
    
    async def handle_comprehension_failure(
        self,
        context: TurnContext,
        attempt: int
    ) -> str:
        """Handle NLU comprehension failures."""
        
        if attempt == 1:
            return "I didn't quite catch that. Could you please repeat?"
        elif attempt == 2:
            return "Let me try again. Could you say that differently?"
        else:
            # Escalate after multiple failures
            return "I'm having trouble understanding. Let me connect you with a human agent."