Tutorial: How to Build a Voice Assistant with Whisper 3, PyAudio 0.2.14, and Python 3.13

mkdir whisper-voice-assistant && cd whisper-voice-assistant\n\n# Create Python 3.13 virtual environment\npython3.13 -m venv venv\n\n# Activate virtual environment (macOS/Linux)\nsource venv/bin/activate\n\n# Activate virtual environment (Windows)\n# .\\venv\\Scripts\\activate\n\n# Verify Python version\npython --version  # Should output Python 3.13.0

# requirements.txt\nopenai-whisper==3.1.0\npyaudio==0.2.14\ntorch==2.3.0  # Whisper 3 requires PyTorch 2.0+\nnumpy==1.26.4\npytest==8.2.0\npytest-benchmark==4.0.0

pip install -r requirements.txt

import sys\nimport pyaudio\nimport whisper\nimport torch\n\nprint(f\"Python version: {sys.version}\")\nprint(f\"PyAudio version: {pyaudio.__version__}\")\nprint(f\"Whisper version: {whisper.__version__}\")\nprint(f\"PyTorch version: {torch.__version__}\")\nprint(f\"CUDA available: {torch.cuda.is_available()}\")\n\n# List PyAudio input devices\naudio = pyaudio.PyAudio()\nprint(\"\\nAvailable input devices:\")\nfor i in range(audio.get_device_count()):\n    device_info = audio.get_device_info_by_index(i)\n    if device_info[\"maxInputChannels\"] > 0:\n        print(f\"  {i}: {device_info['name']}\")\naudio.terminate()

Python version: 3.13.0 (main, Oct  2 2024, 12:00:00) [Clang 15.0.0]\nPyAudio version: 0.2.14\nWhisper version: 3.1.0\nPyTorch version: 2.3.0\nCUDA available: False\nAvailable input devices:\n  0: MacBook Pro Microphone (Built-in)

import pyaudio\nimport wave\nimport time\nfrom typing import Optional, Generator\nimport logging\n\n# Configure logging for audio capture debugging\nlogging.basicConfig(\n    level=logging.INFO,\n    format=\"%(asctime)s - %(levelname)s - %(message)s\"\n)\nlogger = logging.getLogger(__name__)\n\nclass AudioCapture:\n    \"\"\"Handles real-time audio capture via PyAudio 0.2.14 with error handling.\"\"\"\n    \n    def __init__(\n        self,\n        sample_rate: int = 16000,\n        channels: int = 1,\n        chunk_size: int = 1024,\n        format_: int = pyaudio.paInt16\n    ):\n        self.sample_rate = sample_rate\n        self.channels = channels\n        self.chunk_size = chunk_size\n        self.format_ = format_\n        self.audio = pyaudio.PyAudio()\n        self.stream: Optional[pyaudio.Stream] = None\n        logger.info(f\"Initialized AudioCapture with sample rate {sample_rate}, channels {channels}\")\n\n    def list_input_devices(self) -> Generator[dict, None, None]:\n        \"\"\"List all available input audio devices with error handling.\"\"\"\n        device_count = self.audio.get_device_count()\n        for i in range(device_count):\n            try:\n                device_info = self.audio.get_device_info_by_index(i)\n                if device_info.get(\"maxInputChannels\") > 0:\n                    yield {\n                        \"index\": i,\n                        \"name\": device_info.get(\"name\"),\n                        \"channels\": device_info.get(\"maxInputChannels\"),\n                        \"sample_rate\": int(device_info.get(\"defaultSampleRate\"))\n                    }\n            except Exception as e:\n                logger.error(f\"Failed to get info for device {i}: {e}\")\n                continue\n\n    def start_capture(self, device_index: Optional[int] = None) -> None:\n        \"\"\"Start audio capture stream with specified input device.\"\"\"\n        try:\n            self.stream = self.audio.open(\n                format=self.format_,\n                channels=self.channels,\n                rate=self.sample_rate,\n                input=True,\n                input_device_index=device_index,\n                frames_per_buffer=self.chunk_size,\n                stream_callback=None  # Blocking mode for simplicity\n            )\n            logger.info(f\"Started audio capture stream on device {device_index or 'default'}\")\n        except Exception as e:\n            logger.error(f\"Failed to start audio stream: {e}\")\n            raise RuntimeError(f\"Audio capture initialization failed: {e}\") from e\n\n    def capture_chunk(self) -> bytes:\n        \"\"\"Capture a single chunk of audio data with underrun handling.\"\"\"\n        if not self.stream or not self.stream.is_active():\n            raise RuntimeError(\"Audio stream is not active. Call start_capture first.\")\n        try:\n            data = self.stream.read(self.chunk_size, exception_on_overflow=False)\n            return data\n        except Exception as e:\n            logger.error(f\"Failed to capture audio chunk: {e}\")\n            raise\n\n    def stop_capture(self) -> None:\n        \"\"\"Stop and close the audio capture stream.\"\"\"\n        if self.stream:\n            try:\n                self.stream.stop_stream()\n                self.stream.close()\n                logger.info(\"Stopped audio capture stream\")\n            except Exception as e:\n                logger.error(f\"Error stopping stream: {e}\")\n        self.audio.terminate()\n        logger.info(\"Terminated PyAudio instance\")\n\n    def save_to_wav(self, audio_data: list[bytes], output_path: str) -> None:\n        \"\"\"Save captured audio chunks to a WAV file for debugging.\"\"\"\n        try:\n            with wave.open(output_path, \"wb\") as wf:\n                wf.setnchannels(self.channels)\n                wf.setsampwidth(self.audio.get_sample_size(self.format_))\n                wf.setframerate(self.sample_rate)\n                wf.writeframes(b\"\".join(audio_data))\n            logger.info(f\"Saved {len(audio_data)} chunks to {output_path}\")\n        except Exception as e:\n            logger.error(f\"Failed to save WAV file: {e}\")\n            raise\n\nif __name__ == \"__main__\":\n    # Example usage: capture 5 seconds of audio and save to file\n    capture = AudioCapture()\n    # List input devices to find your microphone\n    print(\"Available input devices:\")\n    for device in capture.list_input_devices():\n        print(f\"  Index {device['index']}: {device['name']} (Channels: {device['channels']})\")\n    \n    capture.start_capture(device_index=None)  # Use default input device\n    print(\"Capturing 5 seconds of audio...\")\n    audio_chunks = []\n    start_time = time.time()\n    while time.time() - start_time < 5:\n        chunk = capture.capture_chunk()\n        audio_chunks.append(chunk)\n    capture.stop_capture()\n    capture.save_to_wav(audio_chunks, \"test_capture.wav\")\n    print(\"Saved test capture to test_capture.wav\")

import whisper\nimport torch\nimport logging\nfrom pathlib import Path\nfrom typing import Optional, Dict, Any\nimport time\n\n# Configure logging for transcription debugging\nlogging.basicConfig(\n    level=logging.INFO,\n    format=\"%(asctime)s - %(levelname)s - %(message)s\"\n)\nlogger = logging.getLogger(__name__)\n\nclass WhisperTranscriber:\n    \"\"\"Handles audio transcription using Whisper 3 with model caching and error handling.\"\"\"\n    \n    def __init__(\n        self,\n        model_name: str = \"large-v3\",\n        device: Optional[str] = None,\n        compute_type: str = \"float16\"\n    ):\n        \"\"\"\n        Initialize Whisper transcriber.\n        \n        Args:\n            model_name: Whisper 3 model size (tiny, base, small, medium, large-v3)\n            device: Device to run inference on (cuda, cpu, mps)\n            compute_type: Compute precision (float16, int8, float32)\n        \"\"\"\n        self.model_name = model_name\n        self.device = device or (\"cuda\" if torch.cuda.is_available() else \"cpu\")\n        self.compute_type = compute_type\n        self.model: Optional[whisper.Whisper] = None\n        self._load_model()\n        logger.info(f\"Initialized WhisperTranscriber with model {model_name} on {self.device}\")\n\n    def _load_model(self) -> None:\n        \"\"\"Load Whisper 3 model with memory mapping and error handling.\"\"\"\n        try:\n            logger.info(f\"Loading Whisper model {self.model_name}...\")\n            start_time = time.perf_counter()\n            # Use download_root to cache models in project directory\n            self.model = whisper.load_model(\n                name=self.model_name,\n                device=self.device,\n                download_root=Path(\"./whisper_models\"),\n                in_memory=False  # Memory map model to reduce RAM usage\n            )\n            load_time = time.perf_counter() - start_time\n            logger.info(f\"Loaded model in {load_time:.2f}s. Model size: {self._get_model_size():.2f}GB\")\n        except Exception as e:\n            logger.error(f\"Failed to load Whisper model: {e}\")\n            raise RuntimeError(f\"Model loading failed: {e}\") from e\n\n    def _get_model_size(self) -> float:\n        \"\"\"Calculate model size in GB (approximate).\"\"\"\n        model_sizes = {\n            \"tiny\": 0.07,\n            \"base\": 0.14,\n            \"small\": 0.24,\n            \"medium\": 0.76,\n            \"large-v3\": 1.5\n        }\n        return model_sizes.get(self.model_name, 0.0)\n\n    def transcribe(\n        self,\n        audio_path: str,\n        language: Optional[str] = \"en\",\n        beam_size: int = 5,\n        best_of: int = 5\n    ) -> Dict[str, Any]:\n        \"\"\"\n        Transcribe audio file to text with Whisper 3.\n        \n        Args:\n            audio_path: Path to WAV/MP3 audio file\n            language: Target language code (en, es, fr, etc.)\n            beam_size: Beam size for decoding (higher = more accurate, slower)\n            best_of: Number of candidates to consider for best result\n            \n        Returns:\n            Dictionary with transcription text, segments, and language\n        \"\"\"\n        if not Path(audio_path).exists():\n            raise FileNotFoundError(f\"Audio file not found: {audio_path}\")\n        if not self.model:\n            raise RuntimeError(\"Whisper model is not loaded. Call _load_model first.\")\n        \n        try:\n            logger.info(f\"Transcribing {audio_path}...\")\n            start_time = time.perf_counter()\n            result = self.model.transcribe(\n                audio=audio_path,\n                language=language,\n                beam_size=beam_size,\n                best_of=best_of,\n                fp16=self.compute_type == \"float16\" and self.device != \"cpu\"\n            )\n            transcribe_time = time.perf_counter() - start_time\n            logger.info(f\"Transcription completed in {transcribe_time:.2f}s. Text: {result['text'][:50]}...\")\n            return {\n                \"text\": result[\"text\"],\n                \"segments\": result[\"segments\"],\n                \"language\": result[\"language\"],\n                \"transcribe_time_s\": transcribe_time\n            }\n        except Exception as e:\n            logger.error(f\"Transcription failed for {audio_path}: {e}\")\n            raise\n\n    def transcribe_chunk(self, audio_chunk: bytes, sample_rate: int = 16000) -> str:\n        \"\"\"Transcribe a raw audio chunk (bytes) directly without saving to file.\"\"\"\n        try:\n            # Convert raw bytes to numpy array for Whisper\n            import numpy as np\n            audio_np = np.frombuffer(audio_chunk, dtype=np.int16).astype(np.float32) / 32768.0\n            start_time = time.perf_counter()\n            result = self.model.transcribe(\n                audio=audio_np,\n                language=\"en\",\n                beam_size=1  # Faster for real-time chunks\n            )\n            transcribe_time = time.perf_counter() - start_time\n            logger.debug(f\"Chunk transcribed in {transcribe_time:.2f}s\")\n            return result[\"text\"]\n        except Exception as e:\n            logger.error(f\"Chunk transcription failed: {e}\")\n            return \"\"\n\nif __name__ == \"__main__\":\n    # Example usage: transcribe test_capture.wav\n    transcriber = WhisperTranscriber(model_name=\"base\")  # Use base for faster testing\n    try:\n        result = transcriber.transcribe(\"test_capture.wav\", language=\"en\")\n        print(f\"Transcription result: {result['text']}\")\n        print(f\"Transcription time: {result['transcribe_time_s']:.2f}s\")\n    except Exception as e:\n        print(f\"Transcription failed: {e}\")

import pyaudio\nimport whisper\nimport torch\nimport time\nimport logging\nfrom typing import Generator, Optional, Dict, Any\nfrom pathlib import Path\nimport numpy as np\n\n# Configure logging\nlogging.basicConfig(\n    level=logging.INFO,\n    format=\"%(asctime)s - %(levelname)s - %(message)s\"\n)\nlogger = logging.getLogger(__name__)\n\nclass VoiceAssistant:\n    \"\"\"Fully local voice assistant using PyAudio 0.2.14 and Whisper 3.\"\"\"\n    \n    def __init__(\n        self,\n        whisper_model: str = \"base\",\n        sample_rate: int = 16000,\n        chunk_size: int = 1024,\n        silence_threshold: float = 0.01,\n        silence_duration: float = 1.0\n    ):\n        self.sample_rate = sample_rate\n        self.chunk_size = chunk_size\n        self.silence_threshold = silence_threshold\n        self.silence_duration = silence_duration\n        self.audio = pyaudio.PyAudio()\n        self.stream: Optional[pyaudio.Stream] = None\n        self.whisper = whisper.load_model(whisper_model, device=\"cpu\")  # Use CPU for compatibility\n        self.is_running = False\n        self.command_handlers: Dict[str, callable] = {}\n        self._register_default_handlers()\n        logger.info(\"Voice assistant initialized\")\n\n    def _register_default_handlers(self) -> None:\n        \"\"\"Register default command handlers for common actions.\"\"\"\n        self.command_handlers[\"hello\"] = lambda: print(\"Hello! How can I help you?\")\n        self.command_handlers[\"time\"] = lambda: print(f\"Current time: {time.strftime('%H:%M:%S')}\")\n        self.command_handlers[\"exit\"] = lambda: self.stop()\n        logger.info(\"Registered default command handlers\")\n\n    def register_handler(self, command: str, handler: callable) -> None:\n        \"\"\"Register a custom command handler.\"\"\"\n        self.command_handlers[command.lower()] = handler\n        logger.info(f\"Registered handler for command: {command}\")\n\n    def _is_silence(self, audio_chunk: bytes) -> bool:\n        \"\"\"Detect if an audio chunk is silence using RMS amplitude.\"\"\"\n        try:\n            audio_np = np.frombuffer(audio_chunk, dtype=np.int16).astype(np.float32)\n            rms = np.sqrt(np.mean(np.square(audio_np)))\n            return rms < self.silence_threshold * 32768.0  # Normalize to 16-bit range\n        except Exception as e:\n            logger.error(f\"Silence detection failed: {e}\")\n            return False\n\n    def _capture_audio_until_silence(self) -> list[bytes]:\n        \"\"\"Capture audio chunks until silence is detected for specified duration.\"\"\"\n        audio_chunks = []\n        silence_start = None\n        logger.info(\"Listening for speech...\")\n        while self.is_running:\n            try:\n                chunk = self.stream.read(self.chunk_size, exception_on_overflow=False)\n                audio_chunks.append(chunk)\n                if self._is_silence(chunk):\n                    if silence_start is None:\n                        silence_start = time.time()\n                    elif time.time() - silence_start >= self.silence_duration:\n                        logger.info(f\"Silence detected after {len(audio_chunks)} chunks\")\n                        break\n                else:\n                    silence_start = None  # Reset silence timer if speech detected\n            except Exception as e:\n                logger.error(f\"Audio capture error: {e}\")\n                break\n        return audio_chunks\n\n    def _process_command(self, text: str) -> None:\n        \"\"\"Match transcribed text to registered command handlers.\"\"\"\n        text_lower = text.lower().strip()\n        for command, handler in self.command_handlers.items():\n            if command in text_lower:\n                logger.info(f\"Executing command: {command}\")\n                handler()\n                return\n        logger.info(f\"No handler found for: {text_lower}\")\n        print(f\"You said: {text}\")\n\n    def start(self, device_index: Optional[int] = None) -> None:\n        \"\"\"Start the voice assistant main loop.\"\"\"\n        try:\n            self.stream = self.audio.open(\n                format=pyaudio.paInt16,\n                channels=1,\n                rate=self.sample_rate,\n                input=True,\n                input_device_index=device_index,\n                frames_per_buffer=self.chunk_size\n            )\n            self.is_running = True\n            logger.info(\"Voice assistant started. Say 'exit' to stop.\")\n            print(\"Voice assistant is listening... (say 'exit' to quit)\")\n            \n            while self.is_running:\n                audio_chunks = self._capture_audio_until_silence()\n                if not audio_chunks:\n                    continue\n                # Convert chunks to numpy array for Whisper\n                audio_np = np.frombuffer(b\"\".join(audio_chunks), dtype=np.int16).astype(np.float32) / 32768.0\n                # Transcribe audio\n                start_time = time.perf_counter()\n                result = self.whisper.transcribe(audio_np, language=\"en\", beam_size=1)\n                transcribe_time = time.perf_counter() - start_time\n                text = result[\"text\"].strip()\n                logger.info(f\"Transcribed in {transcribe_time:.2f}s: {text}\")\n                # Process command\n                self._process_command(text)\n        except Exception as e:\n            logger.error(f\"Voice assistant error: {e}\")\n            self.stop()\n        finally:\n            self.stop()\n\n    def stop(self) -> None:\n        \"\"\"Stop the voice assistant and clean up resources.\"\"\"\n        self.is_running = False\n        if self.stream:\n            try:\n                self.stream.stop_stream()\n                self.stream.close()\n                logger.info(\"Closed audio stream\")\n            except Exception as e:\n                logger.error(f\"Error closing stream: {e}\")\n        self.audio.terminate()\n        logger.info(\"Voice assistant stopped\")\n\nif __name__ == \"__main__\":\n    # Initialize and run voice assistant\n    assistant = VoiceAssistant(whisper_model=\"base\")\n    # Register custom handler\n    assistant.register_handler(\"weather\", lambda: print(\"Weather is sunny, 22°C\"))\n    try:\n        assistant.start()\n    except KeyboardInterrupt:\n        assistant.stop()

# Pre-warm Whisper model in Docker entrypoint\nimport whisper\nfrom pathlib import Path\n\ndef prewarm_whisper():\n    model_path = Path(\"./whisper_models\")\n    model_path.mkdir(exist_ok=True)\n    # Load model with memory mapping, cache in project directory\n    model = whisper.load_model(\"large-v3\", download_root=model_path, in_memory=False)\n    print(f\"Pre-warmed Whisper model: {model.name}\")

# Decouple audio capture and transcription with threading\nimport threading\nimport queue\n\nclass ThreadedAudioCapture:\n    def __init__(self, chunk_size=1024):\n        self.chunk_size = chunk_size\n        self.audio_queue = queue.Queue(maxsize=10)\n        self.capture_thread = None\n        self.is_running = False\n\n    def _capture_loop(self):\n        while self.is_running:\n            chunk = self.stream.read(self.chunk_size, exception_on_overflow=False)\n            if not self.audio_queue.full():\n                self.audio_queue.put(chunk)\n\n    def start(self):\n        self.is_running = True\n        self.capture_thread = threading.Thread(target=self._capture_loop, daemon=True)\n        self.capture_thread.start()

# Benchmark end-to-end voice assistant latency\nimport time\nimport csv\n\ndef benchmark_latency(assistant, num_runs=100):\n    results = []\n    for _ in range(num_runs):\n        start = time.perf_counter()\n        # Simulate speech input\n        audio_chunks = assistant._capture_audio_until_silence()\n        silence_time = time.perf_counter()\n        # Transcribe\n        text = assistant.transcribe_chunks(audio_chunks)\n        transcribe_time = time.perf_counter()\n        # Process command\n        assistant._process_command(text)\n        end_time = time.perf_counter()\n        results.append({\n            \"silence_detect_ms\": (silence_time - start) * 1000,\n            \"transcribe_ms\": (transcribe_time - silence_time) * 1000,\n            \"total_ms\": (end_time - start) * 1000\n        })\n    # Write to CSV\n    with open(\"latency_benchmarks.csv\", \"w\") as f:\n        writer = csv.DictWriter(f, fieldnames=[\"silence_detect_ms\", \"transcribe_ms\", \"total_ms\"])\n        writer.writeheader()\n        writer.writerows(results)

whisper-voice-assistant/\n├── src/\n│   ├── audio_capture.py       # PyAudio 0.2.14 audio capture class\n│   ├── transcriber.py         # Whisper 3 transcription class\n│   ├── voice_assistant.py     # Main voice assistant loop\n│   └── command_handlers.py    # Custom command handlers\n├── tests/\n│   ├── test_audio_capture.py  # PyAudio capture unit tests\n│   ├── test_transcriber.py    # Whisper transcription tests\n│   └── benchmark_latency.py   # Latency benchmarking script\n├── requirements.txt           # Python 3.13 dependencies\n├── Dockerfile                 # Container build file\n├── .python-version            # Python 3.13 version pin\n└── README.md                  # Tutorial instructions