azure-ai-voicelive-py

バンドルされたアップストリームのサポート

サポートファイルは、スキルからの相対パスに含まれています。固定されたソース、ライセンス、ハッシュ、インポート時の変更については、UPSTREAM-NOTICE.md を参照してください。

Azure AI Voice Live SDK

双方向の WebSocket 通信を使用して、リアルタイムの音声 AI アプリケーションを構築します。

インストール

pip install azure-ai-voicelive aiohttp azure-identity

環境変数

AZURE_COGNITIVE_SERVICES_ENDPOINT=https://<region>.api.cognitive.microsoft.com
# For API key auth (not recommended for production)
AZURE_COGNITIVE_SERVICES_KEY=<api-key>

実行時とプライバシーの境界

これらのスニペットはバージョンに依存する概要であり、完全な音声クライアントではありません。インストール済みの SDK／API バージョン、リソースのエンドポイント、モデルの利用可否、トークンの対象を併せて確認してください。参照資料の既定値は、以下に示す一部の明示的なスコープと競合します。非同期認証情報は明示的に閉じ、セッション時間と音声メモリの使用量に上限を設け、finally 内でキャプチャタスクをキャンセルして完了を待ってください。音声／文字起こしデータは Azure に送信されるため、録音／アップロードの許可を取得し、保持期間とログを最小限に抑え、設定済みの認証情報でシークレットを管理してください。ツール名／JSON スキーマとユーザーの権限をサーバー側で検証してください。生成された関数呼び出しはリクエストであり、許可ではありません。

認証

DefaultAzureCredential（推奨）：

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

async with connect(
    endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
    credential=DefaultAzureCredential(),
    model="gpt-4o-realtime-preview",
    credential_scopes=["https://cognitiveservices.azure.com/.default"]
) as conn:
    ...

API Key：

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

async with connect(
    endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
    credential=AzureKeyCredential(os.environ["AZURE_COGNITIVE_SERVICES_KEY"]),
    model="gpt-4o-realtime-preview"
) as conn:
    ...

クイックスタート

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

async def main():
    async with connect(
        endpoint=os.environ["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
        credential=DefaultAzureCredential(),
        model="gpt-4o-realtime-preview",
        credential_scopes=["https://cognitiveservices.azure.com/.default"]
    ) as conn:
        # Update session with instructions
        await conn.session.update(session={
            "instructions": "You are a helpful assistant.",
            "modalities": ["text", "audio"],
            "voice": "alloy"
        })
        
        # Supply a user turn, then explicitly request a response.
        await conn.conversation.item.create(item={"type": "message", "role": "user",
            "content": [{"type": "input_text", "text": "Hello"}]})
        await conn.response.create()
        # Listen for events
        async for event in conn:
            print(f"Event: {event.type}")
            if event.type == "response.audio_transcript.done":
                print(f"Transcript: {event.transcript}")
            elif event.type == "response.done":
                break

asyncio.run(main())

コアアーキテクチャ

接続リソース

VoiceLiveConnection は、以下のリソースを公開します。

リソース	用途	主なメソッド
conn.session	セッションの設定	update(session=...)
conn.response	モデルの応答	create()、cancel()
conn.input_audio_buffer	音声入力	append()、commit()、clear()
conn.output_audio_buffer	音声出力	clear()
conn.conversation	会話の状態	item.create()、item.delete()、item.truncate()
conn.transcription_session	文字起こしの設定	update(session=...)

セッションの設定

from azure.ai.voicelive.models import RequestSession, FunctionTool

await conn.session.update(session=RequestSession(
    instructions="You are a helpful voice assistant.",
    modalities=["text", "audio"],
    voice="alloy",  # or "echo", "shimmer", "sage", etc.
    input_audio_format="pcm16",
    output_audio_format="pcm16",
    turn_detection={
        "type": "server_vad",
        "threshold": 0.5,
        "prefix_padding_ms": 300,
        "silence_duration_ms": 500
    },
    tools=[
        FunctionTool(
            type="function",
            name="get_weather",
            description="Get current weather",
            parameters={
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        )
    ]
))

音声ストリーミング

音声を送信する（Base64 PCM16）

import base64

# Read audio chunk (16-bit PCM, 24kHz mono)
audio_chunk = await read_audio_from_microphone()
b64_audio = base64.b64encode(audio_chunk).decode()

await conn.input_audio_buffer.append(audio=b64_audio)

音声を受信する

async for event in conn:
    if event.type == "response.audio.delta":
        audio_bytes = base64.b64decode(event.delta)
        await play_audio(audio_bytes)
    elif event.type == "response.audio.done":
        print("Audio complete")

イベント処理

async for event in conn:
    match event.type:
        # Session events
        case "session.created":
            print(f"Session: {event.session}")
        case "session.updated":
            print("Session updated")
        
        # Audio input events
        case "input_audio_buffer.speech_started":
            print(f"Speech started at {event.audio_start_ms}ms")
        case "input_audio_buffer.speech_stopped":
            print(f"Speech stopped at {event.audio_end_ms}ms")
        
        # Transcription events
        case "conversation.item.input_audio_transcription.completed":
            print(f"User said: {event.transcript}")
        case "conversation.item.input_audio_transcription.delta":
            print(f"Partial: {event.delta}")
        
        # Response events
        case "response.created":
            print(f"Response started: {event.response.id}")
        case "response.audio_transcript.delta":
            print(event.delta, end="", flush=True)
        case "response.audio.delta":
            audio = base64.b64decode(event.delta)
        case "response.done":
            print(f"Response complete: {event.response.status}")
        
        # Function calls
        case "response.function_call_arguments.done":
            result = handle_function(event.name, event.arguments)
            await conn.conversation.item.create(item={
                "type": "function_call_output",
                "call_id": event.call_id,
                "output": json.dumps(result)
            })
            await conn.response.create()
        
        # Errors
        case "error":
            print(f"Error: {event.error.message}")

一般的なパターン

手動ターンモード（VAD なし）

await conn.session.update(session={"turn_detection": None})

# Manually control turns
await conn.input_audio_buffer.append(audio=b64_audio)
await conn.input_audio_buffer.commit()  # End of user turn
await conn.response.create()  # Trigger response

割り込み処理

async for event in conn:
    if event.type == "input_audio_buffer.speech_started":
        # User interrupted - cancel current response
        await conn.response.cancel()
        await conn.output_audio_buffer.clear()

会話履歴

# Add system message
await conn.conversation.item.create(item={
    "type": "message",
    "role": "system",
    "content": [{"type": "input_text", "text": "Be concise."}]
})

# Add user message
await conn.conversation.item.create(item={
    "type": "message",
    "role": "user", 
    "content": [{"type": "input_text", "text": "Hello!"}]
})

await conn.response.create()

音声オプション

音声	説明
alloy	中立的でバランスが取れています
echo	温かみがあり、会話向きです
shimmer	明瞭で、プロフェッショナルです
sage	落ち着きがあり、威厳があります
coral	親しみやすく、明るい印象です
ash	低く、落ち着いた印象です
ballad	表現力豊かです
verse	物語の語りに適しています

Azure の音声には、AzureStandardVoice、AzureCustomVoice、または AzurePersonalVoice モデルを使用してください。

音声フォーマット

フォーマット	サンプルレート	用途
pcm16	24kHz	既定、高音質
pcm16-8000hz	8kHz	電話
pcm16-16000hz	16kHz	音声アシスタント
g711_ulaw	8kHz	電話（米国）
g711_alaw	8kHz	電話（EU）

ターン検出オプション

# Server VAD (default)
{"type": "server_vad", "threshold": 0.5, "silence_duration_ms": 500}

# Azure Semantic VAD (smarter detection)
{"type": "azure_semantic_vad"}
{"type": "azure_semantic_vad_en"}  # English optimized
{"type": "azure_semantic_vad_multilingual"}

エラー処理

from azure.ai.voicelive.aio import ConnectionError, ConnectionClosed

try:
    async with connect(...) as conn:
        async for event in conn:
            if event.type == "error":
                print(f"API Error: {event.error.code} - {event.error.message}")
except ConnectionClosed as e:
    print(f"Connection closed: {e.code} - {e.reason}")
except ConnectionError as e:
    print(f"Connection error: {e}")

参考資料

• 詳細な API リファレンス：references/api-reference.md を参照してください
• 完全なサンプル：references/examples.md を参照してください
• すべてのモデルと型：references/models.md を参照してください

使用する場面

このスキルは、概要で説明されているワークフローやアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記の対象範囲に明確に一致する場合にのみ使用してください。
• 出力を、環境に応じた検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中断して確認を求めてください。