Explorar a biblioteca de clientes do AI Voice Live para Python

A biblioteca de clientes do Serviço de Voz Ao vivo do IA do Azure para Python fornece um cliente de conversão de fala para fala em tempo real para a API do Serviço de Voz Ao vivo do IA do Azure. Ele inicia uma sessão WebSocket para transmitir áudio do microfone ao serviço e receber eventos do servidor, permitindo conversas interativas.

Nesta unidade, você aprenderá a usar o SDK para implementar a autenticação e manipular eventos. Você também verá um exemplo mínimo de criação de uma sessão. Para uma referência completa ao pacote Voice Live, visite a referência do pacote Voice Live.

Implementar autenticação

Você pode implementar a autenticação com uma chave de API ou um token de ID do Microsoft Entra. O exemplo de código a seguir mostra uma implementação de chave de API. Ele pressupõe que as variáveis de ambiente sejam definidas em um .env arquivo ou diretamente em seu ambiente.

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

Para aplicativos de produção, a autenticação do Microsoft Entra é recomendada. O exemplo de código a seguir mostra a implementação do DefaultAzureCredential para autenticação:

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

Manipular eventos

O tratamento adequado de eventos garante uma interação mais perfeita entre o cliente e o agente. Por exemplo, ao manipular um usuário interrompendo o agente de voz, você precisa cancelar a reprodução de áudio do agente imediatamente no cliente. Se você não fizer isso, o cliente continuará reproduzindo a última resposta do agente até que a interrupção seja processada na API, resultando no agente "falando sobre" o usuário.

O exemplo de código a seguir mostra alguns tratamentos básicos de eventos:

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

Exemplo mínimo

O exemplo de código a seguir mostra a autenticação na API e a configuração da sessão.

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