Exploración de la biblioteca cliente de AI Voice Live para Python

La biblioteca cliente de Azure AI Voice Live para Python proporciona un cliente de voz a voz en tiempo real para Azure AI Voice Live API. Abre una sesión de WebSocket para transmitir audio del micrófono al servicio y recibir eventos del servidor para conversaciones interactivas.

En esta unidad, aprenderá a usar el SDK para implementar la autenticación y controlar eventos. También verá un ejemplo mínimo de creación de una sesión. Para obtener una referencia completa al paquete de Voice Live, visite la Referencia del Paquete Voice Live.

Implementación de la autenticación

Puede implementar la autenticación con una clave de API o un token de identificador de Microsoft Entra. En el ejemplo de código siguiente se muestra una implementación de clave de API. Se supone que las variables de entorno se establecen en un .env archivo o directamente en el entorno.

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())

En el caso de las aplicaciones de producción, se recomienda la autenticación de Microsoft Entra. En el ejemplo de código siguiente se muestra la implementación del DefaultAzureCredential para la autenticación.

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())

Control de eventos

El control adecuado de los eventos garantiza una interacción más fluida entre el cliente y el agente. Por ejemplo, al controlar un usuario que interrumpe el agente de voz, debe cancelar la reproducción de audio del agente inmediatamente en el cliente. Si no lo hace, el cliente sigue reproduciendo la última respuesta del agente hasta que la interrupción se procesa en la API, lo que da lugar a que el agente "hable sobre" el usuario.

En el ejemplo de código siguiente se muestran algunos control de eventos básicos:

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}")

Ejemplo mínimo

En el ejemplo de código siguiente se muestra la autenticación en la API y la configuración de la sesión.

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())