Qwen3.8-Omni-Flash-Realtime
复制成功！
立即体验加入对比
概述
Qwen3.8-Omni-Flash-Realtime 通过多种实时协议提供音视频双工交互，面向智能硬件、机器人与实时交互 Agent。支持多通道音频输入与多种通道布局，视频表征可在细粒度与聚合之间按精度与开销选择，支持 60+ 语言音频输入和 30+ 语言语音输出；工具侧支持 Function Call 与 MCP 服务接入，具备工具发现、调用审批与结果续轮的完整事件链。可以在高速实时响应的同时，编排复杂业务逻辑和支持上百个工具的调用，同时拥有生动、多样的声音生成能力。
输入
文本图像音频视频
输出
文本音频
功能
定价
- 输入：音频¥6/M tokens
- 输出：文本¥4.5/M tokens
- 输出：文本+音频（输出的文本不计费）¥12/M tokens
- 输入：文本/图片/视频¥1.5/M tokens
速率限制
- TPM每分钟 Token 数2M
- RPM每分钟请求数60
API 参考
调用API复制成功！
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374
# 依赖：dashscope >= 1.23.9，pyaudio
import os
import json
import sys
import base64
import time
import pyaudio
from dashscope.audio.qwen_omni import MultiModality, AudioFormat,OmniRealtimeCallback,OmniRealtimeConversation
import dashscope
url = f'wss://cn-beijing.maas.qianwenaiapi.com/api-ws/v1/realtime'
# 配置 API Key，若没有设置环境变量，请用 API Key 将下行替换为 dashscope.api_key = "sk-xxx"
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
# 指定音色
voice = 'Tina'
# 指定模型
model = 'qwen3.8-omni-flash-realtime'
# 指定模型角色
instructions = "你是个人助理小云，请用幽默风趣的方式回答用户的问题"
class SimpleCallback(OmniRealtimeCallback):
def __init__(self, pya):
self.pya = pya
self.out = None
def on_open(self):
# 初始化音频输出流
self.out = self.pya.open(
format=pyaudio.paInt16,
channels=1,
rate=24000,
output=True
)
def on_close(self, close_status_code, close_msg):
print(f"[Connection closed] code={close_status_code}, reason={close_msg}", file=sys.stderr, flush=True)
def on_event(self, response):
if response['type'] == 'error':
print(json.dumps(response, ensure_ascii=False), file=sys.stderr, flush=True)
return
if response['type'] == 'response.audio.delta':
# 播放音频
self.out.write(base64.b64decode(response['delta']))
elif response['type'] == 'conversation.item.input_audio_transcription.completed':
# 打印转录文本
print(f"[User] {response['transcript']}")
elif response['type'] == 'response.audio_transcript.done':
# 打印助手回复文本
print(f"[LLM] {response['transcript']}")
# 1. 初始化音频设备
pya = pyaudio.PyAudio()
# 2. 创建回调函数和会话
callback = SimpleCallback(pya)
conv = OmniRealtimeConversation(model=model, callback=callback, url=url)
# 3. 建立连接并配置会话
conv.connect()
conv.update_session(output_modalities=[MultiModality.AUDIO, MultiModality.TEXT], voice=voice, instructions=instructions)
# 4. 初始化音频输入流
mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
# 5. 主循环处理音频输入
print("对话已开始，对着麦克风说话 (Ctrl+C 退出)...")
try:
while True:
audio_data = mic.read(3200, exception_on_overflow=False)
conv.append_audio(base64.b64encode(audio_data).decode())
time.sleep(0.01)
except KeyboardInterrupt:
# 清理资源
conv.close()
mic.close()
callback.out.close()
pya.terminate()
print("\n对话结束")