voice-ai

What this skill does

<objective>
Build production voice AI agents with sub-500ms latency:

1. **STT** - Deepgram Nova-3 streaming transcription (~150ms)
2. **LLM** - Groq llama-3.1-8b-instant for fastest inference (~220ms)
3. **TTS** - Cartesia Sonic for ultra-realistic voice (~90ms)
4. **Telephony** - Twilio Media Streams for real-time bidirectional audio

**CRITICAL: NO OPENAI - Never use `from openai import OpenAI`**

Key deliverables:
- Streaming STT with voice activity detection
- Low-latency LLM responses optimized for voice
- Expressive TTS with emotion controls
- Twilio Media Streams WebSocket handler
</objective>

<quick_start>
**Minimal Voice Pipeline (~50 lines, <500ms):**
```python
import os
import asyncio
from groq import AsyncGroq
from deepgram import AsyncDeepgramClient
from cartesia import AsyncCartesia

# NEVER: from openai import OpenAI

async def voice_pipeline(user_audio: bytes) -> bytes:
    """Process audio input, return audio response."""

    # 1. STT: Deepgram Nova-3 (~150ms)
    dg = AsyncDeepgramClient(api_key=os.getenv("DEEPGRAM_API_KEY"))
    result = await dg.listen.rest.v1.transcribe(
        {"buffer": user_audio, "mimetype": "audio/wav"},
        {"model": "nova-3", "language": "en-US"}
    )
    user_text = result.results.channels[0].alternatives[0].transcript

    # 2. LLM: Groq (~220ms) - NOT OpenAI
    groq = AsyncGroq(api_key=os.getenv("GROQ_API_KEY"))
    response = await groq.chat.completions.create(
        model="llama-3.1-8b-instant",
        messages=[
            {"role": "system", "content": "Keep responses under 2 sentences."},
            {"role": "user", "content": user_text}
        ],
        max_tokens=150
    )
    response_text = response.choices[0].message.content

    # 3. TTS: Cartesia Sonic-2 (~90ms)
    cartesia = AsyncCartesia(api_key=os.getenv("CARTESIA_API_KEY"))
    audio_chunks = []
    for chunk in cartesia.tts.sse(
        model_id="sonic-2",
        transcript=response_text,
        voice={"id": "f9836c6e-a0bd-460e-9d3c-f7299fa60f94"},
        output_format={"container": "raw", "encoding": "pcm_s16le", "sample_rate": 8000}
    ):
        if chunk.audio:
            audio_chunks.append(chunk.audio)

    return b"".join(audio_chunks)  # Total: ~460ms
```
</quick_start>

<success_criteria>
A voice AI agent is successful when:
- Total latency is under 500ms (STT + LLM + TTS)
- STT correctly transcribes with utterance end detection
- TTS sounds natural and conversational
- Barge-in (interruption) works smoothly (Enterprise tier)
- Bilingual support handles language switching
</success_criteria>

<optimal_stack>
## VozLux-Tested Stack

| Component | Provider | Model | Latency | Notes |
|-----------|----------|-------|---------|-------|
| **STT** | Deepgram | Nova-3 | ~150ms | Streaming, VAD, utterance detection |
| **LLM** | Groq | llama-3.1-8b-instant | ~220ms | LPU hardware, fastest inference |
| **TTS** | Cartesia | Sonic-2 | ~90ms | Streaming, emotions, bilingual |
| **TOTAL** | - | - | **~460ms** | Sub-500ms target achieved |

### LLM Priority (Never OpenAI)
```python
LLM_PRIORITY = [
    ("groq", "GROQ_API_KEY", "~220ms"),      # Primary
    ("cerebras", "CEREBRAS_API_KEY", "~200ms"),  # Fallback
    ("anthropic", "ANTHROPIC_API_KEY", "~500ms"),  # Quality fallback
]
# NEVER: from openai import OpenAI
```

### Tier Architecture
| Tier | Latency | STT | LLM | TTS | Features |
|------|---------|-----|-----|-----|----------|
| Free | 3000ms | TwiML Gather | Groq | Polly | Basic IVR |
| Pro | 600ms | Deepgram Nova | Groq | Cartesia | Media Streams |
| Enterprise | 400ms | Deepgram + VAD | Groq | Cartesia | Barge-in |
</optimal_stack>

<deepgram_stt>
## Deepgram STT (v5 SDK)

### Streaming WebSocket Pattern
```python
from deepgram import AsyncDeepgramClient
from deepgram.core.events import EventType
from deepgram.extensions.types.sockets import (
    ListenV1SocketClientResponse,
    ListenV1MediaMessage,
    ListenV1ControlMessage
)

async def streaming_stt():
    client = AsyncDeepgramClient(api_key=os.getenv("DEEPGRAM_API_KEY"))

    async with client.listen.v1.connect(model="nova-3") as connection:
        def on_message(message: ListenV1SocketClientResponse):
            msg_type = getattr(message, "type", None)

            if msg_type == "Results":
                channel = getattr(message, "channel", None)
                if channel and channel.alternatives:
                    text = channel.alternatives[0].transcript
                    is_final = getattr(message, "is_final", False)
                    if text:
                        print(f"{'[FINAL]' if is_final else '[INTERIM]'} {text}")

            elif msg_type == "UtteranceEnd":
                print("[USER FINISHED SPEAKING]")

            elif msg_type == "SpeechStarted":
                print("[USER STARTED SPEAKING - barge-in trigger]")

        connection.on(EventType.MESSAGE, on_message)
        await connection.start_listening()

        # Send audio chunks
        await connection.send_media(ListenV1MediaMessage(data=audio_bytes))

        # Keep alive for long sessions
        await connection.send_control(ListenV1ControlMessage(type="KeepAlive"))
```

### Connection Options
```python
options = {
    "model": "nova-3",
    "language": "en-US",
    "encoding": "mulaw",      # Twilio format
    "sample_rate": 8000,      # Telephony standard
    "interim_results": True,   # Get partial transcripts
    "utterance_end_ms": 1000,  # Silence to end utterance
    "vad_events": True,        # Voice activity detection
}
```

> See `reference/deepgram-setup.md` for full streaming setup.
</deepgram_stt>

<groq_llm>
## Groq LLM (Fastest Inference)

### Voice-Optimized Pattern
```python
from groq import AsyncGroq

class GroqVoiceLLM:
    def __init__(self, model: str = "llama-3.1-8b-instant"):
        self.client = AsyncGroq()
        self.model = model
        self.system_prompt = (
            "You are a helpful voice assistant. "
            "Keep responses to 2-3 sentences max. "
            "Speak naturally as if on a phone call."
        )

    async def generate_stream(self, user_input: str):
        """Streaming for lowest TTFB."""
        stream = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": user_input}
            ],
            max_tokens=150,
            temperature=0.7,
            stream=True,
        )

        async for chunk in stream:
            content = chunk.choices[0].delta.content
            if content:
                yield content  # Pipe to TTS immediately
```

### Model Selection
| Model | Speed | Quality | Use Case |
|-------|-------|---------|----------|
| llama-3.1-8b-instant | ~220ms | Good | Primary voice |
| llama-3.3-70b-versatile | ~500ms | Best | Complex queries |
| mixtral-8x7b-32768 | ~300ms | Good | Long context |

> See `reference/groq-voice-llm.md` for context management.
</groq_llm>

<cartesia_tts>
## Cartesia TTS (Sonic-2)

### Streaming Pattern
```python
from cartesia import AsyncCartesia

class CartesiaTTS:
    VOICES = {
        "en": "f9836c6e-a0bd-460e-9d3c-f7299fa60f94",  # Warm female
        "es": "5c5ad5e7-1020-476b-8b91-fdcbe9cc313c",  # Mexican Spanish
    }

    EMOTIONS = {
        "greeting": "excited",
        "confirmation": "grateful",
        "info": "calm",
        "complaint": "sympathetic",
        "apology": "apologetic",
    }

    def __init__(self, api_key: str):
        self.client = AsyncCartesia(api_key=api_key)

    async def synthesize_stream(
        self,
        text: str,
        language: str = "en",
        emotion: str = "neutral"
    ):
        voice_id = self.VOICES.get(language, self.VOICES["en"])

        response = self.client.tts.sse(
            model_id="sonic-2",
            transcript=text,
            voice={
                "id": voice_id,
                "experimental_controls": {