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 密钥：

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	电话通信（欧洲）

轮次检测选项

# 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

适用场景

当需要执行概述中描述的工作流或操作时，可使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为针对特定环境进行验证、测试或专家审查的替代品。
• 如果缺少必需输入、权限、安全边界或成功标准，请暂停并请求澄清。