voice-ai-development について
リアルタイム音声アプリ開発者向けのスキルです。音声エージェントや音声パイプラインを対象に、プロバイダー選定、ストリーミング、遅延、割り込み処理を扱います。
主な機能
- 01OpenAI Realtime API の実装パターン
- 02Vapi 音声エージェントの開発
- 03Deepgram の音声認識と ElevenLabs の音声合成
- 04WebRTC の音声処理
- 05ストリーミングと遅延の最適化
- 06割り込み検知と処理
活用シーン
- 01電話対応の音声エージェント構築
- 02リアルタイム音声対音声アプリの作成
- 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