Streaming Token Delivery for Voice Interfaces

In real-time voice AI, waiting for complete responses before speaking feels unnatural. Streaming token delivery solves this by sending LLM tokens directly to TTS engines as they’re generated—enabling immediate audio playback while the model is still thinking.

The Latency Problem

Sequential Pipeline (Slow)

User speaks → STT processes → LLM generates complete response → TTS synthesizes → Audio plays

This creates dead air while each stage completes:

# Sequential processing (high latency)
async def sequential_pipeline(audio: bytes):
    # STT: 200ms
    transcript = await stt.transcribe(audio)
    
    # LLM: 500ms
    response = await llm.generate(transcript)
    
    # TTS: 300ms
    audio = await tts.synthesize(response)
    
    # Total: ~1000ms delay
    return audio

Streaming Pipeline (Fast)

User speaks → STT streams → LLM streams tokens → TTS streams audio → Audio plays immediately

Each stage starts before the previous finishes:

# Streaming pipeline (low latency)
async def streaming_pipeline(audio_stream):
    # Stream STT
    async for transcript_chunk in stt.stream_transcribe(audio_stream):
        # Stream LLM tokens
        async for token in llm.stream_generate(transcript_chunk):
            # Stream TTS audio
            audio_chunk = await tts.stream_synthesize(token)
            yield audio_chunk  # Play immediately
    
    # Total: ~200ms to first audio byte

Architecture Components

1. LLM Token Streaming

class StreamingLLM:
    def __init__(self, model: str):
        self.model = model
        
    async def stream_tokens(self, prompt: str) -> AsyncGenerator[str, None]:
        """Stream tokens as they're generated."""
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            stream=True
        )
        
        async for chunk in response:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

2. Intelligent Chunking

class TokenChunker:
    def __init__(self, min_chunk_size: int = 20, 
                 max_chunk_size: int = 100):
        self.min_chunk_size = min_chunk_size
        self.max_chunk_size = max_chunk_size
        self.buffer = ""
        
    def add_token(self, token: str) -> Optional[str]:
        """Add token and return chunk when ready."""
        self.buffer += token
        
        # Check for sentence boundary
        if self._is_sentence_boundary():
            chunk = self.buffer
            self.buffer = ""
            return chunk
        
        # Check for minimum size
        if len(self.buffer) >= self.max_chunk_size:
            chunk = self.buffer
            self.buffer = ""
            return chunk
        
        return None
    
    def _is_sentence_boundary(self) -> bool:
        """Check if buffer ends at sentence boundary."""
        boundaries = ['.', '!', '?', ',', ';', ':']
        return any(self.buffer.endswith(b) for b in boundaries)

3. WebSocket Transport

import websockets
import json

class StreamingTTSClient:
    def __init__(self, tts_url: str):
        self.url = tts_url
        self.ws = None
        
    async def connect(self):
        """Establish WebSocket connection."""
        self.ws = await websockets.connect(self.url)
        
    async def stream_text(self, text_stream):
        """Stream text tokens to TTS."""
        async for chunk in text_stream:
            await self.ws.send(json.dumps({
                "text": chunk,
                "voice": "default"
            }))
            
    async def receive_audio(self) -> AsyncGenerator[bytes, None]:
        """Receive audio chunks from TTS."""
        async for message in self.ws:
            data = json.loads(message)
            if "audio" in data:
                yield bytes(data["audio"])

Complete Streaming Pipeline

class StreamingVoicePipeline:
    def __init__(self):
        self.stt = StreamingSTT()
        self.llm = StreamingLLM()
        self.tts = StreamingTTS()
        self.chunker = TokenChunker()
        
    async def process_stream(self, audio_input):
        """End-to-end streaming pipeline."""
        # Stream STT
        transcript_stream = self.stt.stream_transcribe(audio_input)
        
        # Stream LLM tokens
        token_stream = self.llm.stream_tokens(transcript_stream)
        
        # Chunk tokens for TTS
        async for token in token_stream:
            chunk = self.chunker.add_token(token)
            
            if chunk:
                # Stream chunk to TTS
                audio_chunk = await self.tts.synthesize_chunk(chunk)
                
                # Play immediately
                yield audio_chunk
        
        # Flush remaining buffer
        if self.chunker.buffer:
            audio_chunk = await self.tts.synthesize_chunk(
                self.chunker.buffer
            )
            yield audio_chunk

Handling Interruptions

Barge-In Support

class InterruptibleStreamingPipeline:
    def __init__(self):
        self.is_playing = False
        self.current_tts_task = None
        
    async def handle_interruption(self):
        """Handle user interruption during playback."""
        if self.current_tts_task:
            # Cancel current TTS generation
            self.current_tts_task.cancel()
            
            # Clear audio buffer
            self.clear_playback_buffer()
            
            # Reset pipeline state
            self.reset_state()
    
    async def stream_with_interruption(self, audio_stream):
        """Stream with interruption support."""
        try:
            async for chunk in self.process_stream(audio_stream):
                self.is_playing = True
                yield chunk
        except asyncio.CancelledError:
            # Interruption occurred
            self.is_playing = False
            return

Latency Optimization Techniques

1. Connection Persistence

class PersistentConnectionManager:
    def __init__(self):
        self.connections = {}
        
    async def get_connection(self, service: str):
        """Get or create persistent connection."""
        if service not in self.connections or \
           self.connections[service].closed:
            self.connections[service] = await self.create_connection(service)
        
        return self.connections[service]
    
    async def create_connection(self, service: str):
        """Create new WebSocket connection."""
        return await websockets.connect(
            self.get_service_url(service)
        )

2. Token Buffering

class AdaptiveBuffer:
    def __init__(self, target_latency_ms: int = 200):
        self.target_latency = target_latency_ms
        self.buffer = []
        self.last_flush_time = 0
        
    def add_token(self, token: str) -> Optional[list]:
        """Add token and return buffer if ready."""
        self.buffer.append(token)
        
        current_time = time.time() * 1000
        time_since_flush = current_time - self.last_flush_time
        
        # Flush if buffer is large enough or timeout
        if len(self.buffer) >= 10 or \
           time_since_flush > self.target_latency:
            chunk = self.buffer.copy()
            self.buffer.clear()
            self.last_flush_time = current_time
            return chunk
        
        return None

3. Predictive Prefetching

class PredictivePrefetch:
    def __init__(self):
        self.prediction_model = None
        
    async def predict_next_tokens(self, current_tokens: list):
        """Predict likely next tokens for prefetching."""
        # Use small model to predict next few tokens
        prediction = await self.prediction_model.predict(
            current_tokens[-10:]  # Last 10 tokens
        )
        
        return prediction.get('next_tokens', [])

Monitoring and Metrics

class StreamingMetrics:
    def __init__(self):
        self.metrics = {
            'time_to_first_byte_ms': [],
            'chunk_to_chunk_ms': [],
            'total_pipeline_latency_ms': [],
            'interruption_recovery_ms': []
        }
    
    def record_streaming_performance(self, ttfa_ms: float,
                                    chunk_latency_ms: float):
        """Record streaming metrics."""
        self.metrics['time_to_first_byte_ms'].append(ttfa_ms)
        self.metrics['chunk_to_chunk_ms'].append(chunk_latency_ms)
    
    def get_percentiles(self, metric: str) -> dict:
        """Calculate percentiles for a metric."""
        values = sorted(self.metrics[metric])
        n = len(values)
        
        return {
            'p50': values[int(n * 0.5)],
            'p95': values[int(n * 0.95)],
            'p99': values[int(n * 0.99)]
        }

Best Practices

1. Optimize Chunk Sizes

# Too small: High overhead, potential gaps
chunk_size = 10  # tokens

# Too large: High latency
chunk_size = 100  # tokens

# Sweet spot: Balance latency and naturalness
chunk_size = 30  # tokens (or sentence boundaries)

2. Handle Network Jitter

class JitterBuffer:
    def __init__(self, buffer_ms: int = 200):
        self.buffer_size = buffer_ms
        self.audio_buffer = []
        
    def add_chunk(self, chunk: bytes, timestamp: float):
        """Add audio chunk to jitter buffer."""
        self.audio_buffer.append({
            'data': chunk,
            'timestamp': timestamp
        })
        
        # Sort by timestamp
        self.audio_buffer.sort(key=lambda x: x['timestamp'])
    
    def get_playable_chunk(self) -> Optional[bytes]:
        """Get next chunk when ready for playback."""
        if not self.audio_buffer:
            return None
            
        current_time = time.time() * 1000
        next_chunk = self.audio_buffer[0]
        
        if current_time >= next_chunk['timestamp'] + self.buffer_size:
            self.audio_buffer.pop(0)
            return next_chunk['data']
        
        return None

3. Monitor Real Performance

# Don't just measure TTS in isolation
# Measure end-to-end pipeline

async def measure_real_latency(pipeline, test_audio):
    """Measure actual user-perceived latency."""
    start_time = time.time()
    
    first_audio_received = False
    async for chunk in pipeline.process_stream(test_audio):
        if not first_audio_received:
            ttfa_ms = (time.time() - start_time) * 1000
            print(f"Time to first audio: {ttfa_ms}ms")
            first_audio_received = True
    
    total_time = (time.time() - start_time) * 1000
    print(f"Total pipeline time: {total_time}ms")

Conclusion

Streaming token delivery transforms voice AI from “thinking” to “talking”:

1. Stream everything: STT → LLM → TTS should all stream
2. Chunk intelligently: Balance latency with natural prosody
3. Handle interruptions: Users should be able to barge in
4. Use persistent connections: Avoid reconnection overhead
5. Monitor end-to-end: Isolate component metrics don’t show user experience

The key insight: the biggest latency win isn’t faster models—it’s eliminating sequential waiting. By streaming tokens directly from LLM to TTS, you can achieve sub-300ms time-to-first-audio, making voice agents feel truly conversational.