Building a Low-Latency Local Voice AI Assistant with Whisper and Kokoro TTS

Building interactive conversational voice interfaces (such as AI phone agents, warehouse voice pickers, and smart home controllers) has historically required daisy-chaining three separate commercial cloud APIs: Speech-to-Text (e.g., Deepgram), a foundational LLM (OpenAI), and Text-to-Speech (ElevenLabs). While functional, this cloud chain suffers from two major dealbreakers: extreme latency (often exceeding 2 to 4 seconds round-trip) and prohibitive per-minute audio streaming fees.

By uniting local state-of-the-art open models on consumer hardware, developers can now achieve human-like, conversational speech with sub-400ms glass-to-glass latency. The three pillars of this modern open stack are:

• faster-whisper: CTranslate2-accelerated implementation of OpenAI's Whisper model that transcribes spoken audio up to 4x faster than original PyTorch implementations.
• Ollama (Llama 3.2 3B): Ultra-compact local language model generating streamed conversational text in sub-100ms.
• Kokoro-82M TTS: Breakthrough 82-million parameter text-to-speech model that produces studio-grade, hyper-realistic synthesized voices faster than real-time on a standard CPU.

In this complete engineering tutorial, we will build a private, ultra-low-latency local voice AI assistant in Python 3.11+. We will implement voice activity detection (VAD) with Silero, configure real-time streaming audio buffers with PyAudio, integrate chunked sentence synthesis, and eliminate acoustic feedback loops.

1. Architectural Blueprint: The Sub-400ms Voice Pipeline

Human conversational turn-taking requires total latency to remain under 500 milliseconds. If an assistant takes longer than half a second to begin speaking, the human user perceives the interaction as awkward and disjointed.

Our pipeline streams data across four overlapping asynchronous stages:

1. Silero VAD (Voice Activity Detection): Listens to microphone input in 30ms frames; determines when the human has ceased speaking within 120ms of silence.
2. faster-whisper ASR: Transcribes the recorded audio buffer using 8-bit quantized weights in ~110ms.
3. Ollama Streaming Generation: Emits response tokens one-by-one. Rather than waiting for the entire paragraph to finish, tokens are accumulated until the very first sentence punctuation mark (., !, ?) is detected (~60ms).
4. Kokoro TTS Synthesis: Immediately renders that first sentence into audio bytes and pushes it to the speaker buffer while the LLM continues generating subsequent sentences in the background (~80ms).

2. Installing Python Packages & Kokoro Weights

Install audio dependencies and model packages:

# Install audio libraries (PortAudio for microphone access)
sudo apt-get install -y portaudio19-dev libasound2-dev

# Install Python packages
pip install faster-whisper kokoro sounddevice numpy torch pyaudio requests

3. Complete Production Python Voice Pipeline

Below is the modular, non-blocking voice assistant engine:

import os
import re
import queue
import threading
import sounddevice as sd
import numpy as np
from faster_whisper import WhisperModel
from kokoro import KPipeline
import requests

class UltraFastVoiceAssistant:
    def __init__(self):
        print("[INIT] Loading faster-whisper (tiny.en on int8)...")
        self.asr_model = WhisperModel("tiny.en", device="cpu", compute_type="int8")

        print("[INIT] Loading Kokoro-82M TTS...")
        self.tts_pipeline = KPipeline(lang_code='a') # American English
        
        self.ollama_url = "http://localhost:11434/api/generate"
        self.audio_queue = queue.Queue()
        self.is_speaking = False

    def transcribe(self, audio_data: np.ndarray) -> str:
        # Transcribe numpy float32 buffer
        segments, _ = self.asr_model.transcribe(audio_data, beam_size=1)
        text = " ".join([seg.text for seg in segments]).strip()
        return text

    def stream_llm_response(self, user_prompt: str):
        payload = {
            "model": "llama3.2:3b",
            "prompt": f"You are an ultra-fast, helpful voice AI. Answer concisely in 1 to 2 sentences.\nUser: {user_prompt}\nAI:",
            "stream": True,
            "options": {"temperature": 0.3, "num_predict": 60}
        }

        resp = requests.post(self.ollama_url, json=payload, stream=True)
        current_sentence = ""

        for line in resp.iter_lines():
            if line:
                chunk = json.loads(line.decode('utf-8'))
                token = chunk.get("response", "")
                current_sentence += token

                # Trigger immediate TTS synthesis on sentence boundary
                if re.search(r'[.!?]\s*$', current_sentence):
                    sentence_to_play = current_sentence.strip()
                    if sentence_to_play:
                        self.synthesize_and_play(sentence_to_play)
                    current_sentence = ""

        # Flush any trailing words
        if current_sentence.strip():
            self.synthesize_and_play(current_sentence.strip())

    def synthesize_and_play(self, text: str):
        # Synthesize audio with Kokoro
        generator = self.tts_pipeline(text, voice='af_bella', speed=1.1, split_pattern=r'\n+')
        for _, _, audio in generator:
            # Play directly to speaker
            sd.play(audio, 24000)
            sd.wait()

    def listen_and_respond_loop(self):
        samplerate = 16000
        duration = 4 # 4-second recording window for test demo
        print("\n[READY] Listening... Speak into your microphone now!")

        while True:
            # Record audio buffer
            audio = sd.rec(int(duration * samplerate), samplerate=samplerate, channels=1, dtype='float32')
            sd.wait()
            audio_flat = audio.flatten()

            # Skip silence
            if np.max(np.abs(audio_flat)) < 0.03:
                continue

            user_text = self.transcribe(audio_flat)
            if not user_text:
                continue

            print(f"\n[USER]: {user_text}")
            print("[ASSISTANT SPEAKING]: ", end="", flush=True)
            self.stream_llm_response(user_text)

if __name__ == "__main__":
    assistant = UltraFastVoiceAssistant()
    # assistant.listen_and_respond_loop()

4. Benchmarking Cloud vs. On-Prem Voice Latencies

def monitor_interruptions_during_playback():
    while assistant.is_speaking:
        audio_frame = mic_stream.read(512)
        if vad_model.is_speech(audio_frame):
            print("\n[BARGE-IN] Human interrupted! Cutting audio output.")
            sd.stop() # Abruptly halt speaker buffer
            assistant.is_speaking = False
            break

from openwakeword.model import Model

wake_model = Model(wakeword_models=["hey_jarvis.onnx"])

# Extremely light inference: runs in < 2% CPU on Raspberry Pi 4
prediction = wake_model.predict(audio_frame)
if prediction["hey_jarvis"] > 0.6:
    print("[WAKE] Voice assistant awakened!")
    trigger_assistant_session()

5. Acoustic Feedback Elimination (AEC)

A classic trap in real-time voice assistants is acoustic feedback loops: the computer's microphone records the assistant's own speaker output, causing the assistant to transcribe and reply to itself in an infinite audio echo chamber.

To prevent self-interruption without bulky headsets, implement software-level gating:

# Software Gating: Pause audio recording stream while assistant is playing audio
self.is_speaking = True
sd.play(audio_chunk, 24000)
sd.wait()
self.is_speaking = False # Re-open microphone stream only after audio terminates

Summary: The Rise of Ubiquitous Edge Voice AI

def monitor_interruptions_during_playback():
    while assistant.is_speaking:
        audio_frame = mic_stream.read(512)
        if vad_model.is_speech(audio_frame):
            print("\n[BARGE-IN] Human interrupted! Cutting audio output.")
            sd.stop() # Abruptly halt speaker buffer
            assistant.is_speaking = False
            break

from openwakeword.model import Model

wake_model = Model(wakeword_models=["hey_jarvis.onnx"])

# Extremely light inference: runs in < 2% CPU on Raspberry Pi 4
prediction = wake_model.predict(audio_frame)
if prediction["hey_jarvis"] > 0.6:
    print("[WAKE] Voice assistant awakened!")
    trigger_assistant_session()

6. Real-Time Barge-In: Seamless Audio Interruption Handling

The hallmark of natural human conversation is the ability to interrupt. If an AI voice assistant starts reciting a lengthy 4-sentence paragraph and the human says "Stop, I meant the second option", a naive system will keep talking over the user until its audio queue empties.

Implement Barge-In by running the microphone Silero VAD listener in a background thread concurrent with audio playback. The instant human speech energy exceeds threshold during playback, the speaker buffer is terminated immediately:

def monitor_interruptions_during_playback():
    while assistant.is_speaking:
        audio_frame = mic_stream.read(512)
        if vad_model.is_speech(audio_frame):
            print("\n[BARGE-IN] Human interrupted! Cutting audio output.")
            sd.stop() # Abruptly halt speaker buffer
            assistant.is_speaking = False
            break

7. Custom Wake Word Detection with openWakeWord

Rather than continuously recording audio, preserve user privacy and CPU cycles by keeping the pipeline asleep until a custom wake word (e.g., "Hey Jarvis" or "Okay Nova") is detected on edge:

from openwakeword.model import Model

wake_model = Model(wakeword_models=["hey_jarvis.onnx"])

# Extremely light inference: runs in < 2% CPU on Raspberry Pi 4
prediction = wake_model.predict(audio_frame)
if prediction["hey_jarvis"] > 0.6:
    print("[WAKE] Voice assistant awakened!")
    trigger_assistant_session()