Voice AI Assistants: Speech-to-Text and Natural Language Processing

Published on December 15, 2024 • 15 min read

Voice AI assistants are revolutionizing human-computer interaction. This comprehensive guide teaches you how to build intelligent voice assistants using speech recognition, natural language processing, and text-to-speech technologies.

Table of Contents

1. Introduction to Voice AI
2. Speech Recognition Setup
3. Natural Language Processing
4. Text-to-Speech Implementation
5. Building a Complete Voice Assistant
6. Advanced Features
7. Production Deployment
8. Best Practices

Introduction to Voice AI {#introduction}

Voice AI combines multiple technologies to create natural conversational interfaces. Modern voice assistants can understand speech, process natural language, and respond with synthesized speech.

Core Components

• Speech Recognition: Converting audio to text
• Natural Language Understanding: Extracting intent and entities
• Dialog Management: Managing conversation flow
• Text-to-Speech: Converting text responses to audio

Speech Recognition Setup {#speech-recognition}

Using OpenAI Whisper

1import whisper
2import pyaudio
3import wave
4import numpy as np
5from scipy.io import wavfile
6import speech_recognition as sr
7
8class SpeechRecognizer:
9    def __init__(self, model_size="base"):
10        self.whisper_model = whisper.load_model(model_size)
11        self.recognizer = sr.Recognizer()
12        self.microphone = sr.Microphone()
13        
14    def record_audio(self, duration=5, sample_rate=16000):
15        """Record audio from microphone"""
16        chunk = 1024
17        format = pyaudio.paInt16
18        channels = 1
19        
20        audio = pyaudio.PyAudio()
21        
22        stream = audio.open(
23            format=format,
24            channels=channels,
25            rate=sample_rate,
26            input=True,
27            frames_per_buffer=chunk
28        )
29        
30        print("Recording...")
31        frames = []
32        
33        for _ in range(0, int(sample_rate / chunk * duration)):
34            data = stream.read(chunk)
35            frames.append(data)
36        
37        print("Recording finished")
38        
39        stream.stop_stream()
40        stream.close()
41        audio.terminate()
42        
43        # Convert to numpy array
44        audio_data = b''.join(frames)
45        audio_np = np.frombuffer(audio_data, dtype=np.int16)
46        
47        return audio_np.astype(np.float32) / 32768.0
48    
49    def transcribe_with_whisper(self, audio_file):
50        """Transcribe audio using Whisper"""
51        result = self.whisper_model.transcribe(audio_file)
52        return result["text"]
53    
54    def transcribe_realtime(self):
55        """Real-time speech recognition"""
56        with self.microphone as source:
57            self.recognizer.adjust_for_ambient_noise(source)
58        
59        print("Listening...")
60        
61        while True:
62            try:
63                with self.microphone as source:
64                    audio = self.recognizer.listen(source, timeout=1)
65                
66                # Use Google Speech Recognition for real-time
67                text = self.recognizer.recognize_google(audio)
68                print(f"You said: {text}")
69                return text
70                
71            except sr.WaitTimeoutError:
72                continue
73            except sr.UnknownValueError:
74                print("Could not understand audio")
75                continue
76            except sr.RequestError as e:
77                print(f"Error: {e}")
78                continue
79
80# Usage
81recognizer = SpeechRecognizer()
82audio = recognizer.record_audio(duration=5)
83text = recognizer.transcribe_with_whisper("recorded_audio.wav")
84print(f"Transcribed: {text}")

Advanced Speech Processing

1import librosa
2import noisereduce as nr
3from scipy.signal import butter, filtfilt
4
5class AudioProcessor:
6    def __init__(self, sample_rate=16000):
7        self.sample_rate = sample_rate
8    
9    def load_audio(self, file_path):
10        """Load audio file"""
11        audio, sr = librosa.load(file_path, sr=self.sample_rate)
12        return audio
13    
14    def remove_noise(self, audio):
15        """Remove background noise"""
16        reduced_noise = nr.reduce_noise(y=audio, sr=self.sample_rate)
17        return reduced_noise
18    
19    def normalize_audio(self, audio):
20        """Normalize audio levels"""
21        return librosa.util.normalize(audio)
22    
23    def apply_bandpass_filter(self, audio, low_freq=300, high_freq=3400):
24        """Apply bandpass filter for speech frequencies"""
25        nyquist = self.sample_rate * 0.5
26        low = low_freq / nyquist
27        high = high_freq / nyquist
28        
29        b, a = butter(4, [low, high], btype='band')
30        filtered_audio = filtfilt(b, a, audio)
31        
32        return filtered_audio
33    
34    def detect_speech_segments(self, audio, frame_length=2048, hop_length=512):
35        """Detect speech segments using energy-based VAD"""
36        # Compute short-time energy
37        energy = librosa.feature.rms(
38            y=audio, 
39            frame_length=frame_length, 
40            hop_length=hop_length
41        )[0]
42        
43        # Threshold for speech detection
44        threshold = np.mean(energy) * 0.5
45        speech_frames = energy > threshold
46        
47        # Convert frame indices to time
48        times = librosa.frames_to_time(
49            np.arange(len(speech_frames)), 
50            sr=self.sample_rate, 
51            hop_length=hop_length
52        )
53        
54        return times[speech_frames]
55
56# Usage
57processor = AudioProcessor()
58audio = processor.load_audio("speech.wav")
59clean_audio = processor.remove_noise(audio)
60normalized_audio = processor.normalize_audio(clean_audio)

Natural Language Processing {#nlp}

Intent Recognition and Entity Extraction

1import spacy
2from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
3import re
4from datetime import datetime, timedelta
5
6class NLPProcessor:
7    def __init__(self):
8        # Load spaCy model for NER
9        self.nlp = spacy.load("en_core_web_sm")
10        
11        # Load intent classification model
12        self.intent_classifier = pipeline(
13            "text-classification",
14            model="microsoft/DialoGPT-medium"
15        )
16        
17        # Define intent patterns
18        self.intent_patterns = {
19            'weather': [
20                r'weather', r'temperature', r'forecast', r'rain', r'sunny'
21            ],
22            'time': [
23                r'time', r'clock', r'hour', r'minute'
24            ],
25            'reminder': [
26                r'remind', r'reminder', r'schedule', r'appointment'
27            ],
28            'music': [
29                r'play', r'music', r'song', r'artist', r'album'
30            ],
31            'search': [
32                r'search', r'find', r'look up', r'google'
33            ]
34        }
35    
36    def extract_entities(self, text):
37        """Extract named entities from text"""
38        doc = self.nlp(text)
39        entities = []
40        
41        for ent in doc.ents:
42            entities.append({
43                'text': ent.text,
44                'label': ent.label_,
45                'start': ent.start_char,
46                'end': ent.end_char
47            })
48        
49        return entities
50    
51    def classify_intent(self, text):
52        """Classify user intent"""
53        text_lower = text.lower()
54        
55        # Rule-based intent classification
56        for intent, patterns in self.intent_patterns.items():
57            for pattern in patterns:
58                if re.search(pattern, text_lower):
59                    return intent
60        
61        return 'unknown'
62    
63    def extract_datetime(self, text):
64        """Extract date and time information"""
65        doc = self.nlp(text)
66        datetime_entities = []
67        
68        for ent in doc.ents:
69            if ent.label_ in ['DATE', 'TIME']:
70                datetime_entities.append({
71                    'text': ent.text,
72                    'label': ent.label_
73                })
74        
75        # Parse relative time expressions
76        now = datetime.now()
77        
78        if 'tomorrow' in text.lower():
79            target_date = now + timedelta(days=1)
80            datetime_entities.append({
81                'text': 'tomorrow',
82                'parsed_date': target_date.strftime('%Y-%m-%d')
83            })
84        
85        if 'next week' in text.lower():
86            target_date = now + timedelta(weeks=1)
87            datetime_entities.append({
88                'text': 'next week',
89                'parsed_date': target_date.strftime('%Y-%m-%d')
90            })
91        
92        return datetime_entities
93    
94    def process_command(self, text):
95        """Process complete user command"""
96        result = {
97            'original_text': text,
98            'intent': self.classify_intent(text),
99            'entities': self.extract_entities(text),
100            'datetime': self.extract_datetime(text)
101        }
102        
103        return result
104
105# Usage
106nlp_processor = NLPProcessor()
107result = nlp_processor.process_command("Remind me to call John tomorrow at 3 PM")
108print(result)

Dialog Management

1class DialogManager:
2    def __init__(self):
3        self.conversation_history = []
4        self.current_context = {}
5        self.state = 'idle'
6        
7    def add_to_history(self, user_input, assistant_response):
8        """Add exchange to conversation history"""
9        self.conversation_history.append({
10            'timestamp': datetime.now(),
11            'user': user_input,
12            'assistant': assistant_response
13        })
14    
15    def update_context(self, entities):
16        """Update conversation context with new entities"""
17        for entity in entities:
18            self.current_context[entity['label']] = entity['text']
19    
20    def handle_intent(self, intent, entities, text):
21        """Handle different user intents"""
22        if intent == 'weather':
23            return self.handle_weather_request(entities, text)
24        elif intent == 'time':
25            return self.handle_time_request()
26        elif intent == 'reminder':
27            return self.handle_reminder_request(entities, text)
28        elif intent == 'music':
29            return self.handle_music_request(entities, text)
30        else:
31            return "I'm not sure how to help with that. Can you rephrase?"
32    
33    def handle_weather_request(self, entities, text):
34        """Handle weather-related requests"""
35        location = None
36        for entity in entities:
37            if entity['label'] in ['GPE', 'LOC']:  # Geographic entities
38                location = entity['text']
39                break
40        
41        if not location:
42            location = "your location"
43        
44        return f"I'll check the weather for {location}. Let me get that information for you."
45    
46    def handle_time_request(self):
47        """Handle time-related requests"""
48        current_time = datetime.now().strftime("%I:%M %p")
49        return f"The current time is {current_time}."
50    
51    def handle_reminder_request(self, entities, text):
52        """Handle reminder requests"""
53        # Extract reminder details
54        reminder_text = text
55        datetime_info = None
56        
57        for entity in entities:
58            if entity['label'] in ['DATE', 'TIME']:
59                datetime_info = entity['text']
60        
61        if datetime_info:
62            return f"I'll remind you about: {reminder_text} at {datetime_info}"
63        else:
64            return "When would you like me to remind you?"
65    
66    def handle_music_request(self, entities, text):
67        """Handle music playback requests"""
68        artist = None
69        song = None
70        
71        # Simple extraction - in production, use more sophisticated NER
72        if 'by' in text.lower():
73            parts = text.lower().split('by')
74            if len(parts) > 1:
75                artist = parts[1].strip()
76        
77        return f"I'll play music for you. Searching for {artist if artist else 'music'}..."
78
79# Usage
80dialog_manager = DialogManager()
81response = dialog_manager.handle_intent('weather', [], "What's the weather in New York?")
82print(response)

Text-to-Speech Implementation {#tts}

Using Multiple TTS Engines

1import pyttsx3
2import pygame
3from gtts import gTTS
4import io
5import tempfile
6import os
7from elevenlabs import generate, play, set_api_key
8
9class TextToSpeech:
10    def __init__(self, engine='pyttsx3'):
11        self.engine_type = engine
12        
13        if engine == 'pyttsx3':
14            self.engine = pyttsx3.init()
15            self.setup_pyttsx3()
16        elif engine == 'elevenlabs':
17            # Set your ElevenLabs API key
18            set_api_key("your_api_key_here")
19    
20    def setup_pyttsx3(self):
21        """Configure pyttsx3 engine"""
22        voices = self.engine.getProperty('voices')
23        
24        # Set voice (0 for male, 1 for female typically)
25        if len(voices) > 1:
26            self.engine.setProperty('voice', voices[1].id)
27        
28        # Set speech rate
29        self.engine.setProperty('rate', 150)
30        
31        # Set volume
32        self.engine.setProperty('volume', 0.9)
33    
34    def speak_pyttsx3(self, text):
35        """Speak using pyttsx3"""
36        self.engine.say(text)
37        self.engine.runAndWait()
38    
39    def speak_gtts(self, text, lang='en'):
40        """Speak using Google Text-to-Speech"""
41        tts = gTTS(text=text, lang=lang, slow=False)
42        
43        # Save to temporary file
44        with tempfile.NamedTemporaryFile(delete=False, suffix='.mp3') as tmp_file:
45            tts.save(tmp_file.name)
46            
47            # Play audio
48            pygame.mixer.init()
49            pygame.mixer.music.load(tmp_file.name)
50            pygame.mixer.music.play()
51            
52            # Wait for playback to finish
53            while pygame.mixer.music.get_busy():
54                pygame.time.wait(100)
55            
56            # Clean up
57            os.unlink(tmp_file.name)
58    
59    def speak_elevenlabs(self, text, voice="Bella"):
60        """Speak using ElevenLabs (premium quality)"""
61        audio = generate(
62            text=text,
63            voice=voice,
64            model="eleven_monolingual_v1"
65        )
66        play(audio)
67    
68    def speak(self, text):
69        """Main speak method"""
70        if self.engine_type == 'pyttsx3':
71            self.speak_pyttsx3(text)
72        elif self.engine_type == 'gtts':
73            self.speak_gtts(text)
74        elif self.engine_type == 'elevenlabs':
75            self.speak_elevenlabs(text)
76
77# Usage
78tts = TextToSpeech(engine='gtts')
79tts.speak("Hello! I'm your voice assistant. How can I help you today?")

Building a Complete Voice Assistant {#complete-assistant}

Main Assistant Class

1import threading
2import queue
3import time
4
5class VoiceAssistant:
6    def __init__(self):
7        self.speech_recognizer = SpeechRecognizer()
8        self.nlp_processor = NLPProcessor()
9        self.dialog_manager = DialogManager()
10        self.tts = TextToSpeech(engine='gtts')
11        
12        self.listening = False
13        self.wake_word = "hey assistant"
14        
15    def listen_for_wake_word(self):
16        """Continuously listen for wake word"""
17        print("Listening for wake word...")
18        
19        while True:
20            try:
21                text = self.speech_recognizer.transcribe_realtime()
22                if text and self.wake_word.lower() in text.lower():
23                    print("Wake word detected!")
24                    self.tts.speak("Yes, how can I help you?")
25                    self.process_command()
26                    
27            except Exception as e:
28                print(f"Error in wake word detection: {e}")
29                time.sleep(1)
30    
31    def process_command(self):
32        """Process user command after wake word"""
33        try:
34            # Listen for command
35            print("Listening for command...")
36            command_text = self.speech_recognizer.transcribe_realtime()
37            
38            if command_text:
39                print(f"Command received: {command_text}")
40                
41                # Process with NLP
42                nlp_result = self.nlp_processor.process_command(command_text)
43                
44                # Update dialog context
45                self.dialog_manager.update_context(nlp_result['entities'])
46                
47                # Generate response
48                response = self.dialog_manager.handle_intent(
49                    nlp_result['intent'],
50                    nlp_result['entities'],
51                    command_text
52                )
53                
54                # Speak response
55                print(f"Assistant: {response}")
56                self.tts.speak(response)
57                
58                # Add to conversation history
59                self.dialog_manager.add_to_history(command_text, response)
60                
61        except Exception as e:
62            print(f"Error processing command: {e}")
63            self.tts.speak("Sorry, I didn't understand that. Could you repeat?")
64    
65    def start(self):
66        """Start the voice assistant"""
67        print("Voice Assistant starting...")
68        self.tts.speak("Voice assistant is ready. Say 'hey assistant' to wake me up.")
69        
70        # Start listening in a separate thread
71        listen_thread = threading.Thread(target=self.listen_for_wake_word)
72        listen_thread.daemon = True
73        listen_thread.start()
74        
75        try:
76            # Keep main thread alive
77            while True:
78                time.sleep(1)
79        except KeyboardInterrupt:
80            print("Voice Assistant shutting down...")
81            self.tts.speak("Goodbye!")
82
83# Usage
84if __name__ == "__main__":
85    assistant = VoiceAssistant()
86    assistant.start()

Advanced Features {#advanced}

Emotion Recognition

1from transformers import pipeline
2import librosa
3import numpy as np
4
5class EmotionRecognizer:
6    def __init__(self):
7        self.text_emotion_classifier = pipeline(
8            "text-classification",
9            model="j-hartmann/emotion-english-distilroberta-base"
10        )
11    
12    def analyze_text_emotion(self, text):
13        """Analyze emotion from text"""
14        result = self.text_emotion_classifier(text)
15        return result[0]
16    
17    def analyze_voice_emotion(self, audio_file):
18        """Analyze emotion from voice characteristics"""
19        # Load audio
20        y, sr = librosa.load(audio_file)
21        
22        # Extract features
23        features = {
24            'mfcc': np.mean(librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13), axis=1),
25            'spectral_centroid': np.mean(librosa.feature.spectral_centroid(y=y, sr=sr)),
26            'zero_crossing_rate': np.mean(librosa.feature.zero_crossing_rate(y)),
27            'tempo': librosa.beat.tempo(y=y, sr=sr)[0]
28        }
29        
30        # Simple emotion classification based on features
31        if features['spectral_centroid'] > 2000 and features['tempo'] > 120:
32            return {'emotion': 'excited', 'confidence': 0.8}
33        elif features['spectral_centroid'] < 1000 and features['tempo'] < 80:
34            return {'emotion': 'sad', 'confidence': 0.7}
35        else:
36            return {'emotion': 'neutral', 'confidence': 0.6}
37
38# Usage
39emotion_recognizer = EmotionRecognizer()
40text_emotion = emotion_recognizer.analyze_text_emotion("I'm feeling great today!")
41print(f"Text emotion: {text_emotion}")

Multi-language Support

1class MultiLanguageAssistant:
2    def __init__(self):
3        self.supported_languages = {
4            'en': 'English',
5            'es': 'Spanish',
6            'fr': 'French',
7            'de': 'German',
8            'it': 'Italian'
9        }
10        self.current_language = 'en'
11        
12        # Language detection
13        self.language_detector = pipeline(
14            "text-classification",
15            model="papluca/xlm-roberta-base-language-detection"
16        )
17    
18    def detect_language(self, text):
19        """Detect language of input text"""
20        result = self.language_detector(text)
21        detected_lang = result[0]['label'].lower()
22        
23        if detected_lang in self.supported_languages:
24            return detected_lang
25        return 'en'  # Default to English
26    
27    def translate_text(self, text, target_lang='en'):
28        """Translate text to target language"""
29        from googletrans import Translator
30        
31        translator = Translator()
32        result = translator.translate(text, dest=target_lang)
33        return result.text
34    
35    def process_multilingual_command(self, text):
36        """Process command in any supported language"""
37        # Detect language
38        detected_lang = self.detect_language(text)
39        
40        # Translate to English for processing if needed
41        if detected_lang != 'en':
42            english_text = self.translate_text(text, 'en')
43        else:
44            english_text = text
45        
46        # Process command (using existing NLP processor)
47        # ... processing logic ...
48        
49        response = "I understand your request."
50        
51        # Translate response back to original language
52        if detected_lang != 'en':
53            response = self.translate_text(response, detected_lang)
54        
55        return response, detected_lang
56
57# Usage
58multilang_assistant = MultiLanguageAssistant()
59response, lang = multilang_assistant.process_multilingual_command("¿Qué hora es?")
60print(f"Response in {lang}: {response}")

Production Deployment {#deployment}

WebSocket-based Voice Assistant

1import asyncio
2import websockets
3import json
4import base64
5import wave
6
7class WebSocketVoiceAssistant:
8    def __init__(self):
9        self.assistant = VoiceAssistant()
10        
11    async def handle_client(self, websocket, path):
12        """Handle WebSocket client connection"""
13        print("Client connected")
14        
15        try:
16            async for message in websocket:
17                data = json.loads(message)
18                
19                if data['type'] == 'audio':
20                    # Decode base64 audio
21                    audio_data = base64.b64decode(data['audio'])
22                    
23                    # Save to temporary file
24                    with wave.open('temp_audio.wav', 'wb') as wf:
25                        wf.setnchannels(1)
26                        wf.setsampwidth(2)
27                        wf.setframerate(16000)
28                        wf.writeframes(audio_data)
29                    
30                    # Process audio
31                    text = self.assistant.speech_recognizer.transcribe_with_whisper('temp_audio.wav')
32                    
33                    if text:
34                        # Process command
35                        nlp_result = self.assistant.nlp_processor.process_command(text)
36                        response = self.assistant.dialog_manager.handle_intent(
37                            nlp_result['intent'],
38                            nlp_result['entities'],
39                            text
40                        )
41                        
42                        # Send response
43                        await websocket.send(json.dumps({
44                            'type': 'response',
45                            'text': response,
46                            'intent': nlp_result['intent']
47                        }))
48                
49        except websockets.exceptions.ConnectionClosed:
50            print("Client disconnected")
51    
52    def start_server(self, host='localhost', port=8765):
53        """Start WebSocket server"""
54        start_server = websockets.serve(self.handle_client, host, port)
55        
56        print(f"Voice Assistant WebSocket server starting on {host}:{port}")
57        asyncio.get_event_loop().run_until_complete(start_server)
58        asyncio.get_event_loop().run_forever()
59
60# Usage
61ws_assistant = WebSocketVoiceAssistant()
62ws_assistant.start_server()

Best Practices {#best-practices}

Performance Optimization

1. Audio Processing

   • Use appropriate sample rates (16kHz for speech)
   • Implement noise reduction
   • Use voice activity detection

2. Model Optimization

   • Use quantized models for faster inference
   • Implement model caching
   • Use streaming recognition for real-time applications

3. Error Handling

   • Implement robust error recovery
   • Provide fallback responses
   • Log errors for debugging

Security Considerations

• Privacy: Process audio locally when possible
• Authentication: Implement user authentication for personal assistants
• Data Protection: Encrypt stored conversation data
• Rate Limiting: Prevent abuse of API endpoints

Conclusion

Voice AI assistants represent the future of human-computer interaction. By combining speech recognition, natural language processing, and text-to-speech technologies, you can create powerful conversational interfaces.

Start with basic functionality and gradually add advanced features like emotion recognition and multi-language support as your system matures.

---

This guide provides a comprehensive foundation for building voice AI assistants. Remember to test thoroughly with diverse users and accents for robust performance.