スキル一覧に戻る
pascalvanderheiden

voice-live

by pascalvanderheiden

A list of re-useable agent skills I created for my own purpose.

0🍴 0📅 2026年1月20日
GitHubで見るManusで実行

SKILL.md


name: voice-live description: Implement real-time voice agent capabilities using Azure Voice Live API. Use when building conversational AI voice agents, speech-to-speech applications, real-time audio processing, voice-enabled bots, or when users mention Voice Live, real-time voice, speech synthesis with avatars, or WebSocket-based voice interactions. Covers WebSocket setup, session configuration, audio streaming, noise suppression, echo cancellation, turn detection, function calling (VoiceRAG), avatar integration, and custom voice support.

Voice Live API Skill

Build real-time voice agents using Azure Voice Live API with WebSocket connections, Python backend, and advanced conversational features.

Overview

The Voice Live API enables low-latency, high-quality speech-to-speech interactions for voice agents. It integrates speech recognition, generative AI, and text-to-speech into a unified WebSocket interface.

Key Capabilities

  • Broad locale coverage: 140+ locales for STT, 600+ voices across 150+ locales for TTS
  • Customizable input/output: Phrase lists, custom speech models, custom voices
  • Flexible AI models: GPT-5, GPT-4.1, GPT-4o, Phi, and more
  • Advanced conversational features:
    • Noise suppression (azure_deep_noise_suppression)
    • Echo cancellation (server_echo_cancellation)
    • Robust interruption detection
    • Advanced end-of-turn detection
  • Avatar integration: Standard or customizable avatars synchronized with audio
  • Function calling: VoiceRAG pattern for external actions and grounded responses

Quick Start

Prerequisites

  1. Azure subscription
  2. Microsoft Foundry resource or Azure Speech Services resource
  3. Python 3.9+ with websockets and azure-identity packages

Installation

pip install websockets azure-identity pyaudio

Basic Connection

import asyncio
import websockets
import json
import base64
from azure.identity import DefaultAzureCredential

# Configuration
RESOURCE_NAME = "<your-ai-foundry-resource-name>"
MODEL = "gpt-4o"  # or gpt-realtime, gpt-5, phi4-mini, etc.
API_VERSION = "2025-10-01"

async def connect_voice_live():
    # Get auth token
    credential = DefaultAzureCredential()
    token = credential.get_token("https://ai.azure.com/.default").token
    
    # Build WebSocket URL
    ws_url = f"wss://{RESOURCE_NAME}.services.ai.azure.com/voice-live/realtime?api-version={API_VERSION}&model={MODEL}"
    
    headers = {"Authorization": f"Bearer {token}"}
    
    async with websockets.connect(ws_url, additional_headers=headers) as ws:
        # Wait for session.created
        response = await ws.recv()
        session = json.loads(response)
        print(f"Session created: {session['session']['id']}")
        
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": "You are a helpful AI assistant.",
                "voice": {
                    "name": "en-US-Ava:DragonHDLatestNeural",
                    "type": "azure-standard"
                },
                "turn_detection": {
                    "type": "azure_semantic_vad",
                    "threshold": 0.3,
                    "prefix_padding_ms": 200,
                    "silence_duration_ms": 200,
                    "end_of_utterance_detection": {
                        "model": "semantic_detection_v1",
                        "threshold_level": "default",
                        "timeout_ms": 1000
                    }
                },
                "input_audio_noise_reduction": {"type": "azure_deep_noise_suppression"},
                "input_audio_echo_cancellation": {"type": "server_echo_cancellation"}
            }
        }))
        
        # Handle messages
        async for message in ws:
            event = json.loads(message)
            await handle_event(event)

async def handle_event(event):
    event_type = event.get("type")
    
    if event_type == "session.updated":
        print("Session configured successfully")
    elif event_type == "response.audio.delta":
        # Decode and play audio
        audio_data = base64.b64decode(event["delta"])
        # Play audio_data through speakers
    elif event_type == "response.text.delta":
        print(event["delta"], end="", flush=True)
    elif event_type == "response.done":
        print("\n[Response complete]")
    elif event_type == "error":
        print(f"Error: {event['error']['message']}")

asyncio.run(connect_voice_live())

Session Configuration

Voice Options

# Azure Standard Voice
"voice": {
    "name": "en-US-AvaNeural",
    "type": "azure-standard"
}

# Azure HD Voice (with expressiveness)
"voice": {
    "name": "en-US-Ava:DragonHDLatestNeural",
    "type": "azure-standard",
    "temperature": 0.8,
    "rate": "1.0"
}

# Azure Custom Voice
"voice": {
    "name": "my-custom-voice",
    "type": "azure-custom",
    "endpoint_id": "12345678-1234-1234-1234-123456789012"
}

# OpenAI Voice (only with gpt-realtime models)
"voice": {
    "type": "openai",
    "name": "alloy"  # alloy, ash, ballad, coral, echo, sage, shimmer, verse
}

Turn Detection Options

# Azure Semantic VAD (recommended)
"turn_detection": {
    "type": "azure_semantic_vad",
    "threshold": 0.3,
    "prefix_padding_ms": 200,
    "silence_duration_ms": 500,
    "remove_filler_words": True,
    "end_of_utterance_detection": {
        "model": "semantic_detection_v1",
        "threshold_level": "default",
        "timeout_ms": 1000
    }
}

# Multilingual Semantic VAD
"turn_detection": {
    "type": "azure_semantic_vad_multilingual",
    "languages": ["en", "es", "fr", "de", "ja", "zh"]
}

# Server VAD (basic)
"turn_detection": {
    "type": "server_vad",
    "threshold": 0.5,
    "prefix_padding_ms": 300,
    "silence_duration_ms": 500
}

Audio Configuration

"input_audio_format": "pcm16",  # pcm16, g711_ulaw, g711_alaw
"output_audio_format": "pcm16",
"input_audio_sampling_rate": 24000,  # 16000 or 24000
"input_audio_noise_reduction": {"type": "azure_deep_noise_suppression"},
"input_audio_echo_cancellation": {"type": "server_echo_cancellation"}

Sending Audio

import pyaudio

async def stream_audio(ws):
    p = pyaudio.PyAudio()
    stream = p.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=24000,
        input=True,
        frames_per_buffer=4800  # 200ms chunks
    )
    
    try:
        while True:
            audio_data = stream.read(4800)
            audio_b64 = base64.b64encode(audio_data).decode()
            
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": audio_b64
            }))
            await asyncio.sleep(0.2)
    finally:
        stream.stop_stream()
        stream.close()
        p.terminate()

Function Calling (VoiceRAG Pattern)

Enable external actions and grounded responses:

# Configure tools in session
"tools": [
    {
        "type": "function",
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    },
    {
        "type": "function",
        "name": "search_knowledge_base",
        "description": "Search internal knowledge base",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string"}
            },
            "required": ["query"]
        }
    }
]

Handle function calls:

async def handle_event(event, ws):
    if event["type"] == "response.function_call_arguments.done":
        call_id = event["call_id"]
        name = event["name"] if "name" in event else None
        arguments = json.loads(event["arguments"])
        
        # Execute the function
        result = await execute_function(name, arguments)
        
        # Send function result back
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": json.dumps(result)
            }
        }))
        
        # Trigger response generation
        await ws.send(json.dumps({"type": "response.create"}))

async def execute_function(name, args):
    if name == "get_weather":
        # Call weather API
        return {"temperature": 22, "condition": "sunny"}
    elif name == "search_knowledge_base":
        # Search your RAG system
        return {"results": ["Document 1", "Document 2"]}

Avatar Integration

"modalities": ["text", "audio", "avatar"],
"avatar": {
    "character": "lisa",  # or custom avatar ID
    "style": "casual-sitting",
    "customized": False,
    "video": {
        "resolution": {"width": 1920, "height": 1080},
        "bitrate": 2000000,
        "codec": "h264"
    }
}

After session.updated, establish WebRTC connection:

# Send client SDP
await ws.send(json.dumps({
    "type": "session.avatar.connect",
    "client_sdp": "<your-client-sdp>"
}))

# Receive server SDP in session.avatar.connecting event

Custom Speech Input

Phrase List (Just-in-time customization)

"input_audio_transcription": {
    "model": "azure-speech",
    "phrase_list": ["Neo QLED TV", "TUF Gaming", "AutoQuote Explorer"]
}

Custom Speech Model

"input_audio_transcription": {
    "model": "azure-speech",
    "language": "en",
    "custom_speech": {
        "zh-CN": "847cb03d-7f22-4b11-444-e1be1d77bf17"
    }
}

Supported Models

ModelDescription
gpt-realtimeGPT real-time with Azure TTS option
gpt-realtime-miniGPT mini real-time with Azure TTS
gpt-4oGPT-4o with Azure STT/TTS
gpt-4o-miniGPT-4o mini with Azure STT/TTS
gpt-4.1GPT-4.1 with Azure STT/TTS
gpt-5GPT-5 with Azure STT/TTS
phi4-miniPhi4 with Azure STT/TTS

WebSocket Event Reference

See references/events.md for complete event documentation.

Key Client Events

  • session.update - Configure session
  • input_audio_buffer.append - Send audio
  • input_audio_buffer.commit - Commit audio buffer
  • conversation.item.create - Add items (messages, function outputs)
  • response.create - Request response
  • response.cancel - Cancel response

Key Server Events

  • session.created - Session established
  • session.updated - Configuration confirmed
  • response.audio.delta - Streaming audio
  • response.text.delta - Streaming text
  • response.function_call_arguments.done - Function call ready
  • input_audio_buffer.speech_started - Speech detected
  • input_audio_buffer.speech_stopped - Speech ended

Best Practices

  1. Audio quality: Use 24kHz sample rate for best quality
  2. Noise suppression: Always enable for better accuracy
  3. Echo cancellation: Enable when playing audio through speakers
  4. Turn detection: Use azure_semantic_vad for natural conversations
  5. Buffer size: Send audio in 200ms chunks for low latency
  6. Error handling: Monitor error events and implement reconnection
  7. Function calling: Keep functions focused and well-documented

Complete Example

See scripts/voice_live_client.py for a full-featured implementation.

スコア

総合スコア

60/100

リポジトリの品質指標に基づく評価

SKILL.md

SKILL.mdファイルが含まれている

+20
LICENSE

ライセンスが設定されている

+10
説明文

100文字以上の説明がある

0/10
人気

GitHub Stars 100以上

0/15
最近の活動

3ヶ月以内に更新がある

0/10
フォーク

10回以上フォークされている

0/5
Issue管理

オープンIssueが50未満

+5
言語

プログラミング言語が設定されている

+5
タグ

1つ以上のタグが設定されている

0/5

レビュー

💬

レビュー機能は近日公開予定です