身份认证

DefaultAzureCredential（推荐）：
python
from azure.ai.voicelive.aio import connect
from azure.identity.aio import DefaultAzureCredential

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

API密钥：
python
from azure.ai.voicelive.aio import connect
from azure.core.credentials import AzureKeyCredential

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

快速开始

python
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[AZURECOGNITIVESERVICES_ENDPOINT],
credential=DefaultAzureCredential(),
model=gpt-4o-realtime-preview,
credential_scopes=[https://cognitiveservices.azure.com/.default]
) as conn:
# 更新会话指令
await conn.session.update(session={
instructions: 你是一个有用的助手。,
modalities: [text, audio],
voice: alloy
})

# 监听事件
async for event in conn:
print(f事件：{event.type})
if event.type == response.audio_transcript.done:
print(f转录：{event.transcript})
elif event.type == response.done:
break

asyncio.run(main())

核心架构

连接资源

VoiceLiveConnection 公开以下资源：

会话配置

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

await conn.session.update(session=RequestSession(
instructions=你是一个有用的语音助手。,
modalities=[text, audio],
voice=alloy, # 或 echo, shimmer, sage 等
inputaudioformat=pcm16,
outputaudioformat=pcm16,
turn_detection={
type: server_vad,
threshold: 0.5,
prefixpaddingms: 300,
silencedurationms: 500
},
tools=[
FunctionTool(
type=function,
name=get_weather,
description=获取当前天气,
parameters={
type: object,
properties: {
location: {type: string}
},
required: [location]
}
)
]
))

音频流

发送音频（Base64 PCM16）

python
import base64

读取音频块（16位PCM，24kHz单声道）

await conn.inputaudiobuffer.append(audio=b64_audio)

接收音频

python
async for event in conn:
if event.type == response.audio.delta:
audio_bytes = base64.b64decode(event.delta)
await playaudio(audiobytes)
elif event.type == response.audio.done:
print(音频完成)

事件处理

python
async for event in conn:
match event.type:
# 会话事件
case session.created:
print(f会话：{event.session})
case session.updated:
print(会话已更新)

# 音频输入事件
case inputaudiobuffer.speech_started:
print(f语音在{event.audiostartms}毫秒开始)
case inputaudiobuffer.speech_stopped:
print(f语音在{event.audioendms}毫秒停止)

# 转录事件
case conversation.item.inputaudiotranscription.completed:
print(f用户说：{event.transcript})
case conversation.item.inputaudiotranscription.delta:
print(f部分：{event.delta})

# 响应事件
case response.created:
print(f响应开始：{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响应完成：{event.response.status})

# 函数调用
case response.functioncallarguments.done:
result = handle_function(event.name, event.arguments)
await conn.conversation.item.create(item={
type: functioncalloutput,
callid: event.callid,
output: json.dumps(result)
})
await conn.response.create()

# 错误
case error:
print(f错误：{event.error.message})

常见模式

手动轮次模式（无VAD）

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

手动控制轮次

中断处理

python
async for event in conn:
if event.type == inputaudiobuffer.speech_started:
# 用户中断 - 取消当前响应
await conn.response.cancel()
await conn.outputaudiobuffer.clear()

对话历史

python

添加系统消息

添加用户消息

await conn.response.create()

语音选项

Azure语音：使用 AzureStandardVoice、AzureCustomVoice 或 AzurePersonalVoice 模型。

音频格式

轮次检测选项

python

服务器端VAD（默认）

Azure语义VAD（更智能的检测）