VAD and Audio Chunking: Segmenting Speech for Streaming Agents

Real-time voice agents face a fundamental challenge: how do you know when a user starts and stops speaking? The answer lies in Voice Activity Detection (VAD) and audio chunking—the gating mechanism that transforms continuous audio streams into discrete, processable utterances.

The Problem: Continuous Audio vs. Discrete Utterances

Voice pipelines are designed to operate on utterances—bounded chunks of audio representing complete thoughts or commands. But microphone input is an unbounded stream. VAD bridges this gap by converting continuous audio into discrete events.

Raw Audio Stream:  [speech] [silence] [speech] [silence] ...
                         ↓
VAD Events:        speech_start → speech_end → speech_start → ...
                         ↓
Buffered Audio:    [complete utterance 1] [complete utterance 2] ...

How VAD Works

The Neural Model Approach

Modern VAD systems like Silero VAD use compact neural networks to analyze audio chunks:

import torch
import numpy as np

class SileroVAD:
    def __init__(self, threshold=0.5, sample_rate=16000):
        self.model, _ = torch.hub.load(
            repo_or_dir='snakers4/silero-vad',
            model='silero_vad',
            force_reload=False
        )
        self.threshold = threshold
        self.sample_rate = sample_rate
        self.chunk_size = 512  # 32ms at 16kHz
        
    def process_chunk(self, audio_chunk: np.ndarray) -> float:
        """Process a single chunk and return speech probability."""
        # Ensure correct chunk size
        if len(audio_chunk) != self.chunk_size:
            raise ValueError(f"Chunk must be {self.chunk_size} samples")
        
        # Convert to tensor
        tensor = torch.from_numpy(audio_chunk).float()
        
        # Get speech probability
        probability = self.model(tensor, self.sample_rate).item()
        return probability

The State Machine

VAD uses a state machine with hysteresis to prevent oscillation:

from enum import Enum
from dataclasses import dataclass
import time

class VADState(Enum):
    SILENCE = "silence"
    SPEECH = "speech"
    UNCERTAIN = "uncertain"

@dataclass
class VADConfig:
    start_threshold: float = 0.5
    end_threshold: float = 0.35  # Lower than start for hysteresis
    min_silence_duration_ms: int = 300
    min_speech_duration_ms: int = 250
    prefix_padding_ms: int = 500

class VADStateMachine:
    def __init__(self, config: VADConfig):
        self.config = config
        self.state = VADState.SILENCE
        self.last_speech_time = 0
        self.last_silence_time = 0
        self.speech_buffer = []
        
    def update(self, probability: float, timestamp: float) -> dict:
        """Update state machine with new probability."""
        events = []
        
        if self.state == VADState.SILENCE:
            if probability > self.config.start_threshold:
                self.state = VADState.SPEECH
                self.last_speech_time = timestamp
                events.append("speech_start")
                
        elif self.state == VADState.SPEECH:
            if probability < self.config.end_threshold:
                silence_duration = (timestamp - self.last_speech_time) * 1000
                
                if silence_duration >= self.config.min_silence_duration_ms:
                    self.state = VADState.SILENCE
                    self.last_silence_time = timestamp
                    events.append("speech_end")
        
        return {
            "state": self.state,
            "events": events,
            "probability": probability
        }

Audio Chunking Strategies

Fixed-Size Chunking

The simplest approach—process audio in fixed-size windows:

class FixedChunkProcessor:
    def __init__(self, chunk_size: int = 512, sample_rate: int = 16000):
        self.chunk_size = chunk_size
        self.sample_rate = sample_rate
        self.buffer = np.array([], dtype=np.float32)
        
    def add_audio(self, audio: np.ndarray) -> list:
        """Add audio and return complete chunks."""
        self.buffer = np.concatenate([self.buffer, audio])
        chunks = []
        
        while len(self.buffer) >= self.chunk_size:
            chunk = self.buffer[:self.chunk_size]
            self.buffer = self.buffer[self.chunk_size:]
            chunks.append(chunk)
            
        return chunks

Ring Buffer with Pre-roll

Capture leading audio that arrives before speech detection:

from collections import deque

class PreRollBuffer:
    def __init__(self, pre_roll_ms: int = 500, sample_rate: int = 16000):
        self.sample_rate = sample_rate
        pre_roll_samples = int(pre_roll_ms * sample_rate / 1000)
        self.ring_buffer = deque(maxlen=pre_roll_samples)
        self.capture_buffer = []
        self.is_capturing = False
        
    def add_chunk(self, chunk: np.ndarray):
        """Add audio chunk to buffer."""
        self.ring_buffer.extend(chunk)
        
        if self.is_capturing:
            self.capture_buffer.extend(chunk)
    
    def start_capture(self):
        """Start capturing audio (on speech_start event)."""
        self.is_capturing = True
        # Prepend ring buffer contents
        self.capture_buffer = list(self.ring_buffer)
    
    def stop_capture(self) -> np.ndarray:
        """Stop capturing and return complete utterance."""
        self.is_capturing = False
        utterance = np.array(self.capture_buffer)
        self.capture_buffer = []
        return utterance
    
    def get_current_capture(self) -> np.ndarray:
        """Get current capture buffer contents."""
        return np.array(self.capture_buffer)

Integration with Streaming Agents

Complete VAD Pipeline

import asyncio
from typing import AsyncGenerator, Callable

class StreamingVADPipeline:
    def __init__(
        self,
        vad_model: SileroVAD,
        config: VADConfig,
        on_speech_start: Callable = None,
        on_speech_end: Callable = None,
        on_audio_chunk: Callable = None
    ):
        self.vad = vad_model
        self.config = config
        self.state_machine = VADStateMachine(config)
        self.pre_roll = PreRollBuffer(config.prefix_padding_ms)
        
        self.on_speech_start = on_speech_start
        self.on_speech_end = on_speech_end
        self.on_audio_chunk = on_audio_chunk
        
    async def process_audio_stream(
        self, 
        audio_stream: AsyncGenerator[np.ndarray, None]
    ) -> AsyncGenerator[dict, None]:
        """Process continuous audio stream and yield events."""
        async for audio_chunk in audio_stream:
            # Process through VAD
            probability = self.vad.process_chunk(audio_chunk)
            
            # Update state machine
            result = self.state_machine.update(
                probability, 
                time.time()
            )
            
            # Handle events
            for event in result["events"]:
                if event == "speech_start":
                    self.pre_roll.start_capture()
                    if self.on_speech_start:
                        await self.on_speech_start()
                        
                elif event == "speech_end":
                    utterance = self.pre_roll.stop_capture()
                    if self.on_speech_end:
                        await self.on_speech_end(utterance)
            
            # Add to pre-roll buffer
            self.pre_roll.add_chunk(audio_chunk)
            
            # Yield chunk for processing
            yield {
                "chunk": audio_chunk,
                "probability": probability,
                "state": result["state"],
                "is_speaking": result["state"] == VADState.SPEECH
            }

Semantic VAD

Advanced VAD uses language models to detect utterance boundaries semantically:

class SemanticVAD:
    """VAD that uses semantic understanding to detect end of turn."""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.transcript_buffer = []
        
    async def should_end_turn(self, transcript: str) -> bool:
        """Determine if the user has finished speaking."""
        # Check for completion signals
        completion_signals = [
            "um", "uh", "hmm",  # Hesitation markers
            ".", "?", "!",       # Punctuation
        ]
        
        # Use LLM to assess completion probability
        prompt = f"""
        Given this partial transcript: "{transcript}"
        
        Rate the probability that the user has finished speaking (0.0 to 1.0).
        Consider:
        - Grammatical completeness
        - Semantic completeness
        - Presence of hesitation markers
        - Context of the conversation
        
        Return only the probability as a float.
        """
        
        response = await self.llm.complete(prompt)
        probability = float(response.strip())
        
        return probability > 0.7  # Threshold for end of turn

Configuration Parameters

Tuning VAD for Different Environments

# Noisy environment (e.g., public space)
noisy_config = VADConfig(
    start_threshold=0.7,      # Higher threshold to avoid false triggers
    end_threshold=0.5,
    min_silence_duration_ms=500,  # Longer silence to confirm end
    min_speech_duration_ms=300,
    prefix_padding_ms=600
)

# Quiet environment (e.g., home office)
quiet_config = VADConfig(
    start_threshold=0.4,      # Lower threshold for soft speech
    end_threshold=0.25,
    min_silence_duration_ms=200,  # Shorter silence for faster response
    min_speech_duration_ms=150,
    prefix_padding_ms=400
)

# Conference call (multiple speakers)
conference_config = VADConfig(
    start_threshold=0.6,
    end_threshold=0.4,
    min_silence_duration_ms=400,
    min_speech_duration_ms=250,
    prefix_padding_ms=500
)

Best Practices

1. Always Use Pre-roll Buffers

Without pre-roll, you’ll clip the beginning of utterances:

# Bad: No pre-roll
speech_start → start_buffering → [MISSING AUDIO] utterance

# Good: With pre-roll
ring_buffer → speech_start → [PRE-ROLL + FULL UTTERANCE]

2. Reset State Between Sessions

class VADSession:
    def __init__(self):
        self.vad = SileroVAD()
        
    def start_new_session(self):
        """Reset VAD state for new session."""
        self.vad.model.reset_states()  # Reset LSTM state
        self.state_machine.reset()
        self.pre_roll.clear()

3. Handle Edge Cases

class RobustVAD:
    def __init__(self):
        self.max_speech_duration = 30.0  # seconds
        self.min_speech_duration = 0.1   # seconds
        
    def validate_utterance(self, utterance: np.ndarray, 
                          sample_rate: int) -> bool:
        """Validate utterance before processing."""
        duration = len(utterance) / sample_rate
        
        if duration < self.min_speech_duration:
            return False  # Too short, likely noise
            
        if duration > self.max_speech_duration:
            return False  # Too long, likely VAD failure
            
        # Check for silence
        if np.max(np.abs(utterance)) < 0.01:
            return False  # Silent utterance
            
        return True

4. Monitor Performance

class VADMetrics:
    def __init__(self):
        self.metrics = {
            "utterances_detected": 0,
            "avg_utterance_duration": 0,
            "false_positives": 0,
            "missed_speech": 0
        }
    
    def record_utterance(self, duration: float):
        self.metrics["utterances_detected"] += 1
        n = self.metrics["utterances_detected"]
        self.metrics["avg_utterance_duration"] = (
            (self.metrics["avg_utterance_duration"] * (n - 1) + duration) / n
        )

Conclusion

VAD and audio chunking are foundational to real-time voice agents. The key principles:

1. Event-driven architecture: VAD converts continuous audio to discrete events
2. Hysteresis prevents oscillation: Different thresholds for start/end detection
3. Pre-roll buffers capture leading audio: Never clip the beginning of utterances
4. State management matters: Reset between sessions, handle edge cases
5. Tune for your environment: Adjust thresholds based on noise levels

With proper VAD implementation, your voice agents can accurately detect when users speak, leading to more natural and responsive conversational experiences.