coco-research/voice-ai
Voice AI architecture and implementation guide. Covers two architectures: speech-to-speech (OpenAI Realtime API, lowest latency) and pipeline (STT->LLM->TTS, more control). Includes provider-specific patterns for OpenAI Realtime, Vapi, Deepgram, ElevenLabs, and LiveKit. Use when building voice agents, voice-enabled apps, or real-time conversational AI.
npx skills add https://github.com/coco-research/coco --skill voice-aiYou 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.
| Architecture | Latency | Control | Best For |
|-------------|---------|---------|----------|
| Speech-to-Speech (S2S) | Lowest (~200-400ms) | Less controllable | Natural conversation, emotion preservation |
| Pipeline (STT->LLM->TTS) | Higher (~600-1200ms) | Full control at each step | Custom logic, debugging, provider mixing |
Direct audio-to-audio processing for lowest latency. Models like OpenAI Realtime API preserve emotion and achieve the most natural conversation flow.
Strengths:
Weaknesses:
Separate STT -> LLM -> TTS for maximum control at each step.
Strengths:
Weaknesses:
Detect when user starts/stops speaking. Critical for natural turn-taking.
Key metrics:
Native voice-to-voice with GPT-4o. Best for 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",
"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"] == "response.audio.delta":
# Play audio chunk
audio_bytes = base64.b64decode(event["delta"])
# send to speaker...
Build voice agents with Vapi platform. Best for phone-based agents and quick deployment.
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":
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":
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. Use when you want the highest quality 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:
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
})
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
# WebSocket for lowest latency TTS
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
final_audio = await tts.flush()
yield final_audio
Target: < 800ms total round-trip for natural conversation feel.
| Component | Target | Notes |
|-----------|--------|-------|
| STT | 100-200ms | Use interim results |
| LLM | 200-400ms | Stream tokens |
| TTS | 100-200ms | Stream audio chunks |
| Network | 50-100ms | Choose nearest region |
The single most important optimization: stream every component.
Allow users to interrupt the AI mid-response:
Why bad: Adds seconds of latency. User perceives as slow. Loses conversation flow.
Instead: Stream everything — STT interim results, LLM token streaming, TTS chunk streaming. Start TTS before LLM finishes.
Why bad: Frustrating user experience. Feels like talking to a machine.
Instead: Implement barge-in detection. Use VAD to detect user speech. Stop TTS immediately. Clear audio queue.
Why bad: Misses conversational cues. Cuts off users who pause to think.
Instead: Use semantic VAD that considers context, not just silence duration.
Why bad: Voice responses over 2-3 sentences feel like lectures.
Instead: Constrain response length in system prompts. Prompt for spoken format (concise, conversational).
Why bad: May not be best quality for each component. Single point of failure.
Instead: Mix best providers — Deepgram for STT (speed + accuracy), ElevenLabs for TTS (voice quality), OpenAI/Anthropic for LLM.
| Issue | Severity | Solution |
|-------|----------|----------|
| Latency exceeds budget | Critical | Measure and budget latency for each component |
| Jitter in response time | High | Target jitter metrics, use buffering |
| Poor turn detection | High | Use semantic VAD with context awareness |
| No barge-in support | High | Implement barge-in detection with VAD |
| Overly long responses | Medium | Constrain response length in prompts |
| Unnatural phrasing | Medium | Prompt for spoken format |
| Background noise issues | Medium | Implement noise handling / filtering |
| STT transcription errors | Medium | Mitigate with prompt hints and context |
Works well with: openai-api, openai-agents, openai-whisper, ai-product
Take coco-research/voice-ai from the repository into ~/.claude/skills for personal
use, or into .claude/skills inside a project.
The agent identifies a skill by the name field in its header. Two skills with the
same name cannot sit side by side — one of them will be ignored.