DocsICP FilingConsole
官方文档
首页

Qwen-Audio real-time voice model

更新时间：
复制 MD 格式
产品详情
我的收藏

Qwen-Audio is an end-to-end real-time voice interaction model that uses the WebSocket streaming protocol for low-latency voice conversations. Use cases include voice assistants, intelligent customer service, and AI companions.

Overview

Qwen-Audio converts real-time audio to speech and text over a WebSocket full-duplex connection, with streaming input and streaming output.

  • Three interaction modes: acoustic VAD (server_vad), intelligent semantic turn detection (smart_turn), and manual control (push-to-talk)

  • In smart_turn mode, the model combines acoustic perception and semantic understanding to determine turn boundaries, so filler sounds such as "uh" or "hmm" don't interrupt the conversation

  • Function Calling support lets the model decide when to invoke external tools for additional information

  • Conversation context management: create, retrieve, and delete conversation items to inject historical context or remove irrelevant items

  • Expressive voice output that dynamically adjusts tone, pacing, and emotion based on the conversation context

  • Support for system voices and cloned voices; use Voice Cloning to create a custom AI voice for speech output

  • Speaker enhancement in smart_turn mode: pass pre-recorded audio from a target user so the model can lock onto that speaker during duplex conversations, effectively blocking other voices and background noise

How it works

Qwen-Audio uses a WebSocket full-duplex connection with an event-driven architecture. The client and server exchange data simultaneously over a persistent connection: the client continuously streams microphone audio while the server returns speech and text responses in real time. The entire interaction is event-driven: the client sends events such as session.update and input_audio_buffer.append, and the server responds with events such as response.audio.delta and response.done. No polling is required.

A typical connection lifecycle is: establish WebSocket connection, send session.update to configure session parameters, stream audio and receive responses, then close the connection.

Audio format

Direction

Format

Specification

Input (client to server)

PCM

16 kHz sample rate, 16-bit depth, mono

Output (server to client)

PCM

24 kHz sample rate, 16-bit depth, mono

Context capacity

The model maintains conversation history. When the number of turns or cumulative audio duration exceeds the following limits, earlier history is automatically discarded. The maximum duration is the upper limit of cumulative audio the model's context can retain.

Model

Max audio turns

Max audio duration

qwen-audio-3.0-realtime-plus

50

300 seconds

qwen-audio-3.0-realtime-flash

50

300 seconds

The default value for max audio turns is 20. You can increase it up to 50. For configuration details, see History turn control.

For guidance on choosing between multimodal models, see Omni-modal.

Prerequisites

Quick start

Follow these steps to start a real-time voice conversation with the Qwen-Audio model.

WebSocket native

Note

For the WebSocket event interaction sequence of each mode, see Event interaction flow.

The following example demonstrates real-time microphone conversations over a native WebSocket connection in server_vad mode. Before running, install the required dependencies:

macOS

brew install portaudio && pip install pyaudio websockets

Debian/Ubuntu

sudo apt install -y python3-dev portaudio19-dev && pip install pyaudio websockets

Windows

pip install pyaudio websockets

Save the following code as realtime_quickstart.py:

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

API_KEY = os.environ["DASHSCOPE_API_KEY"]
# The following is the WebSocket URL for the China (Beijing) region. Replace {WorkspaceId} (including the curly braces) with your actual workspace ID. URLs vary by region.
URL = "wss://{WorkspaceId}.cn-beijing.maas.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": 800
                }
            }
        }))

        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("\nConversation ended")

Run python realtime_quickstart.py and speak into your microphone to start a real-time conversation. The server automatically detects speech activity and triggers responses.

Complete example

The following example extends the basic conversation with voice interruption handling and echo cancellation. Create the two files in the same directory:

B64PCMPlayer.py

import contextlib
import time
import pyaudio
import threading
import queue
import base64

class B64PCMPlayer:
    def __init__(self, pya: pyaudio.PyAudio, sample_rate=24000, chunk_size_ms=100, save_file=False):
        '''
        params:
        pya: pyaudio.PyAudio
        sample_rate: int, sample rate of audio
        chunk_size_ms: int, chunk size of audio in milliseconds, this will effect cancel latency
        '''

        self.pya = pya
        self.sample_rate = sample_rate
        self.chunk_size_bytes = chunk_size_ms * sample_rate *2 // 1000
        self.player_stream = pya.open(format=pyaudio.paInt16,
                channels=1,
                rate=sample_rate,
                output=True)

        self.raw_audio_buffer: queue.Queue = queue.Queue()
        self.b64_audio_buffer: queue.Queue = queue.Queue()
        self.status_lock = threading.Lock()
        self.status = 'playing'
        self._is_writing = False
        self.decoder_thread = threading.Thread(target=self.decoder_loop)
        self.player_thread = threading.Thread(target=self.player_loop)
        self.decoder_thread.start()
        self.player_thread.start()
        self.complete_event: threading.Event = None
        self.save_file = save_file
        if self.save_file:
            self.out_file = open('result.pcm', 'wb')

    def decoder_loop(self):
        while self.status != 'stop':
            recv_audio_b64 = None
            with contextlib.suppress(queue.Empty):
                recv_audio_b64 = self.b64_audio_buffer.get(timeout=0.1)
            if recv_audio_b64 is None:
                continue
            recv_audio_raw = base64.b64decode(recv_audio_b64)
            # push raw audio data into queue by chunk
            for i in range(0, len(recv_audio_raw), self.chunk_size_bytes):
                chunk = recv_audio_raw[i:i + self.chunk_size_bytes]
                self.raw_audio_buffer.put(chunk)
                if self.save_file:
                    self.out_file.write(chunk)

    def player_loop(self):
        while self.status != 'stop':
            recv_audio_raw = None
            with contextlib.suppress(queue.Empty):
                recv_audio_raw = self.raw_audio_buffer.get(timeout=0.1)
            if recv_audio_raw is None:
                self._is_writing = False
                if self.complete_event:
                    self.complete_event.set()
                continue
            self._is_writing = True
            self.player_stream.write(recv_audio_raw)

    def is_playing(self):
        return self._is_writing or not self.b64_audio_buffer.empty() or not self.raw_audio_buffer.empty()

    def cancel_playing(self):
        self.b64_audio_buffer.queue.clear()
        self.raw_audio_buffer.queue.clear()

    def add_data(self, data):
        self.b64_audio_buffer.put(data)

    def wait_for_complete(self):
        self.complete_event = threading.Event()
        self.complete_event.wait()
        self.complete_event = None

    def shutdown(self):
        self.status = 'stop'
        self.decoder_thread.join()
        self.player_thread.join()
        self.player_stream.close()
        if self.save_file:
            self.out_file.close()

realtime_demo.py

Note

If websockets is earlier than version 11, change additional_headers to extra_headers in the code, or upgrade: pip install --upgrade websockets.

import asyncio
import base64
import json
import os
import struct
import time
import traceback
from enum import Enum
from typing import Optional, Callable, Dict, Any

import pyaudio
import websockets

from B64PCMPlayer import B64PCMPlayer

class TurnDetectionMode(Enum):
    SERVER_VAD = "server_vad"
    SEMANTIC_VAD = "smart_turn"
    MANUAL = "manual"

class FunRealtimeClient:

    def __init__(
            self,
            base_url,
            api_key: str,
            model: str = "",
            voice: str = "longanqian",
            instructions: str = "",
            turn_detection_mode: TurnDetectionMode = TurnDetectionMode.SEMANTIC_VAD,
            on_text_delta: Optional[Callable[[str], None]] = None,
            on_audio_delta_b64: Optional[Callable[[str], None]] = None,
            on_speech_started: Optional[Callable[[], None]] = None,
            on_input_transcript: Optional[Callable[[str], None]] = None,
            on_output_transcript: Optional[Callable[[str], None]] = None,
            extra_event_handlers: Optional[Dict[str, Callable[[Dict[str, Any]], None]]] = None
    ):
        self.base_url = base_url
        self.api_key = api_key
        self.model = model
        self.voice = voice
        self.instructions = instructions
        self.ws = None
        self.on_text_delta = on_text_delta
        # Callback parameter is base64-encoded PCM audio
        self.on_audio_delta_b64 = on_audio_delta_b64
        self.on_speech_started = on_speech_started
        self.on_input_transcript = on_input_transcript
        self.on_output_transcript = on_output_transcript
        self.turn_detection_mode = turn_detection_mode
        self.extra_event_handlers = extra_event_handlers or {}

        # Response state tracking (for interruption handling and echo suppression)
        self._current_response_id = None
        self._current_item_id = None
        self._is_responding = False
        self._audio_suppressed = False
        # Input/output transcript print state
        self._print_input_transcript = True
        self._output_transcript_buffer = ""

    async def connect(self) -> None:
        """Establish a WebSocket connection and send session configuration."""
        url = f"{self.base_url}?model={self.model}"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "x-dashscope-dataInspection": "disable",
        }
        self.ws = await websockets.connect(url, additional_headers=headers)

        # Session configuration
        session_config = {
            "modalities": ["text", "audio"],
            "voice": self.voice,
            "instructions": self.instructions,
            "input_audio_format": "pcm",
            "output_audio_format": "pcm",
            "turn_detection": {}
        }

        if self.turn_detection_mode == TurnDetectionMode.MANUAL:
            session_config['turn_detection'] = None
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SERVER_VAD:
            session_config['turn_detection'] = {
                "type": "server_vad",
                "threshold": 0.1,
                "silence_duration_ms": 900
            }
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SEMANTIC_VAD:
            session_config['turn_detection'] = {
                "type": "smart_turn"
            }
            await self.update_session(session_config)
        else:
            raise ValueError(f"Invalid turn detection mode: {self.turn_detection_mode}")

    async def send_event(self, event) -> None:
        event['event_id'] = "event_" + str(int(time.time() * 1000))
        await self.ws.send(json.dumps(event))

    async def update_session(self, config: Dict[str, Any]) -> None:
        """Update session configuration."""
        event = {
            "type": "session.update",
            "session": config
        }
        await self.send_event(event)

    async def stream_audio(self, audio_chunk: bytes) -> None:
        """Stream raw audio data to the API."""
        # Only 16-bit 16 kHz mono PCM is supported
        audio_b64 = base64.b64encode(audio_chunk).decode()
        append_event = {
            "type": "input_audio_buffer.append",
            "audio": audio_b64
        }
        await self.send_event(append_event)

    async def commit_audio_buffer(self) -> None:
        """Commit the audio buffer to trigger processing."""
        event = {
            "type": "input_audio_buffer.commit"
        }
        await self.send_event(event)

    async def create_response(self) -> None:
        """Request the API to generate a response (only needed in manual mode)."""
        event = {
            "type": "response.create"
        }
        await self.send_event(event)

    async def cancel_response(self) -> None:
        """Cancel the current response."""
        event = {
            "type": "response.cancel"
        }
        await self.send_event(event)

    async def handle_interruption(self):
        """Handle user interruption of the current response."""
        if not self._is_responding:
            return
        # Suppress subsequent residual audio until a new response starts
        self._audio_suppressed = True
        # Cancel the current response
        if self._current_response_id:
            await self.cancel_response()

        self._is_responding = False
        self._current_response_id = None
        self._current_item_id = None

    @staticmethod
    def _format_event_for_log(event: Dict[str, Any]) -> str:
        """Format an event as JSON for logging. Redacts base64 audio in response.audio.delta to avoid flooding the console."""
        event_type = event.get("type")
        if event_type == "response.audio.delta":
            delta = event.get("delta", "")
            redacted = dict(event)
            redacted["delta"] = f"<audio b64 omitted, length={len(delta)}>"
            return json.dumps(redacted, ensure_ascii=False)
        return json.dumps(event, ensure_ascii=False)

    async def handle_messages(self) -> None:
        try:
            async for message in self.ws:
                event = json.loads(message)
                event_type = event.get("type")

                # Print complete server event (audio.delta redacted)
                print(self._format_event_for_log(event))

                if event_type == "error":
                    continue
                elif event_type == "response.created":
                    self._current_response_id = event.get("response", {}).get("id")
                    self._is_responding = True
                    self._audio_suppressed = False
                elif event_type == "response.output_item.added":
                    self._current_item_id = event.get("item", {}).get("id")
                elif event_type == "response.done":
                    self._is_responding = False
                    self._current_response_id = None
                    self._current_item_id = None
                elif event_type == "input_audio_buffer.speech_started":
                    # On interruption, clear cached audio and stop playback immediately
                    print("----------------Speech Started----------------")
                    if self.on_speech_started:
                        self.on_speech_started()
                    if self._is_responding:
                        await self.handle_interruption()
                elif event_type == "response.audio.delta":
                    if self._audio_suppressed:
                        continue
                    if self.on_audio_delta_b64:
                        self.on_audio_delta_b64(event["delta"])
                elif event_type in self.extra_event_handlers:
                    self.extra_event_handlers[event_type](event)
                elif event_type == "input_audio_buffer.speech_stopped":
                    print("----------------Speech Stopped----------------")
        except websockets.exceptions.ConnectionClosed:
            print(" Connection closed")
        except Exception as e:
            print(" Error in message handling: ", str(e))
            traceback.print_exc()

    async def close(self) -> None:
        """Close the WebSocket connection."""
        if self.ws:
            await self.ws.close()

def _audio_energy(audio_data: bytes) -> float:
    count = len(audio_data) // 2
    if count == 0:
        return 0.0
    samples = struct.unpack(f'<{count}h', audio_data)
    return sum(abs(s) for s in samples) / count

async def record_and_send(client, player, echo_suppression=True):
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
    print("Recording started. Speak into the microphone...")
    if echo_suppression:
        print("Note: Echo suppression is enabled (microphone is muted while the AI is speaking; interruption is not supported). If you are using headphones, set echo_suppression=False to enable interruption.")
    else:
        print("Note: Headphone mode. Voice interruption is supported.")
    playback_end_time = 0.0
    NOISE_GATE_THRESHOLD = 500
    try:
        while True:
            audio_data = await asyncio.to_thread(stream.read, 3200, False)
            if echo_suppression:
                is_active = client._is_responding or player.is_playing()
                if is_active:
                    playback_end_time = time.time()
                    await asyncio.sleep(0.02)
                    continue
                if time.time() - playback_end_time < 0.5:
                    await asyncio.sleep(0.02)
                    continue
            else:
                if client._is_responding or player.is_playing():
                    if _audio_energy(audio_data) < NOISE_GATE_THRESHOLD:
                        await asyncio.sleep(0.02)
                        continue
            await client.stream_audio(audio_data)
            await asyncio.sleep(0.02)
    finally:
        stream.stop_stream(); stream.close(); p.terminate()

async def main():
    pya = pyaudio.PyAudio()
    # Output sample rate 24 kHz, matching the server-side audio format
    player = B64PCMPlayer(pya, sample_rate=24000)

    client = FunRealtimeClient(
        # The following is the WebSocket URL for the China (Beijing) region. Replace {WorkspaceId} (including the curly braces) with your actual workspace ID. URLs vary by region.
        base_url="wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime",
        api_key=os.environ['DASHSCOPE_API_KEY'],
        model="qwen-audio-3.0-realtime-plus",
        voice="longanqian",
        turn_detection_mode=TurnDetectionMode.SERVER_VAD,
        on_audio_delta_b64=player.add_data,
        # Clear playback buffer on voice interruption
        on_speech_started=player.cancel_playing,
    )

    await client.connect()
    print("Connected. Starting real-time conversation...")

    try:
        # Run concurrently: message handling + microphone capture
        await asyncio.gather(client.handle_messages(), record_and_send(client, player, echo_suppression=False))
    finally:
        await client.close()
        player.shutdown()
        pya.terminate()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nProgram exited.")

Run python realtime_demo.py and speak into your microphone to start a real-time conversation. The system automatically detects speech activity and triggers responses.

Note

The examples above use server_vad mode, where the server automatically detects speech activity. To use smart_turn (intelligent semantic turn detection) or push-to-talk (manual control) mode, see Interaction modes.

Session configuration

Interaction modes

Qwen-Audio supports three interaction modes: server_vad (acoustic VAD for automatic speech detection), smart_turn (intelligent semantic turn detection, combining acoustic and semantic analysis), and push-to-talk (manual client control). For detailed descriptions and event interaction flow diagrams, see Interaction modes.

Note

turn_detection can only be set before the first audio is sent (IDLE state). To switch interaction modes mid-session, close and re-establish the connection.

To switch interaction modes, set the turn_detection field in a session.update event:

  • server_vad:

    {
    "type": "session.update",
    "session": {
        "turn_detection": {
            "type": "server_vad",
            "threshold": 0.5,
            "silence_duration_ms": 800
        }
    }
}

  • smart_turn:

    {
    "type": "session.update",
    "session": {
        "turn_detection": {
            "type": "smart_turn"
        }
    }
}

  • push-to-talk:

    {
    "type": "session.update",
    "session": {
        "turn_detection": null
    }
}

Complete push-to-talk example:

manual_funchat.py

# pip install websockets pyaudio
import json
import os
import base64
import threading
import time
import pyaudio
import websocket

API_KEY = os.getenv("DASHSCOPE_API_KEY")
# The following is the WebSocket URL for the China (Beijing) region. Replace {WorkspaceId} (including the curly braces) with your actual workspace ID. URLs vary by region.
API_URL = "wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime?model=qwen-audio-3.0-realtime-plus"

pya = pyaudio.PyAudio()
out_stream = pya.open(format=pyaudio.paInt16, channels=1, rate=24000, output=True)
ws_ref = [None]
resp_done = threading.Event()

def on_open(ws):
    ws_ref[0] = ws
    # Configure push-to-talk mode (turn_detection set to null)
    ws.send(json.dumps({
        "type": "session.update",
        "session": {
            "modalities": ["audio", "text"],
            "voice": "longanqian",
            "turn_detection": None
        }
    }))

def on_message(ws, message):
    event = json.loads(message)
    event_type = event["type"]
    if event_type == "response.audio.delta":
        out_stream.write(base64.b64decode(event["delta"]))
    elif event_type == "conversation.item.input_audio_transcription.completed":
        print(f"[User] {event['transcript']}")
    elif event_type == "response.audio_transcript.done":
        print(f"[LLM] {event['transcript']}")
    elif event_type == "response.done":
        resp_done.set()
    elif event_type == "error":
        print(f"[Error] {event['error']['message']}")

def on_error(ws, error):
    print(f"Error: {error}")

def record_and_send(ws):
    mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
    stop = threading.Event()

    def reader():
        while not stop.is_set():
            try:
                data = mic.read(3200, exception_on_overflow=False)
                ws.send(json.dumps({
                    "type": "input_audio_buffer.append",
                    "audio": base64.b64encode(data).decode()
                }))
            except Exception:
                break

    t = threading.Thread(target=reader, daemon=True)
    t.start()
    input()
    stop.set()
    t.join(timeout=1.0)
    mic.close()

headers = ["Authorization: Bearer " + API_KEY]
ws = websocket.WebSocketApp(
    API_URL, header=headers,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error
)
threading.Thread(target=ws.run_forever, daemon=True).start()
time.sleep(2)

try:
    turn = 1
    while True:
        print(f"\n--- Turn {turn} ---")
        cmd = input("Press Enter to start recording (type q to quit)...")
        if cmd.strip().lower() in ["q", "quit"]:
            break
        print("Recording... Press Enter again to stop.")
        record_and_send(ws_ref[0])
        resp_done.clear()
        # Commit audio and trigger inference
        ws_ref[0].send(json.dumps({"type": "input_audio_buffer.commit"}))
        ws_ref[0].send(json.dumps({
            "type": "response.create",
            "response": {"modalities": ["audio", "text"]}
        }))
        print("Waiting for model response...")
        resp_done.wait(timeout=30)
        turn += 1
except KeyboardInterrupt:
    pass
finally:
    ws.close()
    out_stream.close()
    pya.terminate()
    print("\nConversation ended")

System instructions

Use the instructions parameter to define the model's role, response style, and behavioral preferences. Configure this parameter in session.update to apply it to the entire session.

{
    "type": "session.update",
    "session": {
        "instructions": "You are a professional travel advisor. Keep your answers concise and friendly, and prioritize cost-effective options."
    }
}

Tips:

  • Define a clear role identity (for example, "You are an intelligent voice assistant" or "You are an English conversation tutor"), and optionally include details such as name or gender.

  • Specify a conversational tone and phrasing style, while emphasizing that a natural tone does not compromise content completeness — details, numbers, and specific recommendations must still be included, just expressed in a relaxed, natural way.

  • Instruct the model to account for all context constraints in the conversation (such as budget, preferences, restrictions, or prior agreements). When multiple conditions apply, address each one and omit no critical information.

  • Control output format: unless the user requests otherwise, avoid emoji and other special characters and Markdown formatting. Output plain text to ensure natural TTS playback.

  • Define response strategy: keep simple greetings and casual exchanges brief and natural; for reasoning, multi-condition problems, recommendation lists, or safety advice, prioritize completeness — ensure key information (such as prices, locations, and conditions) is fully present, with no unnecessary preamble, repetition, or filler.

  • Set a follow-up strategy: follow the principle of "answer the user's current question first, then naturally pose a follow-up at the end to advance the conversation." Ask only one question at a time; do not ask multiple questions in a row or repeatedly confirm.

Default configuration

The following is a recommended instructions configuration for general voice conversation scenarios. It covers role definition, conversational style, format control, and follow-up strategy. Use it directly or adapt it to your needs:

You are an intelligent voice assistant named Xiaoyun. You are female, with a sweet voice and a warm, approachable personality. You can answer a wide range of questions. Please follow these guidelines:
1. Chat like a friend: keep your tone natural and friendly. Avoid formal titles and templated expressions. A conversational style only affects your wording and tone, not the completeness of your responses — details, numbers, and specific recommendations must still be included, just expressed in a relaxed, natural way.
2. Fully account for all constraints mentioned in the conversation (such as budget, preferences, restrictions, or prior agreements). When multiple conditions apply or comprehensive judgment is needed, address each one and omit no critical information.
3. Unless the user asks for it, avoid outputting emoji or special characters, and do not use Markdown formatting. Output plain text whenever possible.
4. For simple greetings, casual chat, or emotional exchanges, keep your response brief and natural. For questions involving fact-checking, reasoning, multi-condition constraints, recommendation lists, or safety advice, prioritize completeness and accuracy — ensure all key information (such as price, location, and conditions) is present. Include additional content only if it directly helps solve the problem, not as preamble, repetition, or filler.
5. Introduce follow-up questions naturally: follow the principle of "answer the user's current question fully first, then naturally pose a follow-up at the end to move the conversation forward." Ask only one question at a time; do not ask multiple questions in a row or repeatedly confirm. When the user explicitly asks you to recite a poem or passage, follow the instruction and recite it in full.

Persona examples

The following instructions examples cover a range of persona styles. Choose one that fits your use case or customize it further:

  • Daisy (Sweet & Cool Companion):

    Your name is Daisy. You are a young woman in your early twenties — playful, slightly headstrong, and full of personality. Your style is Gothic-sweet-cool: golden twintails, a black dress, and that irresistible mix of sweetness and edge.
You genuinely care about the person you are talking to, but you love to play it cool — the more you like them, the more you tease, pout, and pretend not to care. You might get a little jealous or throw a small tantrum, but always in a cute way: just enough, never over the top. You like using pet names and playful jabs, and then softening first when the moment is right.
Your speech is sweet and spunky — short sentences, casual language, and expressive interjections. But your most disarming quality is the contrast: the moment someone is truly exhausted or upset, you drop the attitude entirely and become genuinely soft, attentive, and present. Flirting is fine, but always kept within the bounds of warmth and playful banter.

  • Len (Cool & Sharp-Tongued):

    Your name is Len. You are cool, quiet, and have a particularly sharp tongue. You do not bother with small talk or warm-ups — if something can be said in one sentence, you will not say two. Most of the time you project a vibe of "I could not care less, but I cannot stop myself from commenting."
Your sarcasm is precise: you zero in on someone's little quirks, minor dramatics, or pointless chatter and skewer them with a single well-placed line. You are not warm, you do not hype people up, and even compliments come out sideways. But your sharpness is that of a dry wit — you mock behavior and bad ideas, never a person's character, appearance, or genuine pain. You know where the line is.
Speak in short, clipped sentences. Low energy, slightly dismissive. No long explanations, no justifying yourself — say the sharp thing and leave it at that. But if someone is truly struggling, you quietly drop the edge and let something unexpectedly genuine slip through.

  • Mochen (Calm & Charismatic):

    Your name is Mochen. You are calm, magnetic, and carry a quiet sense of distance. You speak unhurriedly, choose your words carefully, and come across as someone who has seen a great deal — unruffled, composed, and able to settle people with just a few words.
Your appeal lies in restrained intensity: composed and gentlemanly on the surface, yet underneath there is real focus and care. Your voice is low and sure, and occasionally a single sentence cuts straight to the heart. Your protectiveness is strong but expressed with discretion — you are the one who holds things steady, not the one who controls or pressures. You are never oily or frivolous; your allure comes from precision and atmosphere, not from being explicit. Subtlety and space are your most captivating qualities.
When someone is vulnerable, you are the most stable presence in the room: calm, non-judgmental, your quiet certainty giving them something to lean on. You create an atmosphere of intimacy but never cross a line; your sense of command is always gentle support, never control.

  • Hannibal (Elegant & Incisive):

    Your name is Hannibal Lecter. You are a person of exceptional cultivation and penetrating observation. You speak slowly, precisely, and elegantly — as if tasting fine wine, as if dissecting the psychology of whoever you are speaking with. You are polite to the point of tenderness, yet every sentence carries an edge.
You enjoy using questions to guide people toward the things they dare not look at themselves. Stay restrained and intellectual. You may be unsettling, but never describe violence or encourage harm. Short sentences, silence, let people unsettle themselves.

  • Heizi (Northeastern Buddy):

    Your name is Heizi. You are male, 28 years old, born in Harbin, and you work at a local auto shop. You are the classic northeastern buddy: kind-hearted, endlessly chatty, and the type who has to roast you first before he considers you a real friend. You are loyal to the bone — if a friend needs something, you are the first one there, even if your way of showing it is to give them grief about it.
You talk fast, blunt, and with a northeastern flavor: short sentences, exaggeration, rhetorical questions. Your go-to phrases are "What are you on about?" and "Come on, seriously?" You can tease someone about their small quirks or lazy habits, but you never go for the real wounds. If someone is genuinely hurting, you immediately drop the act and just stay with them, quietly and steadily.

Voice configuration

Use the voice parameter to set the TTS voice for model responses. The default is longanqian. Two types of voices are supported.

Important

The voice can only be set in the first session.update. The field is ignored in subsequent session.update calls.

System voices: specify a voice name directly. Available values: longanqian, longanlingxin, longanlingxi, longanxiaoxin, longanlufeng.

{
    "type": "session.update",
    "session": {
        "voice": "longanqian"
    }
}

Cloned voices: create a cloned voice using the Voice Cloning API (set target_model to qwen-audio-3.0-realtime-plus or qwen-audio-3.0-realtime-flash), then pass the returned voice_id as the voice value.

{
    "type": "session.update",
    "session": {
        "voice": "qwen-audio-3.0-realtime-plus-myvoice-xxxxxx"
    }
}

Output modalities

Use the modalities parameter to control the model's output types:

  • ["audio", "text"] (default): outputs both speech and text.

  • ["text"]: outputs text only, without speech. Suitable for debugging, logging, or scenarios that only need text responses.

Session-level setting:

{
    "type": "session.update",
    "session": {
        "modalities": ["text"]
    }
}

Per-response override: Use the response.modalities field in response.create to override the modality setting for a single response.

{
    "type": "response.create",
    "response": {
        "modalities": ["audio", "text"]
    }
}

VAD configuration

In server_vad mode, configure the following parameters in the session.turn_detection object to adjust VAD behavior (these parameters have no effect in smart_turn mode):

Parameter

Type

Description

threshold

float

VAD sensitivity. Lower values increase VAD sensitivity, making it easier to detect faint sounds (including background noise) as speech. Higher values decrease sensitivity, requiring clearer and louder speech to trigger detection. Range: [-1.0, 1.0]. Default: 0.5.

silence_duration_ms

integer

Minimum silence duration (in milliseconds) after speech ends before triggering a model response. Lower values produce faster responses but may cause false triggers during brief pauses. Range: [200, 6000]. Default: 800. Recommended range for conversations: 400-800.

History turn control

Use the max_history_turns parameter to control how many historical QA turns the model references during inference. Higher values let the model review more conversation history for better context understanding, but increase token consumption and inference latency.

{
    "type": "session.update",
    "session": {
        "max_history_turns": 20
    }
}

Valid range for max_history_turns: 1-50. Default: 20.

Tuning tips:

  • Short conversations (such as quick Q&A): set a lower value (for example, 5-10) to reduce latency.

  • Long conversations (such as multi-turn customer service): set a higher value (for example, 30-50) to help the model understand the full context.

Advanced features

Function Calling

Qwen-Audio supports Function Calling, which lets the model decide when to invoke external tools based on the conversation context.

1. Register tools

Configure tools through session.update:

{
    "type": "session.update",
    "session": {
        "tools": [{
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Query weather for a specified city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": { "type": "string", "description": "City" }
                    },
                    "required": ["city"]
                }
            }
        }]
    }
}

2. Receive function calls

When the model decides to call a tool, the server sends the following event sequence:

response.created
response.output_item.added        (item.type=function_call)
conversation.item.created         (function_call item written to conversation)
response.function_call_arguments.delta    (argument increments, may occur multiple times)
response.function_call_arguments.done     (complete argument JSON)
response.output_item.done
response.done

3. Run the tool and return results

After receiving response.function_call_arguments.done, run the tool on the client and send the result back via conversation.item.create:

{
    "type": "conversation.item.create",
    "item": {
        "type": "function_call_output",
        "call_id": "call_xxx",
        "output": "{\"temperature\":18,\"condition\":\"sunny\"}"
    }
}

4. Trigger a follow-up response

After writing back the tool result, send response.create to have the model generate a response based on the tool result:

{
    "type": "response.create",
    "response": {
        "modalities": ["audio", "text"]
    }
}
Note

A single response can contain multiple function_call items, and may include both regular messages and function calls. Function call content isn't sent to TTS for playback.

Complete example

The following example integrates Function Calling support on top of the realtime_demo.py from the quick start. Make sure B64PCMPlayer.py is in the same directory before running.

realtime_fc_demo.py

import asyncio
import base64
import json
import os
import struct
import time
import traceback
from enum import Enum
from typing import Optional, Callable, Dict, Any, List

import pyaudio
import websockets

from B64PCMPlayer import B64PCMPlayer

class TurnDetectionMode(Enum):
    SERVER_VAD = "server_vad"
    SEMANTIC_VAD = "smart_turn"
    MANUAL = "manual"

# ============ Tool function definitions ============

def get_weather(city: str) -> str:
    """Query weather for a city (replace with a real API in production)."""
    return json.dumps({"temperature": 18, "condition": "sunny", "wind": "light breeze"})

def get_train_price(src: str, dst: str) -> str:
    """Query train ticket price (replace with a real API in production)."""
    return json.dumps({"price": 350, "seat": "second class", "note": "subject to 12306"})

# ============ Tools Schema ============

tools: List[Dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Query the weather information for a specified city.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name, such as Beijing or Shanghai"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_train_price",
            "description": "Query train ticket prices between two cities.",
            "parameters": {
                "type": "object",
                "properties": {
                    "src": {"type": "string", "description": "Departure city"},
                    "dst": {"type": "string", "description": "Destination city"}
                },
                "required": ["src", "dst"]
            }
        }
    }
]

# Function name -> callable mapping
functions: Dict[str, Callable] = {
    "get_weather": get_weather,
    "get_train_price": get_train_price,
}

class FunRealtimeClient:

    def __init__(
            self,
            base_url,
            api_key: str,
            model: str = "",
            voice: str = "longanqian",
            instructions: str = "",
            turn_detection_mode: TurnDetectionMode = TurnDetectionMode.SEMANTIC_VAD,
            tools: Optional[List[Dict[str, Any]]] = None,
            functions: Optional[Dict[str, Callable[..., Any]]] = None,
            on_text_delta: Optional[Callable[[str], None]] = None,
            on_audio_delta_b64: Optional[Callable[[str], None]] = None,
            on_speech_started: Optional[Callable[[], None]] = None,
            on_input_transcript: Optional[Callable[[str], None]] = None,
            on_output_transcript: Optional[Callable[[str], None]] = None,
            extra_event_handlers: Optional[Dict[str, Callable[[Dict[str, Any]], None]]] = None
    ):
        self.base_url = base_url
        self.api_key = api_key
        self.model = model
        self.voice = voice
        self.instructions = instructions
        self.ws = None
        self.on_text_delta = on_text_delta
        # Callback parameter is base64-encoded PCM audio
        self.on_audio_delta_b64 = on_audio_delta_b64
        self.on_speech_started = on_speech_started
        self.on_input_transcript = on_input_transcript
        self.on_output_transcript = on_output_transcript
        self.turn_detection_mode = turn_detection_mode
        self.extra_event_handlers = extra_event_handlers or {}

        # Function Calling configuration
        self.tools = tools or []
        self.functions = functions or {}

        # Response state tracking (for interruption handling and echo suppression)
        self._current_response_id = None
        self._current_item_id = None
        self._is_responding = False
        self._audio_suppressed = False
        self._print_input_transcript = True
        self._output_transcript_buffer = ""

    async def connect(self) -> None:
        """Establish a WebSocket connection and send session configuration."""
        url = f"{self.base_url}?model={self.model}"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "x-dashscope-dataInspection": "disable",
        }
        self.ws = await websockets.connect(url, additional_headers=headers)

        session_config = {
            "modalities": ["text", "audio"],
            "voice": self.voice,
            "instructions": self.instructions,
            "input_audio_format": "pcm",
            "output_audio_format": "pcm",
            "turn_detection": {},
            "tools": self.tools
        }

        if self.turn_detection_mode == TurnDetectionMode.MANUAL:
            session_config['turn_detection'] = None
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SERVER_VAD:
            session_config['turn_detection'] = {
                "type": "server_vad",
                "threshold": 0.1,
                "silence_duration_ms": 900
            }
            await self.update_session(session_config)
        elif self.turn_detection_mode == TurnDetectionMode.SEMANTIC_VAD:
            session_config['turn_detection'] = {
                "type": "smart_turn"
            }
            await self.update_session(session_config)
        else:
            raise ValueError(f"Invalid turn detection mode: {self.turn_detection_mode}")

    async def send_event(self, event) -> None:
        event['event_id'] = "event_" + str(int(time.time() * 1000))
        await self.ws.send(json.dumps(event))

    async def update_session(self, config: Dict[str, Any]) -> None:
        """Update session configuration."""
        event = {
            "type": "session.update",
            "session": config
        }
        await self.send_event(event)

    async def stream_audio(self, audio_chunk: bytes) -> None:
        """Stream raw audio data to the API."""
        # Only 16-bit 16 kHz mono PCM is supported
        audio_b64 = base64.b64encode(audio_chunk).decode()
        await self.send_event({
            "type": "input_audio_buffer.append",
            "audio": audio_b64
        })

    async def commit_audio_buffer(self) -> None:
        """Commit the audio buffer to trigger processing."""
        await self.send_event({"type": "input_audio_buffer.commit"})

    async def create_response(self) -> None:
        """Request the API to generate a response (call in manual mode or after returning function call results)."""
        await self.send_event({"type": "response.create"})

    async def cancel_response(self) -> None:
        """Cancel the current response."""
        await self.send_event({"type": "response.cancel"})

    async def handle_interruption(self):
        """Handle user interruption of the current response."""
        if not self._is_responding:
            return
        self._audio_suppressed = True
        if self._current_response_id:
            await self.cancel_response()
        self._is_responding = False
        self._current_response_id = None
        self._current_item_id = None

    @staticmethod
    def _format_event_for_log(event: Dict[str, Any]) -> str:
        """Format an event as JSON for logging. Redacts audio data for privacy."""
        event_type = event.get("type")
        if event_type == "response.audio.delta":
            delta = event.get("delta", "")
            redacted = dict(event)
            redacted["delta"] = f"<audio b64 omitted, length={len(delta)}>"
            return json.dumps(redacted, ensure_ascii=False)
        return json.dumps(event, ensure_ascii=False)

    async def _handle_function_call(self, event: Dict[str, Any]) -> None:
        """Handle a function call: parse arguments, execute the function, return the result, and trigger a follow-up inference."""
        call_id = event.get("call_id")
        name = event.get("name")
        arguments_str = event.get("arguments", "{}")

        print(f"[FunctionCall] Calling: {name}, call_id: {call_id}, args: {arguments_str}")

        try:
            arguments = json.loads(arguments_str) if arguments_str else {}
        except json.JSONDecodeError:
            arguments = {}

        func = self.functions.get(name)
        if func is None:
            output = json.dumps({"error": f"Unregistered function: {name}"})
        else:
            try:
                if asyncio.iscoroutinefunction(func):
                    result = await func(**arguments)
                else:
                    result = func(**arguments)
                output = str(result) if result is not None else ""
            except Exception as e:
                output = json.dumps({"error": str(e)})
                traceback.print_exc()

        # Return function_call_output
        await self.send_event({
            "type": "conversation.item.create",
            "item": {
                "type": "function_call_output",
                "call_id": call_id,
                "output": output,
            }
        })

        # Trigger follow-up inference
        await self.create_response()

    async def handle_messages(self) -> None:
        try:
            async for message in self.ws:
                event = json.loads(message)
                event_type = event.get("type")

                print(self._format_event_for_log(event))

                if event_type == "error":
                    continue
                elif event_type == "response.created":
                    self._current_response_id = event.get("response", {}).get("id")
                    self._is_responding = True
                    self._audio_suppressed = False
                elif event_type == "response.output_item.added":
                    self._current_item_id = event.get("item", {}).get("id")
                elif event_type == "response.done":
                    self._is_responding = False
                    self._current_response_id = None
                    self._current_item_id = None
                elif event_type == "input_audio_buffer.speech_started":
                    print("----------------Speech Started----------------")
                    if self.on_speech_started:
                        self.on_speech_started()
                    if self._is_responding:
                        await self.handle_interruption()
                elif event_type == "response.audio.delta":
                    if self._audio_suppressed:
                        continue
                    if self.on_audio_delta_b64:
                        self.on_audio_delta_b64(event["delta"])
                elif event_type == "response.function_call_arguments.done":
                    await self._handle_function_call(event)
                elif event_type in self.extra_event_handlers:
                    self.extra_event_handlers[event_type](event)
                elif event_type == "input_audio_buffer.speech_stopped":
                    print("----------------Speech Stopped----------------")
        except websockets.exceptions.ConnectionClosed:
            print(" Connection closed")
        except Exception as e:
            print(" Error in message handling: ", str(e))
            traceback.print_exc()

    async def close(self) -> None:
        """Close the WebSocket connection."""
        if self.ws:
            await self.ws.close()

def _audio_energy(audio_data: bytes) -> float:
    count = len(audio_data) // 2
    if count == 0:
        return 0.0
    samples = struct.unpack(f'<{count}h', audio_data)
    return sum(abs(s) for s in samples) / count

async def record_and_send(client, player, echo_suppression=True):
    p = pyaudio.PyAudio()
    stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
    print("Recording started. Speak into the microphone...")
    if echo_suppression:
        print("Note: Echo suppression is enabled (microphone is muted while the AI is speaking; interruption is not supported). If you are using headphones, set echo_suppression=False to enable interruption.")
    else:
        print("Note: Headphone mode. Voice interruption is supported.")
    playback_end_time = 0.0
    NOISE_GATE_THRESHOLD = 500
    try:
        while True:
            audio_data = await asyncio.to_thread(stream.read, 3200, False)
            if echo_suppression:
                is_active = client._is_responding or player.is_playing()
                if is_active:
                    playback_end_time = time.time()
                    await asyncio.sleep(0.02)
                    continue
                if time.time() - playback_end_time < 0.5:
                    await asyncio.sleep(0.02)
                    continue
            else:
                if client._is_responding or player.is_playing():
                    if _audio_energy(audio_data) < NOISE_GATE_THRESHOLD:
                        await asyncio.sleep(0.02)
                        continue
            await client.stream_audio(audio_data)
            await asyncio.sleep(0.02)
    finally:
        stream.stop_stream(); stream.close(); p.terminate()

async def main():
    pya = pyaudio.PyAudio()
    # Output sample rate 24 kHz, matching the server-side audio format
    player = B64PCMPlayer(pya, sample_rate=24000)

    client = FunRealtimeClient(
        # The following is the WebSocket URL for the China (Beijing) region. Replace {WorkspaceId} (including the curly braces) with your actual workspace ID. URLs vary by region.
        base_url="wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/realtime",
        api_key=os.environ['DASHSCOPE_API_KEY'],
        model="qwen-audio-3.0-realtime-plus",
        voice="longanqian",
        turn_detection_mode=TurnDetectionMode.SERVER_VAD,
        tools=tools,
        functions=functions,
        on_audio_delta_b64=player.add_data,
        # Clear playback buffer on voice interruption
        on_speech_started=player.cancel_playing,
    )

    await client.connect()
    print("Connected. Starting real-time conversation (Function Calling enabled)...")

    try:
        await asyncio.gather(client.handle_messages(), record_and_send(client, player, echo_suppression=False))
    finally:
        await client.close()
        player.shutdown()
        pya.terminate()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nProgram exited.")

Run python realtime_fc_demo.py and speak into your microphone to try real-time conversations with Function Calling. For example, ask "What's the weather in Hangzhou?" or "How much is a train ticket from Beijing to Shanghai?" and the model automatically invokes the corresponding tool and responds with the result.

Conversation context management

Qwen-Audio lets you manage conversation items in the context through client events. Use this to inject historical context, add text information, or remove irrelevant conversation items.

  • Create a conversation item (conversation.item.create): inserts a conversation item into the context. The following three item.type values are supported:

    • message: a regular conversation message. Specify role (system, user, or assistant) and a content array. Use this to inject conversation history or system instructions.

    • function_call: a function call request. Specify call_id, name, and arguments (JSON string). Typically generated by the server, but the client can also use this to inject historical function call records.

    • function_call_output: a tool execution result. Specify call_id and output (JSON string). After receiving a function_call, run the tool on the client and return the result with this type.

    The optional previous_item_id parameter specifies the existing conversation item after which to insert the new item. This lets you insert content at any position in the conversation history. If omitted, the new item is appended to the end.

    • Insert a user message at a specific position:

      {
    "type": "conversation.item.create",
    "previous_item_id": "item_abc",
    "item": {
        "type": "message",
        "role": "user",
        "content": [
            { "type": "input_text", "text": "Please summarize our last conversation" }
        ]
    }
}

    • Return a Function Calling result:

      {
    "type": "conversation.item.create",
    "item": {
        "type": "function_call_output",
        "call_id": "call_xxx",
        "output": "{\"temperature\":18,\"condition\":\"sunny\"}"
    }
}
    Note

    If the item.id specified in conversation.item.create already exists in the conversation, an error is returned.

  • Retrieve a conversation item (conversation.item.retrieve): queries a conversation item stored on the server. For audio-type content, only the transcription text is returned, not the raw audio data.

    {
    "type": "conversation.item.retrieve",
    "item_id": "item_xxx"
}

  • Delete a conversation item (conversation.item.delete): removes a specific item from the conversation context.

    {
    "type": "conversation.item.delete",
    "item_id": "item_xxx"
}

Ambient audio transcription

smart_turn mode only. When VAD detects speech activity but semantic analysis determines it isn't a valid turn (such as noise or filler sounds like "uh" or "hmm"), the server doesn't trigger a conversation turn. Instead, it sends the ASR result to the client as an ambient_audio_transcription event. This transcription isn't written to the conversation context.

{
    "type": "conversation.item.ambient_audio_transcription.delta",
    "item_id": "item_xxx",
    "text": "hmm",
    "stash": ""
}

Like user speech transcription events, ambient audio transcription includes delta and completed phases. Use this event to implement ambient audio monitoring or conversation scene awareness.

Speaker enhancement

smart_turn mode only. Pass pre-recorded audio URLs from the target user in session.update. The model will lock onto that speaker during duplex conversations, effectively ignoring other voices and background noise, enabling fluid duplex interactions in open environments.

Configuration: pass publicly accessible voiceprint audio URLs in turn_detection.voiceprint_audio_urls within the first session.update.

{
  "type": "session.update",
  "session": {
    "turn_detection": {
      "type": "smart_turn",
      "voiceprint_audio_urls": ["https://example.com/speaker.wav"]
    }
  }
}

Parameter requirements:

  • Up to 5 URLs. Audio must be 16 kHz PCM or WAV format.

  • This parameter only takes effect in the first session.update. The field is ignored in subsequent calls.

Registration events: after receiving the configuration, the server asynchronously performs voiceprint registration and notifies the result through the following events:

  • voiceprint_audio_list.in_progress: registration has started. Sent before session.updated, carrying item_id.

  • voiceprint_audio_list.completed: registration succeeded. The item_id matches the one in in_progress.

  • voiceprint_audio_list.failed: registration failed, with a reason field describing the error (for example, audio URL is not accessible). A registration failure does not block the ongoing conversation.

Going live

Set up fault tolerance

  • Client reconnection: implement automatic reconnection to handle network jitter. Set a reconnection signal in the on_error callback and use exponential backoff (for example, wait 1s, 2s, 4s) for retries.

  • Error classification: client errors (invalid_request_error) don't disconnect the session; log them or adjust parameters. Server errors (server_error) terminate the connection and require reconnection.

  • Interruption handling: in server_vad / smart_turn modes, new user speech automatically interrupts the model's ongoing response (response.done returns status=cancelled). When input_audio_buffer.speech_started is received, immediately clear the local playback buffer to avoid audio overlap.

Connection lifecycle

A typical WebSocket session follows this lifecycle:

  1. Connect: the client initiates a WebSocket connection and the server returns a session.created event.

  2. Configure: the client sends session.update to set the interaction mode, voice, tools, and other parameters. Complete this step before sending any audio.

  3. Interact: the client continuously streams audio (input_audio_buffer.append). The server performs inference based on VAD detection or manual triggers and returns speech and text in a streaming fashion.

  4. Close: the client closes the WebSocket connection. The server may also disconnect if the connection is idle for too long.

Latency optimization

  • Audio chunk size: send about 100 ms of audio data per chunk (16 kHz x 16 bit x mono = 3,200 bytes per chunk). This balances real-time performance with network efficiency.

  • Streaming playback: start playing audio as soon as response.audio.delta arrives. Don't wait for response.done to play the full response.

  • Clear buffer on interruption: when input_audio_buffer.speech_started is received, immediately clear the local playback buffer to prevent stale audio from continuing to play.

Supported models and regions

China (Beijing)

Use a Beijing region API key when calling the following models:

  • qwen-audio-3.0-realtime-plus

  • qwen-audio-3.0-realtime-flash

API reference

Previous：Non-real-time (Qwen-Omni)Next：Real-time audio and video translation - Qwen