Poznawanie biblioteki klienta AI Voice Live dla języka Python

Biblioteka klienta azure AI Voice Live dla języka Python udostępnia klienta mowy na mowę w czasie rzeczywistym dla interfejsu API azure AI Voice Live. Otwiera sesję WebSocket w celu przesyłania strumieniowego dźwięku z mikrofonu do usługi i odbiera zdarzenia z serwera na potrzeby interaktywnych rozmów.

W tej lekcji dowiesz się, jak za pomocą zestawu SDK zaimplementować uwierzytelnianie i obsługiwać zdarzenia. Zostanie również wyświetlony minimalny przykład tworzenia sesji. Aby zapoznać się z pełną dokumentacją pakietu Voice Live, odwiedź dokumentację pakietu Voice Live.

Implementowanie uwierzytelniania

Uwierzytelnianie można zaimplementować przy użyciu klucza API lub tokenu Microsoft Entra ID. Poniższy przykładowy kod przedstawia implementację klucza interfejsu API. Zakłada się, że zmienne środowiskowe są ustawiane w .env pliku lub bezpośrednio w środowisku.

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

W przypadku aplikacji produkcyjnych zaleca się uwierzytelnianie Microsoft Entra. Poniższy przykładowy kod przedstawia implementację DefaultAzureCredential do uwierzytelniania.

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

Obsługa zdarzeń

Właściwa obsługa zdarzeń zapewnia bardziej bezproblemową interakcję między klientem a agentem. Na przykład w przypadku obsługi użytkownika przerywającego agenta głosowego należy natychmiast anulować odtwarzanie dźwięku agenta w kliencie. Jeśli tego nie zrobisz, klient będzie nadal odtwarzać ostatnią odpowiedź agenta, aż do momentu, gdy przerwanie zostanie przetworzone przez interfejs API, co może spowodować, że agent będzie mówić pomimo użytkownika.

Poniższy przykładowy kod przedstawia podstawową obsługę zdarzeń:

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

Minimalny przykład

Poniższy przykładowy kod przedstawia uwierzytelnianie w interfejsie API i konfigurowanie sesji.

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