Whisper Voice-to-Intent Pipeline
Chapter Objectives
- Understand OpenAI Whisper architecture and capabilities
- Implement voice-to-intent processing pipeline
- Integrate speech recognition with humanoid robot control
- Optimize Whisper for real-time robotic applications
- Create intent classification and action mapping systems
Introduction to Whisper for Robotics
OpenAI Whisper is a state-of-the-art speech recognition model that can transcribe speech to text with remarkable accuracy. For humanoid robotics applications, Whisper provides:
- Multilingual Support: Recognition of multiple languages
- Robustness: Works well in various acoustic conditions
- Real-time Processing: Capable of near real-time transcription
- Open Source: Access to model weights and architecture
- Customization: Fine-tuning capabilities for domain-specific tasks
Why Whisper for Humanoid Robots?
- Natural Interaction: Enable natural voice-based communication
- Accessibility: Make robots accessible to users who prefer voice
- Hands-free Operation: Allow operation without physical interaction
- Multimodal Integration: Combine with vision and action systems
- Scalability: Handle multiple languages and dialects
Whisper Model Architecture
Whisper uses a Transformer-based architecture with both encoder and decoder components:
- Encoder: Processes audio input using a convolutional neural network and Transformer layers
- Decoder: Generates text output using autoregressive Transformer architecture
- Multilingual Capability: Trained on 98+ languages for universal recognition
Setting Up Whisper for Robotics
Installation and Dependencies
# Install Whisper and related dependencies
pip install openai-whisper
pip install torch torchvision torchaudio
pip install pyaudio sounddevice webrtcvad
pip install transformers datasets
Basic Whisper Implementation
# python/whisper_robotics.py
import whisper
import torch
import numpy as np
import pyaudio
import wave
import threading
import queue
import time
from typing import Dict, List, Optional, Tuple, Any
import json
import os
from dataclasses import dataclass
from enum import Enum
class IntentType(Enum):
"""Types of intents that can be recognized from speech"""
MOVE = "move"
GRASP = "grasp"
GREET = "greet"
FOLLOW = "follow"
STOP = "stop"
SEARCH = "search"
FETCH = "fetch"
DANCE = "dance"
SLEEP = "sleep"
WAKE = "wake"
OTHER = "other"
@dataclass
class SpeechIntent:
"""Data structure for speech intent results"""
text: str
intent_type: IntentType
confidence: float
timestamp: float
entities: Dict[str, Any]
raw_transcription: str
class WhisperRobotInterface:
"""Interface between Whisper and robotic systems"""
def __init__(self, model_size: str = "base", device: str = "cuda", language: str = "en"):
"""
Initialize Whisper interface for robotics
Args:
model_size: Size of Whisper model (tiny, base, small, medium, large)
device: Device to run model on (cuda, cpu)
language: Language for recognition (en, es, fr, etc.)
"""
self.device = device if torch.cuda.is_available() and device == "cuda" else "cpu"
self.language = language
# Load Whisper model
self.model = whisper.load_model(model_size, device=self.device)
self.model_size = model_size
# Audio parameters
self.sample_rate = 16000
self.chunk_size = 1024
self.audio_format = pyaudio.paInt16
self.channels = 1
# Audio stream
self.audio = pyaudio.PyAudio()
self.stream = None
self.is_listening = False
self.audio_queue = queue.Queue()
# Intent classification
self.intent_classifier = IntentClassifier()
# Statistics
self.stats = {
'total_transcriptions': 0,
'average_latency': 0.0,
'transcription_rate': 0.0
}
print(f"Whisper model loaded on {self.device}")
print(f"Model size: {model_size}, Language: {language}")
def start_listening(self):
"""Start audio recording and processing"""
if self.stream is None:
self.stream = self.audio.open(
format=self.audio_format,
channels=self.channels,
rate=self.sample_rate,
input=True,
frames_per_buffer=self.chunk_size
)
self.is_listening = True
self.listen_thread = threading.Thread(target=self._audio_loop, daemon=True)
self.listen_thread.start()
print("Started listening for speech input")
def stop_listening(self):
"""Stop audio recording and processing"""
self.is_listening = False
if self.stream:
self.stream.stop_stream()
self.stream.close()
self.stream = None
print("Stopped listening for speech input")
def _audio_loop(self):
"""Main audio processing loop"""
audio_buffer = np.array([], dtype=np.float32)
while self.is_listening:
try:
# Read audio chunk
data = self.stream.read(self.chunk_size, exception_on_overflow=False)
audio_chunk = np.frombuffer(data, dtype=np.int16).astype(np.float32) / 32768.0
# Add to buffer
audio_buffer = np.concatenate([audio_buffer, audio_chunk])
# Process when buffer is large enough
if len(audio_buffer) >= self.sample_rate * 2: # 2 seconds of audio
# Transcribe the audio
intent = self.process_audio(audio_buffer)
if intent:
self.handle_intent(intent)
# Keep last 0.5 seconds to maintain context
audio_buffer = audio_buffer[-int(self.sample_rate * 0.5):]
time.sleep(0.01) # Small delay to prevent excessive CPU usage
except Exception as e:
print(f"Error in audio loop: {e}")
time.sleep(0.1)
def process_audio(self, audio_data: np.ndarray) -> Optional[SpeechIntent]:
"""Process audio data and extract intent"""
start_time = time.time()
try:
# Transcribe audio using Whisper
result = self.model.transcribe(
audio_data,
language=self.language,
task="transcribe",
temperature=0.0 # For consistent results
)
transcription = result["text"].strip()
if transcription and len(transcription) > 3: # Filter out very short transcriptions
# Classify intent
intent_type, confidence, entities = self.intent_classifier.classify_intent(transcription)
# Create speech intent object
speech_intent = SpeechIntent(
text=self._extract_command(transcription),
intent_type=intent_type,
confidence=confidence,
timestamp=time.time(),
entities=entities,
raw_transcription=transcription
)
# Update statistics
processing_time = time.time() - start_time
self.stats['total_transcriptions'] += 1
self.stats['average_latency'] = (
(self.stats['average_latency'] * (self.stats['total_transcriptions'] - 1) + processing_time) /
self.stats['total_transcriptions']
)
return speech_intent
except Exception as e:
print(f"Error processing audio: {e}")
return None
def _extract_command(self, transcription: str) -> str:
"""Extract the main command from transcription"""
# Remove common filler words and normalize
command = transcription.lower().strip()
fillers = ["um", "uh", "like", "so", "well", "you know"]
for filler in fillers:
command = command.replace(filler, "").strip()
# Remove extra spaces
import re
command = re.sub(r'\s+', ' ', command).strip()
return command
def handle_intent(self, intent: SpeechIntent):
"""Handle the recognized intent"""
print(f"Recognized intent: {intent.intent_type.value} - '{intent.text}' (confidence: {intent.confidence:.2f})")
# Here you would typically send the intent to the robot's action system
# For now, we'll just log it
self._execute_robot_action(intent)
def _execute_robot_action(self, intent: SpeechIntent):
"""Execute robot action based on intent"""
# This would interface with the robot's control system
# For demonstration, we'll just print what would happen
print(f"Robot would execute: {intent.intent_type.value} with entities: {intent.entities}")
def transcribe_audio_file(self, audio_path: str) -> str:
"""Transcribe a pre-recorded audio file"""
result = self.model.transcribe(audio_path, language=self.language)
return result["text"]
def transcribe_audio_buffer(self, audio_buffer: np.ndarray) -> str:
"""Transcribe an audio buffer directly"""
result = self.model.transcribe(audio_buffer, language=self.language)
return result["text"]
def get_stats(self) -> Dict:
"""Get current statistics"""
return self.stats.copy()
class IntentClassifier:
"""Classify intents from transcribed speech"""
def __init__(self):
# Define intent patterns and keywords
self.intent_patterns = {
IntentType.MOVE: [
r"move\s+(?P<direction>\w+)\s*(?P<distance>\d+\.?\d*)?\s*(?P<unit>meters?|steps?)?",
r"go\s+(?P<direction>\w+)\s*(?P<distance>\d+\.?\d*)?\s*(?P<unit>meters?|steps?)?",
r"walk\s+(?P<direction>\w+)\s*(?P<distance>\d+\.?\d*)?\s*(?P<unit>meters?|steps?)?",
r"step\s+(?P<direction>\w+)\s*(?P<distance>\d+\.?\d*)?\s*(?P<unit>meters?|steps?)?",
],
IntentType.GRASP: [
r"pick\s+up\s+(?P<object>\w+)",
r"grab\s+(?P<object>\w+)",
r"take\s+(?P<object>\w+)",
r"lift\s+(?P<object>\w+)",
r"hold\s+(?P<object>\w+)",
],
IntentType.GREET: [
r"hello",
r"hi",
r"greetings",
r"good\s+(morning|afternoon|evening)",
r"hey",
],
IntentType.FOLLOW: [
r"follow\s+(?P<target>\w+)",
r"come\s+with\s+(?P<target>\w+)",
r"stay\s+with\s+(?P<target>\w+)",
],
IntentType.STOP: [
r"stop",
r"halt",
r"pause",
r"wait",
r"freeze",
],
IntentType.SEARCH: [
r"find\s+(?P<object>\w+)",
r"look\s+for\s+(?P<object>\w+)",
r"search\s+for\s+(?P<object>\w+)",
r"locate\s+(?P<object>\w+)",
],
IntentType.FETCH: [
r"bring\s+(?P<object>\w+)",
r"get\s+(?P<object>\w+)",
r"fetch\s+(?P<object>\w+)",
r"carry\s+(?P<object>\w+)",
],
IntentType.DANCE: [
r"dance",
r"move\s+to\s+the\s+beat",
r"boogie",
r"groove",
r"shake",
],
IntentType.SLEEP: [
r"sleep",
r"rest",
r"power\s+down",
r"shut\s+down",
r"turn\s+off",
],
IntentType.WAKE: [
r"wake\s+up",
r"start",
r"activate",
r"turn\s+on",
r"wake",
],
}
# Keyword mappings for entities
self.direction_keywords = {
'forward': ['forward', 'ahead', 'straight'],
'backward': ['backward', 'back', 'reverse'],
'left': ['left', 'west'],
'right': ['right', 'east'],
'up': ['up', 'top', 'above'],
'down': ['down', 'bottom', 'below'],
}
self.object_keywords = {
'object': ['ball', 'box', 'cup', 'bottle', 'book', 'phone', 'toy', 'tool'],
'person': ['person', 'human', 'you', 'me', 'us', 'someone'],
'location': ['kitchen', 'living room', 'bedroom', 'office', 'hallway', 'door', 'table'],
}
def classify_intent(self, text: str) -> Tuple[IntentType, float, Dict[str, Any]]:
"""Classify intent from text with confidence and entities"""
text_lower = text.lower().strip()
best_intent = IntentType.OTHER
best_confidence = 0.0
best_entities = {}
# Check each intent type
for intent_type, patterns in self.intent_patterns.items():
for pattern in patterns:
import re
match = re.search(pattern, text_lower, re.IGNORECASE)
if match:
confidence = 0.9 # High confidence for pattern matches
entities = match.groupdict()
# Process entities to normalize them
normalized_entities = self._normalize_entities(entities, text_lower)
# If this intent has higher confidence, update
if confidence > best_confidence:
best_intent = intent_type
best_confidence = confidence
best_entities = normalized_entities
# If no pattern matched, try keyword-based classification
if best_confidence < 0.5:
keyword_intent, keyword_confidence, keyword_entities = self._classify_by_keywords(text_lower)
if keyword_confidence > best_confidence:
best_intent = keyword_intent
best_confidence = keyword_confidence
best_entities = keyword_entities
return best_intent, best_confidence, best_entities
def _normalize_entities(self, entities: Dict[str, str], text: str) -> Dict[str, Any]:
"""Normalize extracted entities"""
normalized = {}
for key, value in entities.items():
if value:
# Normalize based on entity type
if key == 'direction':
normalized[key] = self._normalize_direction(value)
elif key == 'object':
normalized[key] = self._normalize_object(value)
elif key == 'distance':
normalized[key] = float(value) if value.replace('.', '').isdigit() else 1.0
elif key == 'unit':
normalized[key] = value if value else 'meters'
else:
normalized[key] = value
return normalized
def _normalize_direction(self, direction: str) -> str:
"""Normalize direction keywords"""
direction = direction.lower().strip()
for norm_dir, keywords in self.direction_keywords.items():
if direction in keywords or direction.startswith(norm_dir):
return norm_dir
return direction # Return original if no match
def _normalize_object(self, obj: str) -> str:
"""Normalize object keywords"""
obj = obj.lower().strip()
for obj_type, keywords in self.object_keywords.items():
if obj in keywords:
return obj
return obj # Return original if no match
def _classify_by_keywords(self, text: str) -> Tuple[IntentType, float, Dict[str, Any]]:
"""Classify intent based on keywords if patterns don't match"""
# Simple keyword-based classification
text_lower = text.lower()
# Count keywords for each intent type
intent_scores = {}
for intent_type, patterns in self.intent_patterns.items():
score = 0
for pattern in patterns:
import re
# Extract keywords from pattern for matching
keywords = re.findall(r'\w+', pattern)
for keyword in keywords:
if keyword in ['(?P', 'direction', 'object', 'distance', 'unit', 'target']:
continue
if keyword in text_lower:
score += 1
intent_scores[intent_type] = score
# Find intent with highest score
if intent_scores:
best_intent = max(intent_scores, key=intent_scores.get)
best_score = intent_scores[best_intent]
if best_score > 0:
confidence = min(0.8, best_score * 0.2) # Scale confidence
return best_intent, confidence, {}
return IntentType.OTHER, 0.1, {} # Low confidence for other
def main():
"""Main function to demonstrate Whisper robotics interface"""
print("Initializing Whisper Robotics Interface...")
# Initialize Whisper interface
whisper_interface = WhisperRobotInterface(
model_size="base", # Use "tiny" for faster processing, "large" for better accuracy
device="cuda" if torch.cuda.is_available() else "cpu",
language="en"
)
# Example: Transcribe a sample audio file
# whisper_interface.transcribe_audio_file("sample_speech.wav")
# Example: Start real-time listening
# whisper_interface.start_listening()
# For demonstration, let's test the intent classifier
test_sentences = [
"Move forward 2 meters",
"Pick up the red ball",
"Hello robot",
"Follow me",
"Stop moving",
"Find the blue cup",
"Bring me the book",
"Dance for me",
"Go to sleep",
"Wake up now"
]
classifier = IntentClassifier()
print("\nTesting intent classification:")
for sentence in test_sentences:
intent_type, confidence, entities = classifier.classify_intent(sentence)
print(f"'{sentence}' -> {intent_type.value} (confidence: {confidence:.2f}, entities: {entities})")
print(f"\nWhisper interface initialized successfully!")
print(f"Model: {whisper_interface.model_size}")
print(f"Device: {whisper_interface.device}")
print(f"Language: {whisper_interface.language}")
if __name__ == "__main__":
main()
Real-time Voice Processing
Audio Capture and Preprocessing
# python/audio_preprocessing.py
import pyaudio
import numpy as np
import threading
import queue
import time
import webrtcvad
import collections
from scipy import signal
from typing import Callable, Optional
class AudioProcessor:
"""Advanced audio processor for robotics applications"""
def __init__(self, sample_rate: int = 16000, chunk_size: int = 1024):
self.sample_rate = sample_rate
self.chunk_size = chunk_size
# Voice activity detection
self.vad = webrtcvad.Vad(2) # Aggressiveness mode 2 (0-3)
# Audio processing parameters
self.silence_threshold = 0.01
self.min_speech_duration = 0.3 # seconds
self.max_silence_duration = 1.0 # seconds before stopping
# Audio queues
self.audio_queue = queue.Queue()
self.processed_queue = queue.Queue()
# State tracking
self.is_recording = False
self.speech_buffer = np.array([], dtype=np.float32)
self.silence_duration = 0.0
self.speech_duration = 0.0
# Callbacks
self.speech_detected_callback: Optional[Callable] = None
self.audio_processed_callback: Optional[Callable] = None
def set_callbacks(self,
speech_detected: Optional[Callable] = None,
audio_processed: Optional[Callable] = None):
"""Set callbacks for audio events"""
self.speech_detected_callback = speech_detected
self.audio_processed_callback = audio_processed
def preprocess_audio(self, audio_chunk: np.ndarray) -> np.ndarray:
"""Preprocess audio chunk for better recognition"""
# Convert to float32 if needed
if audio_chunk.dtype != np.float32:
audio_chunk = audio_chunk.astype(np.float32)
# Normalize audio
audio_chunk = audio_chunk / np.max(np.abs(audio_chunk)) if np.max(np.abs(audio_chunk)) > 0 else audio_chunk
# Apply noise reduction (simple spectral subtraction approach)
audio_chunk = self._reduce_noise(audio_chunk)
# Apply pre-emphasis filter to boost high frequencies
audio_chunk = self._pre_emphasis_filter(audio_chunk)
return audio_chunk
def _reduce_noise(self, audio: np.ndarray) -> np.ndarray:
"""Simple noise reduction using spectral subtraction"""
# For real applications, use more sophisticated noise reduction
# This is a simplified version
if len(audio) < 1024:
return audio
# Estimate noise from beginning of audio (assumed to be silence)
noise_segment = audio[:512]
noise_mean = np.mean(np.abs(noise_segment))
# Apply simple noise thresholding
audio_denoised = np.copy(audio)
mask = np.abs(audio) > noise_mean * 2
audio_denoised[~mask] = 0
return audio_denoised
def _pre_emphasis_filter(self, audio: np.ndarray, coeff: float = 0.97) -> np.ndarray:
"""Apply pre-emphasis filter to boost high frequencies"""
return np.append(audio[0], audio[1:] - coeff * audio[:-1])
def detect_voice_activity(self, audio_chunk: np.ndarray) -> bool:
"""Detect voice activity in audio chunk using WebRTC VAD"""
# Convert to 16-bit PCM for WebRTC VAD
audio_int16 = (audio_chunk * 32767).astype(np.int16)
# WebRTC VAD requires 10, 20, or 30 ms frames
frame_size = int(self.sample_rate * 0.01) # 10ms frame
if len(audio_int16) >= frame_size:
frames = self._split_audio_frames(audio_int16, frame_size)
voice_activity = any(self.vad.is_speech(frame.tobytes(), self.sample_rate) for frame in frames)
return voice_activity
return False
def _split_audio_frames(self, audio: np.ndarray, frame_size: int):
"""Split audio into frames for VAD"""
for i in range(0, len(audio) - frame_size + 1, frame_size):
yield audio[i:i + frame_size]
def process_audio_stream(self, audio_chunk: np.ndarray) -> Optional[np.ndarray]:
"""Process continuous audio stream for speech detection and segmentation"""
# Preprocess the chunk
processed_chunk = self.preprocess_audio(audio_chunk)
# Detect voice activity
is_speech = self.detect_voice_activity(processed_chunk)
if is_speech:
# Add to speech buffer
self.speech_buffer = np.concatenate([self.speech_buffer, processed_chunk])
self.speech_duration += len(processed_chunk) / self.sample_rate
self.silence_duration = 0.0 # Reset silence timer
# Check if speech is long enough to process
if self.speech_duration >= self.min_speech_duration:
speech_data = self.speech_buffer.copy()
self.speech_buffer = np.array([], dtype=np.float32) # Clear buffer
self.speech_duration = 0.0
if self.audio_processed_callback:
self.audio_processed_callback(speech_data)
return speech_data
else:
# Update silence duration
self.silence_duration += len(processed_chunk) / self.sample_rate
# If we have accumulated speech and now have sufficient silence, return it
if (len(self.speech_buffer) > 0 and
self.silence_duration >= self.max_silence_duration and
self.speech_duration >= self.min_speech_duration):
speech_data = self.speech_buffer.copy()
self.speech_buffer = np.array([], dtype=np.float32)
self.speech_duration = 0.0
self.silence_duration = 0.0
if self.audio_processed_callback:
self.audio_processed_callback(speech_data)
return speech_data
return None
class RealTimeAudioProcessor:
"""Real-time audio processor with Whisper integration"""
def __init__(self, whisper_interface: WhisperRobotInterface):
self.whisper_interface = whisper_interface
self.audio_processor = AudioProcessor()
self.is_active = False
self.processing_thread = None
# Set up callbacks
self.audio_processor.set_callbacks(
audio_processed=self._on_audio_processed
)
def _on_audio_processed(self, audio_data: np.ndarray):
"""Callback when audio is processed and ready for transcription"""
# Send to Whisper for transcription
intent = self.whisper_interface.process_audio(audio_data)
if intent:
self.whisper_interface.handle_intent(intent)
def start_processing(self):
"""Start real-time audio processing"""
self.is_active = True
# Initialize PyAudio
self.audio = pyaudio.PyAudio()
self.stream = self.audio.open(
format=pyaudio.paFloat32,
channels=1,
rate=self.audio_processor.sample_rate,
input=True,
frames_per_buffer=self.audio_processor.chunk_size
)
# Start processing thread
self.processing_thread = threading.Thread(target=self._processing_loop, daemon=True)
self.processing_thread.start()
print("Real-time audio processing started")
def stop_processing(self):
"""Stop real-time audio processing"""
self.is_active = False
if hasattr(self, 'stream'):
self.stream.stop_stream()
self.stream.close()
if hasattr(self, 'audio'):
self.audio.terminate()
print("Real-time audio processing stopped")
def _processing_loop(self):
"""Main processing loop"""
while self.is_active:
try:
# Read audio chunk
data = self.stream.read(self.audio_processor.chunk_size, exception_on_overflow=False)
audio_chunk = np.frombuffer(data, dtype=np.float32)
# Process the chunk
result = self.audio_processor.process_audio_stream(audio_chunk)
time.sleep(0.001) # Small delay to prevent excessive CPU usage
except Exception as e:
print(f"Error in processing loop: {e}")
time.sleep(0.1)
def demonstrate_real_time_processing():
"""Demonstrate real-time audio processing with Whisper"""
print("Setting up real-time Whisper processing...")
# Initialize Whisper interface
whisper_interface = WhisperRobotInterface(
model_size="base",
device="cuda" if torch.cuda.is_available() else "cpu",
language="en"
)
# Initialize real-time processor
rt_processor = RealTimeAudioProcessor(whisper_interface)
print("Starting real-time processing...")
print("Speak to the microphone. The system will detect speech and process it with Whisper.")
print("Press Ctrl+C to stop.")
try:
rt_processor.start_processing()
# Keep the main thread alive
while True:
time.sleep(1)
# Print stats periodically
stats = whisper_interface.get_stats()
if stats['total_transcriptions'] > 0:
print(f"Stats - Transcriptions: {stats['total_transcriptions']}, "
f"Average latency: {stats['average_latency']:.3f}s")
except KeyboardInterrupt:
print("\nStopping real-time processing...")
rt_processor.stop_processing()
print("Processing stopped.")
# Example usage
if __name__ == "__main__":
demonstrate_real_time_processing()
Intent Classification and Action Mapping
Advanced Intent Classification
# python/intent_classification.py
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoTokenizer, AutoModel
import numpy as np
from typing import Dict, List, Tuple, Optional, Any
import re
import spacy
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
import pickle
import os
class AdvancedIntentClassifier:
"""Advanced intent classifier using transformer models and rule-based systems"""
def __init__(self, model_path: Optional[str] = None):
# Load spaCy model for NLP processing
try:
self.nlp = spacy.load("en_core_web_sm")
except OSError:
print("Please install en_core_web_sm: python -m spacy download en_core_web_sm")
self.nlp = None
# Initialize rule-based classifier
self.rule_classifier = RuleBasedIntentClassifier()
# Initialize ML-based classifier
self.ml_classifier = MLIntentClassifier()
# Intent confidence thresholds
self.confidence_thresholds = {
IntentType.MOVE: 0.7,
IntentType.GRASP: 0.75,
IntentType.GREET: 0.6,
IntentType.FOLLOW: 0.7,
IntentType.STOP: 0.65,
IntentType.SEARCH: 0.7,
IntentType.FETCH: 0.75,
IntentType.DANCE: 0.6,
IntentType.SLEEP: 0.6,
IntentType.WAKE: 0.6,
IntentType.OTHER: 0.5
}
def classify_intent(self, text: str) -> Tuple[IntentType, float, Dict[str, Any]]:
"""Classify intent using multiple approaches and ensemble"""
# Preprocess text
processed_text = self._preprocess_text(text)
# Get results from different classifiers
rule_result = self.rule_classifier.classify_intent(processed_text)
ml_result = self.ml_classifier.classify_intent(processed_text)
# Ensemble the results
final_intent, confidence, entities = self._ensemble_results(
rule_result, ml_result, processed_text
)
return final_intent, confidence, entities
def _preprocess_text(self, text: str) -> str:
"""Preprocess text for better classification"""
# Convert to lowercase
text = text.lower().strip()
# Remove extra whitespace
text = ' '.join(text.split())
# Remove common filler words
fillers = ["um", "uh", "like", "so", "well", "you know", "actually", "basically"]
for filler in fillers:
text = text.replace(filler, "")
# Remove extra spaces again after filler removal
text = ' '.join(text.split())
return text
def _ensemble_results(self, rule_result: Tuple[IntentType, float, Dict],
ml_result: Tuple[IntentType, float, Dict],
text: str) -> Tuple[IntentType, float, Dict[str, Any]]:
"""Combine results from rule-based and ML classifiers"""
rule_intent, rule_conf, rule_entities = rule_result
ml_intent, ml_conf, ml_entities = ml_result
# If both classifiers agree, boost confidence
if rule_intent == ml_intent and rule_conf > 0.5 and ml_conf > 0.5:
combined_conf = min(0.95, (rule_conf + ml_conf) / 2 * 1.2) # Slight boost
combined_entities = {**rule_entities, **ml_entities}
return rule_intent, combined_conf, combined_entities
# Otherwise, use the higher confidence result
if rule_conf >= ml_conf:
return rule_intent, rule_conf, rule_entities
else:
return ml_intent, ml_conf, ml_entities
class RuleBasedIntentClassifier:
"""Rule-based intent classifier using regex patterns and keyword matching"""
def __init__(self):
# Define comprehensive intent patterns
self.intent_patterns = {
IntentType.MOVE: [
# Movement with direction and distance
(r"move\s+(?P<direction>forward|backward|left|right|up|down)\s*(?P<distance>\d+\.?\d*)?\s*(?P<unit>meters?|steps?|cm|feet)?", 0.9),
(r"go\s+(?P<direction>forward|backward|left|right|up|down)\s*(?P<distance>\d+\.?\d*)?\s*(?P<unit>meters?|steps?|cm|feet)?", 0.9),
(r"walk\s+(?P<direction>forward|backward|left|right|up|down)\s*(?P<distance>\d+\.?\d*)?\s*(?P<unit>meters?|steps?|cm|feet)?", 0.9),
(r"step\s+(?P<direction>forward|backward|left|right|up|down)\s*(?P<distance>\d+\.?\d*)?\s*(?P<unit>meters?|steps?|cm|feet)?", 0.9),
# Simple movement commands
(r"move\s+(?P<direction>forward|backward|left|right|up|down)", 0.8),
(r"go\s+(?P<direction>forward|backward|left|right|up|down)", 0.8),
# Turn commands
(r"turn\s+(?P<direction>left|right|around)\s*(?P<angle>\d+\.?\d*)?\s*(?P<unit>degrees?)?", 0.8),
(r"rotate\s+(?P<direction>left|right)\s*(?P<angle>\d+\.?\d*)?\s*(?P<unit>degrees?)?", 0.8),
],
IntentType.GRASP: [
# Grasp with object specification
(r"pick\s+up\s+(?P<object>\w+)", 0.9),
(r"pick\s+(?P<object>\w+)\s+up", 0.9),
(r"grab\s+(?P<object>\w+)", 0.9),
(r"take\s+(?P<object>\w+)", 0.9),
(r"lift\s+(?P<object>\w+)", 0.9),
(r"hold\s+(?P<object>\w+)", 0.8),
(r"pick\s+the\s+(?P<object>\w+)\s+up", 0.9),
# Place commands
(r"put\s+(?P<object>\w+)\s+(?P<location>\w+)", 0.8),
(r"place\s+(?P<object>\w+)\s+(?P<location>\w+)", 0.8),
(r"set\s+(?P<object>\w+)\s+(?P<location>\w+)", 0.8),
],
IntentType.GREET: [
(r"hello|hi|hey|greetings", 0.9),
(r"good\s+(morning|afternoon|evening)", 0.9),
(r"how\s+are\s+you", 0.8),
(r"what's\s+up", 0.7),
(r"nice\s+to\s+meet\s+you", 0.8),
(r"good\s+to\s+see\s+you", 0.8),
],
IntentType.FOLLOW: [
(r"follow\s+(?P<target>\w+)", 0.9),
(r"come\s+with\s+(?P<target>\w+)", 0.8),
(r"stay\s+with\s+(?P<target>\w+)", 0.8),
(r"accompany\s+(?P<target>\w+)", 0.8),
(r"go\s+with\s+(?P<target>\w+)", 0.8),
(r"walk\s+with\s+(?P<target>\w+)", 0.8),
],
IntentType.STOP: [
(r"stop|halt|pause|wait|freeze", 0.9),
(r"don't\s+move", 0.8),
(r"stand\s+still", 0.8),
(r"remain\s+stationary", 0.8),
(r"cease\s+all\s+movement", 0.8),
],
IntentType.SEARCH: [
(r"find\s+(?P<object>\w+)", 0.9),
(r"look\s+for\s+(?P<object>\w+)", 0.9),
(r"search\s+for\s+(?P<object>\w+)", 0.9),
(r"locate\s+(?P<object>\w+)", 0.9),
(r"where\s+is\s+(?P<object>\w+)", 0.8),
(r"find\s+the\s+(?P<object>\w+)", 0.9),
],
IntentType.FETCH: [
(r"bring\s+(?P<object>\w+)", 0.9),
(r"get\s+(?P<object>\w+)", 0.9),
(r"fetch\s+(?P<object>\w+)", 0.9),
(r"carry\s+(?P<object>\w+)", 0.8),
(r"bring\s+(?P<object>\w+)\s+to\s+(?P<destination>\w+)", 0.9),
(r"get\s+(?P<object>\w+)\s+for\s+(?P<recipient>\w+)", 0.9),
],
IntentType.DANCE: [
(r"dance|boogie|groove|shake", 0.8),
(r"move\s+to\s+the\s+beat", 0.8),
(r"show\s+me\s+your\s+dance", 0.8),
(r"perform\s+a\s+dance", 0.8),
(r"dance\s+for\s+(?P<duration>\d+\.?\d*)\s*(?P<unit>seconds?|minutes?)", 0.8),
],
IntentType.SLEEP: [
(r"sleep|rest|power\s+down|shut\s+down|turn\s+off", 0.9),
(r"go\s+to\s+sleep", 0.9),
(r"take\s+a\s+rest", 0.8),
(r"deactivate", 0.8),
(r"enter\s+sleep\s+mode", 0.8),
],
IntentType.WAKE: [
(r"wake\s+up|start|activate|turn\s+on|wake", 0.9),
(r"get\s+up", 0.8),
(r"power\s+on", 0.8),
(r"come\s+online", 0.8),
(r"exit\s+sleep\s+mode", 0.8),
],
}
# Direction normalization
self.direction_map = {
'forward': ['forward', 'ahead', 'straight', 'onward'],
'backward': ['backward', 'back', 'reverse', 'behind'],
'left': ['left', 'west', 'port'],
'right': ['right', 'east', 'starboard'],
'up': ['up', 'top', 'above', 'upward'],
'down': ['down', 'bottom', 'below', 'downward'],
}
# Object category mappings
self.object_categories = {
'container': ['cup', 'bottle', 'box', 'bowl', 'jar', 'glass'],
'tool': ['screwdriver', 'hammer', 'wrench', 'pliers', 'knife', 'fork'],
'food': ['apple', 'banana', 'bread', 'water', 'snack', 'meal'],
'device': ['phone', 'tablet', 'remote', 'controller', 'watch'],
'furniture': ['chair', 'table', 'desk', 'bed', 'sofa', 'cabinet'],
'person': ['person', 'human', 'you', 'me', 'them', 'us'],
'location': ['kitchen', 'bedroom', 'office', 'living room', 'bathroom', 'hallway'],
}
def classify_intent(self, text: str) -> Tuple[IntentType, float, Dict[str, Any]]:
"""Classify intent using rule-based patterns"""
text_lower = text.lower().strip()
best_intent = IntentType.OTHER
best_confidence = 0.0
best_entities = {}
best_pattern = None
# Try each intent type's patterns
for intent_type, patterns in self.intent_patterns.items():
for pattern, base_confidence in patterns:
match = re.search(pattern, text_lower, re.IGNORECASE)
if match:
# Calculate confidence based on match completeness
entities = match.groupdict()
confidence = base_confidence
# Boost confidence if we have complete entity information
if entities:
confidence *= 1.1 # Small boost for having entities
# Update if this is the best match so far
if confidence > best_confidence:
best_intent = intent_type
best_confidence = confidence
best_entities = entities
best_pattern = pattern
# Normalize entities
normalized_entities = self._normalize_entities(best_entities, text_lower)
# Cap confidence at 0.95 to allow for uncertainty
best_confidence = min(0.95, best_confidence)
return best_intent, best_confidence, normalized_entities
def _normalize_entities(self, entities: Dict[str, str], text: str) -> Dict[str, Any]:
"""Normalize extracted entities"""
normalized = {}
for key, value in entities.items():
if value:
if key == 'direction':
normalized[key] = self._normalize_direction(value.lower())
elif key == 'object':
normalized[key] = self._normalize_object(value.lower())
elif key == 'distance':
normalized[key] = float(value) if self._is_float(value) else 1.0
elif key == 'unit':
normalized[key] = value if value else 'meters'
elif key == 'target':
normalized[key] = self._normalize_person(value.lower())
else:
normalized[key] = value.lower()
return normalized
def _normalize_direction(self, direction: str) -> str:
"""Normalize direction keywords"""
for norm_dir, keywords in self.direction_map.items():
if direction in keywords or direction.startswith(tuple(keywords)):
return norm_dir
return direction # Return original if no match
def _normalize_object(self, obj: str) -> str:
"""Normalize object keywords"""
# Check if it matches any category
for category, objects in self.object_categories.items():
if obj in objects:
return obj
# If not in predefined categories, return as is
return obj
def _normalize_person(self, person: str) -> str:
"""Normalize person references"""
person_map = {
'me': 'user',
'you': 'robot',
'yourself': 'robot',
'us': 'user_group'
}
return person_map.get(person, person)
def _is_float(self, value: str) -> bool:
"""Check if value is a float"""
try:
float(value)
return True
except ValueError:
return False
class MLIntentClassifier:
"""Machine learning based intent classifier"""
def __init__(self):
# Create training data for each intent type
self.training_data = self._create_training_data()
# Initialize and train the classifier
self.classifier = Pipeline([
('tfidf', TfidfVectorizer(ngram_range=(1, 2), max_features=10000)),
('nb', MultinomialNB(alpha=0.1))
])
# Train the classifier
self._train_classifier()
# Intent label mapping
self.label_to_intent = {
0: IntentType.OTHER, 1: IntentType.MOVE, 2: IntentType.GRASP,
3: IntentType.GREET, 4: IntentType.FOLLOW, 5: IntentType.STOP,
6: IntentType.SEARCH, 7: IntentType.FETCH, 8: IntentType.DANCE,
9: IntentType.SLEEP, 10: IntentType.WAKE
}
self.intent_to_label = {v: k for k, v in self.label_to_intent.items()}
def _create_training_data(self) -> Tuple[List[str], List[int]]:
"""Create training data for ML classifier"""
texts = []
labels = []
# Training examples for each intent
intent_examples = {
IntentType.MOVE: [
"move forward", "go forward 2 meters", "step forward", "walk forward",
"move backward", "go back", "step back", "move left", "turn left",
"move right", "turn right", "move up", "move down", "go straight"
],
IntentType.GRASP: [
"pick up the ball", "grab the cup", "take the book", "lift the box",
"hold the bottle", "pick up red ball", "grab blue cup", "take green book"
],
IntentType.GREET: [
"hello", "hi", "hey", "good morning", "good afternoon", "good evening",
"how are you", "what's up", "greetings", "nice to meet you"
],
IntentType.FOLLOW: [
"follow me", "come with me", "stay with you", "follow the person",
"accompany me", "go with you", "walk with me"
],
IntentType.STOP: [
"stop", "halt", "pause", "wait", "freeze", "don't move", "stand still"
],
IntentType.SEARCH: [
"find the ball", "look for the cup", "search for the book",
"locate the phone", "where is the ball", "find red ball"
],
IntentType.FETCH: [
"bring me the cup", "get the book", "fetch the phone",
"carry the bottle", "bring the red ball", "get blue cup"
],
IntentType.DANCE: [
"dance", "boogie", "groove", "shake", "move to the beat",
"show me your dance", "perform a dance"
],
IntentType.SLEEP: [
"sleep", "rest", "power down", "shut down", "turn off",
"go to sleep", "take a rest", "deactivate"
],
IntentType.WAKE: [
"wake up", "start", "activate", "turn on", "wake",
"get up", "power on", "come online"
],
IntentType.OTHER: [
"what is your name", "tell me a joke", "what time is it",
"play music", "tell me about yourself", "how old are you"
]
}
for intent, examples in intent_examples.items():
label = self.intent_to_label[intent]
texts.extend(examples)
labels.extend([label] * len(examples))
return texts, labels
def _train_classifier(self):
"""Train the ML classifier"""
texts, labels = self.training_data
self.classifier.fit(texts, labels)
print("ML Intent Classifier trained successfully")
def classify_intent(self, text: str) -> Tuple[IntentType, float, Dict[str, Any]]:
"""Classify intent using ML classifier"""
try:
# Predict intent and get confidence
predicted_label = self.classifier.predict([text])[0]
prediction_probs = self.classifier.predict_proba([text])[0]
# Get the confidence for the predicted class
confidence = max(prediction_probs)
# Convert label back to intent
intent = self.label_to_intent[predicted_label]
# Extract entities using simple heuristics
entities = self._extract_entities_ml(text)
return intent, confidence, entities
except Exception as e:
print(f"Error in ML classification: {e}")
return IntentType.OTHER, 0.1, {}
def _extract_entities_ml(self, text: str) -> Dict[str, Any]:
"""Extract entities using ML-based approach"""
entities = {}
# Simple entity extraction based on keywords
doc = text.lower()
# Look for objects
object_keywords = ['ball', 'cup', 'book', 'bottle', 'phone', 'box', 'toy']
for obj in object_keywords:
if obj in doc:
entities['object'] = obj
break
# Look for directions
direction_keywords = ['forward', 'backward', 'left', 'right', 'up', 'down']
for direction in direction_keywords:
if direction in doc:
entities['direction'] = direction
break
# Look for distances
import re
distance_match = re.search(r'(\d+\.?\d*)\s*(meters?|steps?|cm|feet?)', doc)
if distance_match:
entities['distance'] = float(distance_match.group(1))
entities['unit'] = distance_match.group(2)
return entities
def test_intent_classifier():
"""Test the intent classifier with various inputs"""
classifier = AdvancedIntentClassifier()
test_inputs = [
"Move forward 2 meters",
"Pick up the red ball",
"Hello robot how are you",
"Follow me to the kitchen",
"Stop moving right now",
"Find the blue cup on the table",
"Bring me the book from the shelf",
"Dance for 30 seconds",
"Go to sleep now",
"Wake up and start working",
"What is the weather like today",
"Turn left and go forward"
]
print("Testing Advanced Intent Classifier:")
print("-" * 50)
for text in test_inputs:
intent, confidence, entities = classifier.classify_intent(text)
print(f"Input: '{text}'")
print(f" Intent: {intent.value}")
print(f" Confidence: {confidence:.3f}")
print(f" Entities: {entities}")
print()
if __name__ == "__main__":
test_intent_classifier()
Robotics Action Mapping
Action Execution System
# python/action_mapping.py
import asyncio
import threading
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass
import time
import math
from enum import Enum
class RobotActionStatus(Enum):
"""Status of robot actions"""
PENDING = "pending"
EXECUTING = "executing"
SUCCESS = "success"
FAILED = "failed"
CANCELLED = "cancelled"
@dataclass
class RobotAction:
"""Represents a robot action to be executed"""
intent_type: IntentType
parameters: Dict[str, Any]
priority: int = 1
timeout: float = 10.0
status: RobotActionStatus = RobotActionStatus.PENDING
start_time: Optional[float] = None
end_time: Optional[float] = None
class RobotActionExecutor:
"""Executes robot actions based on recognized intents"""
def __init__(self):
self.current_action: Optional[RobotAction] = None
self.action_queue = asyncio.Queue()
self.is_running = False
self.executor_thread = None
# Robot state
self.robot_position = [0.0, 0.0, 0.0] # x, y, theta
self.robot_velocity = [0.0, 0.0, 0.0]
self.is_moving = False
self.is_grasping = False
self.held_object = None
def start_execution_loop(self):
"""Start the action execution loop in a separate thread"""
self.is_running = True
self.executor_thread = threading.Thread(target=self._execution_loop, daemon=True)
self.executor_thread.start()
def stop_execution_loop(self):
"""Stop the action execution loop"""
self.is_running = False
if self.executor_thread:
self.executor_thread.join()
async def _execution_loop(self):
"""Main execution loop for robot actions"""
while self.is_running:
try:
# Get next action from queue
action = await asyncio.wait_for(self.action_queue.get(), timeout=0.1)
if action:
await self._execute_action(action)
self.action_queue.task_done()
except asyncio.TimeoutError:
# No action available, continue loop
continue
except Exception as e:
print(f"Error in execution loop: {e}")
async def _execute_action(self, action: RobotAction):
"""Execute a single robot action"""
print(f"Executing action: {action.intent_type.value} with params: {action.parameters}")
action.status = RobotActionStatus.EXECUTING
action.start_time = time.time()
try:
# Execute action based on intent type
if action.intent_type == IntentType.MOVE:
success = await self._execute_move_action(action.parameters)
elif action.intent_type == IntentType.GRASP:
success = await self._execute_grasp_action(action.parameters)
elif action.intent_type == IntentType.GREET:
success = await self._execute_greet_action(action.parameters)
elif action.intent_type == IntentType.FOLLOW:
success = await self._execute_follow_action(action.parameters)
elif action.intent_type == IntentType.STOP:
success = await self._execute_stop_action(action.parameters)
elif action.intent_type == IntentType.SEARCH:
success = await self._execute_search_action(action.parameters)
elif action.intent_type == IntentType.FETCH:
success = await self._execute_fetch_action(action.parameters)
elif action.intent_type == IntentType.DANCE:
success = await self._execute_dance_action(action.parameters)
elif action.intent_type == IntentType.SLEEP:
success = await self._execute_sleep_action(action.parameters)
elif action.intent_type == IntentType.WAKE:
success = await self._execute_wake_action(action.parameters)
else:
success = await self._execute_other_action(action.parameters)
action.status = RobotActionStatus.SUCCESS if success else RobotActionStatus.FAILED
action.end_time = time.time()
except Exception as e:
print(f"Error executing action {action.intent_type.value}: {e}")
action.status = RobotActionStatus.FAILED
action.end_time = time.time()
async def _execute_move_action(self, params: Dict[str, Any]) -> bool:
"""Execute movement action"""
direction = params.get('direction', 'forward')
distance = params.get('distance', 1.0) # Default 1 meter
unit = params.get('unit', 'meters')
print(f"Moving {direction} for {distance} {unit}")
# Convert distance to meters if needed
if unit.lower() in ['cm', 'centimeters']:
distance = distance / 100.0
elif unit.lower() in ['feet', 'ft']:
distance = distance * 0.3048
# Simulate movement
await self._simulate_movement(direction, distance)
# Update robot position
self._update_robot_position(direction, distance)
return True
async def _simulate_movement(self, direction: str, distance: float):
"""Simulate robot movement"""
# Simulate movement time based on distance
movement_time = distance / 0.5 # Assume 0.5 m/s speed
print(f"Simulating movement for {movement_time:.2f} seconds")
await asyncio.sleep(min(movement_time, 5.0)) # Cap at 5 seconds
def _update_robot_position(self, direction: str, distance: float):
"""Update robot position based on movement"""
if direction == 'forward':
self.robot_position[0] += distance
elif direction == 'backward':
self.robot_position[0] -= distance
elif direction == 'left':
self.robot_position[1] += distance
elif direction == 'right':
self.robot_position[1] -= distance
elif direction == 'up':
self.robot_position[2] += distance
elif direction == 'down':
self.robot_position[2] -= distance
async def _execute_grasp_action(self, params: Dict[str, Any]) -> bool:
"""Execute grasp action"""
obj = params.get('object', 'object')
print(f"Attempting to grasp {obj}")
# Simulate grasp action
await asyncio.sleep(2.0) # Simulate grasp time
# Update robot state
self.is_grasping = True
self.held_object = obj
print(f"Successfully grasped {obj}")
return True
async def _execute_greet_action(self, params: Dict[str, Any]) -> bool:
"""Execute greeting action"""
print("Robot is greeting")
# Simulate greeting (head nod, lights, etc.)
await asyncio.sleep(1.0)
print("Greeting completed")
return True
async def _execute_follow_action(self, params: Dict[str, Any]) -> bool:
"""Execute follow action"""
target = params.get('target', 'person')
print(f"Following {target}")
# In real implementation, this would use tracking
# For simulation, just acknowledge
await asyncio.sleep(1.0)
print(f"Started following {target}")
return True
async def _execute_stop_action(self, params: Dict[str, Any]) -> bool:
"""Execute stop action"""
print("Stopping robot movement")
# Stop any ongoing movement
self.is_moving = False
print("Robot stopped")
return True
async def _execute_search_action(self, params: Dict[str, Any]) -> bool:
"""Execute search action"""
obj = params.get('object', 'object')
print(f"Searching for {obj}")
# Simulate search behavior
await asyncio.sleep(3.0) # Simulate search time
print(f"Search for {obj} completed")
return True
async def _execute_fetch_action(self, params: Dict[str, Any]) -> bool:
"""Execute fetch action"""
obj = params.get('object', 'object')
destination = params.get('destination', 'here')
print(f"Fetching {obj} and bringing to {destination}")
# Simulate fetch sequence: search, grasp, move to destination
await asyncio.sleep(1.0) # Search
await asyncio.sleep(1.0) # Grasp
await asyncio.sleep(2.0) # Move to destination
print(f"Fetched {obj} and brought to {destination}")
return True
async def _execute_dance_action(self, params: Dict[str, Any]) -> bool:
"""Execute dance action"""
duration = params.get('duration', 10.0) # Default 10 seconds
print(f"Dancing for {duration} seconds")
# Simulate dance movements
dance_time = min(duration, 30.0) # Cap dance time
await asyncio.sleep(dance_time)
print("Dance completed")
return True
async def _execute_sleep_action(self, params: Dict[str, Any]) -> bool:
"""Execute sleep action"""
print("Robot entering sleep mode")
# Simulate sleep preparation
await asyncio.sleep(1.0)
# Update robot state
self.is_moving = False
self.is_grasping = False
print("Robot is now in sleep mode")
return True
async def _execute_wake_action(self, params: Dict[str, Any]) -> bool:
"""Execute wake action"""
print("Waking up robot")
# Simulate wake sequence
await asyncio.sleep(1.0)
print("Robot is now awake and ready")
return True
async def _execute_other_action(self, params: Dict[str, Any]) -> bool:
"""Execute other/unknown action"""
print("Received unknown command, acknowledging")
await asyncio.sleep(0.5)
return True
def queue_action(self, intent_type: IntentType, parameters: Dict[str, Any],
priority: int = 1, timeout: float = 10.0):
"""Queue an action for execution"""
action = RobotAction(
intent_type=intent_type,
parameters=parameters,
priority=priority,
timeout=timeout
)
# Add to queue (in a thread-safe manner)
asyncio.run_coroutine_threadsafe(
self.action_queue.put(action),
asyncio.get_event_loop()
)
print(f"Action queued: {intent_type.value}")
class RoboticsActionMapper:
"""Maps intents to robot actions"""
def __init__(self):
self.action_executor = RobotActionExecutor()
self.action_executor.start_execution_loop()
def map_intent_to_action(self, intent: SpeechIntent):
"""Map a recognized intent to a robot action"""
# Log the intent
print(f"Mapping intent: {intent.intent_type.value} - '{intent.text}'")
# Queue the action with appropriate parameters
self.action_executor.queue_action(
intent_type=intent.intent_type,
parameters=intent.entities,
priority=self._determine_priority(intent)
)
def _determine_priority(self, intent: SpeechIntent) -> int:
"""Determine action priority based on intent type and context"""
# Higher priority for safety-related commands
if intent.intent_type in [IntentType.STOP, IntentType.SLEEP]:
return 10 # High priority
# Normal priority for most commands
return 5
def shutdown(self):
"""Shutdown the action mapper and executor"""
self.action_executor.stop_execution_loop()
def demonstrate_action_mapping():
"""Demonstrate intent to action mapping"""
print("Setting up Robotics Action Mapper...")
# Initialize action mapper
action_mapper = RoboticsActionMapper()
# Create some sample intents to test
sample_intents = [
SpeechIntent(
text="move forward 2 meters",
intent_type=IntentType.MOVE,
confidence=0.85,
timestamp=time.time(),
entities={'direction': 'forward', 'distance': 2.0, 'unit': 'meters'},
raw_transcription="Move forward 2 meters"
),
SpeechIntent(
text="pick up the red ball",
intent_type=IntentType.GRASP,
confidence=0.90,
timestamp=time.time(),
entities={'object': 'ball'},
raw_transcription="Pick up the red ball"
),
SpeechIntent(
text="hello robot",
intent_type=IntentType.GREET,
confidence=0.95,
timestamp=time.time(),
entities={},
raw_transcription="Hello robot"
),
SpeechIntent(
text="stop moving",
intent_type=IntentType.STOP,
confidence=0.88,
timestamp=time.time(),
entities={},
raw_transcription="Stop moving"
)
]
print("Testing intent to action mapping:")
for intent in sample_intents:
print(f"\nProcessing intent: {intent.intent_type.value}")
action_mapper.map_intent_to_action(intent)
time.sleep(0.5) # Small delay between actions
print("\nActions queued for execution.")
print("Note: In a real system, these would be executed by the robot.")
# Let some actions execute before shutting down
time.sleep(5)
# Shutdown
action_mapper.shutdown()
print("Action mapper shutdown complete.")
if __name__ == "__main__":
demonstrate_action_mapping()
Performance Optimization and Real-time Processing
Optimized Whisper Pipeline
# python/optimized_pipeline.py
import whisper
import torch
import numpy as np
import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor
import queue
import time
from typing import Dict, List, Optional, Callable
import gc
class OptimizedWhisperPipeline:
"""Optimized Whisper pipeline for real-time robotic applications"""
def __init__(self,
model_size: str = "base",
device: str = "cuda",
num_threads: int = 2,
buffer_size: int = 48000): # 3 seconds at 16kHz
self.device = device if torch.cuda.is_available() and device == "cuda" else "cpu"
self.num_threads = num_threads
self.buffer_size = buffer_size
# Load model with optimizations
self.model = whisper.load_model(model_size, device=self.device)
# Use smaller models for faster processing if needed
self.model_size = model_size
# Audio processing parameters
self.sample_rate = 16000
self.chunk_size = 1024
# Processing queues
self.audio_queue = queue.Queue(maxsize=10) # Limit queue size
self.result_queue = queue.Queue(maxsize=10)
# Threading
self.executor = ThreadPoolExecutor(max_workers=num_threads)
self.is_running = False
# Statistics
self.stats = {
'processed_audio': 0,
'transcription_time': 0.0,
'average_latency': 0.0,
'dropped_chunks': 0
}
def start_processing(self):
"""Start the processing pipeline"""
self.is_running = True
# Start processing thread
self.process_thread = threading.Thread(target=self._processing_loop, daemon=True)
self.process_thread.start()
print(f"Optimized Whisper pipeline started on {self.device}")
def stop_processing(self):
"""Stop the processing pipeline"""
self.is_running = False
self.executor.shutdown(wait=True)
def _processing_loop(self):
"""Main processing loop"""
audio_buffer = np.array([], dtype=np.float32)
while self.is_running:
try:
# Process audio in chunks
if not self.audio_queue.empty():
try:
chunk = self.audio_queue.get_nowait()
audio_buffer = np.concatenate([audio_buffer, chunk])
except queue.Empty:
pass
# Process when we have enough audio
if len(audio_buffer) >= self.sample_rate * 0.5: # 0.5 seconds minimum
# Submit for transcription
if len(audio_buffer) > self.buffer_size:
# Trim buffer to prevent excessive memory usage
audio_buffer = audio_buffer[-self.buffer_size:]
# Submit to thread pool for processing
future = self.executor.submit(self._transcribe_audio, audio_buffer.copy())
# Process result when ready
if future.done():
result = future.result()
if result:
self.result_queue.put(result)
# Keep some overlap for continuity
audio_buffer = audio_buffer[-int(self.sample_rate * 0.1):] # Keep last 0.1s
time.sleep(0.01) # Small delay
except Exception as e:
print(f"Error in processing loop: {e}")
time.sleep(0.1)
def _transcribe_audio(self, audio_data: np.ndarray) -> Optional[str]:
"""Transcribe audio data with optimizations"""
start_time = time.time()
try:
# Pad audio to minimum length if needed
if len(audio_data) < self.sample_rate * 0.5: # Minimum 0.5 seconds
pad_length = int(self.sample_rate * 0.5) - len(audio_data)
audio_data = np.pad(audio_data, (0, pad_length), mode='constant')
# Transcribe with optimized settings
result = self.model.transcribe(
audio_data,
language="en",
task="transcribe",
temperature=0.0, # Deterministic results
compression_ratio_threshold=None, # Disable language detection
logprob_threshold=None, # Disable logprob threshold
no_speech_threshold=None # Disable no speech threshold
)
transcription = result["text"].strip()
# Update statistics
processing_time = time.time() - start_time
self.stats['processed_audio'] += 1
self.stats['transcription_time'] += processing_time
self.stats['average_latency'] = (
self.stats['transcription_time'] / self.stats['processed_audio']
)
return transcription
except Exception as e:
print(f"Error transcribing audio: {e}")
self.stats['dropped_chunks'] += 1
return None
def submit_audio_chunk(self, audio_chunk: np.ndarray):
"""Submit an audio chunk for processing"""
try:
self.audio_queue.put_nowait(audio_chunk)
except queue.Full:
# Drop old chunk if queue is full
try:
self.audio_queue.get_nowait() # Remove oldest
self.audio_queue.put_nowait(audio_chunk) # Add new
self.stats['dropped_chunks'] += 1
except queue.Empty:
pass # Queue was empty, just continue
def get_transcription_result(self) -> Optional[str]:
"""Get the next transcription result if available"""
try:
return self.result_queue.get_nowait()
except queue.Empty:
return None
def get_stats(self) -> Dict:
"""Get current statistics"""
return self.stats.copy()
class RealTimeWhisperProcessor:
"""Real-time processor that integrates audio capture with optimized Whisper"""
def __init__(self,
model_size: str = "base",
device: str = "cuda",
enable_vad: bool = True):
# Initialize optimized pipeline
self.pipeline = OptimizedWhisperPipeline(
model_size=model_size,
device=device
)
# Initialize audio processor
self.audio_processor = AudioProcessor()
# Initialize intent classifier
self.intent_classifier = AdvancedIntentClassifier()
# Initialize action mapper
self.action_mapper = RoboticsActionMapper()
# Callbacks
self.transcription_callback: Optional[Callable] = None
self.intent_callback: Optional[Callable] = None
# Setup audio processor callback
self.audio_processor.set_callbacks(
audio_processed=self._on_audio_processed
)
# Processing state
self.is_active = False
self.processing_thread = None
def set_callbacks(self,
transcription: Optional[Callable] = None,
intent: Optional[Callable] = None):
"""Set callbacks for processing events"""
self.transcription_callback = transcription
self.intent_callback = intent
def _on_audio_processed(self, audio_data: np.ndarray):
"""Callback when audio is processed"""
# Submit to Whisper pipeline
self.pipeline.submit_audio_chunk(audio_data)
# Check for results
result = self.pipeline.get_transcription_result()
if result:
self._process_transcription(result)
def _process_transcription(self, transcription: str):
"""Process a transcription result"""
if len(transcription.strip()) > 3: # Filter out very short transcriptions
# Classify intent
intent_type, confidence, entities = self.intent_classifier.classify_intent(transcription)
# Create speech intent
speech_intent = SpeechIntent(
text=self._extract_command(transcription),
intent_type=intent_type,
confidence=confidence,
timestamp=time.time(),
entities=entities,
raw_transcription=transcription
)
# Execute callback if set
if self.intent_callback:
self.intent_callback(speech_intent)
# Map to robot action
self.action_mapper.map_intent_to_action(speech_intent)
def _extract_command(self, transcription: str) -> str:
"""Extract the main command from transcription"""
# Remove common filler words and normalize
command = transcription.lower().strip()
fillers = ["um", "uh", "like", "so", "well", "you know", "actually", "basically"]
for filler in fillers:
command = command.replace(filler, "").strip()
# Remove extra spaces
import re
command = re.sub(r'\s+', ' ', command).strip()
return command
def start_listening(self):
"""Start real-time listening and processing"""
# Start Whisper pipeline
self.pipeline.start_processing()
# Start audio processing
self.audio_processor.start_processing()
self.is_active = True
print("Real-time Whisper processor started")
def stop_listening(self):
"""Stop real-time processing"""
self.is_active = False
# Stop audio processing
self.audio_processor.stop_processing()
# Stop Whisper pipeline
self.pipeline.stop_processing()
# Shutdown action mapper
self.action_mapper.shutdown()
print("Real-time Whisper processor stopped")
def benchmark_pipeline():
"""Benchmark the optimized pipeline"""
print("Benchmarking Optimized Whisper Pipeline...")
# Test with different model sizes
models = ["tiny", "base", "small"]
for model_size in models:
print(f"\nTesting {model_size} model:")
start_time = time.time()
# Initialize pipeline
pipeline = OptimizedWhisperPipeline(
model_size=model_size,
device="cuda" if torch.cuda.is_available() else "cpu"
)
# Generate test audio (silence for speed)
test_audio = np.zeros(16000, dtype=np.float32) # 1 second of silence
# Process multiple chunks
for i in range(10):
pipeline.submit_audio_chunk(test_audio)
time.sleep(0.01) # Small delay
# Wait for processing to complete
time.sleep(2)
stats = pipeline.get_stats()
print(f" Processed: {stats['processed_audio']} chunks")
print(f" Average latency: {stats['average_latency']:.3f}s")
print(f" Dropped chunks: {stats['dropped_chunks']}")
pipeline.stop_processing()
end_time = time.time()
print(f" Total time: {end_time - start_time:.2f}s")
if __name__ == "__main__":
# Run benchmark
benchmark_pipeline()
print("\n" + "="*50)
print("Real-time processing demonstration:")
print("This would start real-time audio processing with Whisper.")
print("In a real application, this would continuously listen")
print("and process speech for robot control.")
print("="*50)
Integration with Humanoid Robot Systems
ROS 2 Integration
# python/ros_integration.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import String, Float32
from sensor_msgs.msg import AudioData
from geometry_msgs.msg import Twist
from builtin_interfaces.msg import Time
from .whisper_robotics import WhisperRobotInterface, SpeechIntent, IntentType
from .action_mapping import RoboticsActionMapper
import numpy as np
import threading
import time
class WhisperROS2Bridge(Node):
"""ROS 2 bridge for Whisper-based voice commands"""
def __init__(self):
super().__init__('whisper_ros2_bridge')
# Initialize Whisper interface
self.whisper_interface = WhisperRobotInterface(
model_size="base",
device="cuda" if torch.cuda.is_available() else "cpu",
language="en"
)
# Initialize action mapper
self.action_mapper = RoboticsActionMapper()
# ROS 2 publishers
self.intent_pub = self.create_publisher(String, '/voice_intent', 10)
self.status_pub = self.create_publisher(String, '/voice_status', 10)
self.confidence_pub = self.create_publisher(Float32, '/voice_confidence', 10)
# ROS 2 subscribers
self.audio_sub = self.create_subscription(
AudioData,
'/audio_input',
self.audio_callback,
10
)
# Robot control publisher
self.cmd_vel_pub = self.create_publisher(Twist, '/cmd_vel', 10)
# Parameters
self.declare_parameter('model_size', 'base')
self.declare_parameter('language', 'en')
self.declare_parameter('confidence_threshold', 0.7)
# Setup audio processing
self.audio_buffer = np.array([], dtype=np.int16)
self.min_audio_length = 16000 * 2 # 2 seconds of audio
# Start Whisper processing
self.whisper_interface.start_listening()
self.get_logger().info("Whisper ROS 2 Bridge initialized")
def audio_callback(self, msg):
"""Handle incoming audio data"""
try:
# Convert audio data to numpy array
audio_data = np.frombuffer(msg.data, dtype=np.int16).astype(np.float32) / 32768.0
# Add to buffer
self.audio_buffer = np.concatenate([self.audio_buffer, audio_data])
# Process when buffer is large enough
if len(self.audio_buffer) >= self.min_audio_length:
# Process with Whisper
intent = self.whisper_interface.process_audio(self.audio_buffer)
if intent:
self.handle_intent(intent)
# Keep overlap
self.audio_buffer = self.audio_buffer[-int(16000 * 0.5):] # Keep last 0.5s
except Exception as e:
self.get_logger().error(f"Error processing audio: {e}")
def handle_intent(self, intent: SpeechIntent):
"""Handle recognized intent"""
self.get_logger().info(f"Recognized intent: {intent.intent_type.value} - '{intent.text}'")
# Publish intent
intent_msg = String()
intent_msg.data = f"{intent.intent_type.value}:{intent.text}"
self.intent_pub.publish(intent_msg)
# Publish confidence
confidence_msg = Float32()
confidence_msg.data = intent.confidence
self.confidence_pub.publish(confidence_msg)
# Map to robot action
self.action_mapper.map_intent_to_action(intent)
# Log status
status_msg = String()
status_msg.data = f"Processed: {intent.intent_type.value} with confidence {intent.confidence:.2f}"
self.status_pub.publish(status_msg)
def destroy_node(self):
"""Cleanup when node is destroyed"""
self.whisper_interface.stop_listening()
self.action_mapper.shutdown()
super().destroy_node()
def main(args=None):
"""Main function to run the Whisper ROS 2 bridge"""
rclpy.init(args=args)
# Create and run the bridge node
bridge = WhisperROS2Bridge()
try:
rclpy.spin(bridge)
except KeyboardInterrupt:
pass
finally:
bridge.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
Best Practices for Whisper in Robotics
Design Guidelines
- Model Selection: Choose model size based on latency vs. accuracy requirements
- Audio Quality: Ensure good audio input quality for best recognition
- Context Awareness: Use context to improve intent classification
- Robustness: Handle failures gracefully and provide feedback
- Privacy: Consider privacy implications of voice data
Performance Considerations
- Latency: Optimize for real-time response requirements
- Resource Usage: Balance computational requirements with robot capabilities
- Battery Life: Consider power consumption for mobile robots
- Network Usage: Plan for offline vs. online processing needs
- Reliability: Implement fallback mechanisms for critical functions
Hands-On Exercise
Exercise: Building a Voice-Controlled Robot
-
Setup Whisper Environment
- Install Whisper and dependencies
- Configure for robotics application
-
Implement Audio Processing
- Set up real-time audio capture
- Implement voice activity detection
- Add noise reduction
-
Create Intent Classification
- Build rule-based classifier
- Implement ML-based classifier
- Create ensemble approach
-
Develop Action Mapping
- Map intents to robot actions
- Implement action execution system
- Add safety checks and validation
-
Test and Optimize
- Test with various commands
- Optimize for real-time performance
- Validate in real robotic scenarios
Summary
The Whisper voice-to-intent pipeline provides a powerful foundation for natural human-robot interaction. By combining state-of-the-art speech recognition with intelligent intent classification and action mapping, we can create intuitive voice interfaces for humanoid robots. Proper optimization for real-time performance and integration with robotic systems enables responsive and natural voice control. The key is balancing accuracy with responsiveness while ensuring robust operation in real-world environments.
Learning Path Adjustment
Based on your experience level, you may want to focus on:
- Beginner: Focus on basic Whisper setup and simple intent classification
- Intermediate: Dive deeper into audio preprocessing and real-time optimization
- Advanced: Explore custom model fine-tuning and complex multimodal integration