Building a Voice Conversation Loop

The Voice Conversation Loop

Recording from the Microphone

import sounddevice as sd
import numpy as np
import tempfile
import soundfile as sf

SAMPLE_RATE = 16000  # 16kHz mono — optimal for Whisper
DURATION = 5         # seconds

def record_audio(duration=DURATION, sample_rate=SAMPLE_RATE):
    print(f'Recording for {duration} seconds...')
    audio_data = sd.rec(
        int(duration * sample_rate),
        samplerate=sample_rate,
        channels=1,
        dtype='float32'
    )
    sd.wait()  # block until recording finishes
    print('Recording complete')
    return audio_data, sample_rate

def audio_to_file(audio_data, sample_rate, filepath='/tmp/recording.wav'):
    sf.write(filepath, audio_data, sample_rate)
    return filepath

Silence Detection for End-of-Speech

import sounddevice as sd
import numpy as np
from collections import deque

SAMPLE_RATE = 16000
CHUNK_SIZE = 1024         # samples per chunk
SILENCE_THRESHOLD = 0.02  # RMS volume below this = silence
SILENCE_DURATION = 1.5   # seconds of silence to end recording
MAX_DURATION = 30         # max recording length in seconds

def record_until_silence():
    chunks = []
    silent_chunks = 0
    max_silent = int(SILENCE_DURATION * SAMPLE_RATE / CHUNK_SIZE)
    max_chunks = int(MAX_DURATION * SAMPLE_RATE / CHUNK_SIZE)

    print('Listening... (speak now)')
    with sd.InputStream(samplerate=SAMPLE_RATE, channels=1,
                        blocksize=CHUNK_SIZE, dtype='float32') as stream:
        while True:
            chunk, _ = stream.read(CHUNK_SIZE)
            chunks.append(chunk.copy())

            rms = np.sqrt(np.mean(chunk**2))
            if rms < SILENCE_THRESHOLD:
                silent_chunks += 1
            else:
                silent_chunks = 0  # speech detected — reset counter

            if silent_chunks >= max_silent and len(chunks) > max_silent:
                break
            if len(chunks) >= max_chunks:
                break

    audio = np.concatenate(chunks, axis=0)
    return audio, SAMPLE_RATE

The Full Voice Loop

import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

conversation_history = []

def voice_conversation_loop():
    print('Voice agent started. Say something (Ctrl+C to exit).')

    while True:
        # 1. Record user
        audio_data, sample_rate = record_until_silence()
        audio_path = audio_to_file(audio_data, sample_rate)

        # 2. Transcribe
        user_text = transcribe_file(audio_path)
        print(f'You: {user_text}')

        if not user_text.strip():
            continue  # empty — listen again

        # 3. Run agent
        conversation_history.append({'role': 'user', 'content': user_text})
        response = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[{'role': 'system', 'content': 'You are a helpful voice assistant. Keep responses concise.'}] + conversation_history[-10:]
        )
        agent_text = response.choices[0].message.content
        conversation_history.append({'role': 'assistant', 'content': agent_text})
        print(f'Agent: {agent_text}')

        # 4. TTS and play
        say(agent_text)  # speaks sentence by sentence

Wake Word Detection Concept

# Wake word detection with pvporcupine (Picovoice)
# pip install pvporcupine sounddevice

import pvporcupine
import sounddevice as sd
import numpy as np
import os

def listen_for_wake_word(wake_word='computer'):
    porcupine = pvporcupine.create(
        access_key=os.getenv('PICOVOICE_KEY'),
        keywords=[wake_word]  # built-in: computer, hey barista, hey google, jarvis...
    )

    print(f'Listening for wake word: "{wake_word}"...')

    with sd.InputStream(
        samplerate=porcupine.sample_rate,
        channels=1,
        dtype='int16',
        blocksize=porcupine.frame_length
    ) as stream:
        while True:
            frame, _ = stream.read(porcupine.frame_length)
            pcm = frame.flatten().astype('int16')
            result = porcupine.process(pcm)
            if result >= 0:
                print(f'Wake word detected!')
                porcupine.delete()
                return True

Interrupt Handling

import threading
import sounddevice as sd
import numpy as np

interrupt_event = threading.Event()

def monitor_for_interrupt(threshold=0.03):
    def audio_callback(indata, frames, time_info, status):
        rms = np.sqrt(np.mean(indata**2))
        if rms > threshold:
            print('Interrupt detected!')
            interrupt_event.set()

    with sd.InputStream(callback=audio_callback, channels=1,
                        samplerate=16000, blocksize=1024):
        while not interrupt_event.is_set():
            sd.sleep(50)

def speak_with_interrupt(text, voice='nova'):
    interrupt_event.clear()

    # Start interrupt monitor in background
    monitor_thread = threading.Thread(target=monitor_for_interrupt, daemon=True)
    monitor_thread.start()

    # Play audio sentence by sentence
    for sentence in split_into_sentences(text):
        if interrupt_event.is_set():
            print('Playback interrupted')
            break
        audio_path = cached_tts(sentence, voice)
        play_audio_file(audio_path)

Noise Filtering

import numpy as np

NOISE_GATE_THRESHOLD = 0.015  # RMS threshold

def apply_noise_gate(audio_data, threshold=NOISE_GATE_THRESHOLD):
    rms = np.sqrt(np.mean(audio_data**2))
    if rms < threshold:
        return np.zeros_like(audio_data)  # silence below threshold
    return audio_data

def has_speech_content(audio_data, threshold=0.02):
    rms_values = [
        np.sqrt(np.mean(audio_data[i:i+1600]**2))
        for i in range(0, len(audio_data), 1600)
    ]
    speech_frames = sum(1 for r in rms_values if r > threshold)
    speech_ratio = speech_frames / max(len(rms_values), 1)
    return speech_ratio > 0.1  # at least 10% of frames have speech

# Use in loop before transcribing
audio, sr = record_until_silence()
if has_speech_content(audio):
    text = transcribe_file(audio_to_file(audio, sr))
else:
    print('No speech detected — listening again')

State Machine for the Conversation Loop

from enum import Enum

class AgentState(Enum):
    IDLE           = 'idle'
    LISTENING      = 'listening'
    TRANSCRIBING   = 'transcribing'
    THINKING       = 'thinking'
    SPEAKING       = 'speaking'

current_state = AgentState.IDLE

def set_state(new_state):
    global current_state
    print(f'State: {current_state.value} -> {new_state.value}')
    current_state = new_state

def voice_loop_with_state():
    while True:
        set_state(AgentState.LISTENING)
        audio, sr = record_until_silence()

        if not has_speech_content(audio):
            continue

        set_state(AgentState.TRANSCRIBING)
        text = transcribe_file(audio_to_file(audio, sr))
        if not text.strip():
            continue
        print(f'You: {text}')

        set_state(AgentState.THINKING)
        agent_response = run_agent(text)
        print(f'Agent: {agent_response}')

        set_state(AgentState.SPEAKING)
        speak_with_interrupt(agent_response)

Conversation Context Management

import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

MAX_HISTORY_TURNS = 10
SYSTEM_PROMPT = 'You are a concise voice assistant. Keep responses under 3 sentences unless asked for detail.'

class VoiceAgent:
    def __init__(self):
        self.history = []

    def respond(self, user_text):
        self.history.append({'role': 'user', 'content': user_text})

        # Trim history to last N turns
        recent = self.history[-(MAX_HISTORY_TURNS * 2):]

        response = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[{'role': 'system', 'content': SYSTEM_PROMPT}] + recent
        )
        agent_text = response.choices[0].message.content
        self.history.append({'role': 'assistant', 'content': agent_text})
        return agent_text

    def reset(self):
        self.history = []
        print('Conversation history cleared')

Graceful Shutdown and Cleanup

import signal
import sys
import glob
import os

TMP_DIR = '/tmp'
agent = None

def cleanup_temp_files():
    for f in glob.glob(os.path.join(TMP_DIR, 'recording_*.wav')):
        os.unlink(f)
    print('Temp files cleaned up')

def graceful_shutdown(signum, frame):
    print('\nShutting down voice agent...')
    if agent:
        say('Goodbye! Have a great day.')
    cleanup_temp_files()
    sys.exit(0)

# Register signal handler
signal.signal(signal.SIGINT, graceful_shutdown)
signal.signal(signal.SIGTERM, graceful_shutdown)

# Main entry point
agent = VoiceAgent()
print('Voice agent ready. Press Ctrl+C to exit.')
voice_loop_with_state()

Logging the Conversation

import json
import os
from datetime import datetime

LOG_DIR = '/tmp/voice_agent_logs'
os.makedirs(LOG_DIR, exist_ok=True)

conversation_log = []

def log_turn(speaker, text, latency_ms=None, error=None):
    entry = {
        'timestamp': datetime.now().isoformat(),
        'speaker': speaker,   # 'user' or 'agent'
        'text': text,
        'latency_ms': latency_ms
    }
    if error:
        entry['error'] = error
    conversation_log.append(entry)

def save_conversation_log(session_id=None):
    if not session_id:
        session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
    log_path = os.path.join(LOG_DIR, f'session_{session_id}.json')
    with open(log_path, 'w') as f:
        json.dump({
            'session_id': session_id,
            'turns': len(conversation_log),
            'log': conversation_log
        }, f, indent=2)
    print(f'Log saved: {log_path}')
    return log_path

# Usage in voice loop
log_turn('user', transcribed_text, latency_ms=420)
log_turn('agent', agent_response, latency_ms=850)

Knowledge Check

Recap: Building a Voice Conversation Loop