Alibaba Cloud Model Studio:Qwen-Omni-Realtime

DashScope Python SDK

• 実行環境

Python 3.10 以降がインストールされていることを確認してください。

オペレーティングシステム用の PyAudio をインストールします。

brew install portaudio && pip install pyaudio

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

pip install pyaudio

その他の依存関係をインストールします。

pip install websocket-client dashscope

• インタラクションモード

  • VAD モード (音声区間検出、発話の開始と終了を自動検出)

    ユーザーの発話終了を検出した後、サーバーが応答します。

  • 手動モード (押して話す、離して送信)

    クライアントが発話の開始と終了を制御します。発話後、アプリケーションがサーバーに通知する必要があります。

  VAD モード

  vad_dash.py という名前の Python ファイルを作成し、以下のコードをコピーします。

  vad_dash.py

  # Dependencies: dashscope >= 1.23.9, pyaudio
  import os
  import base64
  import time
  import pyaudio
  from dashscope.audio.qwen_omni import MultiModality, AudioFormat,OmniRealtimeCallback,OmniRealtimeConversation
  import dashscope
  
  # Configuration: URL, API key, voice, model, model role
  # Specify the region. 'intl' for Singapore region, 'cn' for China (Beijing) region.
  region = 'intl'
  base_domain = '{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com' if region == 'intl' else 'dashscope.aliyuncs.com'
  url = f'wss://{base_domain}/api-ws/v1/realtime'
  # Configure the API key. If the environment variable is not set, replace the following line with your API key: dashscope.api_key = "sk-xxx"
  dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
  # Specify the voice.
  voice = 'Ethan'
  # Specify the model.
  model = 'qwen3.5-omni-plus-realtime'
  # Specify the model role.
  instructions = "You are Xiaoyun, a personal assistant. Answer the user's questions in a humorous and witty way."
  class SimpleCallback(OmniRealtimeCallback):
      def __init__(self, pya):
          self.pya = pya
          self.out = None
      def on_open(self):
          # Initialize the audio output stream.
          self.out = self.pya.open(
              format=pyaudio.paInt16,
              channels=1,
              rate=24000,
              output=True
          )
      def on_event(self, response):
          if response['type'] == 'response.audio.delta':
              # Play the audio.
              self.out.write(base64.b64decode(response['delta']))
          elif response['type'] == 'conversation.item.input_audio_transcription.delta':
              # Streaming preview: text is the confirmed prefix, stash is the unconfirmed suffix.
              preview = response.get('text', '') + response.get('stash', '')
              print(f"\r[User] {preview}", end='', flush=True)
          elif response['type'] == 'conversation.item.input_audio_transcription.completed':
              # Transcription completed. Print the final text and move to a new line.
              print(f"\r[User] {response['transcript']}")
          elif response['type'] == 'response.audio_transcript.done':
              # Print the assistant's response text.
              print(f"[LLM] {response['transcript']}")
  
  # 1. Initialize the audio device.
  pya = pyaudio.PyAudio()
  # 2. Create the callback function and conversation.
  callback = SimpleCallback(pya)
  conv = OmniRealtimeConversation(model=model, callback=callback, url=url)
  # 3. Connect and configure the session.
  conv.connect()
  conv.update_session(output_modalities=[MultiModality.AUDIO, MultiModality.TEXT], voice=voice, instructions=instructions)
  # 4. Initialize the audio input stream.
  mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
  # 5. Main loop to process audio input.
  print("Conversation started. Speak into the microphone (Ctrl+C to exit)...")
  try:
      while True:
          audio_data = mic.read(3200, exception_on_overflow=False)
          conv.append_audio(base64.b64encode(audio_data).decode())
          time.sleep(0.01)
  except KeyboardInterrupt:
      # Clean up resources.
      conv.close()
      mic.close()
      callback.out.close()
      pya.terminate()
      print("\nConversation ended")

  vad_dash.py を実行して、マイクを通じてリアルタイム会話を開始します。システムは発話を検出し、音声をサーバーにストリーミングします。

  手動モード

  manual_dash.py という名前の Python ファイルを作成し、以下のコードをコピーします。

  manual_dash.py

  # Dependencies: dashscope >= 1.23.9, pyaudio
  import os
  import base64
  import sys
  import threading
  import pyaudio
  from dashscope.audio.qwen_omni import *
  import dashscope
  
  # If the environment variable is not set, replace the next line with your API key: dashscope.api_key = "sk-xxx"
  dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
  voice = 'Ethan'
  
  class MyCallback(OmniRealtimeCallback):
      """A minimal callback: initializes the speaker on connection and plays the returned audio directly in the event handler."""
      def __init__(self, ctx):
          super().__init__()
          self.ctx = ctx
  
      def on_open(self) -> None:
          # Initialize PyAudio and the speaker (24 kHz, mono, 16-bit) after the connection is established.
          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.delta': lambda r: print('\rquestion: {}'.format(r.get('text', '') + r.get('stash', '')), end='', flush=True),
                  'conversation.item.input_audio_transcription.completed': self._transcription_completed,
                  '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 _transcription_completed(self, response):
          print()
          self.ctx['transcription_done'].set()
  
      def _play_audio(self, response):
          # Decode the base64 data and write it directly 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 current turn as complete, allowing the main loop to proceed.
          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 stream_record_and_send(pya_inst, conversation, sample_rate=16000, chunk_size=3200):
      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:
                  data = stream.read(chunk_size, exception_on_overflow=False)
                  conversation.append_audio(base64.b64encode(data).decode())
              except Exception:
                  break
  
      t = threading.Thread(target=_reader, daemon=True)
      t.start()
      input()
      stop_evt.set()
      t.join(timeout=1.0)
      stream.close()
  
  
  if __name__  == '__main__':
      print('Initializing ...')
      # Runtime context: stores audio and conversation handles.
      ctx = {'pya': None, 'out': None, 'conv': None, 'resp_done': threading.Event(), 'transcription_done': threading.Event()}
      callback = MyCallback(ctx)
      conversation = OmniRealtimeConversation(
          model='qwen3.5-omni-plus-realtime',
          callback=callback,
          # The following URL is for the Singapore region. When calling, replace WorkspaceId with your actual workspace ID. URLs vary by region.
          url="wss://{WorkspaceId}.ap-southeast-1.maas.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 and audio output, and disable server-side VAD for manual recording.
      conversation.update_session(
          output_modalities=[MultiModality.AUDIO, MultiModality.TEXT],
          voice=voice,
          enable_input_audio_transcription=True,
          input_audio_transcription_model='qwen3-asr-flash-realtime',
          enable_turn_detection=False,
          instructions="You are Xiaoyun, a personal assistant. Please answer the user's questions accurately and in a friendly manner, always responding with a helpful attitude."
      )
  
      try:
          turn = 1
          while True:
              print(f"\n--- turn {turn} ---")
              print("Press Enter to start recording (enter q and press Enter to exit)...")
              user_input = input()
              if user_input.strip().lower() in ['q', 'quit']:
                  print("User requested to exit...")
                  break
              print("Recording... Press Enter again to stop recording.")
              if ctx['pya'] is None:
                  ctx['pya'] = pyaudio.PyAudio()
              stream_record_and_send(ctx['pya'], conversation)
  
              ctx['transcription_done'].clear()
              ctx['resp_done'].clear()
              conversation.commit()
              ctx['transcription_done'].wait(timeout=10)
              print("Waiting for model response...")
              conversation.create_response()
              ctx['resp_done'].wait()
              turn += 1
      except KeyboardInterrupt:
          print("\nProgram interrupted by user.")
      finally:
          shutdown_ctx(ctx)
          print("Program exited.")

  manual_dash.py を実行します。Enter キーを押して録音を開始し、再度 Enter キーを押して停止および送信します。モデルの音声応答は自動的に再生されます。

DashScope Java SDK

インタラクションモードの選択

• VAD モード (音声区間検出、発話の開始と終了を自動検出)

  Realtime API はユーザーの発話開始と終了を検出し、応答します。

• 手動モード (押して話す、離して送信)

  クライアントが発話の開始と終了を制御します。発話後、クライアントがサーバーにメッセージを送信する必要があります。

import com.alibaba.dashscope.audio.omni.*;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.google.gson.JsonObject;
import javax.sound.sampled.*;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Base64;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class OmniServerVad {
    static class SequentialAudioPlayer {
        private final SourceDataLine line;
        private final Queue<byte[]> audioQueue = new ConcurrentLinkedQueue<>();
        private final Thread playerThread;
        private final AtomicBoolean shouldStop = new AtomicBoolean(false);

        public SequentialAudioPlayer() throws LineUnavailableException {
            AudioFormat format = new AudioFormat(24000, 16, 1, true, false);
            line = AudioSystem.getSourceDataLine(format);
            line.open(format);
            line.start();

            playerThread = new Thread(() -> {
                while (!shouldStop.get()) {
                    byte[] audio = audioQueue.poll();
                    if (audio != null) {
                        line.write(audio, 0, audio.length);
                    } else {
                        try { Thread.sleep(10); } catch (InterruptedException ignored) {}
                    }
                }
            }, "AudioPlayer");
            playerThread.start();
        }

        public void play(String base64Audio) {
            try {
                byte[] audio = Base64.getDecoder().decode(base64Audio);
                audioQueue.add(audio);
            } catch (Exception e) {
                System.err.println("Failed to decode audio: " + e.getMessage());
            }
        }

        public void cancel() {
            audioQueue.clear();
            line.flush();
        }

        public void close() {
            shouldStop.set(true);
            try { playerThread.join(1000); } catch (InterruptedException ignored) {}
            line.drain();
            line.close();
        }
    }

    public static void main(String[] args) {
        try {
            SequentialAudioPlayer player = new SequentialAudioPlayer();
            AtomicBoolean userIsSpeaking = new AtomicBoolean(false);
            AtomicBoolean shouldStop = new AtomicBoolean(false);

            OmniRealtimeParam param = OmniRealtimeParam.builder()
                    .model("qwen3.5-omni-plus-realtime")
                    .apikey(System.getenv("DASHSCOPE_API_KEY"))
                    // The following URL is for the Singapore region. When calling, replace WorkspaceId with your actual workspace ID. URLs vary by region.
                    .url("wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime")
                    .build();

            OmniRealtimeConversation conversation = new OmniRealtimeConversation(param, new OmniRealtimeCallback() {
                @Override public void onOpen() {
                    System.out.println("Connection established.");
                }
                @Override public void onClose(int code, String reason) {
                    System.out.println("Connection closed (" + code + "): " + reason);
                    shouldStop.set(true);
                }
                @Override public void onEvent(JsonObject event) {
                    handleEvent(event, player, userIsSpeaking);
                }
            });

            conversation.connect();
            conversation.updateSession(OmniRealtimeConfig.builder()
                    .modalities(Arrays.asList(OmniRealtimeModality.AUDIO, OmniRealtimeModality.TEXT))
                    .voice("Ethan")
                    .enableTurnDetection(true)
                    .enableInputAudioTranscription(true)
                    .parameters(Map.of("instructions",
                            "You are an AI customer service agent for a five-star hotel. Answer customer inquiries about room types, facilities, prices, and booking policies accurately and in a friendly manner. Always respond with a professional and helpful attitude. Do not provide unconfirmed information or information beyond the scope of the hotel's services."))
                    .build()
            );

            System.out.println("Start speaking (speech start/end is automatically detected, press Ctrl+C to exit)...");
            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("Conversation closed. Stopping recording.");
                            break;
                        }
                    }
                }
                Thread.sleep(20);
            }

            conversation.close(1000, "Normal termination");
            player.close();
            mic.close();
            System.out.println("\nProgram exited.");

        } catch (NoApiKeyException e) {
            System.err.println("API key not found: Set the DASHSCOPE_API_KEY environment variable.");
            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[User started speaking]");
                player.cancel();
                userIsSpeaking.set(true);
                break;
            case "input_audio_buffer.speech_stopped":
                System.out.println("[User stopped speaking]");
                userIsSpeaking.set(false);
                break;
            case "response.audio.delta":
                if (!userIsSpeaking.get()) {
                    player.play(event.get("delta").getAsString());
                }
                break;
            case "conversation.item.input_audio_transcription.delta":
                // Streaming preview: text is confirmed prefix, stash is unconfirmed suffix
                String preview2 = event.get("text").getAsString() + event.get("stash").getAsString();
                System.out.print("\rUser: " + preview2);
                break;
            case "conversation.item.input_audio_transcription.completed":
                System.out.println();
                break;
            case "response.audio_transcript.done":
                System.out.println("Assistant: " + event.get("transcript").getAsString());
                break;
            case "response.done":
                System.out.println("Response completed.");
                break;
        }
    }
}

// DashScope Java SDK 2.20.9 or later is required.

import com.alibaba.dashscope.audio.omni.*;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.google.gson.JsonObject;
import javax.sound.sampled.*;
import java.io.IOException;
import java.util.Arrays;
import java.util.Base64;
import java.util.HashMap;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class Main {
    public static class RealtimePcmPlayer {
        private int sampleRate;
        private SourceDataLine line;
        private AudioFormat audioFormat;
        private Thread decoderThread;
        private Thread playerThread;
        private AtomicBoolean stopped = new AtomicBoolean(false);
        private Queue<String> b64AudioBuffer = new ConcurrentLinkedQueue<>();
        private Queue<byte[]> RawAudioBuffer = new ConcurrentLinkedQueue<>();

        public RealtimePcmPlayer(int sampleRate) throws LineUnavailableException {
            this.sampleRate = sampleRate;
            this.audioFormat = new AudioFormat(this.sampleRate, 16, 1, true, false);
            DataLine.Info info = new DataLine.Info(SourceDataLine.class, audioFormat);
            line = (SourceDataLine) AudioSystem.getLine(info);
            line.open(audioFormat);
            line.start();
            decoderThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (!stopped.get()) {
                        String b64Audio = b64AudioBuffer.poll();
                        if (b64Audio != null) {
                            byte[] rawAudio = Base64.getDecoder().decode(b64Audio);
                            RawAudioBuffer.add(rawAudio);
                        } 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();
        }

        // Plays an audio chunk and blocks until playback is complete.
        private void playChunk(byte[] chunk) throws IOException, InterruptedException {
            if (chunk == null || chunk.length == 0) return;

            int bytesWritten = 0;
            while (bytesWritten < chunk.length) {
                bytesWritten += line.write(chunk, bytesWritten, chunk.length - bytesWritten);
            }
            int audioLength = chunk.length / (this.sampleRate*2/1000);
            // Waits for the audio in the buffer to finish playing.
            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);
            }
            line.drain();
        }

        public void shutdown() throws InterruptedException {
            stopped.set(true);
            decoderThread.join();
            playerThread.join();
            if (line != null && line.isRunning()) {
                line.drain();
                line.close();
            }
        }
    } 
    // Records audio and streams it to the conversation in real time.
    private static void recordAndSend(TargetDataLine line, OmniRealtimeConversation conversation) {
        byte[] buffer = new byte[3200];
        AtomicBoolean stopRecording = new AtomicBoolean(false);

        // Starts a thread to listen for the Enter key.
        Thread enterKeyListener = new Thread(() -> {
            try {
                System.in.read();
                stopRecording.set(true);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        enterKeyListener.start();

        // Records and sends audio in real time.
        while (!stopRecording.get()) {
            int count = line.read(buffer, 0, buffer.length);
            if (count > 0) {
                byte[] chunk = new byte[count];
                System.arraycopy(buffer, 0, chunk, 0, count);
                conversation.appendAudio(Base64.getEncoder().encodeToString(chunk));
            }
        }
    }

    public static void main(String[] args) throws InterruptedException, LineUnavailableException {
        OmniRealtimeParam param = OmniRealtimeParam.builder()
                .model("qwen3.5-omni-plus-realtime")
                // API keys for Singapore and China (Beijing) are different. Get an API key from: https://www.alibabacloud.com/help/en/model-studio/get-api-key
                // If the DASHSCOPE_API_KEY environment variable is not set, replace the apikey call below with your Model Studio API key, for example: .apikey("sk-xxx")
                .apikey(System.getenv("DASHSCOPE_API_KEY"))
                // The following URL is for the Singapore region. When calling, replace WorkspaceId with your actual workspace ID. URLs vary by region.
                .url("wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime")
                .build();
        AtomicReference<CountDownLatch> responseDoneLatch = new AtomicReference<>(null);
        responseDoneLatch.set(new CountDownLatch(1));
        AtomicReference<CountDownLatch> transcriptionDoneLatch = new AtomicReference<>(null);
        transcriptionDoneLatch.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.delta":
                        // Streaming preview: text is confirmed prefix, stash is unconfirmed suffix
                        String transcriptPreview = message.get("text").getAsString() + message.get("stash").getAsString();
                        System.out.print("\rquestion: " + transcriptPreview);
                        break;
                    case "conversation.item.input_audio_transcription.completed":
                        System.out.println();
                        transcriptionDoneLatch.get().countDown();
                        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("Ethan")
                .enableTurnDetection(false)
                // Set the model's role.
                .parameters(new HashMap<String, Object>() {{
                    put("instructions","You are Xiaoyun, a personal assistant. Please answer the user's questions accurately and in a friendly manner, always responding with a helpful attitude.");
                }})
                .build();
        conversation.updateSession(config);

        // Set up microphone for recording.
        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("Press Enter to start recording...");
                try {
                    System.in.read();
                } catch (IOException e) {
                    System.err.println("Error reading input: " + e.getMessage());
                    break; // Exit the loop if an error occurs.
                }
                System.out.println("Recording started. Speak now... Press Enter again to stop recording and send.");
                recordAndSend(line, conversation);
                conversation.commit();
                // Wait for transcription to complete before triggering model response to avoid interleaved output
                transcriptionDoneLatch.get().await(10, TimeUnit.SECONDS);
                System.out.println("Waiting for model response...");
                conversation.createResponse(null, null);
                responseDoneLatch.get().await();
                // Resets the latches for the next turn.
                responseDoneLatch.set(new CountDownLatch(1));
                transcriptionDoneLatch.set(new CountDownLatch(1));
            }
        } catch (LineUnavailableException e) {
            e.printStackTrace();
        } finally {
            if (line != null) {
                line.stop();
                line.close();
            }
        }
    }}

WebSocket (Python)

• 実行環境の準備

  Python 3.10 以降がインストールされていることを確認してください。

  オペレーティングシステム用の pyaudio をインストールします。

  macOS

  brew install portaudio && pip install pyaudio

  Debian/Ubuntu

  sudo apt-get install python3-pyaudio
  
  or
  
  pip install pyaudio

  pip install pyaudio の使用を推奨します。インストールに失敗した場合は、オペレーティングシステム用の portaudio 依存関係を先にインストールしてください。

  CentOS

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

  Windows

  pip install pyaudio

  WebSocket 依存関係をインストールします。

  pip install websockets==15.0.1

• クライアントの作成

  omni_realtime_client.py という名前のファイルを作成し、以下のコードをコピーします。

  omni_realtime_client.py

  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"
      SEMANTIC_VAD = "semantic_vad"  # Recommended for the qwen3.5-omni-realtime model
      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,
              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
          self.on_audio_delta = on_audio_delta
          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 {}
  
          # Current response status
          self._current_response_id = None
          self._current_item_id = None
          self._is_responding = False
          # Input/output transcript printing status
          self._print_input_transcript = True
          self._output_transcript_buffer = ""
  
      async def connect(self) -> None:
          """Establish a WebSocket connection with the Realtime API."""
          url = f"{self.base_url}?model={self.model}"
          headers = {
              "Authorization": f"Bearer {self.api_key}"
          }
          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",
              "input_audio_transcription": {
                  "model": "qwen3-asr-flash-realtime"
              }
          }
  
          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,
                  "prefix_padding_ms": 500,
                  "silence_duration_ms": 900
              }
              await self.update_session(session_config)
          elif self.turn_detection_mode == TurnDetectionMode.SEMANTIC_VAD:
              session_config['turn_detection'] = {
                  "type": "semantic_vad",
                  "threshold": 0.1,
                  "prefix_padding_ms": 500,
                  "silence_duration_ms": 900
              }
              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 the 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 append_image(self, image_chunk: bytes) -> None:
          """Append image data to the image buffer.
          Image data can come from local files or a real-time video stream.
          Note:
              - The image format must be JPG or JPEG. We recommend a resolution of 480p or 720p. The maximum supported resolution is 1080p.
              - A single image after Base64 encoding must not exceed 256 KB. We recommend keeping the raw image size below 190 KB before encoding.
              - Encode the image data into Base64 before sending.
              - We recommend sending images at one frame per second.
              - You must send audio data at least once before you send image data.
          """
          image_b64 = base64.b64encode(image_chunk).decode()
          event = {
              "type": "input_image_buffer.append",
              "image": image_b64
          }
          await self.send_event(event)
  
      async def create_response(self) -> None:
          """Request the API to generate a response. This is required only 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 a user interruption of the current response."""
          if not self._is_responding:
              return
          # 1. 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
  
      async def handle_messages(self) -> None:
          try:
              async for message in self.ws:
                  event = json.loads(message)
                  event_type = event.get("type")
                  if event_type == "error":
                      print(" Error: ", event['error'])
                      continue
                  elif event_type == "response.created":
                      self._current_response_id = event.get("response", {}).get("id")
                      self._is_responding = True
                  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 start detected")
                      if self._is_responding:
                          print("Handling interruption")
                          await self.handle_interruption()
                  elif event_type == "input_audio_buffer.speech_stopped":
                      print("Speech end detected")
                  elif event_type == "response.text.delta":
                      if self.on_text_delta:
                          self.on_text_delta(event["delta"])
                  elif event_type == "response.audio.delta":
                      if self.on_audio_delta:
                          audio_bytes = base64.b64decode(event["delta"])
                          self.on_audio_delta(audio_bytes)
                  elif event_type == "conversation.item.input_audio_transcription.delta":
                      preview = event.get("text", "") + event.get("stash", "")
                      print(f"\rUser: {preview}", end='', flush=True)
                  elif event_type == "conversation.item.input_audio_transcription.completed":
                      transcript = event.get("transcript", "")
                      print()
                      if self.on_input_transcript:
                          await asyncio.to_thread(self.on_input_transcript, transcript)
                          self._print_input_transcript = True
                  elif event_type == "response.audio_transcript.delta":
                      if self.on_output_transcript:
                          delta = event.get("delta", "")
                          if not self._print_input_transcript:
                              self._output_transcript_buffer += delta
                          else:
                              if self._output_transcript_buffer:
                                  await asyncio.to_thread(self.on_output_transcript, self._output_transcript_buffer)
                                  self._output_transcript_buffer = ""
                              await asyncio.to_thread(self.on_output_transcript, delta)
                  elif event_type == "response.audio_transcript.done":
                      print(f"assistant: {event.get('transcript', '')}")
                      self._print_input_transcript = False
                  elif event_type in self.extra_event_handlers:
                      self.extra_event_handlers[event_type](event)
          except websockets.exceptions.ConnectionClosed:
              print(" Connection closed")
          except Exception as e:
              print(" Error in message handling: ", str(e))
      async def close(self) -> None:
          """Close the WebSocket connection."""
          if self.ws:
              await self.ws.close()

• インタラクションモードの選択

  • VAD モード (音声区間検出、発話の開始と終了を自動検出)

    Realtime API はユーザーの発話開始と終了を検出し、応答を生成します。

  • 手動モード (押して話す、離して送信)

    ユーザーが音声送信の開始と終了を制御します。発話後、クライアントがサーバーに応答生成を要求するメッセージを送信する必要があります。

  VAD モード

  omni_realtime_client.py と同じディレクトリに vad_mode.py という名前のファイルを作成し、以下のコードをコピーします。

  vad_mode.py

  # -- coding: utf-8 --
  import os, asyncio, pyaudio, queue, threading
  from omni_realtime_client import OmniRealtimeClient, TurnDetectionMode
  
  # Audio player class that handles interruptions
  class AudioPlayer:
      def __init__(self, pyaudio_instance, rate=24000):
          self.stream = pyaudio_instance.open(format=pyaudio.paInt16, channels=1, rate=rate, output=True)
          self.queue = queue.Queue()
          self.stop_evt = threading.Event()
          self.interrupt_evt = threading.Event()
          threading.Thread(target=self._run, daemon=True).start()
  
      def _run(self):
          while not self.stop_evt.is_set():
              try:
                  data = self.queue.get(timeout=0.5)
                  if data is None: break
                  if not self.interrupt_evt.is_set(): self.stream.write(data)
                  self.queue.task_done()
              except queue.Empty: continue
  
      def add_audio(self, data): self.queue.put(data)
      def handle_interrupt(self): self.interrupt_evt.set(); self.queue.queue.clear()
      def stop(self): self.stop_evt.set(); self.queue.put(None); self.stream.stop_stream(); self.stream.close()
  
  # Record from the microphone and send the audio
  async def record_and_send(client):
      p = pyaudio.PyAudio()
      stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=3200)
      print("Recording started. Please speak...")
      try:
          while True:
              audio_data = stream.read(3200)
              await client.stream_audio(audio_data)
              await asyncio.sleep(0.02)
      finally:
          stream.stop_stream(); stream.close(); p.terminate()
  
  async def main():
      p = pyaudio.PyAudio()
      player = AudioPlayer(pyaudio_instance=p)
  
      client = OmniRealtimeClient(
          # This is the base_url for the Singapore region. The base_url for the China (Beijing) region is wss://dashscope.aliyuncs.com/api-ws/v1/realtime.
          base_url="wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime",
          api_key=os.environ.get("DASHSCOPE_API_KEY"),
          model="qwen3.5-omni-plus-realtime",
          voice="Ethan",
          instructions="You are Xiaoyun, a witty and humorous assistant.",
          # SEMANTIC_VAD is recommended for models like qwen3.5-omni-realtime.
          turn_detection_mode=TurnDetectionMode.SEMANTIC_VAD,
          on_text_delta=lambda t: print(f"\nassistant: {t}", end="", flush=True),
          on_audio_delta=player.add_audio,
      )
  
      await client.connect()
      print("Connection successful. Starting the real-time conversation...")
  
      # Run concurrently
      await asyncio.gather(client.handle_messages(), record_and_send(client))
  
  if __name__ == "__main__":
      try:
          asyncio.run(main())
      except KeyboardInterrupt:
          print("\nProgram exited.")

  vad_mode.py を実行して、マイクを通じてリアルタイム会話を開始します。システムは発話を検出し、音声をサーバーにストリーミングします。

  手動モード

  omni_realtime_client.py と同じディレクトリに manual_mode.py という名前のファイルを作成し、以下のコードをコピーします。

  manual_mode.py

  # -- coding: utf-8 --
  import os
  import asyncio
  import time
  import threading
  import queue
  import pyaudio
  from omni_realtime_client import OmniRealtimeClient, TurnDetectionMode
  
  
  class AudioPlayer:
      """Real-time audio player class."""
  
      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()  # Add a lock for synchronized access.
          self._last_data_time = time.time()  # Record the time when the last data was received.
          self._response_done = False  # Add a flag to indicate that the response is complete.
          self._waiting_for_response = False  # Flag to indicate whether the client is waiting for a server response.
          # Record the time of the last write to the audio stream and the duration of the last audio chunk for more accurate playback end detection.
          self._last_play_time = time.time()
          self._last_chunk_duration = 0.0
  
      def start(self):
          """Start the audio player."""
          with self._lock:
              if self.is_playing:
                  return
  
              self.is_playing = True
  
              try:
                  self.pyaudio_instance = pyaudio.PyAudio()
  
                  # Create an audio output stream.
                  self.stream = self.pyaudio_instance.open(
                      format=pyaudio.paInt16,  # 16-bit
                      channels=self.channels,
                      rate=self.sample_rate,
                      output=True,
                      frames_per_buffer=1024
                  )
  
                  # Start the playback thread.
                  self.play_thread = threading.Thread(target=self._play_audio)
                  self.play_thread.daemon = True
                  self.play_thread.start()
  
                  print("Audio player started")
              except Exception as e:
                  print(f"Failed to start audio player: {e}")
                  self._cleanup_resources()
                  raise
  
      def stop(self):
          """Stop the audio player."""
          with self._lock:
              if not self.is_playing:
                  return
  
              self.is_playing = False
  
          # Clear the queue.
          while not self.audio_queue.empty():
              try:
                  self.audio_queue.get_nowait()
              except queue.Empty:
                  break
  
          # Wait for the playback thread to finish. Wait outside the lock to avoid a deadlock.
          if self.play_thread and self.play_thread.is_alive():
              self.play_thread.join(timeout=2.0)
  
          # Acquire the lock again to clean up resources.
          with self._lock:
              self._cleanup_resources()
  
          print("Audio player stopped")
  
      def _cleanup_resources(self):
          """Clean up audio resources. This must be called within the lock."""
          try:
              # Close the audio stream.
              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"Error closing audio stream: {e}")
  
          try:
              if self.pyaudio_instance:
                  self.pyaudio_instance.terminate()
                  self.pyaudio_instance = None
          except Exception as e:
              print(f"Error terminating PyAudio: {e}")
  
      def add_audio_data(self, audio_data):
          """Add audio data to the playback queue."""
          if self.is_playing and audio_data:
              self.audio_queue.put(audio_data)
              with self._lock:
                  self._last_data_time = time.time()  # Update the time when the last data was received.
                  self._waiting_for_response = False  # Data received, no longer waiting.
  
      def stop_receiving_data(self):
          """Mark that no more new audio data will be received."""
          with self._lock:
              self._response_done = True
              self._waiting_for_response = False  # Response ended, no longer waiting.
  
      def prepare_for_next_turn(self):
          """Reset the player state for the next conversation turn."""
          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  # Start waiting for the next response.
  
          # Clear any remaining audio data from the previous turn.
          while not self.audio_queue.empty():
              try:
                  self.audio_queue.get_nowait()
              except queue.Empty:
                  break
  
      def is_finished_playing(self):
          """Check if all audio data has been played."""
          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
  
              # ---------------------- Smart end detection ----------------------
              # 1. Preferred method: If the server has marked completion and the playback queue is empty,
              #    wait for the most recent audio chunk to finish playing (chunk duration + 0.1s tolerance).
              if self._response_done and queue_size == 0:
                  min_wait = max(self._last_chunk_duration + 0.1, 0.5)  # Wait at least 0.5s.
                  if time_since_last_play >= min_wait:
                      return True
  
              # 2. Fallback method: If no new data has been received for over a second and the playback queue is empty.
              #    This logic serves as a safeguard if the server does not explicitly send `response.done`.
              if not self._waiting_for_response and queue_size == 0 and time_since_last_data > 1.0:
                  print("\n(No new audio received for a while, assuming playback is finished)")
                  return True
  
              return False
  
      def _play_audio(self):
          """Worker thread for playing audio data."""
          while True:
              # Check if it should stop.
              with self._lock:
                  if not self.is_playing:
                      break
                  stream_ref = self.stream  # Get a reference to the stream.
  
              try:
                  # Get audio data from the queue, with a timeout of 0.1 seconds.
                  audio_data = self.audio_queue.get(timeout=0.1)
  
                  # Check the status and stream validity again.
                  with self._lock:
                      if self.is_playing and stream_ref and not stream_ref.is_stopped():
                          try:
                              # Play the audio data.
                              stream_ref.write(audio_data)
                              # Update the latest playback information.
                              self._last_play_time = time.time()
                              self._last_chunk_duration = len(audio_data) / (
                                          self.channels * self.sample_width) / self.sample_rate
                          except Exception as e:
                              print(f"Error writing to audio stream: {e}")
                              break
  
                  # Mark this data block as processed.
                  self.audio_queue.task_done()
  
              except queue.Empty:
                  # Continue waiting if the queue is empty.
                  continue
              except Exception as e:
                  print(f"Error playing audio: {e}")
                  break
  
  
  class MicrophoneRecorder:
      """Real-time microphone recorder."""
  
      def __init__(self, sample_rate=16000, channels=1, chunk_size=3200):
          self.sample_rate = sample_rate
          self.channels = channels
          self.chunk_size = chunk_size
          self.pyaudio_instance = None
          self.stream = None
          self.frames = []
          self._is_recording = False
          self._record_thread = None
  
      def _recording_thread(self):
          """Recording worker thread."""
          # Continuously read data from the audio stream while _is_recording is True.
          while self._is_recording:
              try:
                  # Use exception_on_overflow=False to avoid crashing due to a buffer overflow.
                  data = self.stream.read(self.chunk_size, exception_on_overflow=False)
                  self.frames.append(data)
              except (IOError, OSError) as e:
                  # When the stream is closed, a read operation may raise an error.
                  print(f"Error reading from recording stream, it might be closed: {e}")
                  break
  
      def start(self):
          """Start recording."""
          if self._is_recording:
              print("Recording is already in progress.")
              return
  
          self.frames = []
          self._is_recording = True
  
          try:
              self.pyaudio_instance = pyaudio.PyAudio()
              self.stream = self.pyaudio_instance.open(
                  format=pyaudio.paInt16,
                  channels=self.channels,
                  rate=self.sample_rate,
                  input=True,
                  frames_per_buffer=self.chunk_size
              )
  
              self._record_thread = threading.Thread(target=self._recording_thread)
              self._record_thread.daemon = True
              self._record_thread.start()
              print("Microphone recording started...")
          except Exception as e:
              print(f"Failed to start microphone: {e}")
              self._is_recording = False
              self._cleanup()
              raise
  
      def stop(self):
          """Stop recording and return the audio data."""
          if not self._is_recording:
              return None
  
          self._is_recording = False
  
          # Wait for the recording thread to exit safely.
          if self._record_thread:
              self._record_thread.join(timeout=1.0)
  
          self._cleanup()
  
          print("Microphone recording stopped.")
          return b''.join(self.frames)
  
      def _cleanup(self):
          """Safely clean up PyAudio resources."""
          if self.stream:
              try:
                  if self.stream.is_active():
                      self.stream.stop_stream()
                  self.stream.close()
              except Exception as e:
                  print(f"Error closing audio stream: {e}")
  
          if self.pyaudio_instance:
              try:
                  self.pyaudio_instance.terminate()
              except Exception as e:
                  print(f"Error terminating PyAudio instance: {e}")
  
          self.stream = None
          self.pyaudio_instance = None
  
  
  async def interactive_test():
      """
      Interactive test for multi-turn conversations with audio and image support.
      """
      # ------------------- 1. Initialization and connection (one-time) -------------------
      # API keys for the Singapore and China (Beijing) regions are different. To get an API key, see https://www.alibabacloud.com/help/en/model-studio/get-api-key
      api_key = os.environ.get("DASHSCOPE_API_KEY")
      if not api_key:
          print("Please set the DASHSCOPE_API_KEY environment variable.")
          return
  
      print("--- Real-time Multimodal Audio/Video Chat Client ---")
      print("Initializing audio player and client...")
  
      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(Received response end marker)")
          audio_player.stop_receiving_data()
  
      realtime_client = OmniRealtimeClient(
          # This is the base_url for the Singapore region. If you use a model in the China (Beijing) region, replace the base_url with wss://dashscope.aliyuncs.com/api-ws/v1/realtime.
          base_url="wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/realtime",
          api_key=api_key,
          model="qwen3.5-omni-plus-realtime",
          voice="Ethan",
          instructions="You are Xiaoyun, a personal assistant. Please answer the user's questions accurately and in a friendly manner, always responding with a helpful attitude.", # Set the model's role.
          on_text_delta=lambda text: print(f"assistant: {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("Connected to the server. Enter 'q' or 'quit' to exit at any time.")
          message_handler_task = asyncio.create_task(realtime_client.handle_messages())
          await asyncio.sleep(0.5)
  
          turn_counter = 1
          # ------------------- 2. Multi-turn conversation loop -------------------
          while True:
              print(f"\n--- Turn {turn_counter} ---")
              audio_player.prepare_for_next_turn()
  
              recorded_audio = None
              image_paths = []
  
              # --- Get user input: Record from microphone ---
              loop = asyncio.get_event_loop()
              recorder = MicrophoneRecorder(sample_rate=16000)  # We recommend a 16k sample rate for speech recognition.
  
              print("Ready to record. Press Enter to start recording (or enter 'q' to exit)...")
              user_input = await loop.run_in_executor(None, input)
              if user_input.strip().lower() in ['q', 'quit']:
                  print("User requested to exit...")
                  return
  
              try:
                  recorder.start()
              except Exception:
                  print("Could not start recording. Please check your microphone permissions and device. Skipping this turn.")
                  continue
  
              print("Recording... Press Enter again to stop recording.")
              await loop.run_in_executor(None, input)
  
              recorded_audio = recorder.stop()
  
              if not recorded_audio or len(recorded_audio) == 0:
                  print("No valid audio was recorded. Please start this turn again.")
                  continue
  
              # --- Get image input (optional) ---
              # The image input feature is disabled by default. Uncomment the code below to enable it.
              # print("\nEnter the absolute path of an [image file] on each line (optional). When finished, enter 's' or press Enter to send the request.")
              # while True:
              #     path = input("Image path: ").strip()
              #     if path.lower() == 's' or path == '':
              #         break
              #     if path.lower() in ['q', 'quit']:
              #         print("User requested to exit...")
              #         return
              #
              #     if not os.path.isabs(path):
              #         print("Error: Please enter an absolute path.")
              #         continue
              #     if not os.path.exists(path):
              #         print(f"Error: File not found -> {path}")
              #         continue
              #     image_paths.append(path)
              #     print(f"Image added: {os.path.basename(path)}")
  
              # --- 3. Send data and get response ---
              print("\n--- Input Confirmation ---")
              print(f"Audio to process: 1 (from microphone), Images: {len(image_paths)}")
              print("------------------")
  
              # 3.1 Send the recorded audio.
              try:
                  print(f"Sending microphone recording ({len(recorded_audio)} bytes)")
                  await realtime_client.stream_audio(recorded_audio)
                  await asyncio.sleep(0.1)
              except Exception as e:
                  print(f"Failed to send microphone recording: {e}")
                  continue
  
              # 3.2 Send all image files.
              # The image input feature is disabled by default. Uncomment the code below to enable it.
              # for i, path in enumerate(image_paths):
              #     try:
              #         with open(path, "rb") as f:
              #             data = f.read()
              #         print(f"Sending image {i+1}: {os.path.basename(path)} ({len(data)} bytes)")
              #         await realtime_client.append_image(data)
              #         await asyncio.sleep(0.1)
              #     except Exception as e:
              #         print(f"Failed to send image {os.path.basename(path)}: {e}")
  
              # 3.3 Submit and wait for the response.
              print("Submitting all inputs, requesting server response...")
              await realtime_client.commit_audio_buffer()
              await realtime_client.create_response()
  
              print("Waiting for and playing server response audio...")
              start_time = time.time()
              max_wait_time = 60
              while not audio_player.is_finished_playing():
                  if time.time() - start_time > max_wait_time:
                      print(f"\nWait timed out ({max_wait_time} seconds). Moving to the next turn.")
                      break
                  await asyncio.sleep(0.2)
              
              print("\nAudio playback for this turn is complete!")
              turn_counter += 1
  
      except (asyncio.CancelledError, KeyboardInterrupt):
          print("\nProgram was interrupted.")
      except Exception as e:
          print(f"An unhandled error occurred: {e}")
      finally:
          # ------------------- 4. Clean up resources -------------------
          print("\nClosing connection and cleaning up resources...")
          if message_handler_task and not message_handler_task.done():
              message_handler_task.cancel()
  
          if 'realtime_client' in locals() and realtime_client.ws and not realtime_client.ws.close:
              await realtime_client.close()
              print("Connection closed.")
  
          audio_player.stop()
          print("Program exited.")
  
  
  if __name__ == "__main__":
      try:
          asyncio.run(interactive_test())
      except KeyboardInterrupt:
          print("\nProgram was forcibly exited by the user.")

  manual_mode.py を実行します。Enter キーを押して録音を開始し、再度 Enter キーを押して停止および送信します。

Python

• 実行環境

  Python 3.10 以降が必要です。以下の依存関係をインストールします。

  pip install aiortc aiohttp sounddevice numpy certifi av

• デモの実行

  webrtc_demo.py という名前の Python ファイルを作成し、以下のコードを貼り付けます。

  webrtc_demo.py

  # Dependencies: pip install aiortc aiohttp sounddevice numpy certifi av
  import asyncio
  import json
  import os
  import queue
  import ssl
  import threading
  
  import aiohttp
  import certifi
  import numpy as np
  import sounddevice as sd
  from aiortc import RTCPeerConnection, RTCConfiguration, RTCSessionDescription
  from aiortc.contrib.media import MediaPlayer
  from av import AudioFrame
  
  # Replace with your API key, or set the DASHSCOPE_API_KEY environment variable
  API_KEY = os.getenv("DASHSCOPE_API_KEY", "your-api-key")
  MODEL = "qwen3.5-omni-plus-realtime"
  # Replace {endpoint} with the endpoint obtained from your account manager
  SIGNALING_URL = f"https://{{endpoint}}/api/v1/webrtc/realtime?model={MODEL}"
  
  
  # --------------- Audio frame parsing ---------------
  
  def _nb_channels(frame: AudioFrame) -> int:
      """Get the number of channels in an audio frame, compatible with different PyAV versions"""
      if hasattr(frame.layout, "nb_channels"):
          return int(frame.layout.nb_channels)
      ch = getattr(frame.layout, "channels", 1)
      if isinstance(ch, (tuple, list)):
          return len(ch)
      return int(ch)
  
  
  def audioframe_to_s16_samples(frame: AudioFrame) -> np.ndarray:
      """
      Server audio frames are stereo interleaved. Direct reshape causes channel misalignment.
      Rearrange to (samples, channels) based on actual channel count.
      Different aiortc decoder versions return different array shapes for the same audio,
      so unified handling is needed here.
      """
      arr = np.asarray(frame.to_ndarray())
      ch = _nb_channels(frame)
      samples = int(frame.samples)
  
      if arr.ndim == 2 and arr.shape[0] == ch and arr.shape[1] == samples:
          return arr.T.copy()
      if arr.ndim == 2 and arr.shape[0] == 1 and arr.shape[1] == samples * ch:
          return arr.reshape(-1).reshape(samples, ch).copy()
      if arr.ndim == 1 and arr.shape[0] == samples * ch:
          return arr.reshape(samples, ch).copy()
  
      flat = arr.reshape(-1)
      if ch > 0 and flat.size % ch == 0:
          return flat.reshape(flat.size // ch, ch).copy()
      raise ValueError(f"unexpected shape={arr.shape}, ch={ch}, samples={samples}")
  
  
  # --------------- Low-latency audio player ---------------
  
  class RemoteAudioPlayer:
      """
      Low-latency audio player that plays 5ms audio blocks to minimize delay.
      Supports voice interruption: clears the buffer when the user starts speaking,
      stopping playback of old model responses.
      Merges stereo server audio to mono (average of left and right channels) for playback.
      """
      def __init__(self, samplerate=48000, out_channels=1, blocksize=240, max_seconds=0.2):
          self.samplerate = samplerate
          self.out_channels = out_channels
          self.blocksize = blocksize
          self._q = queue.Queue(maxsize=max(5, int(max_seconds * samplerate / blocksize) + 5))
          self._lock = threading.Lock()
          self._rb_size = max(1, int(max_seconds * samplerate))
          self._rb = np.zeros((self._rb_size, out_channels), dtype=np.int16)
          self._rb_w = 0
          self._rb_r = 0
          self._rb_len = 0
          self._stream = None
          self._closed = False
  
      def start(self):
          if self._stream:
              return
  
          def callback(outdata, frames, _time, status):
              if self._closed:
                  outdata[:] = np.zeros((frames, self.out_channels), dtype=np.int16)
                  return
              while True:
                  try:
                      chunk = self._q.get_nowait()
                  except queue.Empty:
                      break
                  with self._lock:
                      self._write_rb(chunk)
              with self._lock:
                  out = self._read_rb(frames)
              outdata[:] = out
  
          self._stream = sd.OutputStream(
              samplerate=self.samplerate,
              channels=self.out_channels,
              dtype="int16",
              blocksize=self.blocksize,
              callback=callback,
          )
          self._stream.start()
  
      def clear(self):
          """Clear playback buffer for voice interruption"""
          try:
              while True:
                  self._q.get_nowait()
          except queue.Empty:
              pass
          with self._lock:
              self._rb_w = 0
              self._rb_r = 0
              self._rb_len = 0
              self._rb[:] = 0
  
      def _write_rb(self, chunk: np.ndarray):
          n = int(chunk.shape[0])
          if n <= 0:
              return
          overflow = max(0, self._rb_len + n - self._rb_size)
          if overflow > 0:
              self._rb_r = (self._rb_r + overflow) % self._rb_size
              self._rb_len -= overflow
          end = self._rb_size - self._rb_w
          if n <= end:
              self._rb[self._rb_w:self._rb_w + n] = chunk
          else:
              self._rb[self._rb_w:] = chunk[:end]
              self._rb[:n - end] = chunk[end:]
          self._rb_w = (self._rb_w + n) % self._rb_size
          self._rb_len += n
  
      def _read_rb(self, frames: int) -> np.ndarray:
          if self._rb_len <= 0:
              return np.zeros((frames, self.out_channels), dtype=np.int16)
          n = min(frames, self._rb_len)
          out = np.zeros((frames, self.out_channels), dtype=np.int16)
          end = self._rb_size - self._rb_r
          if n <= end:
              out[:n] = self._rb[self._rb_r:self._rb_r + n]
          else:
              out[:end] = self._rb[self._rb_r:]
              out[end:n] = self._rb[:n - end]
          self._rb_r = (self._rb_r + n) % self._rb_size
          self._rb_len -= n
          return out
  
      async def push_frame(self, frame: AudioFrame):
          """Receive audio frames, auto-merge channels and enqueue"""
          if self._closed:
              return
          pcm = audioframe_to_s16_samples(frame)
          in_ch = pcm.shape[1]
          if self.out_channels == 1:
              if in_ch == 1:
                  out = pcm
              else:
                  out = np.mean(pcm.astype(np.int32), axis=1).astype(np.int16).reshape(-1, 1)
          else:
              if in_ch == self.out_channels:
                  out = pcm
              elif in_ch == 1 and self.out_channels == 2:
                  out = np.repeat(pcm, 2, axis=1)
              else:
                  out = pcm[:, :self.out_channels]
          try:
              self._q.put_nowait(out)
          except queue.Full:
              try:
                  self._q.get_nowait()
              except queue.Empty:
                  pass
              try:
                  self._q.put_nowait(out)
              except queue.Full:
                  pass
  
      async def close(self):
          self._closed = True
          if self._stream:
              self._stream.stop()
              self._stream.close()
              self._stream = None
  
  
  # --------------- main ---------------
  
  async def main():
      pc = RTCPeerConnection(RTCConfiguration(iceServers=[]))
  
      # Initialize audio player (mono output, 5ms blocksize for low latency)
      speaker = RemoteAudioPlayer(samplerate=48000, out_channels=1, blocksize=240, max_seconds=0.2)
      speaker.start()
  
      # Initialize microphone (macOS avfoundation; for Linux use pulse or alsa)
      mic = MediaPlayer("none:0", format="avfoundation",
                        options={"sample_rate": "48000", "channels": "1"})
      if not mic.audio:
          raise RuntimeError("No microphone detected. Check the avfoundation audio device index.")
      pc.addTrack(mic.audio)
  
      # Client creates DataChannel (name is customizable); server pushes events through a channel named "txt"
      pc.createDataChannel("oai-events")
  
      remote_dc = None
      got_first_txt_msg = False
  
      def make_session_update() -> dict:
          """Build session.update config: voice, audio format, VAD strategy, inference parameters"""
          return {
              "type": "session.update",
              "session": {
                  "modalities": ["text", "audio"],
                  "voice": "Tina",
                  "input_audio_format": "pcm",
                  "output_audio_format": "pcm",
                  "instructions": "You are a friendly AI assistant.",
                  "turn_detection": {"type": "server_vad", "threshold": 0.5, "silence_duration_ms": 800},
                  "max_tokens": 16384,
                  "temperature": 0.9,
              },
          }
  
      # Handle server-pushed DataChannel events
      @pc.on("datachannel")
      def on_datachannel(ch):
          nonlocal remote_dc, got_first_txt_msg
          print(f"[DC] Received server DataChannel: {ch.label}")
          if ch.label == "txt":
              remote_dc = ch
  
          @ch.on("message")
          def on_msg(msg):
              nonlocal got_first_txt_msg
              try:
                  evt = json.loads(msg)
              except Exception:
                  return
              print(f"[{ch.label}] {evt.get('type')}")
  
              # Clear playback buffer when user starts speaking (voice interruption)
              if isinstance(evt, dict) and evt.get("type") == "input_audio_buffer.speech_started":
                  speaker.clear()
                  print("[Playback] User speech detected, clearing buffer (interruption)")
  
              # Send session.update after receiving first message on txt channel
              if ch.label == "txt" and not got_first_txt_msg:
                  got_first_txt_msg = True
                  if remote_dc and remote_dc.readyState == "open":
                      remote_dc.send(json.dumps(make_session_update(), ensure_ascii=False))
                      print("[DC] session.update sent")
  
      # Receive server audio and play with low latency
      @pc.on("track")
      async def on_track(track):
          if track.kind == "audio":
              async def _play():
                  try:
                      while True:
                          frame = await track.recv()
                          await speaker.push_frame(frame)
                  except Exception:
                      pass
              asyncio.create_task(_play())
  
      @pc.on("iceconnectionstatechange")
      def on_ice():
          print(f"[ICE] {pc.iceConnectionState}")
  
      @pc.on("connectionstatechange")
      async def on_conn():
          print(f"[Connection] {pc.connectionState}")
          if pc.connectionState in ("failed", "closed", "disconnected"):
              await pc.close()
  
      # SDP exchange: create Offer and POST to signaling server, get Answer
      offer = await pc.createOffer()
      await pc.setLocalDescription(offer)
  
      async with aiohttp.ClientSession() as session:
          async with session.post(
              SIGNALING_URL,
              ssl=ssl.create_default_context(cafile=certifi.where()),
              data=offer.sdp.encode("utf-8"),
              headers={
                  "Content-Type": "application/sdp",
                  "Authorization": f"Bearer {API_KEY}",
              },
              timeout=aiohttp.ClientTimeout(total=10),
          ) as resp:
              if not resp.ok:
                  raise Exception(f"SDP exchange failed: {resp.status} {await resp.text()}")
              answer_sdp = await resp.text()
  
      await pc.setRemoteDescription(RTCSessionDescription(sdp=answer_sdp, type="answer"))
      print("SDP exchange complete, waiting for connection...")
  
      try:
          await asyncio.Event().wait()
      except (KeyboardInterrupt, asyncio.CancelledError):
          pass
      finally:
          print(f"\nExiting. Final state: connection={pc.connectionState}, ICE={pc.iceConnectionState}")
          await speaker.close()
          try:
              if mic and mic.audio:
                  mic.audio.stop()
          except Exception:
              pass
          await pc.close()
  
  asyncio.run(main())

  webrtc_demo.py を実行して、マイクを通じて Qwen-Omni-Realtime モデルとのリアルタイム会話を開始します。システムはユーザーの発話開始を検出し、音声を自動的にサーバーに送信します。

JavaScript

• 前提条件

  • WebRTC をサポートする最新のブラウザ (Chrome、Edge、Firefox、Safari など) を使用します。

  • ブラウザはマイクへのアクセス許可を必要とします。

  • ブラウザのクロスオリジンセキュリティポリシーにより、ブラウザはサーバーへの直接接続リクエストを送信できません。ターミナルで curl コマンドを実行して、接続設定を完了する必要があります。

• デモの実行

  webrtc_demo.html という名前の HTML ファイルを作成し、以下のコードを貼り付けます。

  webrtc_demo.html

  <!DOCTYPE html>
  <html lang="en">
  <head>
      <meta charset="UTF-8" />
      <title>WebRTC Realtime Voice Chat</title>
      <style>
          * { box-sizing: border-box; margin: 0; padding: 0; }
          body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif; background: #f5f7fa; color: #1d2129; padding: 24px; line-height: 1.6; }
  
          .container { max-width: 800px; margin: 0 auto; }
  
          h1 { font-size: 22px; font-weight: 600; margin-bottom: 20px; color: #1d2129; }
  
          /* Sticky top bar */
          .sticky-top { position: sticky; top: 0; z-index: 100; background: #f5f7fa; margin: 0 -24px 16px; padding: 12px 24px; border-bottom: 1px solid transparent; transition: border-color .2s; }
          .sticky-top.scrolled { border-bottom-color: #e5e6eb; }
  
          /* Toolbar */
          .toolbar { display: flex; align-items: center; gap: 10px; flex-wrap: wrap; margin-bottom: 12px; }
          .toolbar label { display: flex; align-items: center; gap: 6px; font-size: 13px; color: #4e5969; cursor: pointer; }
  
          /* Buttons */
          button { padding: 8px 18px; font-size: 13px; font-weight: 500; border: 1px solid #c9cdd4; border-radius: 6px; background: #fff; color: #1d2129; cursor: pointer; transition: all .15s; }
          button:hover:not(:disabled) { border-color: #165dff; color: #165dff; }
          button:disabled { opacity: .4; cursor: not-allowed; }
          .btn-primary { background: #165dff; border-color: #165dff; color: #fff; }
          .btn-primary:hover:not(:disabled) { background: #4080ff; border-color: #4080ff; color: #fff; }
          .btn-danger { border-color: #f53f3f; color: #f53f3f; }
          .btn-danger:hover:not(:disabled) { background: #f53f3f; color: #fff; }
  
          /* Status indicator */
          .status-bar { display: flex; align-items: center; gap: 8px; padding: 10px 14px; border-radius: 8px; background: #fff; border: 1px solid #e5e6eb; font-size: 13px; }
          .status-dot { width: 8px; height: 8px; border-radius: 50%; background: #c9cdd4; flex-shrink: 0; }
          .status-dot.connected { background: #00b42a; }
          .status-dot.connecting { background: #ff7d00; animation: pulse 1s infinite; }
          .status-dot.error { background: #f53f3f; }
          @keyframes pulse { 0%,100% { opacity: 1; } 50% { opacity: .4; } }
  
          /* SDP card */
          .card { background: #fff; border: 1px solid #e5e6eb; border-radius: 10px; padding: 16px; margin-bottom: 16px; }
          .card-title { font-size: 13px; font-weight: 600; color: #4e5969; margin-bottom: 8px; }
          .step-num { display: inline-flex; align-items: center; justify-content: center; width: 20px; height: 20px; border-radius: 50%; background: #165dff; color: #fff; font-size: 11px; font-weight: 600; margin-right: 6px; }
          .card-hint { font-size: 12px; color: #86909c; margin-top: 6px; }
  
          textarea { width: 100%; font-family: "SF Mono", "Fira Code", "Fira Mono", Menlo, Consolas, monospace; font-size: 12px; padding: 10px; border: 1px solid #e5e6eb; border-radius: 6px; resize: vertical; background: #f7f8fa; color: #1d2129; transition: border-color .15s; }
          textarea:focus { outline: none; border-color: #165dff; background: #fff; }
  
          /* Video */
          .video-section { margin-bottom: 16px; }
          .video-label { font-size: 13px; color: #86909c; margin-bottom: 6px; }
          video { width: 320px; max-width: 100%; background: #000; border-radius: 8px; display: block; }
  
          /* Events panel */
          .events-title { font-size: 14px; font-weight: 600; color: #1d2129; margin-bottom: 10px; }
          .events-container { display: flex; flex-direction: column; gap: 6px; }
          .event-item { background: #fff; border: 1px solid #e5e6eb; border-radius: 8px; overflow: hidden; }
          .event-header { display: flex; align-items: center; gap: 8px; padding: 8px 12px; cursor: pointer; user-select: none; font-size: 12px; }
          .event-header:hover { background: #f7f8fa; }
          .event-arrow { font-size: 14px; font-weight: 700; width: 18px; text-align: center; }
          .event-arrow.server { color: #00b42a; }
          .event-arrow.client { color: #165dff; }
          .event-label { color: #4e5969; }
          .event-time { color: #c9cdd4; margin-left: auto; font-size: 11px; }
          .event-body { display: none; padding: 10px 12px; background: #f7f8fa; border-top: 1px solid #e5e6eb; }
          .event-body pre { margin: 0; font-size: 11px; font-family: "SF Mono", Menlo, Consolas, monospace; color: #4e5969; white-space: pre-wrap; word-break: break-all; }
          .events-empty { font-size: 13px; color: #c9cdd4; padding: 16px 0; text-align: center; }
      </style>
  </head>
  <body>
  <div class="container">
      <h1>WebRTC Realtime Voice Chat</h1>
  
      <div class="sticky-top">
          <div class="toolbar">
              <button id="startBtn" class="btn-primary">Start Session</button>
              <button id="setAnswerBtn" disabled>Set Answer</button>
              <button id="endBtn" class="btn-danger" disabled>End Session</button>
              <button id="downloadBtn" disabled>Download Remote Audio</button>
              <label>
                  <input id="sendVideoCheckbox" type="checkbox" />
                  Enable Video
              </label>
          </div>
  
          <div class="status-bar">
              
  
          </div>
      </div>
  
      <div class="card">
          <div class="card-title">Offer SDP</div>
          <div style="margin-bottom: 8px;">
              <button id="copyOfferBtn" disabled>Copy Offer SDP</button>
          </div>
          <textarea id="offerBox" rows="6" readonly placeholder="Auto-generated after clicking Start Session"></textarea>
          <div class="card-hint">Auto-generated after ICE gathering completes. Copy and send to the server via curl to get the Answer.</div>
      </div>
  
      <div class="card">
          <div class="card-title">curl Command</div>
          <div style="margin-bottom: 8px;">
              <button id="copyCurlBtn" disabled>Copy curl Command</button>
          </div>
          <textarea id="curlBox" rows="6" readonly placeholder="Auto-filled after Offer SDP is generated"></textarea>
          <div class="card-hint">Copy this command to your terminal. Paste the returned Answer SDP below.</div>
      </div>
  
      <div class="card">
          <div class="card-title">Answer SDP</div>
          <textarea id="answerBox" rows="6" placeholder="Paste the Answer SDP returned by curl here"></textarea>
          <div class="card-hint">After pasting, click Set Answer above to establish the connection.</div>
      </div>
  
      <div class="video-section" id="videoSection" style="display:none;">
          <div class="video-label">Local Video Preview</div>
          <video id="localVideo" autoplay playsinline muted></video>
      </div>
  
      <div class="events-title">Events (DataChannel)</div>
      <div id="events" class="events-container"></div>
  </div>
  
  <script>
      const eventsDiv = document.getElementById('events');
      const startBtn = document.getElementById('startBtn');
      const setAnswerBtn = document.getElementById('setAnswerBtn');
      const endBtn = document.getElementById('endBtn');
      const downloadBtn = document.getElementById('downloadBtn');
      const copyOfferBtn = document.getElementById('copyOfferBtn');
      const statusDot = document.getElementById('statusDot');
      const statusText = document.getElementById('statusText');
  
      const copyCurlBtn = document.getElementById('copyCurlBtn');
      const curlBox = document.getElementById('curlBox');
  
      const sendVideoCheckbox = document.getElementById('sendVideoCheckbox');
      const localVideo = document.getElementById('localVideo');
  
      let pc = null;
      let hiddenRemoteAudioEl = null;
  
      let mediaRecorder = null;
      let recordedChunks = [];
      let audioBlob = null;
  
      let localStream = null;
  
      let sendCanvas = null;
      let sendCanvasCtx = null;
      let sendCanvasStream = null;
      let sendRafId = 0;
  
      let gatedAudioTracks = [];
      let gatedVideoTracks = [];
      let audioSender = null;
      let videoSender = null;
      let audioTrack = null;
      let videoTrack = null;
  
      function setStatus(text, state) {
        statusText.textContent = text;
        statusDot.className = 'status-dot' + (state ? ' ' + state : '');
      }
  
      function gateMedia(on) {
        for (const t of gatedAudioTracks) t.enabled = !!on;
        for (const t of gatedVideoTracks) t.enabled = !!on;
      }
  
      function sendUpdate(channel) {
        const update = {
          event_id: `event_${Date.now()}`,
          type: "session.update",
          session: {
            input_audio_format: "pcm",
            input_audio_transcription: { model: "qwen3-asr-flash-realtime" },
            instructions: "You are a helpful assistant.",
            modalities: ["text", "audio"],
            output_audio_format: "pcm",
            smooth_output: false,
            turn_detection: {
              prefix_padding_ms: 500,
              silence_duration_ms: 800,
              threshold: 0.5,
              type: "server_vad",
            },
          },
        };
        if (channel && channel.readyState === "open") channel.send(JSON.stringify(update));
      }
  
      // ===== Events panel =====
      const events = [];
      function nowTs() { return new Date().toLocaleTimeString(); }
  
      function renderEvents() {
        eventsDiv.innerHTML = "";
        if (events.length === 0) {
          const empty = document.createElement("div");
          empty.className = "events-empty";
          empty.textContent = "Waiting for events...";
          eventsDiv.appendChild(empty);
          return;
        }
  
        for (const item of events) {
          const { event, timestamp } = item;
          const isClient = event?.type?.includes("update") || event?.type?.includes("create");
  
          const wrap = document.createElement("div");
          wrap.className = "event-item";
  
          const header = document.createElement("div");
          header.className = "event-header";
  
          const arrow = document.createElement("span");
          arrow.className = "event-arrow " + (isClient ? "client" : "server");
          arrow.textContent = isClient ? "↓" : "↑";
  
          const label = document.createElement("span");
          label.className = "event-label";
          const who = isClient ? "client" : "server";
          const type = event?.type ?? "message";
          label.textContent = `${who}: ${type}`;
  
          const time = document.createElement("span");
          time.className = "event-time";
          time.textContent = timestamp;
  
          const body = document.createElement("div");
          body.className = "event-body";
          const pre = document.createElement("pre");
          pre.textContent = JSON.stringify(event, null, 2);
          body.appendChild(pre);
  
          header.onclick = () => { body.style.display = body.style.display === "block" ? "none" : "block"; };
  
          header.appendChild(arrow);
          header.appendChild(label);
          header.appendChild(time);
          wrap.appendChild(header);
          wrap.appendChild(body);
  
          eventsDiv.appendChild(wrap);
        }
      }
  
      function clearUIEvents() { events.length = 0; renderEvents(); }
      function pushEventFromDataChannel(eventObj) {
        const ts = eventObj.timestamp || nowTs();
        if (!eventObj.timestamp) eventObj.timestamp = ts;
        events.unshift({ event: eventObj, timestamp: ts });
        renderEvents();
      }
  
      function normalizeSdpForSetRemote(sdp) {
        sdp = String(sdp).trim().replace(/\r?\n/g, "\r\n");
        if (!sdp.endsWith("\r\n")) sdp += "\r\n";
        return sdp;
      }
  
      // ===== WebRTC =====
      startBtn.onclick = () => startSession().catch(err => console.log("startSession error:", err));
      endBtn.onclick = () => endSession();
      setAnswerBtn.onclick = () => setRemoteAnswerFromUI().catch(err => console.log("setRemoteAnswer error:", err));
      copyOfferBtn.onclick = async () => {
        const txt = offerBox.value;
        if (!txt) return;
        await navigator.clipboard.writeText(txt);
        alert("Offer SDP copied");
      };
      copyCurlBtn.onclick = async () => {
        const txt = curlBox.value;
        if (!txt) return;
        await navigator.clipboard.writeText(txt);
        alert("curl command copied. Run it in your terminal.");
      };
      downloadBtn.onclick = () => {
        if (audioBlob) downloadBlob(audioBlob, 'remote-audio.webm');
        else alert('No audio recording available');
      };
  
      async function startSession() {
        if (pc) return;
  
        pc = new RTCPeerConnection({ iceServers: [] });
        clearUIEvents();
        setStatus('Requesting microphone access...', 'connecting');
  
        offerBox.value = "";
        answerBox.value = "";
        curlBox.value = "";
        setAnswerBtn.disabled = true;
        copyOfferBtn.disabled = true;
        copyCurlBtn.disabled = true;
  
        endBtn.disabled = false;
        downloadBtn.disabled = true;
  
        pc.onconnectionstatechange = () => {
          if (!pc) return;
          if (pc.connectionState === 'connected') {
            setStatus('Connected. Start speaking.', 'connected');
          } else if (["failed", "closed", "disconnected"].includes(pc.connectionState)) {
            console.log("onconnectionstatechange:", pc.connectionState);
            endSession(true);
          }
        };
  
        pc.ontrack = async (e) => {
          const stream = e.streams[0];
          ensureHiddenAudioEl();
          hiddenRemoteAudioEl.srcObject = stream;
          try { await hiddenRemoteAudioEl.play(); } catch {}
          startRecordingRemoteStream(stream);
        };
  
        const wantVideo = !!sendVideoCheckbox.checked;
  
        const localPreviewFps = 30;
        const sendFps = 2;
  
        const constraints = wantVideo
          ? {
              audio: true,
              video: {
                facingMode: { ideal: "user" },
                frameRate: { ideal: localPreviewFps, max: localPreviewFps },
                width: { ideal: 640 },
                height: { ideal: 480 },
              }
            }
          : { audio: true };
  
        localStream = await navigator.mediaDevices.getUserMedia(constraints);
  
        const videoSection = document.getElementById('videoSection');
        if (wantVideo) {
          localVideo.srcObject = localStream;
          localVideo.style.display = "block";
          videoSection.style.display = "";
          try { await localVideo.play(); } catch {}
        } else {
          localVideo.srcObject = null;
          localVideo.style.display = "none";
          videoSection.style.display = "none";
        }
  
        gatedAudioTracks = [];
        gatedVideoTracks = [];
  
        localStream.getAudioTracks().forEach(t => {
          pc.addTrack(t, localStream);
          gatedAudioTracks.push(t);
        });
  
        if (wantVideo) {
          if (sendRafId) cancelAnimationFrame(sendRafId);
          sendRafId = 0;
          if (sendCanvasStream) sendCanvasStream.getTracks().forEach(t => t.stop());
          sendCanvasStream = null;
          sendCanvasCtx = null;
          sendCanvas = null;
  
          const settings = localStream.getVideoTracks()[0].getSettings();
          sendCanvas = document.createElement("canvas");
          sendCanvas.width = settings.width || 640;
          sendCanvas.height = settings.height || 480;
          sendCanvasCtx = sendCanvas.getContext("2d", { alpha: false });
  
          sendCanvasStream = sendCanvas.captureStream(sendFps);
          const lowFpsTrack = sendCanvasStream.getVideoTracks()[0];
          pc.addTrack(lowFpsTrack, sendCanvasStream);
          gatedVideoTracks.push(lowFpsTrack);
  
          const pump = () => {
            if (!sendCanvasCtx || !sendCanvas) return;
            try { sendCanvasCtx.drawImage(localVideo, 0, 0, sendCanvas.width, sendCanvas.height); } catch {}
            sendRafId = requestAnimationFrame(pump);
          };
          sendRafId = requestAnimationFrame(pump);
        }
  
        gateMedia(false);
  
        audioSender = pc.getSenders().find(s => s.track?.kind === 'audio');
        videoSender = pc.getSenders().find(s => s.track?.kind === 'video');
        audioTrack = audioSender?.track;
        videoTrack = videoSender?.track;
  
        await audioSender?.replaceTrack(null);
        await videoSender?.replaceTrack(videoTrack ? null : undefined);
  
        const dc = pc.createDataChannel('oai-events');
  
        dc.onopen = () => console.log("DC open");
        dc.onmessage = (e) => {
          handleDcMessage(e.data, dc);
        };
  
        pc.ondatachannel = (event) => {
          const ch = event.channel;
          ch.onmessage = (e) => {
              handleDcMessage(e.data, ch);
          };
        };
  
        function handleDcMessage(data, channel) {
            let obj;
            try { obj = JSON.parse(data); }
            catch (err) {
              pushEventFromDataChannel({ type: "raw", data: String(data), parseError: String(err) });
              return;
            }
            pushEventFromDataChannel(obj);
  
            if (obj?.type === "session.created") {
              console.log("Session created, opening media gate.");
              gateMedia(true);
              if(audioSender) audioSender.replaceTrack(audioTrack);
              if(videoSender && videoTrack) videoSender.replaceTrack(videoTrack);
  
              sendUpdate(channel);
            }
        }
  
        pc.onicegatheringstatechange = () => {
          if (!pc) return;
          if (pc.iceGatheringState === "complete" && pc.localDescription?.sdp) {
            const sdp = pc.localDescription.sdp;
            offerBox.value = sdp;
            copyOfferBtn.disabled = false;
            setAnswerBtn.disabled = false;
  
            const escapedSdp = sdp.replace(/'/g, "'\\''");
            curlBox.value = `curl -X POST 'https://{endpoint}/api/v1/webrtc/realtime?model=qwen3.5-omni-plus-realtime' \\\n  -H 'Content-Type: application/sdp' \\\n  -H 'Authorization: Bearer $DASHSCOPE_API_KEY' \\\n  --data-binary '${escapedSdp}'`;
            copyCurlBtn.disabled = false;
  
            setStatus('Offer SDP generated. Copy the curl command to your terminal to get the Answer SDP.', 'connecting');
            console.log("ICE Gathering Complete. Ready to set remote description.");
          }
        };
  
        const offer = await pc.createOffer();
        await pc.setLocalDescription(offer);
      }
  
      async function setRemoteAnswerFromUI() {
        if (!pc) return alert('Click "Start Session" first to generate the Offer.');
        const txt = answerBox.value.trim();
        if (!txt) return alert("Please paste the Answer SDP");
  
        const answerSdp = normalizeSdpForSetRemote(txt);
        try {
            await pc.setRemoteDescription({ type: 'answer', sdp: answerSdp });
            setStatus('Establishing connection...', 'connecting');
        } catch (e) {
            alert("Failed to set Answer: " + e.message);
            console.error(e);
        }
      }
  
      function endSession(silent = false) {
        if (sendRafId) cancelAnimationFrame(sendRafId);
        sendRafId = 0;
  
        if (sendCanvasStream) {
          sendCanvasStream.getTracks().forEach(t => t.stop());
        }
        sendCanvasStream = null;
        sendCanvasCtx = null;
        sendCanvas = null;
  
        try { if (mediaRecorder && mediaRecorder.state !== "inactive") mediaRecorder.stop(); } catch {}
        mediaRecorder = null;
  
        if (localStream) {
          localStream.getTracks().forEach(t => t.stop());
          localStream = null;
        }
        localVideo.srcObject = null;
        localVideo.style.display = "none";
        document.getElementById('videoSection').style.display = "none";
  
        if (pc) {
          try { pc.close(); } catch {}
          pc = null;
        }
  
        gatedAudioTracks = [];
        gatedVideoTracks = [];
  
        if (hiddenRemoteAudioEl) {
          try { hiddenRemoteAudioEl.pause(); } catch {}
          hiddenRemoteAudioEl.srcObject = null;
          hiddenRemoteAudioEl.remove();
          hiddenRemoteAudioEl = null;
        }
  
        endBtn.disabled = true;
        setAnswerBtn.disabled = true;
        copyOfferBtn.disabled = true;
        copyCurlBtn.disabled = true;
        downloadBtn.disabled = !audioBlob;
  
        setStatus('Disconnected', '');
        if (!silent) console.log("session ended");
      }
  
      function ensureHiddenAudioEl() {
        if (hiddenRemoteAudioEl) return;
        hiddenRemoteAudioEl = document.createElement("audio");
        hiddenRemoteAudioEl.autoplay = true;
        hiddenRemoteAudioEl.playsInline = true;
        hiddenRemoteAudioEl.muted = false;
        hiddenRemoteAudioEl.style.display = "none";
        document.body.appendChild(hiddenRemoteAudioEl);
      }
  
      function startRecordingRemoteStream(remoteStream) {
        const audioTracks = remoteStream.getAudioTracks();
        if (!audioTracks.length) return;
  
        const audioStream = new MediaStream(audioTracks);
        recordedChunks = [];
        audioBlob = null;
        downloadBtn.disabled = true;
  
        try {
          mediaRecorder = new MediaRecorder(audioStream, { mimeType: 'audio/webm' });
        } catch (err) {
          console.log("MediaRecorder create failed:", err);
          return;
        }
  
        mediaRecorder.ondataavailable = (e) => {
          if (e.data && e.data.size > 0) recordedChunks.push(e.data);
        };
  
        mediaRecorder.onstop = () => {
          audioBlob = new Blob(recordedChunks, { type: 'audio/webm' });
          downloadBtn.disabled = !audioBlob || audioBlob.size === 0;
        };
  
        mediaRecorder.start();
      }
  
      function downloadBlob(blob, filename) {
        const url = URL.createObjectURL(blob);
        const a = document.createElement('a');
        a.style.display = 'none';
        a.href = url;
        a.download = filename;
        document.body.appendChild(a);
        a.click();
        URL.revokeObjectURL(url);
        a.remove();
      }
  
      renderEvents();
  
      const stickyTop = document.querySelector('.sticky-top');
      window.addEventListener('scroll', () => {
        stickyTop.classList.toggle('scrolled', window.scrollY > 10);
      }, { passive: true });
  </script>
  </body>
  </html>

  このファイルをブラウザで開き、以下の手順に従います。

  1. Start Session をクリックします。ページが自動的に Offer SDP と対応する curl コマンドを生成します。

  2. Copy curl Command をクリックし、ターミナルで実行します。出力結果が Answer SDP になります。

  3. Answer SDP を Answer SDP テキストボックスに貼り付け、Set Answer をクリックして接続を確立し、音声チャットを開始します。