Azure AI Voice Live SDK

Créez des applications vocales IA en temps réel avec une communication WebSocket bidirectionnelle.

Installation

pip install azure-ai-voicelive aiohttp azure-identity

Variables d'environnement

AZURE_COGNITIVE_SERVICES_ENDPOINT=https://<region>.api.cognitive.microsoft.com  # Requis pour toutes les méthodes d'auth
# Pour l'authentification par clé API (non recommandée en production)
AZURE_COGNITIVE_SERVICES_KEY=<api-key>  # Requis uniquement pour l'authentification AzureKeyCredential
AZURE_TOKEN_CREDENTIALS=prod # Requis uniquement si DefaultAzureCredential est utilisé en production

Authentification

DefaultAzureCredential (préféré) :

import os
from azure.ai.voicelive.aio import connect
from azure.identity.aio import DefaultAzureCredential, ManagedIdentityCredential

# Dev local : DefaultAzureCredential. Production : définir AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Ou utiliser des credentials spécifiques directement en production :
# Voir https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

async with connect(
    endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
    credential=credential,
    model="gpt-4o-realtime-preview",
    credential_scopes=["https://cognitiveservices.azure.com/.default"]
) as conn:
    ...

Clé API :

from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential

async with connect(
    endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
    credential=AzureKeyCredential(os.environ["AZURE_COGNITIVE_SERVICES_KEY"]),
    model="gpt-4o-realtime-preview"
) as conn:
    ...

Démarrage rapide

import asyncio
import os
from azure.ai.voicelive.aio import connect
from azure.identity.aio import DefaultAzureCredential

async def main():
    async with connect(
        endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
        credential=DefaultAzureCredential(),
        model="gpt-4o-realtime-preview",
        credential_scopes=["https://cognitiveservices.azure.com/.default"]
    ) as conn:
        # Mettre à jour la session avec des instructions
        await conn.session.update(session={
            "instructions": "You are a helpful assistant.",
            "modalities": ["text", "audio"],
            "voice": "alloy"
        })

        # Écouter les événements
        async for event in conn:
            print(f"Event: {event.type}")
            if event.type == "response.audio_transcript.done":
                print(f"Transcript: {event.transcript}")
            elif event.type == "response.done":
                break

asyncio.run(main())

Architecture fondamentale

Ressources de connexion

VoiceLiveConnection expose ces ressources :

Configuration de session

from azure.ai.voicelive.models import RequestSession, FunctionTool

await conn.session.update(session=RequestSession(
    instructions="You are a helpful voice assistant.",
    modalities=["text", "audio"],
    voice="alloy",  # ou "echo", "shimmer", "sage", etc.
    input_audio_format="pcm16",
    output_audio_format="pcm16",
    turn_detection={
        "type": "server_vad",
        "threshold": 0.5,
        "prefix_padding_ms": 300,
        "silence_duration_ms": 500
    },
    tools=[
        FunctionTool(
            type="function",
            name="get_weather",
            description="Get current weather",
            parameters={
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        )
    ]
))

Streaming audio

Envoyer de l'audio (Base64 PCM16)

import base64

# Lire un chunk audio (PCM 16-bit, mono 24kHz)
audio_chunk = await read_audio_from_microphone()
b64_audio = base64.b64encode(audio_chunk).decode()

await conn.input_audio_buffer.append(audio=b64_audio)

Recevoir de l'audio

async for event in conn:
    if event.type == "response.audio.delta":
        audio_bytes = base64.b64decode(event.delta)
        await play_audio(audio_bytes)
    elif event.type == "response.audio.done":
        print("Audio complete")

Gestion des événements

async for event in conn:
    match event.type:
        # Événements de session
        case "session.created":
            print(f"Session: {event.session}")
        case "session.updated":
            print("Session updated")

        # Événements d'entrée audio
        case "input_audio_buffer.speech_started":
            print(f"Speech started at {event.audio_start_ms}ms")
        case "input_audio_buffer.speech_stopped":
            print(f"Speech stopped at {event.audio_end_ms}ms")

        # Événements de transcription
        case "conversation.item.input_audio_transcription.completed":
            print(f"User said: {event.transcript}")
        case "conversation.item.input_audio_transcription.delta":
            print(f"Partial: {event.delta}")

        # Événements de réponse
        case "response.created":
            print(f"Response started: {event.response.id}")
        case "response.audio_transcript.delta":
            print(event.delta, end="", flush=True)
        case "response.audio.delta":
            audio = base64.b64decode(event.delta)
        case "response.done":
            print(f"Response complete: {event.response.status}")

        # Appels de fonction
        case "response.function_call_arguments.done":
            result = handle_function(event.name, event.arguments)
            await conn.conversation.item.create(item={
                "type": "function_call_output",
                "call_id": event.call_id,
                "output": json.dumps(result)
            })
            await conn.response.create()

        # Erreurs
        case "error":
            print(f"Error: {event.error.message}")

Patterns courants

Mode tour manuel (sans VAD)

await conn.session.update(session={"turn_detection": None})

# Contrôler les tours manuellement
await conn.input_audio_buffer.append(audio=b64_audio)
await conn.input_audio_buffer.commit()  # Fin du tour utilisateur
await conn.response.create()  # Déclencher une réponse

Gestion des interruptions

async for event in conn:
    if event.type == "input_audio_buffer.speech_started":
        # L'utilisateur a interrompu - annuler la réponse actuelle
        await conn.response.cancel()
        await conn.output_audio_buffer.clear()

Historique de conversation

# Ajouter un message système
await conn.conversation.item.create(item={
    "type": "message",
    "role": "system",
    "content": [{"type": "input_text", "text": "Be concise."}]
})

# Ajouter un message utilisateur
await conn.conversation.item.create(item={
    "type": "message",
    "role": "user", 
    "content": [{"type": "input_text", "text": "Hello!"}]
})

await conn.response.create()

Options de voix

Voix Azure : utilisez les modèles AzureStandardVoice, AzureCustomVoice, ou AzurePersonalVoice.

Formats audio

Options de détection de tour

# VAD serveur (défaut)
{"type": "server_vad", "threshold": 0.5, "silence_duration_ms": 500}

# Azure Semantic VAD (détection plus intelligente)
{"type": "azure_semantic_vad"}
{"type": "azure_semantic_vad_en"}  # Optimisé pour l'anglais
{"type": "azure_semantic_vad_multilingual"}

Gestion des erreurs

from azure.ai.voicelive.aio import ConnectionError, ConnectionClosed

try:
    async with connect(...) as conn:
        async for event in conn:
            if event.type == "error":
                print(f"API Error: {event.error.code} - {event.error.message}")
except ConnectionClosed as e:
    print(f"Connection closed: {e.code} - {e.reason}")
except ConnectionError as e:
    print(f"Connection error: {e}")

Références

• Référence API détaillée : voir references/api-reference.md
• Exemples complets : voir references/examples.md
• Tous les modèles et types : voir references/models.md