Voice AI Engine Development

1---
2name: voice-ai-engine-development
3description: "Build real-time conversational AI voice engines using async worker pipelines, streaming transcription, LLM agents, and TTS synthesis with interrupt handling and multi-provider support"
4---
5 
6# Voice AI Engine Development
7 
8## Overview
9 
10This skill guides you through building production-ready voice AI engines with real-time conversation capabilities. Voice AI engines enable natural, bidirectional conversations between users and AI agents through streaming audio processing, speech-to-text transcription, LLM-powered responses, and text-to-speech synthesis.
11 
12The core architecture uses an async queue-based worker pipeline where each component runs independently and communicates via `asyncio.Queue` objects, enabling concurrent processing, interrupt handling, and real-time streaming at every stage.
13 
14## When to Use This Skill
15 
16Use this skill when:
17- Building real-time voice conversation systems
18- Implementing voice assistants or chatbots
19- Creating voice-enabled customer service agents
20- Developing voice AI applications with interrupt capabilities
21- Integrating multiple transcription, LLM, or TTS providers
22- Working with streaming audio processing pipelines
23- The user mentions Vocode, voice engines, or conversational AI
24 
25## Core Architecture Principles
26 
27### The Worker Pipeline Pattern
28 
29Every voice AI engine follows this pipeline:
30 
31```
32Audio In → Transcriber → Agent → Synthesizer → Audio Out
33           (Worker 1)   (Worker 2)  (Worker 3)
34```
35 
36**Key Benefits:**
37- **Decoupling**: Workers only know about their input/output queues
38- **Concurrency**: All workers run simultaneously via asyncio
39- **Backpressure**: Queues automatically handle rate differences
40- **Interruptibility**: Everything can be stopped mid-stream
41 
42### Base Worker Pattern
43 
44Every worker follows this pattern:
45 
46```python
47class BaseWorker:
48    def __init__(self, input_queue, output_queue):
49        self.input_queue = input_queue   # asyncio.Queue to consume from
50        self.output_queue = output_queue # asyncio.Queue to produce to
51        self.active = False
52    
53    def start(self):
54        """Start the worker's processing loop"""
55        self.active = True
56        asyncio.create_task(self._run_loop())
57    
58    async def _run_loop(self):
59        """Main processing loop - runs forever until terminated"""
60        while self.active:
61            item = await self.input_queue.get()  # Block until item arrives
62            await self.process(item)              # Process the item
63    
64    async def process(self, item):
65        """Override this - does the actual work"""
66        raise NotImplementedError
67    
68    def terminate(self):
69        """Stop the worker"""
70        self.active = False
71```
72 
73## Component Implementation Guide
74 
75### 1. Transcriber (Audio → Text)
76 
77**Purpose**: Converts incoming audio chunks to text transcriptions
78 
79**Interface Requirements**:
80```python
81class BaseTranscriber:
82    def __init__(self, transcriber_config):
83        self.input_queue = asyncio.Queue()   # Audio chunks (bytes)
84        self.output_queue = asyncio.Queue()  # Transcriptions
85        self.is_muted = False
86    
87    def send_audio(self, chunk: bytes):
88        """Client calls this to send audio"""
89        if not self.is_muted:
90            self.input_queue.put_nowait(chunk)
91        else:
92            # Send silence instead (prevents echo during bot speech)
93            self.input_queue.put_nowait(self.create_silent_chunk(len(chunk)))
94    
95    def mute(self):
96        """Called when bot starts speaking (prevents echo)"""
97        self.is_muted = True
98    
99    def unmute(self):
100        """Called when bot stops speaking"""
101        self.is_muted = False
102```
103 
104**Output Format**:
105```python
106class Transcription:
107    message: str          # "Hello, how are you?"
108    confidence: float     # 0.95
109    is_final: bool        # True = complete sentence, False = partial
110    is_interrupt: bool    # Set by TranscriptionsWorker
111```
112 
113**Supported Providers**:
114- **Deepgram** - Fast, accurate, streaming
115- **AssemblyAI** - High accuracy, good for accents
116- **Azure Speech** - Enterprise-grade
117- **Google Cloud Speech** - Multi-language support
118 
119**Critical Implementation Details**:
120- Use WebSocket for bidirectional streaming
121- Run sender and receiver tasks concurrently with `asyncio.gather()`
122- Mute transcriber when bot speaks to prevent echo/feedback loops
123- Handle both final and partial transcriptions
124 
125### 2. Agent (Text → Response)
126 
127**Purpose**: Processes user input and generates conversational responses
128 
129**Interface Requirements**:
130```python
131class BaseAgent:
132    def __init__(self, agent_config):
133        self.input_queue = asyncio.Queue()   # TranscriptionAgentInput
134        self.output_queue = asyncio.Queue()  # AgentResponse
135        self.transcript = None               # Conversation history
136    
137    async def generate_response(self, human_input, is_interrupt, conversation_id):
138        """Override this - returns AsyncGenerator of responses"""
139        raise NotImplementedError
140```
141 
142**Why Streaming Responses?**
143- **Lower latency**: Start speaking as soon as first sentence is ready
144- **Better interrupts**: Can stop mid-response
145- **Sentence-by-sentence**: More natural conversation flow
146 
147**Supported Providers**:
148- **OpenAI** (GPT-4, GPT-3.5) - High quality, fast
149- **Google Gemini** - Multimodal, cost-effective
150- **Anthropic Claude** - Long context, nuanced responses
151 
152**Critical Implementation Details**:
153- Maintain conversation history in `Transcript` object
154- Stream responses using `AsyncGenerator`
155- **IMPORTANT**: Buffer entire LLM response before yielding to synthesizer (prevents audio jumping)
156- Handle interrupts by canceling current generation task
157- Update conversation history with partial messages on interrupt
158 
159### 3. Synthesizer (Text → Audio)
160 
161**Purpose**: Converts agent text responses to speech audio
162 
163**Interface Requirements**:
164```python
165class BaseSynthesizer:
166    async def create_speech(self, message: BaseMessage, chunk_size: int) -> SynthesisResult:
167        """
168        Returns a SynthesisResult containing:
169        - chunk_generator: AsyncGenerator that yields audio chunks
170        - get_message_up_to: Function to get partial text (for interrupts)
171        """
172        raise NotImplementedError
173```
174 
175**SynthesisResult Structure**:
176```python
177class SynthesisResult:
178    chunk_generator: AsyncGenerator[ChunkResult, None]
179    get_message_up_to: Callable[[float], str]  # seconds → partial text
180    
181    class ChunkResult:
182        chunk: bytes          # Raw PCM audio
183        is_last_chunk: bool
184```
185 
186**Supported Providers**:
187- **ElevenLabs** - Most natural voices, streaming
188- **Azure TTS** - Enterprise-grade, many languages
189- **Google Cloud TTS** - Cost-effective, good quality
190- **Amazon Polly** - AWS integration
191- **Play.ht** - Voice cloning
192 
193**Critical Implementation Details**:
194- Stream audio chunks as they're generated
195- Convert audio to LINEAR16 PCM format (16kHz sample rate)
196- Implement `get_message_up_to()` for interrupt handling
197- Handle audio format conversion (MP3 → PCM)
198 
199### 4. Output Device (Audio → Client)
200 
201**Purpose**: Sends synthesized audio back to the client
202 
203**CRITICAL: Rate Limiting for Interrupts**
204 
205```python
206async def send_speech_to_output(self, message, synthesis_result,
207                                stop_event, seconds_per_chunk):
208    chunk_idx = 0
209    async for chunk_result in synthesis_result.chunk_generator:
210        # Check for interrupt
211        if stop_event.is_set():
212            logger.debug(f"Interrupted after {chunk_idx} chunks")
213            message_sent = synthesis_result.get_message_up_to(
214                chunk_idx * seconds_per_chunk
215            )
216            return message_sent, True  # cut_off = True
217        
218        start_time = time.time()
219        
220        # Send chunk to output device
221        self.output_device.consume_nonblocking(chunk_result.chunk)
222        
223        # CRITICAL: Wait for chunk to play before sending next one
224        # This is what makes interrupts work!
225        speech_length = seconds_per_chunk
226        processing_time = time.time() - start_time
227        await asyncio.sleep(max(speech_length - processing_time, 0))
228        
229        chunk_idx += 1
230    
231    return message, False  # cut_off = False
232```
233 
234**Why Rate Limiting?**
235Without rate limiting, all audio chunks would be sent immediately, which would:
236- Buffer entire message on client side
237- Make interrupts impossible (all audio already sent)
238- Cause timing issues
239 
240By sending one chunk every N seconds:
241- Real-time playback is maintained
242- Interrupts can stop mid-sentence
243- Natural conversation flow is preserved
244 
245## The Interrupt System
246 
247The interrupt system is critical for natural conversations.
248 
249### How Interrupts Work
250 
251**Scenario**: Bot is saying "I think the weather will be nice today and tomorrow and—" when user interrupts with "Stop".
252 
253**Step 1: User starts speaking**
254```python
255# TranscriptionsWorker detects new transcription while bot speaking
256async def process(self, transcription):
257    if not self.conversation.is_human_speaking:  # Bot was speaking!
258        # Broadcast interrupt to all in-flight events
259        interrupted = self.conversation.broadcast_interrupt()
260        transcription.is_interrupt = interrupted
261```
262 
263**Step 2: broadcast_interrupt() stops everything**
264```python
265def broadcast_interrupt(self):
266    num_interrupts = 0
267    # Interrupt all queued events
268    while True:
269        try:
270            interruptible_event = self.interruptible_events.get_nowait()
271            if interruptible_event.interrupt():  # Sets interruption_event
272                num_interrupts += 1
273        except queue.Empty:
274            break
275    
276    # Cancel current tasks
277    self.agent.cancel_current_task()              # Stop generating text
278    self.agent_responses_worker.cancel_current_task()  # Stop synthesizing
279    return num_interrupts > 0
280```
281 
282**Step 3: SynthesisResultsWorker detects interrupt**
283```python
284async def send_speech_to_output(self, synthesis_result, stop_event, ...):
285    async for chunk_result in synthesis_result.chunk_generator:
286        # Check stop_event (this is the interruption_event)
287        if stop_event.is_set():
288            logger.debug("Interrupted! Stopping speech.")
289            # Calculate what was actually spoken
290            seconds_spoken = chunk_idx * seconds_per_chunk
291            partial_message = synthesis_result.get_message_up_to(seconds_spoken)
292            # e.g., "I think the weather will be nice today"
293            return partial_message, True  # cut_off = True
294```
295 
296**Step 4: Agent updates history**
297```python
298if cut_off:
299    # Update conversation history with partial message
300    self.agent.update_last_bot_message_on_cut_off(message_sent)
301    # History now shows:
302    # Bot: "I think the weather will be nice today" (incomplete)
303```
304 
305### InterruptibleEvent Pattern
306 
307Every event in the pipeline is wrapped in an `InterruptibleEvent`:
308 
309```python
310class InterruptibleEvent:
311    def __init__(self, payload, is_interruptible=True):
312        self.payload = payload
313        self.is_interruptible = is_interruptible
314        self.interruption_event = threading.Event()  # Initially not set
315        self.interrupted = False
316    
317    def interrupt(self) -> bool:
318        """Interrupt this event"""
319        if not self.is_interruptible:
320            return False
321        if not self.interrupted:
322            self.interruption_event.set()  # Signal to stop!
323            self.interrupted = True
324            return True
325        return False
326    
327    def is_interrupted(self) -> bool:
328        return self.interruption_event.is_set()
329```
330 
331## Multi-Provider Factory Pattern
332 
333Support multiple providers with a factory pattern:
334 
335```python
336class VoiceHandler:
337    """Multi-provider factory for voice components"""
338    
339    def create_transcriber(self, agent_config: Dict):
340        """Create transcriber based on transcriberProvider"""
341        provider = agent_config.get("transcriberProvider", "deepgram")
342        
343        if provider == "deepgram":
344            return self._create_deepgram_transcriber(agent_config)
345        elif provider == "assemblyai":
346            return self._create_assemblyai_transcriber(agent_config)
347        elif provider == "azure":
348            return self._create_azure_transcriber(agent_config)
349        elif provider == "google":
350            return self._create_google_transcriber(agent_config)
351        else:
352            raise ValueError(f"Unknown transcriber provider: {provider}")
353    
354    def create_agent(self, agent_config: Dict):
355        """Create LLM agent based on llmProvider"""
356        provider = agent_config.get("llmProvider", "openai")
357        
358        if provider == "openai":
359            return self._create_openai_agent(agent_config)
360        elif provider == "gemini":
361            return self._create_gemini_agent(agent_config)
362        else:
363            raise ValueError(f"Unknown LLM provider: {provider}")
364    
365    def create_synthesizer(self, agent_config: Dict):
366        """Create voice synthesizer based on voiceProvider"""
367        provider = agent_config.get("voiceProvider", "elevenlabs")
368        
369        if provider == "elevenlabs":
370            return self._create_elevenlabs_synthesizer(agent_config)
371        elif provider == "azure":
372            return self._create_azure_synthesizer(agent_config)
373        elif provider == "google":
374            return self._create_google_synthesizer(agent_config)
375        elif provider == "polly":
376            return self._create_polly_synthesizer(agent_config)
377        elif provider == "playht":
378            return self._create_playht_synthesizer(agent_config)
379        else:
380            raise ValueError(f"Unknown voice provider: {provider}")
381```
382 
383## WebSocket Integration
384 
385Voice AI engines typically use WebSocket for bidirectional audio streaming:
386 
387```python
388@app.websocket("/conversation")
389async def websocket_endpoint(websocket: WebSocket):
390    await websocket.accept()
391    
392    # Create voice components
393    voice_handler = VoiceHandler()
394    transcriber = voice_handler.create_transcriber(agent_config)
395    agent = voice_handler.create_agent(agent_config)
396    synthesizer = voice_handler.create_synthesizer(agent_config)
397    
398    # Create output device
399    output_device = WebsocketOutputDevice(
400        ws=websocket,
401        sampling_rate=16000,
402        audio_encoding=AudioEncoding.LINEAR16
403    )
404    
405    # Create conversation orchestrator
406    conversation = StreamingConversation(
407        output_device=output_device,
408        transcriber=transcriber,
409        agent=agent,
410        synthesizer=synthesizer
411    )
412    
413    # Start all workers
414    await conversation.start()
415    
416    try:
417        # Receive audio from client
418        async for message in websocket.iter_bytes():
419            conversation.receive_audio(message)
420    except WebSocketDisconnect:
421        logger.info("Client disconnected")
422    finally:
423        await conversation.terminate()
424```
425 
426## Common Pitfalls and Solutions
427 
428### 1. Audio Jumping/Cutting Off
429 
430**Problem**: Bot's audio jumps or cuts off mid-response.
431 
432**Cause**: Sending text to synthesizer in small chunks causes multiple TTS calls.
433 
434**Solution**: Buffer the entire LLM response before sending to synthesizer:
435 
436```python
437# ❌ Bad: Yields sentence-by-sentence
438async for sentence in llm_stream:
439    yield GeneratedResponse(message=BaseMessage(text=sentence))
440 
441# ✅ Good: Buffer entire response
442full_response = ""
443async for chunk in llm_stream:
444    full_response += chunk
445yield GeneratedResponse(message=BaseMessage(text=full_response))
446```
447 
448### 2. Echo/Feedback Loop
449 
450**Problem**: Bot hears itself speaking and responds to its own audio.
451 
452**Cause**: Transcriber not muted during bot speech.
453 
454**Solution**: Mute transcriber when bot starts speaking:
455 
456```python
457# Before sending audio to output
458self.transcriber.mute()
459# After audio playback complete
460self.transcriber.unmute()
461```
462 
463### 3. Interrupts Not Working
464 
465**Problem**: User can't interrupt bot mid-sentence.
466 
467**Cause**: All audio chunks sent at once instead of rate-limited.
468 
469**Solution**: Rate-limit audio chunks to match real-time playback:
470 
471```python
472async for chunk in synthesis_result.chunk_generator:
473    start_time = time.time()
474    
475    # Send chunk
476    output_device.consume_nonblocking(chunk)
477    
478    # Wait for chunk duration before sending next
479    processing_time = time.time() - start_time
480    await asyncio.sleep(max(seconds_per_chunk - processing_time, 0))
481```
482 
483### 4. Memory Leaks from Unclosed Streams
484 
485**Problem**: Memory usage grows over time.
486 
487**Cause**: WebSocket connections or API streams not properly closed.
488 
489**Solution**: Always use context managers and cleanup:
490 
491```python
492try:
493    async with websockets.connect(url) as ws:
494        # Use websocket
495        pass
496finally:
497    # Cleanup
498    await conversation.terminate()
499    await transcriber.terminate()
500```
501 
502## Production Considerations
503 
504### 1. Error Handling
505 
506```python
507async def _run_loop(self):
508    while self.active:
509        try:
510            item = await self.input_queue.get()
511            await self.process(item)
512        except Exception as e:
513            logger.error(f"Worker error: {e}", exc_info=True)
514            # Don't crash the worker, continue processing
515```
516 
517### 2. Graceful Shutdown
518 
519```python
520async def terminate(self):
521    """Gracefully shut down all workers"""
522    self.active = False
523    
524    # Stop all workers
525    self.transcriber.terminate()
526    self.agent.terminate()
527    self.synthesizer.terminate()
528    
529    # Wait for queues to drain
530    await asyncio.sleep(0.5)
531    
532    # Close connections
533    if self.websocket:
534        await self.websocket.close()
535```
536 
537### 3. Monitoring and Logging
538 
539```python
540# Log key events
541logger.info(f"🎤 [TRANSCRIBER] Received: '{transcription.message}'")
542logger.info(f"🤖 [AGENT] Generating response...")
543logger.info(f"🔊 [SYNTHESIZER] Synthesizing {len(text)} characters")
544logger.info(f"⚠️ [INTERRUPT] User interrupted bot")
545 
546# Track metrics
547metrics.increment("transcriptions.count")
548metrics.timing("agent.response_time", duration)
549metrics.gauge("active_conversations", count)
550```
551 
552### 4. Rate Limiting and Quotas
553 
554```python
555# Implement rate limiting for API calls
556from aiolimiter import AsyncLimiter
557 
558rate_limiter = AsyncLimiter(max_rate=10, time_period=1)  # 10 calls/second
559 
560async def call_api(self, data):
561    async with rate_limiter:
562        return await self.client.post(data)
563```
564 
565## Key Design Patterns
566 
567### 1. Producer-Consumer with Queues
568 
569```python
570# Producer
571async def producer(queue):
572    while True:
573        item = await generate_item()
574        queue.put_nowait(item)
575 
576# Consumer
577async def consumer(queue):
578    while True:
579        item = await queue.get()
580        await process_item(item)
581```
582 
583### 2. Streaming Generators
584 
585Instead of returning complete results:
586 
587```python
588# ❌ Bad: Wait for entire response
589async def generate_response(prompt):
590    response = await openai.complete(prompt)  # 5 seconds
591    return response
592 
593# ✅ Good: Stream chunks as they arrive
594async def generate_response(prompt):
595    async for chunk in openai.complete(prompt, stream=True):
596        yield chunk  # Yield after 0.1s, 0.2s, etc.
597```
598 
599### 3. Conversation State Management
600 
601Maintain conversation history for context:
602 
603```python
604class Transcript:
605    event_logs: List[Message] = []
606    
607    def add_human_message(self, text):
608        self.event_logs.append(Message(sender=Sender.HUMAN, text=text))
609    
610    def add_bot_message(self, text):
611        self.event_logs.append(Message(sender=Sender.BOT, text=text))
612    
613    def to_openai_messages(self):
614        return [
615            {"role": "user" if msg.sender == Sender.HUMAN else "assistant",
616             "content": msg.text}
617            for msg in self.event_logs
618        ]
619```
620 
621## Testing Strategies
622 
623### 1. Unit Test Workers in Isolation
624 
625```python
626async def test_transcriber():
627    transcriber = DeepgramTranscriber(config)
628    
629    # Mock audio input
630    audio_chunk = b'\x00\x01\x02...'
631    transcriber.send_audio(audio_chunk)
632    
633    # Check output
634    transcription = await transcriber.output_queue.get()
635    assert transcription.message == "expected text"
636```
637 
638### 2. Integration Test Pipeline
639 
640```python
641async def test_full_pipeline():
642    # Create all components
643    conversation = create_test_conversation()
644    
645    # Send test audio
646    conversation.receive_audio(test_audio_chunk)
647    
648    # Wait for response
649    response = await wait_for_audio_output(timeout=5)
650    
651    assert response is not None
652```
653 
654### 3. Test Interrupts
655 
656```python
657async def test_interrupt():
658    conversation = create_test_conversation()
659    
660    # Start bot speaking
661    await conversation.agent.generate_response("Tell me a long story")
662    
663    # Interrupt mid-response
664    await asyncio.sleep(1)  # Let it speak for 1 second
665    conversation.broadcast_interrupt()
666    
667    # Verify partial message in transcript
668    last_message = conversation.transcript.event_logs[-1]
669    assert last_message.text != full_expected_message
670```
671 
672## Implementation Workflow
673 
674When implementing a voice AI engine:
675 
6761. **Start with Base Workers**: Implement the base worker pattern first
6772. **Add Transcriber**: Choose a provider and implement streaming transcription
6783. **Add Agent**: Implement LLM integration with streaming responses
6794. **Add Synthesizer**: Implement TTS with audio streaming
6805. **Connect Pipeline**: Wire all workers together with queues
6816. **Add Interrupts**: Implement the interrupt system
6827. **Add WebSocket**: Create WebSocket endpoint for client communication
6838. **Test Components**: Unit test each worker in isolation
6849. **Test Integration**: Test the full pipeline end-to-end
68510. **Add Error Handling**: Implement robust error handling and logging
68611. **Optimize**: Add rate limiting, monitoring, and performance optimizations
687 
688## Related Skills
689 
690- `@websocket-patterns` - For WebSocket implementation details
691- `@async-python` - For asyncio and async patterns
692- `@streaming-apis` - For streaming API integration
693- `@audio-processing` - For audio format conversion and processing
694- `@systematic-debugging` - For debugging complex async pipelines
695 
696## Resources
697 
698**Libraries**:
699- `asyncio` - Async programming
700- `websockets` - WebSocket client/server
701- `FastAPI` - WebSocket server framework
702- `pydub` - Audio manipulation
703- `numpy` - Audio data processing
704 
705**API Providers**:
706- Transcription: Deepgram, AssemblyAI, Azure Speech, Google Cloud Speech
707- LLM: OpenAI, Google Gemini, Anthropic Claude
708- TTS: ElevenLabs, Azure TTS, Google Cloud TTS, Amazon Polly, Play.ht
709 
710## Summary
711 
712Building a voice AI engine requires:
713- ✅ Async worker pipeline for concurrent processing
714- ✅ Queue-based communication between components
715- ✅ Streaming at every stage (transcription, LLM, synthesis)
716- ✅ Interrupt system for natural conversations
717- ✅ Rate limiting for real-time audio playback
718- ✅ Multi-provider support for flexibility
719- ✅ Proper error handling and graceful shutdown
720 
721**The key insight**: Everything must stream and everything must be interruptible for natural, real-time conversations.
722