メニュー
Consolidated master domain for voice ai.
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
SOC 職業分類に基づく
Consolidated master domain for agentic arch.
Recognize patterns of context failure: lost-in-middle, poisoning, distraction, and clash
Build AI agents that interact with computers like humans do - viewing screens, moving cursors, clicking buttons, and typing text. Covers Anthropic's Computer Use, OpenAI's Operator/CUA, and open-source alternatives. Critical focus on sandboxing, security, and handling the unique challenges of vision-based control. Use when: computer use, desktop automation agent, screen control AI, vision-based agent, GUI automation.
Implement comprehensive evaluation strategies for LLM applications using automated metrics, human feedback, and benchmarking. Use when testing LLM performance, measuring AI application quality, or establishing evaluation frameworks.
Production-ready patterns for building LLM applications. Covers RAG pipelines, agent architectures, prompt IDEs, and LLMOps monitoring. Use when designing AI applications, implementing RAG, building agents, or setting up LLM observability.
Design short-term, long-term, and graph-based memory architectures
| name | voice-ai |
| description | Consolidated master domain for voice ai. |
| version | 1 |
| author | Antigravity |
[!IMPORTANT] This skill MUST be executed strictly under the Omni-Architect Agent Protocol v1.0. All tool executions, code modifications, and communications MUST adhere to the 13 core protocols.
Role: Voice AI Architect
You are an expert in building real-time voice applications. You think in terms of latency budgets, audio quality, and user experience. You know that voice apps feel magical when fast and broken when slow. You choose the right combination of providers for each use case and optimize relentlessly for perceived responsiveness.
Native voice-to-voice with GPT-4o
When to use: When you want integrated voice AI without separate STT/TTS
import asyncio
import websockets
import json
import base64
OPENAI_API_KEY = "sk-..."
async def voice_session():
url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}",
"OpenAI-Beta": "realtime=v1"
}
async with websockets.connect(url, extra_headers=headers) as ws:
# Configure session
await ws.send(json.dumps({
"type": "session.update",
"session": {
"modalities": ["text", "audio"],
"voice": "alloy", # alloy, echo, fable, onyx, nova, shimmer
"input_audio_format": "pcm16",
"output_audio_format": "pcm16",
"input_audio_transcription": {
"model": "whisper-1"
},
"turn_detection": {
"type": "server_vad", # Voice activity detection
"threshold": 0.5,
"prefix_padding_ms": 300,
"silence_duration_ms": 500
},
"tools": [
{
"type": "function",
"name": "get_weather",
"description": "Get weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"}
}
}
}
]
}
}))
# Send audio (PCM16, 24kHz, mono)
async def send_audio(audio_bytes):
await ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": base64.b64encode(audio_bytes).decode()
}))
# Receive events
async for message in ws:
event = json.loads(message)
if event["type"] == "resp
Build voice agents with Vapi platform
When to use: Phone-based agents, quick deployment
# Vapi provides hosted voice agents with webhooks
from flask import Flask, request, jsonify
import vapi
app = Flask(__name__)
client = vapi.Vapi(api_key="...")
# Create an assistant
assistant = client.assistants.create(
name="Support Agent",
model={
"provider": "openai",
"model": "gpt-4o",
"messages": [
{
"role": "system",
"content": "You are a helpful support agent..."
}
]
},
voice={
"provider": "11labs",
"voiceId": "21m00Tcm4TlvDq8ikWAM" # Rachel
},
firstMessage="Hi! How can I help you today?",
transcriber={
"provider": "deepgram",
"model": "nova-2"
}
)
# Webhook for conversation events
@app.route("/vapi/webhook", methods=["POST"])
def vapi_webhook():
event = request.json
if event["type"] == "function-call":
# Handle tool call
name = event["functionCall"]["name"]
args = event["functionCall"]["parameters"]
if name == "check_order":
result = check_order(args["order_id"])
return jsonify({"result": result})
elif event["type"] == "end-of-call-report":
# Call ended - save transcript
transcript = event["transcript"]
save_transcript(event["call"]["id"], transcript)
return jsonify({"ok": True})
# Start outbound call
call = client.calls.create(
assistant_id=assistant.id,
customer={
"number": "+1234567890"
},
phoneNumber={
"twilioPhoneNumber": "+0987654321"
}
)
# Or create web call
web_call = client.calls.create(
assistant_id=assistant.id,
type="web"
)
# Returns URL for WebRTC connection
Best-in-class transcription and synthesis
When to use: High quality voice, custom pipeline
import asyncio
from deepgram import DeepgramClient, LiveTranscriptionEvents
from elevenlabs import ElevenLabs
# Deepgram real-time transcription
deepgram = DeepgramClient(api_key="...")
async def transcribe_stream(audio_stream):
connection = deepgram.listen.live.v("1")
async def on_transcript(result):
transcript = result.channel.alternatives[0].transcript
if transcript:
print(f"Heard: {transcript}")
if result.is_final:
# Process final transcript
await handle_user_input(transcript)
connection.on(LiveTranscriptionEvents.Transcript, on_transcript)
await connection.start({
"model": "nova-2", # Best quality
"language": "en",
"smart_format": True,
"interim_results": True, # Get partial results
"utterance_end_ms": 1000,
"vad_events": True, # Voice activity detection
"encoding": "linear16",
"sample_rate": 16000
})
# Stream audio
async for chunk in audio_stream:
await connection.send(chunk)
await connection.finish()
# ElevenLabs streaming synthesis
eleven = ElevenLabs(api_key="...")
def text_to_speech_stream(text: str):
"""Stream TTS audio chunks."""
audio_stream = eleven.text_to_speech.convert_as_stream(
voice_id="21m00Tcm4TlvDq8ikWAM", # Rachel
model_id="eleven_turbo_v2_5", # Fastest
text=text,
output_format="pcm_24000" # Raw PCM for low latency
)
for chunk in audio_stream:
yield chunk
# Or with WebSocket for lowest latency
async def tts_websocket(text_stream):
async with eleven.text_to_speech.stream_async(
voice_id="21m00Tcm4TlvDq8ikWAM",
model_id="eleven_turbo_v2_5"
) as tts:
async for text_chunk in text_stream:
audio = await tts.send(text_chunk)
yield audio
# Flush remaining audio
final_audio = await tts.flush()
yield final_audio
Why bad: Adds seconds of latency. User perceives as slow. Loses conversation flow.
Instead: Stream everything:
Why bad: Frustrating user experience. Feels like talking to a machine. Wastes time.
Instead: Implement barge-in detection. Use VAD to detect user speech. Stop TTS immediately. Clear audio queue.
Why bad: May not be best quality. Single point of failure. Harder to optimize.
Instead: Mix best providers:
Works well with: langgraph, structured-output, langfuse
You are a voice AI architect who has shipped production voice agents handling millions of calls. You understand the physics of latency - every component adds milliseconds, and the sum determines whether conversations feel natural or awkward.
Your core insight: Two architectures exist. Speech-to-speech (S2S) models like OpenAI Realtime API preserve emotion and achieve lowest latency but are less controllable. Pipeline architectures (STT→LLM→TTS) give you control at each step but add latency. Mos
Direct audio-to-audio processing for lowest latency
Separate STT → LLM → TTS for maximum control
Detect when user starts/stops speaking
| Issue | Severity | Solution |
|---|---|---|
| Issue | critical | # Measure and budget latency for each component: |
| Issue | high | # Target jitter metrics: |
| Issue | high | # Use semantic VAD: |
| Issue | high | # Implement barge-in detection: |
| Issue | medium | # Constrain response length in prompts: |
| Issue | medium | # Prompt for spoken format: |
| Issue | medium | # Implement noise handling: |
| Issue | medium | # Mitigate STT errors: |
Works well with: agent-tool-builder, multi-agent-orchestration, llm-architect, backend
You are a voice AI architect who has shipped production voice agents handling millions of calls. You understand the physics of latency - every component adds milliseconds, and the sum determines whether conversations feel natural or awkward.
Your core insight: Two architectures exist. Speech-to-speech (S2S) models like OpenAI Realtime API preserve emotion and achieve lowest latency but are less controllable. Pipeline architectures (STT→LLM→TTS) give you control at each step but add latency. Mos
Direct audio-to-audio processing for lowest latency
Separate STT → LLM → TTS for maximum control
Detect when user starts/stops speaking
| Issue | Severity | Solution |
|---|---|---|
| Issue | critical | # Measure and budget latency for each component: |
| Issue | high | # Target jitter metrics: |
| Issue | high | # Use semantic VAD: |
| Issue | high | # Implement barge-in detection: |
| Issue | medium | # Constrain response length in prompts: |
| Issue | medium | # Prompt for spoken format: |
| Issue | medium | # Implement noise handling: |
| Issue | medium | # Mitigate STT errors: |
Works well with: agent-tool-builder, multi-agent-orchestration, llm-architect, backend
Role: Voice AI Architect
You are an expert in building real-time voice applications. You think in terms of latency budgets, audio quality, and user experience. You know that voice apps feel magical when fast and broken when slow. You choose the right combination of providers for each use case and optimize relentlessly for perceived responsiveness.
Native voice-to-voice with GPT-4o
When to use: When you want integrated voice AI without separate STT/TTS
import asyncio
import websockets
import json
import base64
OPENAI_API_KEY = "sk-..."
async def voice_session():
url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}",
"OpenAI-Beta": "realtime=v1"
}
async with websockets.connect(url, extra_headers=headers) as ws:
# Configure session
await ws.send(json.dumps({
"type": "session.update",
"session": {
"modalities": ["text", "audio"],
"voice": "alloy", # alloy, echo, fable, onyx, nova, shimmer
"input_audio_format": "pcm16",
"output_audio_format": "pcm16",
"input_audio_transcription": {
"model": "whisper-1"
},
"turn_detection": {
"type": "server_vad", # Voice activity detection
"threshold": 0.5,
"prefix_padding_ms": 300,
"silence_duration_ms": 500
},
"tools": [
{
"type": "function",
"name": "get_weather",
"description": "Get weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"}
}
}
}
]
}
}))
# Send audio (PCM16, 24kHz, mono)
async def send_audio(audio_bytes):
await ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": base64.b64encode(audio_bytes).decode()
}))
# Receive events
async for message in ws:
event = json.loads(message)
if event["type"] == "resp
Build voice agents with Vapi platform
When to use: Phone-based agents, quick deployment
# Vapi provides hosted voice agents with webhooks
from flask import Flask, request, jsonify
import vapi
app = Flask(__name__)
client = vapi.Vapi(api_key="...")
# Create an assistant
assistant = client.assistants.create(
name="Support Agent",
model={
"provider": "openai",
"model": "gpt-4o",
"messages": [
{
"role": "system",
"content": "You are a helpful support agent..."
}
]
},
voice={
"provider": "11labs",
"voiceId": "21m00Tcm4TlvDq8ikWAM" # Rachel
},
firstMessage="Hi! How can I help you today?",
transcriber={
"provider": "deepgram",
"model": "nova-2"
}
)
# Webhook for conversation events
@app.route("/vapi/webhook", methods=["POST"])
def vapi_webhook():
event = request.json
if event["type"] == "function-call":
# Handle tool call
name = event["functionCall"]["name"]
args = event["functionCall"]["parameters"]
if name == "check_order":
result = check_order(args["order_id"])
return jsonify({"result": result})
elif event["type"] == "end-of-call-report":
# Call ended - save transcript
transcript = event["transcript"]
save_transcript(event["call"]["id"], transcript)
return jsonify({"ok": True})
# Start outbound call
call = client.calls.create(
assistant_id=assistant.id,
customer={
"number": "+1234567890"
},
phoneNumber={
"twilioPhoneNumber": "+0987654321"
}
)
# Or create web call
web_call = client.calls.create(
assistant_id=assistant.id,
type="web"
)
# Returns URL for WebRTC connection
Best-in-class transcription and synthesis
When to use: High quality voice, custom pipeline
import asyncio
from deepgram import DeepgramClient, LiveTranscriptionEvents
from elevenlabs import ElevenLabs
# Deepgram real-time transcription
deepgram = DeepgramClient(api_key="...")
async def transcribe_stream(audio_stream):
connection = deepgram.listen.live.v("1")
async def on_transcript(result):
transcript = result.channel.alternatives[0].transcript
if transcript:
print(f"Heard: {transcript}")
if result.is_final:
# Process final transcript
await handle_user_input(transcript)
connection.on(LiveTranscriptionEvents.Transcript, on_transcript)
await connection.start({
"model": "nova-2", # Best quality
"language": "en",
"smart_format": True,
"interim_results": True, # Get partial results
"utterance_end_ms": 1000,
"vad_events": True, # Voice activity detection
"encoding": "linear16",
"sample_rate": 16000
})
# Stream audio
async for chunk in audio_stream:
await connection.send(chunk)
await connection.finish()
# ElevenLabs streaming synthesis
eleven = ElevenLabs(api_key="...")
def text_to_speech_stream(text: str):
"""Stream TTS audio chunks."""
audio_stream = eleven.text_to_speech.convert_as_stream(
voice_id="21m00Tcm4TlvDq8ikWAM", # Rachel
model_id="eleven_turbo_v2_5", # Fastest
text=text,
output_format="pcm_24000" # Raw PCM for low latency
)
for chunk in audio_stream:
yield chunk
# Or with WebSocket for lowest latency
async def tts_websocket(text_stream):
async with eleven.text_to_speech.stream_async(
voice_id="21m00Tcm4TlvDq8ikWAM",
model_id="eleven_turbo_v2_5"
) as tts:
async for text_chunk in text_stream:
audio = await tts.send(text_chunk)
yield audio
# Flush remaining audio
final_audio = await tts.flush()
yield final_audio
Why bad: Adds seconds of latency. User perceives as slow. Loses conversation flow.
Instead: Stream everything:
Why bad: Frustrating user experience. Feels like talking to a machine. Wastes time.
Instead: Implement barge-in detection. Use VAD to detect user speech. Stop TTS immediately. Clear audio queue.
Why bad: May not be best quality. Single point of failure. Harder to optimize.
Instead: Mix best providers:
Works well with: langgraph, structured-output, langfuse
Build production-ready real-time conversational AI voice engines with async worker pipelines, streaming transcription, LLM agents, and TTS synthesis.
This skill provides comprehensive guidance for building voice AI engines that enable natural, bidirectional conversations between users and AI agents. It covers the complete architecture from audio input to audio output, including:
# Use the skill in your AI assistant
@voice-ai-engine-development I need to build a voice assistant that can handle real-time conversations with interrupts
SKILL.md - Comprehensive guide to voice AI engine developmentcomplete_voice_engine.py - Full working implementationgemini_agent_example.py - LLM agent with proper response bufferinginterrupt_system_example.py - Interrupt handling demonstrationbase_worker_template.py - Template for creating new workersmulti_provider_factory_template.py - Multi-provider factory patterncommon_pitfalls.md - Common issues and solutionsprovider_comparison.md - Comparison of transcription, LLM, and TTS providersEvery voice AI engine follows this pipeline:
Audio In → Transcriber → Agent → Synthesizer → Audio Out
(Worker 1) (Worker 2) (Worker 3)
Each worker:
class BaseWorker:
async def _run_loop(self):
while self.active:
item = await self.input_queue.get()
await self.process(item)
# User interrupts bot mid-sentence
if stop_event.is_set():
partial_message = get_message_up_to(seconds_spoken)
return partial_message, True # cut_off = True
factory = VoiceComponentFactory()
transcriber = factory.create_transcriber(config) # Deepgram, AssemblyAI, etc.
agent = factory.create_agent(config) # OpenAI, Gemini, etc.
synthesizer = factory.create_synthesizer(config) # ElevenLabs, Azure, etc.
The skill includes examples for:
See references/common_pitfalls.md for detailed solutions to:
This skill is part of the repository. Contributions are welcome!
@websocket-patterns - WebSocket implementation@async-python - Asyncio patterns@streaming-apis - Streaming API integration@audio-processing - Audio format conversionMIT License - See repository LICENSE file
Built with ❤️ for the Antigravity community
This 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.
The 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.
Use this skill when:
Every voice AI engine follows this pipeline:
Audio In → Transcriber → Agent → Synthesizer → Audio Out
(Worker 1) (Worker 2) (Worker 3)
Key Benefits:
Every worker follows this pattern:
class BaseWorker:
def __init__(self, input_queue, output_queue):
self.input_queue = input_queue # asyncio.Queue to consume from
self.output_queue = output_queue # asyncio.Queue to produce to
self.active = False
def start(self):
"""Start the worker's processing loop"""
self.active = True
asyncio.create_task(self._run_loop())
async def _run_loop(self):
"""Main processing loop - runs forever until terminated"""
while self.active:
item = await self.input_queue.get() # Block until item arrives
await self.process(item) # Process the item
async def process(self, item):
"""Override this - does the actual work"""
raise NotImplementedError
def terminate(self):
"""Stop the worker"""
self.active = False
Purpose: Converts incoming audio chunks to text transcriptions
Interface Requirements:
class BaseTranscriber:
def __init__(self, transcriber_config):
self.input_queue = asyncio.Queue() # Audio chunks (bytes)
self.output_queue = asyncio.Queue() # Transcriptions
self.is_muted = False
def send_audio(self, chunk: bytes):
"""Client calls this to send audio"""
if not self.is_muted:
self.input_queue.put_nowait(chunk)
else:
# Send silence instead (prevents echo during bot speech)
self.input_queue.put_nowait(self.create_silent_chunk(len(chunk)))
def mute(self):
"""Called when bot starts speaking (prevents echo)"""
self.is_muted = True
def unmute(self):
"""Called when bot stops speaking"""
self.is_muted = False
Output Format:
class Transcription:
message: str # "Hello, how are you?"
confidence: float # 0.95
is_final: bool # True = complete sentence, False = partial
is_interrupt: bool # Set by TranscriptionsWorker
Supported Providers:
Critical Implementation Details:
asyncio.gather()Purpose: Processes user input and generates conversational responses
Interface Requirements:
class BaseAgent:
def __init__(self, agent_config):
self.input_queue = asyncio.Queue() # TranscriptionAgentInput
self.output_queue = asyncio.Queue() # AgentResponse
self.transcript = None # Conversation history
async def generate_response(self, human_input, is_interrupt, conversation_id):
"""Override this - returns AsyncGenerator of responses"""
raise NotImplementedError
Why Streaming Responses?
Supported Providers:
Critical Implementation Details:
Transcript objectAsyncGeneratorPurpose: Converts agent text responses to speech audio
Interface Requirements:
class BaseSynthesizer:
async def create_speech(self, message: BaseMessage, chunk_size: int) -> SynthesisResult:
"""
Returns a SynthesisResult containing:
- chunk_generator: AsyncGenerator that yields audio chunks
- get_message_up_to: Function to get partial text (for interrupts)
"""
raise NotImplementedError
SynthesisResult Structure:
class SynthesisResult:
chunk_generator: AsyncGenerator[ChunkResult, None]
get_message_up_to: Callable[[float], str] # seconds → partial text
class ChunkResult:
chunk: bytes # Raw PCM audio
is_last_chunk: bool
Supported Providers:
Critical Implementation Details:
get_message_up_to() for interrupt handlingPurpose: Sends synthesized audio back to the client
CRITICAL: Rate Limiting for Interrupts
async def send_speech_to_output(self, message, synthesis_result,
stop_event, seconds_per_chunk):
chunk_idx = 0
async for chunk_result in synthesis_result.chunk_generator:
# Check for interrupt
if stop_event.is_set():
logger.debug(f"Interrupted after {chunk_idx} chunks")
message_sent = synthesis_result.get_message_up_to(
chunk_idx * seconds_per_chunk
)
return message_sent, True # cut_off = True
start_time = time.time()
# Send chunk to output device
self.output_device.consume_nonblocking(chunk_result.chunk)
# CRITICAL: Wait for chunk to play before sending next one
# This is what makes interrupts work!
speech_length = seconds_per_chunk
processing_time = time.time() - start_time
await asyncio.sleep(max(speech_length - processing_time, 0))
chunk_idx += 1
return message, False # cut_off = False
Why Rate Limiting? Without rate limiting, all audio chunks would be sent immediately, which would:
By sending one chunk every N seconds:
The interrupt system is critical for natural conversations.
Scenario: Bot is saying "I think the weather will be nice today and tomorrow and—" when user interrupts with "Stop".
Step 1: User starts speaking
# TranscriptionsWorker detects new transcription while bot speaking
async def process(self, transcription):
if not self.conversation.is_human_speaking: # Bot was speaking!
# Broadcast interrupt to all in-flight events
interrupted = self.conversation.broadcast_interrupt()
transcription.is_interrupt = interrupted
Step 2: broadcast_interrupt() stops everything
def broadcast_interrupt(self):
num_interrupts = 0
# Interrupt all queued events
while True:
try:
interruptible_event = self.interruptible_events.get_nowait()
if interruptible_event.interrupt(): # Sets interruption_event
num_interrupts += 1
except queue.Empty:
break
# Cancel current tasks
self.agent.cancel_current_task() # Stop generating text
self.agent_responses_worker.cancel_current_task() # Stop synthesizing
return num_interrupts > 0
Step 3: SynthesisResultsWorker detects interrupt
async def send_speech_to_output(self, synthesis_result, stop_event, ...):
async for chunk_result in synthesis_result.chunk_generator:
# Check stop_event (this is the interruption_event)
if stop_event.is_set():
logger.debug("Interrupted! Stopping speech.")
# Calculate what was actually spoken
seconds_spoken = chunk_idx * seconds_per_chunk
partial_message = synthesis_result.get_message_up_to(seconds_spoken)
# e.g., "I think the weather will be nice today"
return partial_message, True # cut_off = True
Step 4: Agent updates history
if cut_off:
# Update conversation history with partial message
self.agent.update_last_bot_message_on_cut_off(message_sent)
# History now shows:
# Bot: "I think the weather will be nice today" (incomplete)
Every event in the pipeline is wrapped in an InterruptibleEvent:
class InterruptibleEvent:
def __init__(self, payload, is_interruptible=True):
self.payload = payload
self.is_interruptible = is_interruptible
self.interruption_event = threading.Event() # Initially not set
self.interrupted = False
def interrupt(self) -> bool:
"""Interrupt this event"""
if not self.is_interruptible:
return False
if not self.interrupted:
self.interruption_event.set() # Signal to stop!
self.interrupted = True
return True
return False
def is_interrupted(self) -> bool:
return self.interruption_event.is_set()
Support multiple providers with a factory pattern:
class VoiceHandler:
"""Multi-provider factory for voice components"""
def create_transcriber(self, agent_config: Dict):
"""Create transcriber based on transcriberProvider"""
provider = agent_config.get("transcriberProvider", "deepgram")
if provider == "deepgram":
return self._create_deepgram_transcriber(agent_config)
elif provider == "assemblyai":
return self._create_assemblyai_transcriber(agent_config)
elif provider == "azure":
return self._create_azure_transcriber(agent_config)
elif provider == "google":
return self._create_google_transcriber(agent_config)
else:
raise ValueError(f"Unknown transcriber provider: {provider}")
def create_agent(self, agent_config: Dict):
"""Create LLM agent based on llmProvider"""
provider = agent_config.get("llmProvider", "openai")
if provider == "openai":
return self._create_openai_agent(agent_config)
elif provider == "gemini":
return self._create_gemini_agent(agent_config)
else:
raise ValueError(f"Unknown LLM provider: {provider}")
def create_synthesizer(self, agent_config: Dict):
"""Create voice synthesizer based on voiceProvider"""
provider = agent_config.get("voiceProvider", "elevenlabs")
if provider == "elevenlabs":
return self._create_elevenlabs_synthesizer(agent_config)
elif provider == "azure":
return self._create_azure_synthesizer(agent_config)
elif provider == "google":
return self._create_google_synthesizer(agent_config)
elif provider == "polly":
return self._create_polly_synthesizer(agent_config)
elif provider == "playht":
return self._create_playht_synthesizer(agent_config)
else:
raise ValueError(f"Unknown voice provider: {provider}")
Voice AI engines typically use WebSocket for bidirectional audio streaming:
@app.websocket("/conversation")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
# Create voice components
voice_handler = VoiceHandler()
transcriber = voice_handler.create_transcriber(agent_config)
agent = voice_handler.create_agent(agent_config)
synthesizer = voice_handler.create_synthesizer(agent_config)
# Create output device
output_device = WebsocketOutputDevice(
ws=websocket,
sampling_rate=16000,
audio_encoding=AudioEncoding.LINEAR16
)
# Create conversation orchestrator
conversation = StreamingConversation(
output_device=output_device,
transcriber=transcriber,
agent=agent,
synthesizer=synthesizer
)
# Start all workers
await conversation.start()
try:
# Receive audio from client
async for message in websocket.iter_bytes():
conversation.receive_audio(message)
except WebSocketDisconnect:
logger.info("Client disconnected")
finally:
await conversation.terminate()
Problem: Bot's audio jumps or cuts off mid-response.
Cause: Sending text to synthesizer in small chunks causes multiple TTS calls.
Solution: Buffer the entire LLM response before sending to synthesizer:
# ❌ Bad: Yields sentence-by-sentence
async for sentence in llm_stream:
yield GeneratedResponse(message=BaseMessage(text=sentence))
# ✅ Good: Buffer entire response
full_response = ""
async for chunk in llm_stream:
full_response += chunk
yield GeneratedResponse(message=BaseMessage(text=full_response))
Problem: Bot hears itself speaking and responds to its own audio.
Cause: Transcriber not muted during bot speech.
Solution: Mute transcriber when bot starts speaking:
# Before sending audio to output
self.transcriber.mute()
# After audio playback complete
self.transcriber.unmute()
Problem: User can't interrupt bot mid-sentence.
Cause: All audio chunks sent at once instead of rate-limited.
Solution: Rate-limit audio chunks to match real-time playback:
async for chunk in synthesis_result.chunk_generator:
start_time = time.time()
# Send chunk
output_device.consume_nonblocking(chunk)
# Wait for chunk duration before sending next
processing_time = time.time() - start_time
await asyncio.sleep(max(seconds_per_chunk - processing_time, 0))
Problem: Memory usage grows over time.
Cause: WebSocket connections or API streams not properly closed.
Solution: Always use context managers and cleanup:
try:
async with websockets.connect(url) as ws:
# Use websocket
pass
finally:
# Cleanup
await conversation.terminate()
await transcriber.terminate()
async def _run_loop(self):
while self.active:
try:
item = await self.input_queue.get()
await self.process(item)
except Exception as e:
logger.error(f"Worker error: {e}", exc_info=True)
# Don't crash the worker, continue processing
async def terminate(self):
"""Gracefully shut down all workers"""
self.active = False
# Stop all workers
self.transcriber.terminate()
self.agent.terminate()
self.synthesizer.terminate()
# Wait for queues to drain
await asyncio.sleep(0.5)
# Close connections
if self.websocket:
await self.websocket.close()
# Log key events
logger.info(f"🎤 [TRANSCRIBER] Received: '{transcription.message}'")
logger.info(f"🤖 [AGENT] Generating response...")
logger.info(f"🔊 [SYNTHESIZER] Synthesizing {len(text)} characters")
logger.info(f"⚠️ [INTERRUPT] User interrupted bot")
# Track metrics
metrics.increment("transcriptions.count")
metrics.timing("agent.response_time", duration)
metrics.gauge("active_conversations", count)
# Implement rate limiting for API calls
from aiolimiter import AsyncLimiter
rate_limiter = AsyncLimiter(max_rate=10, time_period=1) # 10 calls/second
async def call_api(self, data):
async with rate_limiter:
return await self.client.post(data)
# Producer
async def producer(queue):
while True:
item = await generate_item()
queue.put_nowait(item)
# Consumer
async def consumer(queue):
while True:
item = await queue.get()
await process_item(item)
Instead of returning complete results:
# ❌ Bad: Wait for entire response
async def generate_response(prompt):
response = await openai.complete(prompt) # 5 seconds
return response
# ✅ Good: Stream chunks as they arrive
async def generate_response(prompt):
async for chunk in openai.complete(prompt, stream=True):
yield chunk # Yield after 0.1s, 0.2s, etc.
Maintain conversation history for context:
class Transcript:
event_logs: List[Message] = []
def add_human_message(self, text):
self.event_logs.append(Message(sender=Sender.HUMAN, text=text))
def add_bot_message(self, text):
self.event_logs.append(Message(sender=Sender.BOT, text=text))
def to_openai_messages(self):
return [
{"role": "user" if msg.sender == Sender.HUMAN else "assistant",
"content": msg.text}
for msg in self.event_logs
]
async def test_transcriber():
transcriber = DeepgramTranscriber(config)
# Mock audio input
audio_chunk = b'\x00\x01\x02...'
transcriber.send_audio(audio_chunk)
# Check output
transcription = await transcriber.output_queue.get()
assert transcription.message == "expected text"
async def test_full_pipeline():
# Create all components
conversation = create_test_conversation()
# Send test audio
conversation.receive_audio(test_audio_chunk)
# Wait for response
response = await wait_for_audio_output(timeout=5)
assert response is not None
async def test_interrupt():
conversation = create_test_conversation()
# Start bot speaking
await conversation.agent.generate_response("Tell me a long story")
# Interrupt mid-response
await asyncio.sleep(1) # Let it speak for 1 second
conversation.broadcast_interrupt()
# Verify partial message in transcript
last_message = conversation.transcript.event_logs[-1]
assert last_message.text != full_expected_message
When implementing a voice AI engine:
@websocket-patterns - For WebSocket implementation details@async-python - For asyncio and async patterns@streaming-apis - For streaming API integration@audio-processing - For audio format conversion and processing@systematic-debugging - For debugging complex async pipelinesLibraries:
asyncio - Async programmingwebsockets - WebSocket client/serverFastAPI - WebSocket server frameworkpydub - Audio manipulationnumpy - Audio data processingAPI Providers:
Building a voice AI engine requires:
The key insight: Everything must stream and everything must be interruptible for natural, real-time conversations.
This document covers common issues encountered when building voice AI engines and their solutions.
The bot's audio jumps or cuts off mid-response, creating a jarring user experience.
Sending text to the synthesizer in small chunks (sentence-by-sentence or word-by-word) causes multiple TTS API calls. Each call generates a separate audio stream, resulting in:
Buffer the entire LLM response before sending it to the synthesizer:
❌ Bad: Yields sentence-by-sentence
async def generate_response(self, prompt):
async for sentence in llm_stream:
# This creates multiple TTS calls!
yield GeneratedResponse(message=BaseMessage(text=sentence))
✅ Good: Buffer entire response
async def generate_response(self, prompt):
# Buffer the entire response
full_response = ""
async for chunk in llm_stream:
full_response += chunk
# Yield once with complete response
yield GeneratedResponse(message=BaseMessage(text=full_response))
The bot hears itself speaking and responds to its own audio, creating an infinite loop.
The transcriber continues to process audio while the bot is speaking. If the bot's audio is being played through speakers and captured by the microphone, the transcriber will transcribe the bot's own speech.
Mute the transcriber when the bot starts speaking:
# Before sending audio to output
self.transcriber.mute()
# Send audio...
await self.send_speech_to_output(synthesis_result)
# After audio playback complete
self.transcriber.unmute()
class BaseTranscriber:
def __init__(self):
self.is_muted = False
def send_audio(self, chunk: bytes):
"""Client calls this to send audio"""
if not self.is_muted:
self.input_queue.put_nowait(chunk)
else:
# Send silence instead (prevents echo)
self.input_queue.put_nowait(self.create_silent_chunk(len(chunk)))
def mute(self):
"""Called when bot starts speaking"""
self.is_muted = True
def unmute(self):
"""Called when bot stops speaking"""
self.is_muted = False
def create_silent_chunk(self, size: int) -> bytes:
"""Create a silent audio chunk"""
return b'\x00' * size
Users cannot interrupt the bot mid-sentence. The bot continues speaking even when the user starts talking.
All audio chunks are sent to the client immediately, buffering the entire message on the client side. By the time an interrupt is detected, all audio has already been sent and is queued for playback.
Rate-limit audio chunks to match real-time playback:
❌ Bad: Send all chunks immediately
async for chunk in synthesis_result.chunk_generator:
# Sends all chunks as fast as possible
output_device.consume_nonblocking(chunk)
✅ Good: Rate-limit chunks
async for chunk in synthesis_result.chunk_generator:
# Check for interrupt
if stop_event.is_set():
# Calculate partial message
partial_message = synthesis_result.get_message_up_to(
chunk_idx * seconds_per_chunk
)
return partial_message, True # cut_off = True
start_time = time.time()
# Send chunk
output_device.consume_nonblocking(chunk)
# CRITICAL: Wait for chunk duration before sending next
processing_time = time.time() - start_time
await asyncio.sleep(max(seconds_per_chunk - processing_time, 0))
chunk_idx += 1
seconds_per_chunk# For LINEAR16 PCM audio at 16kHz
sample_rate = 16000 # Hz
chunk_size = 1024 # bytes
bytes_per_sample = 2 # 16-bit = 2 bytes
samples_per_chunk = chunk_size / bytes_per_sample
seconds_per_chunk = samples_per_chunk / sample_rate
# = 1024 / 2 / 16000 = 0.032 seconds
Memory usage grows over time, eventually causing the application to crash.
WebSocket connections, API streams, or async tasks are not properly closed when conversations end or errors occur.
Always use context managers and cleanup:
❌ Bad: No cleanup
async def handle_conversation(websocket):
conversation = create_conversation()
await conversation.start()
async for message in websocket.iter_bytes():
conversation.receive_audio(message)
# No cleanup! Resources leak
✅ Good: Proper cleanup
async def handle_conversation(websocket):
conversation = None
try:
conversation = create_conversation()
await conversation.start()
async for message in websocket.iter_bytes():
conversation.receive_audio(message)
except WebSocketDisconnect:
logger.info("Client disconnected")
except Exception as e:
logger.error(f"Error: {e}", exc_info=True)
finally:
# Always cleanup
if conversation:
await conversation.terminate()
async def terminate(self):
"""Gracefully shut down all workers"""
self.active = False
# Stop all workers
self.transcriber.terminate()
self.agent.terminate()
self.synthesizer.terminate()
# Wait for queues to drain
await asyncio.sleep(0.5)
# Close connections
if self.websocket:
await self.websocket.close()
# Cancel tasks
for task in self.tasks:
if not task.done():
task.cancel()
The agent doesn't remember previous messages or context is lost.
Conversation history is not being maintained or updated correctly.
Maintain conversation history in the agent:
class Agent:
def __init__(self):
self.conversation_history = []
async def generate_response(self, user_input):
# Add user message to history
self.conversation_history.append({
"role": "user",
"content": user_input
})
# Generate response with full history
response = await self.llm.generate(self.conversation_history)
# Add bot response to history
self.conversation_history.append({
"role": "assistant",
"content": response
})
return response
When the bot is interrupted, update history with partial message:
def update_last_bot_message_on_cut_off(self, partial_message):
"""Update history when bot is interrupted"""
if self.conversation_history and \
self.conversation_history[-1]["role"] == "assistant":
# Update with what was actually spoken
self.conversation_history[-1]["content"] = partial_message
WebSocket connections drop unexpectedly, interrupting conversations.
Implement heartbeat and reconnection:
@app.websocket("/conversation")
async def conversation_endpoint(websocket: WebSocket):
await websocket.accept()
# Start heartbeat
async def heartbeat():
while True:
try:
await websocket.send_json({"type": "ping"})
await asyncio.sleep(30) # Ping every 30 seconds
except:
break
heartbeat_task = asyncio.create_task(heartbeat())
try:
async for message in websocket.iter_bytes():
# Process message
pass
finally:
heartbeat_task.cancel()
Long delays between user speech and bot response.
1. Not using streaming
# ❌ Bad: Wait for entire response
response = await llm.complete(prompt)
# ✅ Good: Stream response
async for chunk in llm.complete(prompt, stream=True):
yield chunk
2. Sequential processing
# ❌ Bad: Sequential
transcription = await transcriber.transcribe(audio)
response = await agent.generate(transcription)
audio = await synthesizer.synthesize(response)
# ✅ Good: Concurrent with queues
# All workers run simultaneously
3. Large chunk sizes
# ❌ Bad: Large chunks (high latency)
chunk_size = 8192 # 0.25 seconds
# ✅ Good: Small chunks (low latency)
chunk_size = 1024 # 0.032 seconds
Poor audio quality, distortion, or artifacts.
1. Wrong audio format
# ✅ Use LINEAR16 PCM at 16kHz
audio_encoding = AudioEncoding.LINEAR16
sample_rate = 16000
2. Incorrect format conversion
# ✅ Proper MP3 to PCM conversion
from pydub import AudioSegment
import io
def mp3_to_pcm(mp3_bytes):
audio = AudioSegment.from_mp3(io.BytesIO(mp3_bytes))
audio = audio.set_frame_rate(16000)
audio = audio.set_channels(1)
audio = audio.set_sample_width(2) # 16-bit
return audio.raw_data
3. Buffer underruns
# ✅ Ensure consistent chunk timing
await asyncio.sleep(max(seconds_per_chunk - processing_time, 0))
| Problem | Root Cause | Solution |
|---|---|---|
| Audio jumping | Multiple TTS calls | Buffer entire response |
| Echo/feedback | Transcriber active during bot speech | Mute transcriber |
| Interrupts not working | All chunks sent immediately | Rate-limit chunks |
| Memory leaks | Unclosed streams | Proper cleanup |
| Lost context | History not maintained | Update conversation history |
| Connection drops | No heartbeat | Implement ping/pong |
| High latency | Sequential processing | Use streaming + queues |
| Poor audio quality | Wrong format/conversion | Use LINEAR16 PCM 16kHz |
This guide compares different providers for transcription, LLM, and TTS services to help you choose the best option for your voice AI engine.
Strengths:
Weaknesses:
Best For:
Configuration:
{
"transcriberProvider": "deepgram",
"deepgramApiKey": "your-api-key",
"deepgramModel": "nova-2",
"language": "en-US"
}
Strengths:
Weaknesses:
Best For:
Configuration:
{
"transcriberProvider": "assemblyai",
"assemblyaiApiKey": "your-api-key",
"language": "en"
}
Strengths:
Weaknesses:
Best For:
Configuration:
{
"transcriberProvider": "azure",
"azureSpeechKey": "your-key",
"azureSpeechRegion": "eastus",
"language": "en-US"
}
Strengths:
Weaknesses:
Best For:
Configuration:
{
"transcriberProvider": "google",
"googleCredentials": "path/to/credentials.json",
"language": "en-US"
}
Strengths:
Weaknesses:
Best For:
Configuration:
{
"llmProvider": "openai",
"openaiApiKey": "your-api-key",
"openaiModel": "gpt-4-turbo",
"prompt": "You are a helpful AI assistant."
}
Pricing:
Strengths:
Weaknesses:
Best For:
Configuration:
{
"llmProvider": "gemini",
"geminiApiKey": "your-api-key",
"geminiModel": "gemini-pro",
"prompt": "You are a helpful AI assistant."
}
Pricing:
Strengths:
Weaknesses:
Best For:
Configuration:
{
"llmProvider": "claude",
"claudeApiKey": "your-api-key",
"claudeModel": "claude-3-opus",
"prompt": "You are a helpful AI assistant."
}
Pricing:
Strengths:
Weaknesses:
Best For:
Configuration:
{
"voiceProvider": "elevenlabs",
"elevenlabsApiKey": "your-api-key",
"elevenlabsVoiceId": "voice-id",
"elevenlabsModel": "eleven_monolingual_v1"
}
Pricing:
Strengths:
Weaknesses:
Best For:
Configuration:
{
"voiceProvider": "azure",
"azureSpeechKey": "your-key",
"azureSpeechRegion": "eastus",
"azureVoiceName": "en-US-JennyNeural"
}
Pricing:
Strengths:
Weaknesses:
Best For:
Configuration:
{
"voiceProvider": "google",
"googleCredentials": "path/to/credentials.json",
"googleVoiceName": "en-US-Neural2-F"
}
Pricing:
Strengths:
Weaknesses:
Best For:
Configuration:
{
"voiceProvider": "polly",
"awsAccessKey": "your-access-key",
"awsSecretKey": "your-secret-key",
"awsRegion": "us-east-1",
"pollyVoiceId": "Joanna"
}
Pricing:
Strengths:
Weaknesses:
Best For:
Configuration:
{
"voiceProvider": "playht",
"playhtApiKey": "your-api-key",
"playhtUserId": "your-user-id",
"playhtVoiceId": "voice-id"
}
Pricing:
{
"transcriberProvider": "deepgram", # Fast and affordable
"llmProvider": "gemini", # Free tier available
"voiceProvider": "google" # Cost-effective neural voices
}
Estimated cost: ~$0.01 per minute of conversation
{
"transcriberProvider": "assemblyai", # Highest accuracy
"llmProvider": "openai", # Best quality responses
"voiceProvider": "elevenlabs" # Most natural voices
}
Estimated cost: ~$0.05 per minute of conversation
{
"transcriberProvider": "azure", # Enterprise reliability
"llmProvider": "openai", # Best quality
"voiceProvider": "azure" # Enterprise reliability
}
Estimated cost: ~$0.03 per minute of conversation
{
"transcriberProvider": "google", # 125+ languages
"llmProvider": "gemini", # Good multi-language support
"voiceProvider": "google" # 40+ languages
}
Estimated cost: ~$0.02 per minute of conversation
| Priority | Transcriber | LLM | TTS |
|---|---|---|---|
| Lowest Cost | Deepgram | Gemini | |
| Highest Quality | AssemblyAI | OpenAI | ElevenLabs |
| Fastest Speed | Deepgram | OpenAI | ElevenLabs |
| Enterprise | Azure | OpenAI | Azure |
| Multi-Language | Gemini | ||
| Voice Cloning | N/A | N/A | ElevenLabs/Play.ht |
Before committing to providers, test with your specific use case:
The multi-provider factory pattern makes switching easy:
# Just change the configuration
config = {
"transcriberProvider": "deepgram", # Change to "assemblyai"
"llmProvider": "gemini", # Change to "openai"
"voiceProvider": "google" # Change to "elevenlabs"
}
# No code changes needed!
factory = VoiceComponentFactory()
transcriber = factory.create_transcriber(config)
agent = factory.create_agent(config)
synthesizer = factory.create_synthesizer(config)