voice-ai

500ms 未満のレイテンシで本番環境向けの音声 AI エージェントを構築します:

1. STT - Deepgram Nova-3 ストリーミング文字起こし (~150ms)
2. LLM - 最速の推論を実現する Groq llama-3.1-8b-instant (~220ms)
3. TTS - 超リアルな音声を実現する Cartesia Sonic (~90ms)
4. テレフォニー - リアルタイム双方向音声のための Twilio Media Streams

重要: OPENAI は使用しない - from openai import OpenAI を絶対に使わないでください

主な成果物:

• 音声活動検出を備えたストリーミング STT
• 音声用に最適化された低レイテンシの LLM 応答
• 感情コントロールを備えた表現力豊かな TTS
• Twilio Media Streams の WebSocket ハンドラー

<quick_start> 最小構成の音声パイプライン (~50 行、500ms 未満):

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> 音声 AI エージェントは次の条件を満たすときに成功と言えます:

• 合計レイテンシが 500ms 未満 (STT + LLM + TTS)
• STT が発話終了検出を伴って正確に文字起こしする
• TTS が自然で会話的に聞こえる
• 割り込み (バージイン) がスムーズに機能する (Enterprise ティア)
• バイリンガルサポートが言語切り替えに対応する </success_criteria>

<optimal_stack>

VozLux 検証済みスタック

コンポーネント	プロバイダー	モデル	レイテンシ	備考
STT	Deepgram	Nova-3	~150ms	ストリーミング、VAD、発話検出
LLM	Groq	llama-3.1-8b-instant	~220ms	LPU ハードウェア、最速の推論
TTS	Cartesia	Sonic-2	~90ms	ストリーミング、感情、バイリンガル
合計	-	-	~460ms	500ms 未満の目標を達成

LLM の優先順位 (OpenAI は決して使わない)

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

ティアアーキテクチャ

ティア	レイテンシ	STT	LLM	TTS	機能
Free	3000ms	TwiML Gather	Groq	Polly	基本的な IVR
Pro	600ms	Deepgram Nova	Groq	Cartesia	Media Streams
Enterprise	400ms	Deepgram + VAD	Groq	Cartesia	バージイン
</optimal_stack>

<deepgram_stt>

Deepgram STT (v5 SDK)

ストリーミング WebSocket パターン

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"))

接続オプション

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
}

完全なストリーミング設定については reference/deepgram-setup.md を参照してください。 </deepgram_stt>

<groq_llm>

Groq LLM (最速の推論)

音声に最適化されたパターン

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

モデルの選択

モデル	速度	品質	用途
llama-3.1-8b-instant	~220ms	良好	主要な音声
llama-3.3-70b-versatile	~500ms	最高	複雑なクエリ
mixtral-8x7b-32768	~300ms	良好	長いコンテキスト

コンテキスト管理については reference/groq-voice-llm.md を参照してください。 </groq_llm>

<cartesia_tts>

Cartesia TTS (Sonic-2)

ストリーミングパターン

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": {
                    "speed": "normal",
                    "emotion": [emotion] if emotion != "neutral" else []
                }
            },
            language=language,
            output_format={
                "container": "raw",
                "encoding": "pcm_s16le",
                "sample_rate": 8000,  # Telephony
            },
        )

        for chunk in response:
            if chunk.audio:
                yield chunk.audio

タイムスタンプ付き

response = client.tts.sse(
    model_id="sonic-2",
    transcript="Hello, world!",
    voice={"id": voice_id},
    output_format={"container": "raw", "encoding": "pcm_f32le", "sample_rate": 44100},
    add_timestamps=True,
)

for chunk in response:
    if chunk.word_timestamps:
        for word, start, end in zip(
            chunk.word_timestamps.words,
            chunk.word_timestamps.start,
            chunk.word_timestamps.end
        ):
            print(f"'{word}': {start:.2f}s - {end:.2f}s")

全 57 種類の感情については reference/cartesia-tts.md を参照してください。 </cartesia_tts>

<twilio_media_streams>

Twilio Media Streams

WebSocket ハンドラー (FastAPI)

from fastapi import FastAPI, WebSocket, Request
from fastapi.responses import Response
from twilio.twiml.voice_response import VoiceResponse, Connect, Stream
import json, base64, audioop

app = FastAPI()

@app.post("/voice/incoming")
async def incoming_call(request: Request):
    """Route to Media Streams WebSocket."""
    form = await request.form()
    caller = form.get("From", "")
    lang = "es" if caller.startswith("+52") else "en"

    response = VoiceResponse()
    connect = Connect()
    connect.append(Stream(url=f"wss://your-app.com/voice/stream?lang={lang}"))
    response.append(connect)

    return Response(content=str(response), media_type="application/xml")

@app.websocket("/voice/stream")
async def media_stream(websocket: WebSocket, lang: str = "en"):
    await websocket.accept()
    stream_sid = None

    while True:
        message = await websocket.receive_text()
        data = json.loads(message)

        if data["event"] == "start":
            stream_sid = data["start"]["streamSid"]
            # Initialize STT, send greeting

        elif data["event"] == "media":
            audio = base64.b64decode(data["media"]["payload"])
            # Send to Deepgram STT

        elif data["event"] == "stop":
            break

async def send_audio(websocket, stream_sid: str, pcm_audio: bytes):
    """Convert PCM to mu-law and send to Twilio."""
    mulaw = audioop.lin2ulaw(pcm_audio, 2)
    await websocket.send_text(json.dumps({
        "event": "media",
        "streamSid": stream_sid,
        "media": {"payload": base64.b64encode(mulaw).decode()}
    }))

完全なハンドラーについては reference/twilio-webhooks.md を参照してください。 </twilio_media_streams>

<bilingual_support>

バイリンガルサポート (EN/ES)

自動検出

def detect_language(caller_number: str) -> str:
    if caller_number.startswith("+52"):
        return "es"  # Mexico
    elif caller_number.startswith("+1"):
        return "en"  # US/Canada
    return "es"  # Default Spanish

音声プロンプト

GREETINGS = {
    "en": "Hello! How can I help you today?",
    "es": "Hola! En que puedo ayudarle hoy?",  # Use "usted" for respect
}

</bilingual_support>

<voice_prompts>

音声プロンプトエンジニアリング

VOICE_PROMPT = """
# Role
You are a bilingual voice assistant for {business_name}.

# Tone
- 2-3 sentences max for phone clarity
- NEVER use bullet points, lists, or markdown
- Spell out emails: "john at company dot com"
- Phone numbers with pauses: "five one two... eight seven seven..."
- Spanish: Use "usted" for formal respect

# Guardrails
- Never make up information
- Transfer to human after 3 failed attempts
- Match caller's language

# Error Recovery
English: "I want to make sure I got that right. Did you say [repeat]?"
Spanish: "Quiero asegurarme de entender bien. Dijo [repetir]?"
"""

完全なテンプレートについては reference/voice-prompts.md を参照してください。 </voice_prompts>

<file_locations>

参照ファイル

• reference/deepgram-setup.md - 完全なストリーミング STT 設定
• reference/groq-voice-llm.md - 音声向けの Groq パターン
• reference/cartesia-tts.md - 全 57 種類の感情、音声クローニング
• reference/twilio-webhooks.md - 完全な Media Streams ハンドラー
• reference/latency-optimization.md - 500ms 未満を実現するテクニック
• reference/voice-prompts.md - 音声に最適化されたプロンプト </file_locations>

## リクエストルーティング

ユーザーが音声エージェントを希望: → フルスタックを提供 (Deepgram + Groq + Cartesia + Twilio) → quick_start パイプラインから開始

ユーザーが STT のみを希望: → Deepgram ストリーミングパターンを提供 → 参照: reference/deepgram-setup.md

ユーザーが TTS のみを希望: → 感情を伴う Cartesia パターンを提供 → 参照: reference/cartesia-tts.md

ユーザーがレイテンシ最適化を希望: → 現在のスタックを監査し、ボトルネックを特定 → 参照: reference/latency-optimization.md

ユーザーが OpenAI に言及: → 直ちに Groq へリダイレクト → 説明: 「NO OPENAI - 最低レイテンシには Groq を使用」

<env_setup>

環境変数

# Required (NEVER OpenAI)
DEEPGRAM_API_KEY=your_key
GROQ_API_KEY=gsk_xxxx
CARTESIA_API_KEY=your_key

# Twilio
TWILIO_ACCOUNT_SID=ACxxxx
TWILIO_AUTH_TOKEN=your_token
TWILIO_PHONE_NUMBER=+15551234567

# Fallbacks
ANTHROPIC_API_KEY=sk-ant-xxxx
CEREBRAS_API_KEY=csk_xxxx

pip install deepgram-sdk groq cartesia twilio fastapi

</env_setup>

<quick_reference>

クイックリファレンスカード

STACK:
  STT: Deepgram Nova-3 (~150ms)
  LLM: Groq llama-3.1-8b-instant (~220ms) - NOT OPENAI
  TTS: Cartesia Sonic-2 (~90ms)

LATENCY TARGETS:
  Pro: 600ms (Media Streams)
  Enterprise: 400ms (Full streaming + barge-in)

BILINGUAL:
  +52 -> Spanish (es)
  +1 -> English (en)
  Default -> Spanish

EMOTIONS (Cartesia):
  greeting -> excited
  confirmation -> grateful
  complaint -> sympathetic

アウトカムサイドカーを出力

~/.claude/skill-analytics/last-outcome-voice-ai.json に書き込みます: {"ts":"[UTC ISO8601]","skill":"voice-ai","version":"1.0.0","variant":"default","status":"[success|partial|error]","runtime_ms":[ms],"metrics":{"agents_built":[n],"endpoints_configured":[n],"latency_targets_met":[n]},"error":null,"session_id":"[YYYY-MM-DD]"} </quick_reference>