### Install PyAudio for Python

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Instructions for installing the PyAudio library, a cross-platform audio I/O library, for Python. It covers macOS, Debian/Ubuntu, CentOS, and Windows, recommending pip installation and noting potential dependencies like portaudio.

```bash
brew install portaudio && pip install pyaudio
```

```bash
sudo apt-get install python3-pyaudio

或者

pip install pyaudio
```

```bash
sudo yum install -y portaudio portaudio-devel && pip install pyaudio
```

```bash
pip install pyaudio
```

--------------------------------

### Install Websockets for Python

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Installs the 'websockets' Python library version 15.0.1, which is required for establishing WebSocket connections.

```bash
pip install websockets==15.0.1
```

--------------------------------

### Main Method for OmniRealtimeConversation Setup (Java)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

The main method initializes OmniRealtimeParam, sets up a CountDownLatch for response synchronization, creates a RealtimePcmPlayer, and establishes an OmniRealtimeConversation with a callback handler. It demonstrates the initial setup for real-time conversation.

```java
public static void main(String[] args) throws InterruptedException, LineUnavailableException {
        OmniRealtimeParam param = OmniRealtimeParam.builder()
                .model("qwen3-omni-flash-realtime")
                // .apikey("your-dashscope-api-key")
                .build();
        AtomicReference<CountDownLatch> responseDoneLatch = new AtomicReference<>(null);
        responseDoneLatch.set(new CountDownLatch(1));

        RealtimePcmPlayer audioPlayer = new RealtimePcmPlayer(24000);
        final AtomicReference<OmniRealtimeConversation> conversationRef = new AtomicReference<>(null);
        OmniRealtimeConversation conversation = new OmniRealtimeConversation(param, new OmniRealtimeCallback() {
            @Override
            public void onOpen() {
                System.out.println("connection opened");
            }
            @Override
            public void onEvent(JsonObject message) {
                String type = message.get("type").getAsString();
                switch(type) {
                    case "session.created":
                        System.out.println("start session: " + message.get("session").getAsJsonObject().get("id").getAsString());
                        break;
                    case "conversation.item.input_audio_transcription.completed":

```

--------------------------------

### Manual Dashscope Initialization (Python)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

This snippet shows the necessary imports and basic setup for using the dashscope library, specifically for audio-related functionalities. It includes importing os, base64, sys, threading, pyaudio, and components from dashscope.audio.qwen_omni. This is a prerequisite for more complex interactions.

```python
# 依赖：dashscope >= 1.23.9，pyaudio。
import os
import base64
import sys
import threading
import pyaudio
from dashscope.audio.qwen_omni import *
import dashscope
```

--------------------------------

### Install DashScope and WebSocket Client (Shell)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Installs the necessary Python libraries, `websocket-client` and `dashscope`, for integrating with the Realtime model via WebSocket.

```shell
pip install websocket-client dashscope
```

--------------------------------

### Install PyAudio Dependencies (Shell)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Commands to install the PyAudio library and its dependencies on macOS, Debian/Ubuntu, CentOS, and Windows. This is required for audio input processing in the Realtime model integration.

```shell
# macOS
brew install portaudio && pip install pyaudio
```

```shell
# Debian/Ubuntu (system-wide)
sudo apt-get install python3-pyaudio
```

```shell
# Debian/Ubuntu (virtual environment)
sudo apt update
sudo apt install -y python3-dev portaudio19-dev
pip install pyaudio
```

```shell
# CentOS
sudo yum install -y portaudio portaudio-devel && pip install pyaudio
```

```shell
# Windows (assuming Python and pip are in PATH)
pip install pyaudio
```

--------------------------------

### Realtime Model Client Setup and Connection (Python)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

This section shows how to initialize and connect the OmniRealtimeClient for real-time interaction with the model. It configures the client with a base URL, API key, model name, voice, instructions, and turn detection mode. It also sets up callbacks for text and audio deltas and establishes a connection to the server.

```python
async def main():
    p = pyaudio.PyAudio()
    player = AudioPlayer(pyaudio_instance=p)

    client = OmniRealtimeClient(
        # 以下是中国内地（北京）地域 base_url，国际（新加坡）地域base_url为 wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime
        base_url="wss://dashscope.aliyuncs.com/api-ws/v1/realtime",
        api_key=os.environ.get("DASHSCOPE_API_KEY"),
        model="qwen3-omni-flash-realtime",
        voice="Cherry",
        instructions="你是小云，风趣幽默的好助手",
        turn_detection_mode=TurnDetectionMode.SERVER_VAD,
        on_text_delta=lambda t: print(f"\nAssistant: {t}", end="", flush=True),
        on_audio_delta=player.add_audio,
    )

    await client.connect()
    print("连接成功，开始实时对话...")

    # 并发运行
    await asyncio.gather(client.handle_messages(), record_and_send(client))

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n程序已退出。")
```

--------------------------------

### Realtime Conversation Setup and Audio Input (Java)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Configures a real-time conversation with the model, enabling turn detection and input audio transcription. It then captures audio from the microphone, encodes it to Base64, and sends it to the conversation. Handles exceptions and gracefully closes resources.

```java
Conversation conversation = Conversation.builder()
            .model(
```

```java
.enableTurnDetection(true)
            .enableInputAudioTranscription(true)
            .parameters(Map.of("instructions",
                    "你是五星级酒店的AI客服专员，请准确且友好地解答客户关于房型、设施、价格、预订政策的咨询。请始终以专业和乐于助人的态度回应，杜绝提供未经证实或超出酒店服务范围的信息。"))
            .build();

            System.out.println("请开始说话（自动检测语音开始/结束，按Ctrl+C退出）...");
            AudioFormat format = new AudioFormat(16000, 16, 1, true, false);
            TargetDataLine mic = AudioSystem.getTargetDataLine(format);
            mic.open(format);
            mic.start();

            ByteBuffer buffer = ByteBuffer.allocate(3200);
            while (!shouldStop.get()) {
                int bytesRead = mic.read(buffer.array(), 0, buffer.capacity());
                if (bytesRead > 0) {
                    try {
                        conversation.appendAudio(Base64.getEncoder().encodeToString(buffer.array()));
                    } catch (Exception e) {
                        if (e.getMessage() != null && e.getMessage().contains("closed")) {
                            System.out.println("对话已关闭，停止录音");
                            break;
                        }
                    }
                }
                Thread.sleep(20);
            }

            conversation.close(1000, "正常结束");
            player.close();
            mic.close();
            System.out.println("\n程序已退出");

        } catch (NoApiKeyException e) {
            System.err.println("未找到API KEY: 请设置环境变量 DASHSCOPE_API_KEY");
            System.exit(1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void handleEvent(JsonObject event, SequentialAudioPlayer player, AtomicBoolean userIsSpeaking) {
        String type = event.get("type").getAsString();
        switch (type) {
            case "input_audio_buffer.speech_started":
                System.out.println("\n[用户开始说话]");
                player.cancel();
                userIsSpeaking.set(true);
                break;
            case "input_audio_buffer.speech_stopped":
                System.out.println("[用户停止说话]");
                userIsSpeaking.set(false);
                break;
            case "response.audio.delta":
                if (!userIsSpeaking.get()) {
                    player.play(event.get("delta").getAsString());
                }
                break;
            case "conversation.item.input_audio_transcription.completed":
                System.out.println("用户: " + event.get("transcript").getAsString());
                break;
            case "response.audio_transcript.delta":
                System.out.print(event.get("delta").getAsString());
                break;
            case "response.done":
                System.out.println("回复完成");
                break;
        }
    }
}
```

--------------------------------

### Realtime Client Connection and Message Handling

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Establishes a connection to the realtime client, starts a message handler task, and manages the connection lifecycle. It includes error handling for connection issues and ensures resources are cleaned up upon program termination.

```python
voice="Ethan",
        instructions="你是个人助理小云，请你准确且友好地解答用户的问题，始终以乐于助人的态度回应。", # 设定模型角色
        on_text_delta=lambda text: print(f"助手回复: {text}", end="", flush=True),
        on_audio_delta=on_audio_received,
        turn_detection_mode=TurnDetectionMode.MANUAL,
        extra_event_handlers={"response.done": on_response_done}
    )

    message_handler_task = None
    try:
        await realtime_client.connect()
        print("已连接到服务器。输入 'q' 或 'quit' 可随时退出程序。")
        message_handler_task = asyncio.create_task(realtime_client.handle_messages())
        await asyncio.sleep(0.5)

```

--------------------------------

### Create OmniRealtimeClient in Python

Source: https://help.aliyun.com/zh/model-studio/realtime/index

This Python code defines the `OmniRealtimeClient` class for interacting with Aliyun's realtime services. It includes setup for WebSocket connections, handling different turn detection modes, and callback functions for text and audio deltas, as well as input transcripts.

```python
import asyncio
import websockets
import json
import base64
import time
from typing import Optional, Callable, List, Dict, Any
from enum import Enum

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

class OmniRealtimeClient:

    def __init__(
            self,
            base_url,
            api_key: str,
            model: str = "",
            voice: str = "Ethan",
            instructions: str = "You are a helpful assistant.",
            turn_detection_mode: TurnDetectionMode = TurnDetectionMode.SERVER_VAD,
            on_text_delta: Optional[Callable[[str], None]] = None,
            on_audio_delta: Optional[Callable[[bytes], None]] = None,
            on_input_transcript: Optional[Callable[[str], None]] = None,

```

--------------------------------

### Python Realtime Audio Player Class

Source: https://help.aliyun.com/zh/model-studio/realtime/index

The `AudioPlayer` class manages real-time audio playback. It initializes PyAudio, opens an audio stream, and uses a separate thread to play audio data from a queue. It includes methods to start, stop, add audio data, and check playback status, with thread-safe access using locks. Dependencies include `os`, `asyncio`, `time`, `threading`, `queue`, `pyaudio`, and `omni_realtime_client`.

```python
# -- coding: utf-8 --
import os
import asyncio
import time
import threading
import queue
import pyaudio
from omni_realtime_client import OmniRealtimeClient, TurnDetectionMode

class AudioPlayer:
    """实时音频播放器类"""
    def __init__(self, sample_rate=24000, channels=1, sample_width=2):
        self.sample_rate = sample_rate
        self.channels = channels
        self.sample_width = sample_width  # 2 bytes for 16-bit
        self.audio_queue = queue.Queue()
        self.is_playing = False
        self.play_thread = None
        self.pyaudio_instance = None
        self.stream = None
        self._lock = threading.Lock()  # 添加锁来同步访问
        self._last_data_time = time.time()  # 记录最后接收数据的时间
        self._response_done = False  # 添加响应完成标志
        self._waiting_for_response = False # 标记是否正在等待服务器响应
        # 记录最后一次向音频流写入数据的时间及最近一次音频块的时长，用于更精确地判断播放结束
        self._last_play_time = time.time()
        self._last_chunk_duration = 0.0
        
    def start(self):
        """启动音频播放器"""
        with self._lock:
            if self.is_playing:
                return
                
            self.is_playing = True
            
            try:
                self.pyaudio_instance = pyaudio.PyAudio()
                
                # 创建音频输出流
                self.stream = self.pyaudio_instance.open(
                    format=pyaudio.paInt16,  # 16-bit
                    channels=self.channels,
                    rate=self.sample_rate,
                    output=True,
                    frames_per_buffer=1024
                )
                
                # 启动播放线程
                self.play_thread = threading.Thread(target=self._play_audio)
                self.play_thread.daemon = True
                self.play_thread.start()
                
                print("音频播放器已启动")
            except Exception as e:
                print(f"启动音频播放器失败: {e}")
                self._cleanup_resources()
                raise
    
    def stop(self):
        """停止音频播放器"""
        with self._lock:
            if not self.is_playing:
                return
                
            self.is_playing = False
        
        # 清空队列
        while not self.audio_queue.empty():
            try:
                self.audio_queue.get_nowait()
            except queue.Empty:
                break
        
        # 等待播放线程结束（在锁外面等待，避免死锁）
        if self.play_thread and self.play_thread.is_alive():
            self.play_thread.join(timeout=2.0)
        
        # 再次获取锁来清理资源
        with self._lock:
            self._cleanup_resources()
        
        print("音频播放器已停止")
    
    def _cleanup_resources(self):
        """清理音频资源（必须在锁内调用）"""
        try:
            # 关闭音频流
            if self.stream:
                if not self.stream.is_stopped():
                    self.stream.stop_stream()
                self.stream.close()
                self.stream = None
        except Exception as e:
            print(f"关闭音频流时出错: {e}")
        
        try:
            if self.pyaudio_instance:
                self.pyaudio_instance.terminate()
                self.pyaudio_instance = None
        except Exception as e:
            print(f"终止PyAudio时出错: {e}")
    
    def add_audio_data(self, audio_data):
        """添加音频数据到播放队列"""
        if self.is_playing and audio_data:
            self.audio_queue.put(audio_data)
            with self._lock:
                self._last_data_time = time.time()  # 更新最后接收数据的时间
                self._waiting_for_response = False # 收到数据，不再等待
    
    def stop_receiving_data(self):
        """标记不再接收新的音频数据"""
        with self._lock:
            self._response_done = True
            self._waiting_for_response = False # 响应结束，不再等待
    
    def prepare_for_next_turn(self):
        """为下一轮对话重置播放器状态。"""
        with self._lock:
            self._response_done = False
            self._last_data_time = time.time()
            self._last_play_time = time.time()
            self._last_chunk_duration = 0.0
            self._waiting_for_response = True # 开始等待下一轮响应
        
        # 清空上一轮可能残留的音频数据
        while not self.audio_queue.empty():
            try:
                self.audio_queue.get_nowait()
            except queue.Empty:
                break

    def is_finished_playing(self):
        """检查是否已经播放完所有音频数据"""
        with self._lock:
            queue_size = self.audio_queue.qsize()
            time_since_last_data = time.time() - self._last_data_time
            time_since_last_play = time.time() - self._last_play_time
            
            # ---------------------- 智能结束判定 ----------------------
            # 1. 首选：如果服务器已标记完成且播放队列为空
            #    进一步等待最近一块音频播放完毕（音频块时长 + 0.1s 容错）。
            if self._response_done and queue_size == 0:
                min_wait = max(self._last_chunk_duration + 0.1, 0.5)  # 至少等待 0.5s
                if time_since_last_play >= min_wait:
                    return True

            # 2. 备用：如果长时间没有新数据且播放队列为空
            #    当服务器没有明确发出 `response.done` 时，此逻辑作为保障
            #    （此部分逻辑未在提供的代码中完整实现，仅为注释说明）
            # if not self._response_done and queue_size == 0 and time_since_last_data > 5.0: # 示例：超过5秒无数据
            #     return True

            return False

    def _play_audio(self):
        """音频播放线程的入口函数"""
        while self.is_playing:
            try:
                # 尝试从队列获取音频数据
                # 设置一个短的超时时间，以便能及时响应 stop() 调用
                audio_data = self.audio_queue.get(timeout=0.1)
                
                if audio_data:
                    # 记录播放时间，用于判断播放结束
                    start_play_time = time.time()
                    self.stream.write(audio_data)
                    end_play_time = time.time()
                    
                    with self._lock:
                        self._last_play_time = end_play_time
                        # 估算当前音频块的时长
                        # 假设 audio_data 的长度是字节数，需要转换为秒
                        bytes_per_second = self.sample_rate * self.channels * self.sample_width
                        self._last_chunk_duration = len(audio_data) / bytes_per_second
                        
                self.audio_queue.task_done() # 标记任务完成

            except queue.Empty:
                # 队列为空，继续循环，检查 is_playing 状态
                # 如果 is_finished_playing() 返回 True，则可以考虑退出循环
                if self.is_finished_playing():
                    break
                continue
            except Exception as e:
                print(f"播放音频时出错: {e}")
                # 发生错误时，尝试停止播放并清理资源
                self.stop()
                break
        
        print("音频播放线程结束")

```

--------------------------------

### RealtimePcmPlayer Class for Real-time Audio Playback (Java)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

The RealtimePcmPlayer class manages the playback of raw audio data in real-time. It utilizes separate threads for decoding and playing audio chunks, handling buffering, and managing the audio line. It supports starting, stopping, and shutting down the playback process.

```java
class RealtimePcmPlayer {
        private final AtomicBoolean stopped = new AtomicBoolean(false);
        private Thread decoderThread;
        private Thread playerThread;
        private AudioTrack track;
        private int sampleRate;
        private BlockingQueue<String> b64AudioBuffer = new LinkedBlockingQueue<>();
        private static final BlockingQueue<byte[]> RawAudioBuffer = new LinkedBlockingQueue<>();

        public RealtimePcmPlayer(int sampleRate) {
            this.sampleRate = sampleRate;
            decoderThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (!stopped.get()) {
                        String b64Audio = b64AudioBuffer.poll();
                        if (b64Audio != null) {
                            try {
                                byte[] rawAudio = Base64.getDecoder().decode(b64Audio);
                                RawAudioBuffer.put(rawAudio);
                            } catch (InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                        } else {
                            try {
                                Thread.sleep(100);
                            } catch (InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                        }
                    }
                }
            });
            playerThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (!stopped.get()) {
                        byte[] rawAudio = RawAudioBuffer.poll();
                        if (rawAudio != null) {
                            try {
                                playChunk(rawAudio);
                            } catch (IOException e) {
                                throw new RuntimeException(e);
                            } catch (InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                        } else {
                            try {
                                Thread.sleep(100);
                            } catch (InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                        }
                    }
                }
            });
            decoderThread.start();
            playerThread.start();
        }

        // 播放一个音频块并阻塞直到播放完成
        private void playChunk(byte[] chunk) throws IOException, InterruptedException {
            if (chunk == null || chunk.length == 0) return;

            int bytesWritten = 0;
            while (bytesWritten < chunk.length) {
                bytesWritten += track.write(chunk, bytesWritten, chunk.length - bytesWritten);
            }
            int audioLength = chunk.length / (this.sampleRate*2/1000);
            // 等待缓冲区中的音频播放完成
            Thread.sleep(audioLength - 10);
        }

        public void write(String b64Audio) {
            b64AudioBuffer.add(b64Audio);
        }

        public void cancel() {
            b64AudioBuffer.clear();
            RawAudioBuffer.clear();
        }

        public void waitForComplete() throws InterruptedException {
            while (!b64AudioBuffer.isEmpty() || !RawAudioBuffer.isEmpty()) {
                Thread.sleep(100);
            }
            track.drain();
        }

        public void shutdown() throws InterruptedException {
            stopped.set(true);
            decoderThread.join();
            playerThread.join();
            if (track != null && track.isRunning()) {
                track.drain();
                track.close();
            }
        }
    } // RealtimePcmPlayer 类定义结束
```

--------------------------------

### VAD Mode Interaction Flow

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Describes the interaction flow for the VAD (Voice Activity Detection) mode, where the server automatically detects speech start and end.

```APIDOC
## VAD Mode Interaction Flow

### Description
This mode enables server-side automatic detection of speech start and end, suitable for voice call scenarios. The `session.turn_detection` is set to `"server_vad"`.

### Interaction Steps:
1. Server detects speech start: `input_audio_buffer.speech_started` event.
2. Client sends audio and image data: `input_audio_buffer.append` and `input_image_buffer.append` events. Note: `input_image_buffer.append` requires a prior `input_audio_buffer.append`.
3. Server detects speech end: `input_audio_buffer.speech_stopped` event.
4. Server commits audio buffer: `input_audio_buffer.committed` event.
5. Server creates user message item: `conversation.item.created` event.

### Event Lifecycle:
**Client Events** | **Server Events**
---|---
`session.update` | `session.created`, `session.updated`
`input_audio_buffer.append`, `input_image_buffer.append` | `input_audio_buffer.speech_started`, `input_audio_buffer.speech_stopped`, `input_audio_buffer.committed`
 | `response.created`, `response.output_item.added`, `conversation.item.created`, `response.content_part.added`, `response.audio_transcript.delta`, `response.audio.delta`, `response.audio_transcript.done`, `response.audio.done`, `response.content_part.done`, `response.output_item.done`, `response.done`
```

--------------------------------

### Initialize and Configure Realtime Conversation

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Sets up the realtime conversation with specified model, callback, and URL. It then configures session parameters including output modalities, voice, audio formats, and transcription settings.

```python
import os
import sys
import base64
import pyaudio
import threading

from http import HTTPStatus
from dashscope.realtime.conversation.api import OmniRealtimeConversation
from dashscope.realtime.conversation.callback import OmniRealtimeCallback
from dashscope.realtime.conversation.defs import MultiModality, AudioFormat

# If the environment variable is not set, replace the following line with your API Key
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
voice = 'Cherry'

class MyCallback(OmniRealtimeCallback):
    """Simplest callback: initialize speaker on connection, play returned audio directly in events."""
    def __init__(self, ctx):
        super().__init__()
        self.ctx = ctx

    def on_open(self) -> None:
        # Initialize PyAudio and speaker after connection is established (24k/mono/16bit)
        print('connection opened')
        try:
            self.ctx['pya'] = pyaudio.PyAudio()
            self.ctx['out'] = self.ctx['pya'].open(
                format=pyaudio.paInt16,
                channels=1,
                rate=24000,
                output=True
            )
            print('audio output initialized')
        except Exception as e:
            print('[Error] audio init failed: {}'.format(e))

    def on_close(self, close_status_code, close_msg) -> None:
        print('connection closed with code: {}, msg: {}'.format(close_status_code, close_msg))
        sys.exit(0)

    def on_event(self, response: str) -> None:
        try:
            t = response['type']
            handlers = {
                'session.created': lambda r: print('start session: {}'.format(r['session']['id'])),
                'conversation.item.input_audio_transcription.completed': lambda r: print('question: {}'.format(r['transcript'])),
                'response.audio_transcript.delta': lambda r: print('llm text: {}'.format(r['delta'])),
                'response.audio.delta': self._play_audio,
                'response.done': self._response_done,
            }
            h = handlers.get(t)
            if h:
                h(response)
        except Exception as e:
            print('[Error] {}'.format(e))

    def _play_audio(self, response):
        # Decode base64 directly and write to the output stream for playback
        if self.ctx['out'] is None:
            return
        try:
            data = base64.b64decode(response['delta'])
            self.ctx['out'].write(data)
        except Exception as e:
            print('[Error] audio playback failed: {}'.format(e))

    def _response_done(self, response):
        # Mark the end of the current conversation turn, used in the main loop for waiting
        if self.ctx['conv'] is not None:
            print('[Metric] response: {}, first text delay: {}, first audio delay: {}'.format(
                self.ctx['conv'].get_last_response_id(),
                self.ctx['conv'].get_last_first_text_delay(),
                self.ctx['conv'].get_last_first_audio_delay(),
            ))
        if self.ctx['resp_done'] is not None:
            self.ctx['resp_done'].set()

def shutdown_ctx(ctx):
    """Safely release audio and PyAudio resources."""
    try:
        if ctx['out'] is not None:
            ctx['out'].close()
            ctx['out'] = None
    except Exception:
        pass
    try:
        if ctx['pya'] is not None:
            ctx['pya'].terminate()
            ctx['pya'] = None
    except Exception:
        pass

def record_until_enter(pya_inst: pyaudio.PyAudio, sample_rate=16000, chunk_size=3200):
    """Record until Enter is pressed, return PCM bytes."""
    frames = []
    stop_evt = threading.Event()

    stream = pya_inst.open(
        format=pyaudio.paInt16,
        channels=1,
        rate=sample_rate,
        input=True,
        frames_per_buffer=chunk_size
    )

    def _reader():
        while not stop_evt.is_set():
            try:
                frames.append(stream.read(chunk_size, exception_on_overflow=False))
            except Exception:
                break

    t = threading.Thread(target=_reader, daemon=True)
    t.start()
    input()  # User presses Enter again to stop recording
    stop_evt.set()
    t.join(timeout=1.0)
    try:
        stream.close()
    except Exception:
        pass
    return b''.join(frames)


if __name__  == '__main__':
    print('Initializing ...')
    # Runtime context: stores audio and session handles
    ctx = {'pya': None, 'out': None, 'conv': None, 'resp_done': threading.Event()}
    callback = MyCallback(ctx)
    conversation = OmniRealtimeConversation(
        model='qwen3-omni-flash-realtime',
        callback=callback,
        # URL for Beijing region; for Singapore region models, replace with: wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime
        url="wss://dashscope.aliyuncs.com/api-ws/v1/realtime",
    )
    try:
        conversation.connect()
    except Exception as e:
        print('[Error] connect failed: {}'.format(e))
        sys.exit(1)

    ctx['conv'] = conversation
    # Session configuration: enable text + audio output (disable server-side VAD, use manual recording)
    conversation.update_session(
        output_modalities=[MultiModality.AUDIO, MultiModality.TEXT],
        voice=voice,
        input_audio_format=AudioFormat.PCM_16000HZ_MONO_16BIT,
        output_audio_format=AudioFormat.PCM_24000HZ_MONO_16BIT,
        enable_input_audio_transcription=True,
        # Model for transcribing input audio, only gummy-realtime-v1 is supported
        input_audio_transcription_model='gummy-realtime-v1',
        enable_turn_detection=False,
        instructions="You are XiaoYun, a personal assistant. Please accurately and friendly answer user questions, always responding with a helpful attitude."
    )

    try:
        turn = 1
        while True:
            print(f"\n--- Turn {turn} ---")
            print("Press Enter to start recording (type q and Enter to exit)...")

```

--------------------------------

### Commit Audio Buffer to Realtime API (Python)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Submits the audio buffer to the Realtime API for processing. This action triggers the API to start processing the accumulated audio data. It's typically called after streaming audio chunks to signal completion of a segment.

```python
async def commit_audio_buffer(self) -> None:
        """提交音频缓冲区以触发处理。"""
        event = {
            "type": "input_audio_buffer.commit"
        }
        await self.send_event(event)
```

--------------------------------

### Interactive Realtime Dialogue Client - Python

Source: https://help.aliyun.com/zh/model-studio/realtime/index

An asynchronous function demonstrating an interactive client for real-time multimodal dialogue. It initializes an audio player, sets up callbacks for audio reception and response completion, and connects to the Aliyun real-time API. Dependencies include 'os', 'pyaudio', 'threading', and the 'OmniRealtimeClient' class.

```python
async def interactive_test():
    """
    交互式测试脚本：允许多轮连续对话，每轮可以发送音频和图片。
    """
    # ------------------- 1. 初始化和连接 (一次性) -------------------
    api_key = os.environ.get("DASHSCOPE_API_KEY")
    if not api_key:
        print("请设置DASHSCOPE_API_KEY环境变量")
        return

    print("--- 实时多轮音视频对话客户端 ---")
    print("正在初始化音频播放器和客户端...")
    
    audio_player = AudioPlayer()
    audio_player.start()

    def on_audio_received(audio_data):
        audio_player.add_audio_data(audio_data)

    def on_response_done(event):
        print("\n(收到响应结束标记)")
        audio_player.stop_receiving_data()

    realtime_client = OmniRealtimeClient(
        base_url="wss://dashscope.aliyuncs.com/api-ws/v1/realtime",
        api_key=api_key,
        model="qwen3-omni-flash-realtime",
```

--------------------------------

### Establish WebSocket Connection for Qwen-Omni-Realtime using DashScope SDK (Java)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

This Java code demonstrates establishing a real-time conversation with the Qwen-Omni-Realtime model using the DashScope SDK. It implements a callback interface to handle connection events. The code requires the DashScope SDK for Java (version >= 2.20.9) and an API key for authentication.

```java
// SDK 版本不低于 2.20.9
import com.alibaba.dashscope.audio.omni.*;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.google.gson.JsonObject;
import java.util.concurrent.CountDownLatch;

public class Main {
    public static void main(String[] args) throws InterruptedException, NoApiKeyException {
        CountDownLatch latch = new CountDownLatch(1);
        OmniRealtimeParam param = OmniRealtimeParam.builder()
                .model("qwen3-omni-flash-realtime")
                .apikey(System.getenv("DASHSCOPE_API_KEY"))
                // 以下为北京地域url，若使用新加坡地域的模型，需将url替换为：wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime
                .url("wss://dashscope.aliyuncs.com/api-ws/v1/realtime")
                .build();

        OmniRealtimeConversation conversation = new OmniRealtimeConversation(param, new OmniRealtimeCallback() {
            @Override
            public void onOpen() {
                System.out.println("Connected Successfully");
            }
            @Override
            public void onEvent(JsonObject message) {
                System.out.println(message);
            }
            @Override
            public void onClose(int code, String reason) {
                System.out.println("connection closed code: " + code + ", reason: " + reason);
                latch.countDown();
            }
        });
        conversation.connect();
        latch.await();
        conversation.close(1000, "bye");
        System.exit(0);
    }
}

```

--------------------------------

### Establish WebSocket Connection for Qwen-Omni-Realtime (Python)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

This Python code demonstrates how to establish a WebSocket connection to the Qwen-Omni-Realtime model using the 'websocket-client' library. It requires an API key for authentication and specifies the model and server URL. The code handles connection opening, message reception, and error events.

```python
# pip install websocket-client
import json
import websocket
import os

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=qwen3-omni-flash-realtime"

headers = [
    "Authorization: Bearer " + API_KEY
]

def on_open(ws):
    print(f"Connected to server: {API_URL}")
def on_message(ws, message):
    data = json.loads(message)
    print("Received event:", json.dumps(data, indent=2))
def on_error(ws, error):
    print("Error:", error)

ws = websocket.WebSocketApp(
    API_URL,
    header=headers,
    on_open=on_open,
    on_message=on_message,
    on_error=on_error
)

ws.run_forever()

```

--------------------------------

### Real-time Conversation with Qwen Omni (Python)

Source: https://help.aliyun.com/zh/model-studio/realtime/index

This snippet demonstrates a full real-time conversation flow with the Qwen Omni model. It includes initializing audio devices, setting up a callback for audio and text events, connecting to the model, and a main loop for processing microphone input. Dependencies include dashscope and pyaudio.

```python
# 配置 API Key，若没有设置环境变量，请用 API Key 将下行替换为 dashscope.api_key = "sk-xxx"
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
# 指定音色
voice = 'Cherry'
# 指定模型
model = 'qwen3-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_event(self, response):
        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对话结束")
```

--------------------------------

### Configure Session

Source: https://help.aliyun.com/zh/model-studio/realtime/index

Send the `session.update` event to configure the session parameters, including output modalities, voice settings, audio formats, instructions, and turn detection.

```APIDOC
## POST /session/update

### Description
Configures the session parameters for the Realtime Model Studio.

### Method
POST

### Endpoint
/session/update

### Parameters
#### Request Body
- **event_id** (string) - Required - The unique ID generated by the client for this event.
- **type** (string) - Required - Must be set to `session.update`.
- **session** (object) - Required - Contains the session configuration.
  - **modalities** (array of strings) - Required - Specifies the output modalities. Supports `["text"]` or `["text", "audio"]`.
  - **voice** (string) - Optional - The voice to be used for audio output.
  - **input_audio_format** (string) - Required - The format of the input audio. Only `pcm16` is supported.
  - **output_audio_format** (string) - Required - The format of the output audio. Supported formats depend on the model (e.g., `pcm24` for Qwen3-Omni-Flash-Realtime, `pcm16` for Qwen-Omni-Turbo-Realtime).
  - **instructions** (string) - Optional - System messages to set the model's objective or role.
  - **turn_detection** (object or null) - Optional - Configuration for voice activity detection (VAD). If null, the client controls response initiation. 
    - **type** (string) - Required if `turn_detection` is an object - Must be set to `server_vad`.
    - **threshold** (number) - Optional - VAD detection threshold. Recommended to adjust based on environment noise.
    - **silence_duration_ms** (integer) - Optional - Duration of silence to trigger a model response.

### Request Example
```json
{
    "event_id": "event_ToPZqeobitzUJnt3QqtWg",
    "type": "session.update",
    "session": {
        "modalities": [
            "text",
            "audio"
        ],
        "voice": "Cherry",
        "input_audio_format": "pcm16",
        "output_audio_format": "pcm24",
        "instructions": "你是某五星级酒店的AI客服专员，请准确且友好地解答客户关于房型、设施、价格、预订政策的咨询。请始终以专业和乐于助人的态度回应，杜绝提供未经证实或超出酒店服务范围的信息。",
        "turn_detection": {
            "type": "server_vad",
            "threshold": 0.5,
            "silence_duration_ms": 800
        }
    }
}
```

### Response
#### Success Response (200)
- **status** (string) - Indicates the success of the operation.

#### Response Example
```json
{
  "status": "success"
}
```
```

--------------------------------

### Handle WebSocket Messages in Java

Source: https://help.aliyun.com/zh/model-studio/realtime/index

This Java code snippet demonstrates how to handle incoming WebSocket messages. It processes different message types like 'question', 'response.audio_transcript.delta', 'response.audio.delta', and 'response.done', printing relevant information or playing received audio data. It also includes logic for managing conversation state and metrics.

```java
System.out.println("question: " + message.get("transcript").getAsString());
                        break;
                    case "response.audio_transcript.delta":
                        System.out.println("got llm response delta: " + message.get("delta").getAsString());
                        break;
                    case "response.audio.delta":
                        String recvAudioB64 = message.get("delta").getAsString();
                        audioPlayer.write(recvAudioB64);
                        break;
                    case "response.done":
                        System.out.println("======RESPONSE DONE======");
                        if (conversationRef.get() != null) {
                            System.out.println("[Metric] response: " + conversationRef.get().getResponseId() +
                                    ", first text delay: " + conversationRef.get().getFirstTextDelay() +
                                    " ms, first audio delay: " + conversationRef.get().getFirstAudioDelay() + " ms");
                        }
                        responseDoneLatch.get().countDown();
                        break;
                    default:
                        break;
                }
            }
            @Override
            public void onClose(int code, String reason) {
                System.out.println("connection closed code: " + code + ", reason: " + reason);
            }
        });
        conversationRef.set(conversation);
        try {
            conversation.connect();
        } catch (NoApiKeyException e) {
            throw new RuntimeException(e);
        }
        OmniRealtimeConfig config = OmniRealtimeConfig.builder()
                .modalities(Arrays.asList(OmniRealtimeModality.AUDIO, OmniRealtimeModality.TEXT))
                .voice("Cherry")
                .enableTurnDetection(false)
                // 设定模型角色
                .parameters(new HashMap<String, Object>() {{ 
                    put("instructions","你是个人助理小云，请你准确且友好地解答用户的问题，始终以乐于助人的态度回应。");
                }})
                .build();
        conversation.updateSession(config);

        // 新增麦克风录音功能
        AudioFormat format = new AudioFormat(16000, 16, 1, true, false);
        DataLine.Info info = new DataLine.Info(TargetDataLine.class, format);
        
        if (!AudioSystem.isLineSupported(info)) {
            System.out.println("Line not supported");
            return;
        }
        
        TargetDataLine line = null;
        try {
            line = (TargetDataLine) AudioSystem.getLine(info);
            line.open(format);
            line.start();
            
            while (true) {
                System.out.println("按Enter开始录音...");
                try {
                    System.in.read();
                } catch (IOException e) {
                    System.err.println("读取输入时发生错误: " + e.getMessage());
                    break; // 发生错误时退出循环
                }
                System.out.println("开始录音，请说话...再次按Enter停止录音并发送");
                recordAndSend(line, conversation);
                conversation.commit();
                conversation.createResponse(null, null);
                // 重置latch以便下次等待
                responseDoneLatch.set(new CountDownLatch(1));
            }
        } catch (LineUnavailableException | IOException e) {
            e.printStackTrace();
        } finally {
            if (line != null) {
                line.stop();
                line.close();
            }
        }
```