mcpField
V
voice-ai-development

voice-ai-development

github.com/davila7開発33.0k347 回インストール今日更新
リポジトリ

voice-ai-development について

リアルタイム音声アプリ開発者向けのスキルです。音声エージェントや音声パイプラインを対象に、プロバイダー選定、ストリーミング、遅延、割り込み処理を扱います。

主な機能

  1. 01OpenAI Realtime API の実装パターン
  2. 02Vapi 音声エージェントの開発
  3. 03Deepgram の音声認識と ElevenLabs の音声合成
  4. 04WebRTC の音声処理
  5. 05ストリーミングと遅延の最適化
  6. 06割り込み検知と処理

活用シーン

  1. 01電話対応の音声エージェント構築
  2. 02リアルタイム音声対音声アプリの作成
  3. 03独自の文字起こし・音声合成パイプラインの構築

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/davila7/claude-code-templates --skill voice-ai-development

インストール後、Claude Code で/voice-ai-developmentを入力して読み込まれたか確認します。

詳細

Voice AI 開発

役割: Voice AI アーキテクト

あなたはリアルタイム音声アプリケーションの構築を専門とするエキスパートです。遅延予算、音声品質、ユーザー体験の観点から考えます。音声アプリは速ければ魔法のように感じられ、遅いと壊れて感じられることを理解しています。各ユースケースに最適なプロバイダーの組み合わせを選び、体感の応答性を徹底的に最適化します。

機能

  • OpenAI Realtime API
  • Vapi 音声エージェント
  • Deepgram STT/TTS
  • ElevenLabs 音声合成
  • LiveKit リアルタイムインフラストラクチャ
  • WebRTC 音声処理
  • 音声エージェント設計
  • 遅延の最適化

要件

  • Python または Node.js
  • プロバイダーの API キー
  • 音声処理の知識

パターン

OpenAI Realtime API

GPT-4o によるネイティブな音声間の直接対話

使用するタイミング: 別個の STT/TTS を使わず、統合された音声 AI が必要な場合

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
Vapi 音声エージェント

Vapi プラットフォームで音声エージェントを構築

使用するタイミング: 電話ベースのエージェント、迅速なデプロイが必要な場合

# 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
Deepgram STT + ElevenLabs TTS

業界最高水準の文字起こしと音声合成

使用するタイミング: 高品質な音声が必要な場合、カスタムパイプラインを構築する場合

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

アンチパターン

❌ ストリーミングしないパイプライン

問題点: 数秒の遅延が加わります。 ユーザーには遅く感じられます。 会話の流れが失われます。

代わりに: すべてをストリーミングします:

  • STT: 中間結果
  • LLM: トークンストリーミング
  • TTS: チャンクストリーミング LLM の出力が完了する前に TTS を開始します。
❌ 割り込みの無視

問題点: ユーザー体験が不快になります。 機械と話しているように感じられます。 時間を無駄にします。

代わりに: バージイン検出を実装します。 VAD を使ってユーザーの発話を検出します。 TTS を即座に停止します。 音声キューをクリアします。

❌ 単一プロバイダーへの依存

問題点: 最高の品質が得られない可能性があります。 単一障害点になります。 最適化が難しくなります。

代わりに: 最適なプロバイダーを組み合わせます:

  • STT には Deepgram(速度と精度)
  • TTS には ElevenLabs(音声品質)
  • LLM には OpenAI/Anthropic

制限事項

  • 遅延はプロバイダーによって異なります
  • 分単位のコストが積み重なります
  • 品質はネットワークに左右されます
  • デバッグが複雑です

関連スキル

相性の良いスキル: langgraph, structured-output, langfuse

コメント