Erkunden der AI Voice Live-Clientbibliothek für Python

Die Azure AI Voice Live-Clientbibliothek für Python bietet einen Echtzeit-Sprach-zu-Sprache-Client für die Azure AI Voice Live-API. Es wird eine WebSocket-Sitzung zum Streamen von Mikrofonaudio an den Dienst geöffnet, und Serverereignisse werden für reaktionsschnelle Unterhaltungen empfangen.

In dieser Lektion erfahren Sie, wie Sie das SDK verwenden, um Authentifizierung zu implementieren und Ereignisse zu behandeln. Außerdem sehen Sie ein minimales Beispiel für das Erstellen einer Sitzung. Eine vollständige Dokumentation für das Voice Live-Paket finden Sie in der Voice Live-Paketreferenz.

Implementieren der Authentifizierung

Sie können die Authentifizierung mit einem API-Schlüssel oder einem Microsoft Entra-ID-Token implementieren. Das folgende Codebeispiel zeigt eine API-Schlüsselimplementierung. Es wird davon ausgegangen, dass Umgebungsvariablen in einer .env Datei oder direkt in Ihrer Umgebung festgelegt werden.

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

Für Produktionsanwendungen wird die Microsoft Entra-Authentifizierung empfohlen. Das folgende Codebeispiel zeigt die Implementierung der DefaultAzureCredential Authentifizierung:

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

Behandeln von Ereignissen

Die ordnungsgemäße Behandlung von Ereignissen stellt eine nahtlosere Interaktion zwischen Dem Client und Agent sicher. Wenn Sie beispielsweise einen Benutzer behandeln, der den Sprach-Agent unterbricht, müssen Sie die Audiowiedergabe des Agents sofort im Client abbrechen. Wenn Sie dies nicht tun, spielt der Client die letzte Antwort des Agenten so lange ab, bis die Unterbrechung in der API verarbeitet wird – was dazu führt, dass der Agent dem Benutzer „ins Wort redet”.

Das folgende Codebeispiel zeigt einige grundlegende Ereignisbehandlungen:

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

Minimales Beispiel

Das folgende Codebeispiel zeigt die Authentifizierung für die API und das Konfigurieren der Sitzung.

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