Azure realtime voice ai

何时使用

适用场景：

• 用 Python azure-ai-voicelive SDK 通过 WebSocket 搭建低延迟双向语音对话（语音助手、客服机器人、电话语音网关）。
• 需要流式收发音频（边说边听）、服务端/语义 VAD 自动断句、实时转写、语音内函数调用（function calling）。
• 接入 gpt-4o-realtime-preview 等实时模型，且后端为 Azure 认知服务端点。

不该用（负边界）：

• 非 Azure 语音栈（如 OpenAI Realtime 直连、其他云厂商语音 SDK）。
• 离线/批量语音转写、TTS 文件合成等非实时场景——应改用对应批处理服务。
• 纯文本 LLM 对话——无需音频通道时本技能过重。

步骤

1. 安装依赖：pip install azure-ai-voicelive aiohttp azure-identity
2. 配置环境变量：设置端点 AZURE_COGNITIVE_SERVICES_ENDPOINT（形如 https://<region>.api.cognitive.microsoft.com）；生产环境用托管身份认证，避免裸 API Key。
3. 建立异步连接：用 connect() 异步上下文管理器，优先传 DefaultAzureCredential() 并指定 credential_scopes=["https://cognitiveservices.azure.com/.default"]，传入 model。
4. 更新 session：调 conn.session.update(session=...) 设置 instructions、modalities、voice、音频格式、turn_detection、tools。
5. 音频收发：上行用 conn.input_audio_buffer.append(audio=b64)（Base64 PCM16）；下行监听 response.audio.delta 事件并 base64.b64decode 播放。
6. 事件循环：async for event in conn 按 event.type 分发——会话/语音起止/转写/响应/函数调用/错误。
7. 函数调用回填：收到 response.function_call_arguments.done 后，conversation.item.create 写回 function_call_output，再 conn.response.create() 触发后续响应。
8. 错误处理：捕获 ConnectionClosed / ConnectionError，并对 event.type == "error" 单独处理。

指令

• 认证（生产首选托管身份）：

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 方式（不推荐生产）：用 AzureKeyCredential(os.environ["AZURE_COGNITIVE_SERVICES_KEY"]) 替换 credential，可省 scopes。

• 连接资源（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=...)

• 音频格式：pcm16(24kHz 默认) / pcm16-16000hz(语音助手) / pcm16-8000hz、g711_ulaw(美)、g711_alaw(欧)(电话)。
• 语音：alloy/echo/shimmer/sage/coral/ash/ballad/verse；Azure 专属用 AzureStandardVoice / AzureCustomVoice / AzurePersonalVoice。
• 断句（turn_detection）：{"type":"server_vad","threshold":0.5,"silence_duration_ms":500}；更智能用 azure_semantic_vad / azure_semantic_vad_en / azure_semantic_vad_multilingual。

示例

最小可运行骨架（监听转写与响应完成）：

import asyncio, 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:
        await conn.session.update(session={
            "instructions": "你是一个乐于助人的语音助手。",
            "modalities": ["text", "audio"],
            "voice": "alloy",
        })
        async for event in conn:
            if event.type == "response.audio_transcript.done":
                print(f"转写: {event.transcript}")
            elif event.type == "response.done":
                break

asyncio.run(main())

事件分发 + 函数调用回填（核心片段）：

async for event in conn:
    match event.type:
        case "input_audio_buffer.speech_started":   # 用户开始说话（可用于打断）
            await conn.response.cancel()
            await conn.output_audio_buffer.clear()
        case "conversation.item.input_audio_transcription.completed":
            print(f"用户: {event.transcript}")
        case "response.audio.delta":
            await play_audio(base64.b64decode(event.delta))
        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()
        case "error":
            print(f"错误: {event.error.code} - {event.error.message}")

手动断句（关闭 VAD）：session.update(session={"turn_detection": None})，随后 append() → commit()(结束用户轮次) → response.create()(触发响应)。

注意事项

• 生产用托管身份（DefaultAzureCredential），API Key 仅限本地调试，勿入库。
• 上行音频须为 16-bit PCM、单声道，按所选格式匹配采样率（默认 24kHz），再 Base64 编码。
• 打断处理：收到 speech_started 时同时 response.cancel() 与 output_audio_buffer.clear()，否则旧音频会继续播。
• 电话场景选 g711_ulaw/alaw 或 8kHz PCM 以匹配链路带宽。
• 事件类型字符串需精确匹配（如 response.audio.delta vs response.audio_transcript.delta），易混淆。
• 始终捕获 ConnectionClosed / ConnectionError，并区分网络断连与业务 error 事件。
• 本技能产出为骨架，需结合实际麦克风/扬声器 IO 与环境凭据自行联调验证。

互见

• 源技能内置参考：references/api-reference.md（详细 API）、references/examples.md（完整示例）、references/models.md（模型与类型）。
• 同域「智能/misc」下其他 LLM/语音相关技能。
• 如需纯文本 Claude/Anthropic 应用，转 claude-api。

---

采编自 sickn33/antigravity-awesome-skills（MIT License）。