Qwen-Audio-3.1-Realtime
NEW复制成功！
概述
Qwen-Audio-3.1-Realtime 是新一代实时全双工语音大模型，进一步增强口语推理、多轮指令遵循、共情表达与角色扮演能力，兼顾智能应答与自然对话节奏。模型优化噪声拒识、多人对话理解与动态语义感知，让开放场景下的打断、等待与接话更自然；支持多语言交流与智能工具调用，可连接知识库及业务系统，将任务执行结果自然融入对话。同时强化安全拒答与边界控制，为智能客服、办公协作、语音陪伴等应用提供更智能、自然、可靠的交互体验
输入
文本音频
输出
文本音频
功能
定价
- 输入：音频¥40/M tokens
- 输入：文本¥5/M tokens
- 输出：文本¥40/M tokens
- 输出：文本+音频（输出的文本不计费）¥150/M tokens
速率限制与上下文
- 最大输入245K
- 最大输出16K
- 最大输入（思考模式）245K
- 最大输出（思考模式）16K
- 上下文262K
- 最大思维链2K
- TPM每分钟 Token 数100K
- RPM每分钟请求数60
API 参考
调用API复制成功！
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263
import asyncio
import base64
import json
import os
import pyaudio
import websockets
API_KEY = os.environ["DASHSCOPE_API_KEY"]
URL = "wss://maas.qianwenaiapi.com/api-ws/v1/realtime?model=qwen-audio-3.1-realtime-plus"
pya = pyaudio.PyAudio()
mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
spk = pya.open(format=pyaudio.paInt16, channels=1, rate=24000, output=True)
async def main():
headers = {"Authorization": f"Bearer {API_KEY}"}
async with websockets.connect(URL, additional_headers=headers) as ws:
await ws.send(json.dumps({
"type": "session.update",
"session": {
"modalities": ["text", "audio"],
"voice": "longanqian",
"turn_detection": {
"type": "server_vad",
"threshold": 0.5,
"silence_duration_ms": 500
}
}
}))
async def send_audio():
while True:
data = await asyncio.to_thread(mic.read, 3200, False)
await ws.send(json.dumps({
"type": "input_audio_buffer.append",
"audio": base64.b64encode(data).decode()
}))
await asyncio.sleep(0.02)
async def recv_events():
async for msg in ws:
event = json.loads(msg)
t = event["type"]
if t == "response.audio.delta":
audio = base64.b64decode(event["delta"])
await asyncio.to_thread(spk.write, audio)
elif t == "conversation.item.input_audio_transcription.completed":
print(f"[You] {event['transcript']}")
elif t == "response.audio_transcript.done":
print(f"[AI] {event['transcript']}")
elif t == "error":
print(f"[Error] {event['error']['message']}")
await asyncio.gather(send_audio(), recv_events())
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
mic.close()
spk.close()
pya.terminate()
print("\nSession ended.")