Install
openclaw skills install @thcjp/azure-voicelive-2Real-time Azure voice AI SDK for bidirectional audio/text streaming, speech-to-text, microphone input, customization, and multilingual transcription.
openclaw skills install @thcjp/azure-voicelive-2核心功能: 本技能提供多种输入格式、自动化处理、专业的能力支持、专业能力支持、化处理等能力。
Build real-time voice AI applications with bidirectional WebSocket communication.
pip install azure-ai-voicelive aiohttp azure-identity
AZURE_COGNITIVE_SERVICES_ENDPOINT=https://<region>.api.cognitive.microsoft.com
AZURE_COGNITIVE_SERVICES_KEY=<api-key>
DefaultAzureCredential (preferred):
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="o-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="o-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="o-realtime-preview",
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"
})
# 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())
The VoiceLiveConnection exposes these resources:
| Resource | Purpose | Key Methods |
|---|---|---|
conn.session | Session configuration | update(session=...) |
conn.response | Model responses | create(), cancel() |
conn.input_audio_buffer | Audio input | append(), commit(), clear() |
conn.output_audio_buffer | Audio output | clear() |
conn.conversation | Conversation state | item.create(), item.delete(), item.truncate() |
conn.transcription_session | Transcription config | update(session=...) |
边界条件处理:
输入参数:
endpoint: 认知服务端点URL。credential: 认证凭据,可以是环境变量、API密钥或默认Azure凭据。model: 模型名称,例如o-realtime-preview。credential_scopes: 认证作用域。输出参数:
event: 事件对象,包含事件类型、数据等。transcript: 转写文本。audio: 音频数据。error: 错误信息。401 Unauthorized: 检查API密钥是否正确。403 Forbidden: 检查用户权限。404 Not Found: 检查请求的URL是否正确。500 Internal Server Error: 检查网络连接。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"]
}
)
]
))
import base64
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}")
print("Session updated")
# Audio input events
case "input_audio_buffer.speech_started":
print(f"Speech started at {event.audio_start_ms}ms")
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}")
print(f"Partial: {event.delta}")
# Response events
case "response.created":
print(f"Response started: {event.response.id}")
audio_transcript.delta":
print(event.delta, end="", flush=True)
audio = base64.b64decode(event.delta)
print(f"Response complete: {event.response.status}")
# Function calls
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)
})
# Errors
case "error":
print(f"Error: {event.error.message}")
await conn.session.update(session={"turn_detection": None})
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()
await conn.conversation.item.create(item={
"type": "message",
"role": "system",
"content": [{"type": "input_text", "text": "Be concise."}]
})
await conn.conversation.item.create(item={
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "Hello!"}]
})
await conn.response.create()
| Voice | Description |
|---|---|
alloy | Neutral, balanced |
echo | Warm, conversational |
shimmer | Clear, professional |
sage | Calm, authoritative |
coral | Friendly, upbeat |
ash | Deep, measured |
ballad | Expressive |
verse | Storytelling |
Azure voices: Use AzureStandardVoice, AzureCustomVoice, or AzurePersonalVoice models.
| Format | Sample Rate | Use Case |
|---|---|---|
pcm16 | 24kHz | Default, high quality |
pcm16-8000hz | 8kHz | Telephony |
pcm16-16000hz | 16kHz | Voice assistants |
g711_ulaw | 8kHz | Telephony (US) |
g711_alaw | 8kHz | Telephony (EU) |
{"type": "server_vad", "threshold": 0.5, "silence_duration_ms": 500}
{"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:
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}")
| 依赖项 | 类型 | 是否必需 | 获取方式 |
|---|---|---|---|
| LLM API | API | 必需 | 由Agent内置LLM提供 |
| 场景 | 输入 | 输出 |
|---|---|---|
| 基础使用 | 用户请求 | 处理结果 |
不适用于:需要人工判断的复杂决策场景
# 请参考上方使用说明进行配置和调用
result = "ready"
```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["AZURE_COGNITIVE_SERVICES_ENDPOINT"],
credential=DefaultAzureCredential(),
model="o-realtime-preview",
azure.com/.default"]
) as conn:
# Update session with instructions
await conn.session.update(sessio
A: 请先阅读使用流程章节,确认环境满足依赖说明中的要求。
A: 请参考错误处理章节,按照表格中的处理方式操作。
A: 请参考已知限制章节了解具体限制。
Q1: 如何配置环境变量? A: 在操作系统中设置环境变量,例如在Windows中,可以通过系统属性设置环境变量。
Q2: 如何处理认证失败? A: 检查API密钥是否正确,或者使用默认Azure凭据。
Q3: 如何处理音频输入无响应? A: 检查麦克风是否正常工作,或者尝试重新启动应用。
Q4: 如何处理转写错误? A: 检查音频质量,或者尝试使用不同的模型。
Q5: 如何处理网络错误? A: 检查网络连接,或者尝试重新启动应用。