Explorer la bibliothèque de client AI Voice Live pour Python

La bibliothèque de client Azure AI Voice Live pour Python fournit un client de reconnaissance vocale en temps réel pour l’API Azure AI Voice Live. Il ouvre une session WebSocket pour diffuser en continu l’audio du microphone vers le service et reçoit des événements du serveur pour les conversations dynamiques.

Dans cette unité, vous allez apprendre à utiliser le Kit de développement logiciel (SDK) pour implémenter l’authentification et gérer les événements. Vous voyez également un exemple minimal de création d’une session. Pour obtenir une référence complète au package Voice Live, consultez la référence du package vocal live.

Implémenter l’authentification

Vous pouvez implémenter l’authentification avec une clé API ou un jeton d’ID Microsoft Entra. L’exemple de code suivant montre une implémentation de clé API. Il part du principe que les variables d’environnement sont définies dans un .env fichier ou directement dans votre environnement.

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

async def main():
    async with connect(
        endpoint="your-endpoint",
        credential=AzureKeyCredential("your-api-key"),
        model="gpt-4o"
    ) as connection:
        # Your async code here
        pass

asyncio.run(main())

Pour les applications de production, l’authentification Microsoft Entra est recommandée. L’exemple de code suivant montre l’implémentation de l’authentification DefaultAzureCredential :

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

async def main():
    credential = DefaultAzureCredential()
    
    async with connect(
        endpoint="your-endpoint",
        credential=credential,
        model="gpt-4o"
    ) as connection:
        # Your async code here
        pass

asyncio.run(main())

Gestion des événements

La gestion appropriée des événements garantit une interaction plus transparente entre le client et l’agent. Par exemple, lors de la gestion d’un utilisateur interrompant l’agent vocal, vous devez annuler immédiatement la lecture audio de l’agent dans le client. Si ce n’est pas le cas, le client continue à lire la dernière réponse de l’assistant jusqu’à ce que l’interruption soit traitée dans l’API, ce qui entraîne l’agent à parler en même temps que l’utilisateur.

L’exemple de code suivant montre une gestion des événements de base :

async for event in connection:
    if event.type == ServerEventType.SESSION_UPDATED:
        print(f"Session ready: {event.session.id}")
        # Start audio capture
        
    elif event.type == ServerEventType.INPUT_AUDIO_BUFFER_SPEECH_STARTED:
        print("User started speaking")
        # Stop playback and cancel any current response
        
    elif event.type == ServerEventType.RESPONSE_AUDIO_DELTA:
        # Play the audio chunk
        audio_bytes = event.delta
        
    elif event.type == ServerEventType.ERROR:
        print(f"Error: {event.error.message}")

Exemple minimal

L’exemple de code suivant montre l’authentification auprès de l’API et la configuration de la session.

import asyncio
from azure.core.credentials import AzureKeyCredential
from azure.ai.voicelive.aio import connect
from azure.ai.voicelive.models import (
    RequestSession, Modality, InputAudioFormat, OutputAudioFormat, ServerVad, ServerEventType
)

API_KEY = "your-api-key"
ENDPOINT = "your-endpoint"
MODEL = "gpt-4o"

async def main():
    async with connect(
        endpoint=ENDPOINT,
        credential=AzureKeyCredential(API_KEY),
        model=MODEL,
    ) as conn:
        session = RequestSession(
            modalities=[Modality.TEXT, Modality.AUDIO],
            instructions="You are a helpful assistant.",
            input_audio_format=InputAudioFormat.PCM16,
            output_audio_format=OutputAudioFormat.PCM16,
            turn_detection=ServerVad(
                threshold=0.5, 
                prefix_padding_ms=300, 
                silence_duration_ms=500
            ),
        )
        await conn.session.update(session=session)

        # Process events
        async for evt in conn:
            print(f"Event: {evt.type}")
            if evt.type == ServerEventType.RESPONSE_DONE:
                break

asyncio.run(main())