Qwen-Audio-Realtime-Plus

Copy success!

Add to Compare

Overview

Qwen-Audio-3.0-Realtime-Plus is a next-generation, real-time, full-duplex speech foundation model that has topped global authoritative benchmarks, ranking first overall in the "Speech-to-Speech" category on Artificial Analysis—a leading independent evaluation platform. This end-to-end model balances high-level intelligence with the natural pacing of full-duplex conversation, ensuring seamless, fluid real-time interaction without compromising speech reasoning capabilities. Through engineering optimizations such as parallel inference and omnidirectional streaming, it minimizes end-to-end response latency, delivering a conversational experience that is both fast and intelligent. The Plus Edition places greater emphasis on high-quality, in-depth interaction.

Input

AudioText

Output

AudioText

Features

Prefix Completion

Enable Partial Mode when calling the Qwen API to make the model continue strictly from your provided prefix text.View docs

Function Calling

Use function calling to connect large language models with external tools and systems.View docs

Cache

Context Cache stores shared prefixes for long-context requests to reduce repeated computation, improve latency, and lower cost.View docs

Structured Outputs

Structured Outputs help ensure the model returns a JSON string in the expected format.View docs

Batches

Asynchronously process requests in batches to reduce costs.View docs

Web Search

Enable web search so the model can answer with real-time retrieved data.View docs

Fine-tuning

Train models on sample data to better adapt them to specific tasks.View docs

Pricing

Input: Audio
$6.4Per 1M tokens
Input: Text
$0.8Per 1M tokens
Output: Text
$6.4Per 1M tokens
Output: Text&Audio (Output text is not charged)
$24Per 1M tokens

Rate Limits & Context

Max Input
16K
Max Output
8K
Max Input (Thinking)
16K
Max Output (Thinking)
8K
Context
40K
Max Reasoning
1K
TPMTokens Per Minute
100K
RPMRequests Per Minute
60

API Reference

Call API

Copy success!

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263

import asyncio
import base64
import json
import os
import pyaudio
import websockets

API_KEY = os.environ["DASHSCOPE_API_KEY"]
URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen-audio-3.0-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.")