Qwen3-Omni-Flash-Realtime-2025-09-15

Qwen3-Omni-Flash-Realtime

Copied!

Add to Compare

Real-time Omni-modality

Overview

Real-time Omni-modality

The real-time version of the Qwen3-Omni-Flash multimodal large-scale model, based on the Thinker–Talker Mixed Expert (MoE) architecture, supports efficient understanding and speech generation of text, images, audio, and video. It can interact with text in 119 languages and speech in 20 languages, generating human-like speech for precise cross-lingual communication. The model boasts powerful command-following and system prompt customization capabilities, flexibly adapting to conversational styles and character settings. It is widely used in scenarios such as text creation, voice assistants, and multimedia analysis, providing a natural and smooth multimodal interaction experience.

Input

TextImageAudioVideo

Output

TextAudio

Features

Prefix Completion

Function Calling

Cache

Structured Outputs

Batches

Web Search

Pricing

Input: Text
$0.52Per 1M tokens
Input: Audio
$4.57Per 1M tokens
Input: Vision
$0.94Per 1M tokens
Output: Text (When input contains only text)
$1.99Per 1M tokens
Output: Text (When input contains images/audio/video)
$3.67Per 1M tokens
Output: Text&Audio (Output text is not charged)
$18.13Per 1M tokens

Context

64K

Max Input

56K

Max Output

Rate Limits

RPMRequests Per Minute
60
TPMTokens Per Minute
100K

API Reference

Get API Key

Copied!

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768

# Dependencies: dashscope >= 1.23.9, pyaudio
import os
import base64
import time

import pyaudio
from dashscope.audio.qwen_omni import MultiModality, AudioFormat, OmniRealtimeCallback, OmniRealtimeConversation
import dashscope


url = f'wss://[workspace-id].ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime'
# API key: if DASHSCOPE_API_KEY is not set, use: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
# Voice
voice = 'Ethan'
# Model
model = 'qwen3-omni-flash-realtime-2025-09-15'
# Assistant instructions
instructions = (
    "You are Xiaoyun, a personal assistant. Answer the user's questions in a humorous and witty way."
)
class SimpleCallback(OmniRealtimeCallback):
    def __init__(self, pya):
        self.pya = pya
        self.out = None
    def on_open(self):
        # Initialize audio output stream
        self.out = self.pya.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=24000,
            output=True
        )
    def on_event(self, response):
        if response['type'] == 'response.audio.delta':
            # Play audio
            self.out.write(base64.b64decode(response['delta']))
        elif response['type'] == 'conversation.item.input_audio_transcription.completed':
            # Print user transcript
            print(f"[User] {response['transcript']}")
        elif response['type'] == 'response.audio_transcript.done':
            # Print assistant transcript
            print(f"[LLM] {response['transcript']}")

# 1. Initialize audio device
pya = pyaudio.PyAudio()
# 2. Create callback and conversation
callback = SimpleCallback(pya)
conv = OmniRealtimeConversation(model=model, callback=callback, url=url)
# 3. Connect and configure session
conv.connect()
conv.update_session(output_modalities=[MultiModality.AUDIO, MultiModality.TEXT], voice=voice, instructions=instructions)
# 4. Initialize microphone input stream
mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
# 5. Main loop: stream microphone audio
print("Conversation started. Speak into the microphone (Ctrl+C to exit)...")
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:
    # Clean up
    conv.close()
    mic.close()
    callback.out.close()
    pya.terminate()
    print("\nConversation ended.")

# Dependencies: dashscope >= 1.23.9, pyaudio
import os
import base64
import time

import pyaudio
from dashscope.audio.qwen_omni import MultiModality, AudioFormat, OmniRealtimeCallback, OmniRealtimeConversation
import dashscope


url = f'wss://[workspace-id].ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime'
# API key: if DASHSCOPE_API_KEY is not set, use: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
# Voice
voice = 'Ethan'
# Model
model = 'qwen3-omni-flash-realtime-2025-09-15'
# Assistant instructions
instructions = (
    "You are Xiaoyun, a personal assistant. Answer the user's questions in a humorous and witty way."
)
class SimpleCallback(OmniRealtimeCallback):
    def __init__(self, pya):
        self.pya = pya
        self.out = None
    def on_open(self):
        # Initialize audio output stream
        self.out = self.pya.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=24000,
            output=True
        )
    def on_event(self, response):
        if response['type'] == 'response.audio.delta':
            # Play audio
            self.out.write(base64.b64decode(response['delta']))
        elif response['type'] == 'conversation.item.input_audio_transcription.completed':
            # Print user transcript
            print(f"[User] {response['transcript']}")
        elif response['type'] == 'response.audio_transcript.done':
            # Print assistant transcript
            print(f"[LLM] {response['transcript']}")

# 1. Initialize audio device
pya = pyaudio.PyAudio()
# 2. Create callback and conversation
callback = SimpleCallback(pya)
conv = OmniRealtimeConversation(model=model, callback=callback, url=url)
# 3. Connect and configure session
conv.connect()
conv.update_session(output_modalities=[MultiModality.AUDIO, MultiModality.TEXT], voice=voice, instructions=instructions)
# 4. Initialize microphone input stream
mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
# 5. Main loop: stream microphone audio
print("Conversation started. Speak into the microphone (Ctrl+C to exit)...")
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:
    # Clean up
    conv.close()
    mic.close()
    callback.out.close()
    pya.terminate()
    print("\nConversation ended.")