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 を使用して依存関係をインストールします。

pip install websocket-client dashscope

• 対話モードの選択

  • VAD モード（音声区間検出、音声の開始／終了を自動検出）

    サーバーがユーザーの発話の開始および終了を自動検出し、応答します。

  • 手動モード（プッシュ・トゥ・トーク、リリース・トゥ・センド）

    クライアントが発話タイミングを制御します。ユーザーの発話終了後に、クライアントがサーバーにメッセージを送信します。

  VAD モード

  新しい Python ファイル vad_dash.py を作成し、以下のコードをコピーして貼り付けます。

  vad_dash.py

  # 依存関係：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
  
  # 設定パラメーター：エンドポイント、API キー、ボイス、モデル、モデルの役割
  # リージョンを指定します。「intl」は国際（シンガポール）、"cn" は中国（北京）です。
  region = 'intl'
  base_domain = 'dashscope-intl.aliyuncs.com' if region == 'intl' else 'dashscope.aliyuncs.com'
  url = f'wss://{base_domain}/api-ws/v1/realtime'
  # API キーを設定します。環境変数を設定していない場合は、次の行を dashscope.api_key = "sk-xxx" に置き換えてください。
  dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
  # ボイスを指定します。
  voice = 'Cherry'
  # モデルを指定します。
  model = 'qwen3.5-omni-plus-realtime'
  # モデルの役割を指定します。
  instructions = "あなたは個人アシスタントの「小雲」です。ユーザーの質問にユーモアと機知を交えてお答えください。"
  class SimpleCallback(OmniRealtimeCallback):
      def __init__(self, pya):
          self.pya = pya
          self.out = None
      def on_open(self):
          # 音声出力ストリームを初期化します。
          self.out = self.pya.open(
              format=pyaudio.paInt16,
              channels=1,
              rate=24000,
              output=True
          )
      def on_event(self, response):
          if response['type'] == 'response.audio.delta':
              # 音声を再生します。
              self.out.write(base64.b64decode(response['delta']))
          elif response['type'] == 'conversation.item.input_audio_transcription.completed':
              # 文字起こしされたテキストを表示します。
              print(f"[ユーザー] {response['transcript']}")
          elif response['type'] == 'response.audio_transcript.done':
              # アシスタントの返答テキストを表示します。
              print(f"[LLM] {response['transcript']}")
  
  # 1. 音声デバイスを初期化します。
  pya = pyaudio.PyAudio()
  # 2. コールバック関数とセッションを作成します。
  callback = SimpleCallback(pya)
  conv = OmniRealtimeConversation(model=model, callback=callback, url=url)
  # 3. 接続を確立し、セッションを構成します。
  conv.connect()
  conv.update_session(output_modalities=[MultiModality.AUDIO, MultiModality.TEXT], voice=voice, instructions=instructions)
  # 4. 音声入力ストリームを初期化します。
  mic = pya.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True)
  # 5. 音声入力を処理するメインループです。
  print("会話が開始されました。マイクに向かって話してください（Ctrl+C で終了）…")
  try:
      while True:
          audio_data = mic.read(3200, exception_on_overflow=False)
          conv.append_audio(base64.b64encode(audio_data).decode())
          time.sleep(0.01)
  except KeyboardInterrupt:
      # リソースをクリーンアップします。
      conv.close()
      mic.close()
      callback.out.close()
      pya.terminate()
      print("\n会話が終了しました")

  vad_dash.py を実行して、マイク経由で Qwen-Omni-Realtime とリアルタイム会話を行います。システムが音声の開始／終了を検出し、手動介入なしで自動的にデータをサーバーに送信します。

  手動モード

  新しい Python ファイル manual_dash.py を作成し、以下のコードをコピーして貼り付けます。

  manual_dash.py

  # 依存関係：dashscope >= 1.23.9、pyaudio。
  import os
  import base64
  import sys
  import threading
  import pyaudio
  from dashscope.audio.qwen_omni import *
  import dashscope
  
  # 環境変数を設定していない場合は、次の行を API キーに置き換えてください：dashscope.api_key = "sk-xxx"
  dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')
  voice = 'Cherry'
  
  class MyCallback(OmniRealtimeCallback):
      """最小限のコールバック：接続時にスピーカーを初期化し、返された音声を直接再生します。"""
      def __init__(self, ctx):
          super().__init__()
          self.ctx = ctx
  
      def on_open(self) -> None:
          # 接続後に PyAudio およびスピーカー（24k/モノラル/16bit）を初期化します。
          print('接続が確立されました')
          try:
              self.ctx['pya'] = pyaudio.PyAudio()
              self.ctx['out'] = self.ctx['pya'].open(
                  format=pyaudio.paInt16,
                  channels=1,
                  rate=24000,
                  output=True
              )
              print('音声出力が初期化されました')
          except Exception as e:
              print('[エラー] 音声初期化に失敗しました：{}'.format(e))
  
      def on_close(self, close_status_code, close_msg) -> None:
          print('接続が終了しました（コード：{}、メッセージ：{}）'.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('セッションを開始しました：{}'.format(r['session']['id'])),
                  'conversation.item.input_audio_transcription.completed': lambda r: print('質問：{}'.format(r['transcript'])),
                  'response.audio_transcript.delta': lambda r: print('LLM のテキスト：{}'.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('[エラー] {}'.format(e))
  
      def _play_audio(self, response):
          # Base64 をデコードし、再生用に出力ストリームに書き込みます。
          if self.ctx['out'] is None:
              return
          try:
              data = base64.b64decode(response['delta'])
              self.ctx['out'].write(data)
          except Exception as e:
              print('[エラー] 音声再生に失敗しました：{}'.format(e))
  
      def _response_done(self, response):
          # メインループが待機できるように、現在のターンを完了としてマークします。
          if self.ctx['conv'] is not None:
              print('[メトリック] 応答：{}、最初のテキスト遅延：{}、最初の音声遅延：{}'.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):
      """音声および PyAudio リソースを安全に解放します。"""
      try:
          if ctx['out'] is not None:
              ctx['out'].close()
              ctx['out'] = None
      except Exception:
          pass
      try:
          if ctx['pya'] is not None:
              ctx['pya'].terminate()
              ctx['pya'] = None
      except Exception:
          pass
  
  
  def record_until_enter(pya_inst: pyaudio.PyAudio, sample_rate=16000, chunk_size=3200):
      """Enter キーを押すと録音を停止し、PCM バイトを返します。"""
      frames = []
      stop_evt = threading.Event()
  
      stream = pya_inst.open(
          format=pyaudio.paInt16,
          channels=1,
          rate=sample_rate,
          input=True,
          frames_per_buffer=chunk_size
      )
  
      def _reader():
          while not stop_evt.is_set():
              try:
                  frames.append(stream.read(chunk_size, exception_on_overflow=False))
              except Exception:
                  break
  
      t = threading.Thread(target=_reader, daemon=True)
      t.start()
      input()  # ユーザーが再度 Enter キーを押すと録音を停止します。
      stop_evt.set()
      t.join(timeout=1.0)
      try:
          stream.close()
      except Exception:
          pass
      return b''.join(frames)
  
  
  if __name__  == '__main__':
      print('初期化中 …')
      # 実行時コンテキスト：音声およびセッションハンドルを格納します。
      ctx = {'pya': None, 'out': None, 'conv': None, 'resp_done': threading.Event()}
      callback = MyCallback(ctx)
      conversation = OmniRealtimeConversation(
          model='qwen3.5-omni-plus-realtime',
          callback=callback,
          # 以下は国際（シンガポール）リージョンの URL です。中国（北京）リージョンのモデルを使用する場合は、URL を wss://dashscope.aliyuncs.com/api-ws/v1/realtime に置き換えてください。
          url="wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime",
      )
      try:
          conversation.connect()
      except Exception as e:
          print('[エラー] 接続に失敗しました：{}'.format(e))
          sys.exit(1)
  
      ctx['conv'] = conversation
      # セッション構成：テキストおよび音声出力を有効化（サーバー側 VAD を無効化し、手動録音に切り替えます）。
      conversation.update_session(
          output_modalities=[MultiModality.AUDIO, MultiModality.TEXT],
          voice=voice,
          enable_input_audio_transcription=True,
          # 入力音声を文字起こすモデル。gummy-realtime-v1 のみサポートされます。
          input_audio_transcription_model='gummy-realtime-v1',
          enable_turn_detection=False,
          instructions="あなたは個人アシスタントの「小雲」です。ユーザーの質問に正確かつ親切に回答し、常に親切な態度で応答してください。"
      )
  
      try:
          turn = 1
          while True:
              print(f"\n--- ターン {turn} ---")
              print("Enter キーを押して録音を開始します（q を入力すると終了します）…")
              user_input = input()
              if user_input.strip().lower() in ['q', 'quit']:
                  print("ユーザーが終了を要求しました…")
                  break
              print("録音中… 再度 Enter キーを押して停止します。")
              if ctx['pya'] is None:
                  ctx['pya'] = pyaudio.PyAudio()
              recorded = record_until_enter(ctx['pya'])
              if not recorded:
                  print("有効な音声が録音されませんでした。もう一度お試しください。")
                  continue
              print(f"音声を {len(recorded)} バイト録音しました。送信中…")
  
              # 3200 バイトずつのチャンクで送信します（16k/16bit/100ms に対応）。
              chunk_size = 3200
              for i in range(0, len(recorded), chunk_size):
                  chunk = recorded[i:i+chunk_size]
                  conversation.append_audio(base64.b64encode(chunk).decode('ascii'))
  
              print("送信が完了しました。モデルの応答を待っています…")
              ctx['resp_done'].clear()
              conversation.commit()
              conversation.create_response()
              ctx['resp_done'].wait()
              print('音声再生が完了しました。')
              turn += 1
      except KeyboardInterrupt:
          print("\nユーザーによりプログラムが中断されました。")
      finally:
          shutdown_ctx(ctx)
          print("プログラムが終了しました。")

  manual_dash.py を実行します。Enter キーを押して話しかけ、再度 Enter キーを押してモデルの音声応答を受信します。

DashScope Java SDK

対話モードの選択

• VAD モード（音声区間検出、音声の開始／終了を自動検出）

  リアルタイム 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("音声デコードに失敗しました： " + 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"))
                    // 以下は国際（シンガポール）リージョンの URL です。中国（北京）リージョンのモデルを使用する場合は、URL を wss://dashscope.aliyuncs.com/api-ws/v1/realtime に置き換えてください。
                    .url("wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime")
                    .build();

            OmniRealtimeConversation conversation = new OmniRealtimeConversation(param, new OmniRealtimeCallback() {
                @Override public void onOpen() {
                    System.out.println("接続が確立されました");
                }
                @Override public void onClose(int code, String reason) {
                    System.out.println("接続が終了しました (" + 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("Cherry")
                    .enableTurnDetection(true)
                    .enableInputAudioTranscription(true)
                    .parameters(Map.of("instructions",
                            "あなたは五つ星ホテルの AI 顧客サービス担当者です。客室タイプ、施設、料金、予約ポリシーに関するお客様のお問い合わせに正確かつ親切に回答してください。常にプロフェッショナルで親切な態度で応答してください。未確認の情報やホテルのサービス範囲を超えた情報を提供しないでください。"))
                    .build()
            );

            System.out.println("話しかけてください（音声の開始／終了を自動検出、Ctrl+C で終了）…");
            AudioFormat format = new AudioFormat(16000, 16, 1, true, false);
            TargetDataLine mic = AudioSystem.getTargetDataLine(format);
            mic.open(format);
            mic.start();

            ByteBuffer buffer = ByteBuffer.allocate(3200);
            while (!shouldStop.get()) {
                int bytesRead = mic.read(buffer.array(), 0, buffer.capacity());
                if (bytesRead > 0) {
                    try {
                        conversation.appendAudio(Base64.getEncoder().encodeToString(buffer.array()));
                    } catch (Exception e) {
                        if (e.getMessage() != null && e.getMessage().contains("closed")) {
                            System.out.println("会話が終了しました。録音を停止します。");
                            break;
                        }
                    }
                }
                Thread.sleep(20);
            }

            conversation.close(1000, "通常終了");
            player.close();
            mic.close();
            System.out.println("\nプログラムが終了しました。");

        } catch (NoApiKeyException e) {
            System.err.println("API KEY が見つかりません：DASHSCOPE_API_KEY 環境変数を設定してください。");
            System.exit(1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void handleEvent(JsonObject event, SequentialAudioPlayer player, AtomicBoolean userIsSpeaking) {
        String type = event.get("type").getAsString();
        switch (type) {
            case "input_audio_buffer.speech_started":
                System.out.println("\n[ユーザーが話始めました]");
                player.cancel();
                userIsSpeaking.set(true);
                break;
            case "input_audio_buffer.speech_stopped":
                System.out.println("[ユーザーが話終わりました]");
                userIsSpeaking.set(false);
                break;
            case "response.audio.delta":
                if (!userIsSpeaking.get()) {
                    player.play(event.get("delta").getAsString());
                }
                break;
            case "conversation.item.input_audio_transcription.completed":
                System.out.println("ユーザー： " + event.get("transcript").getAsString());
                break;
            case "response.audio_transcript.delta":
                System.out.print(event.get("delta").getAsString());
                break;
            case "response.done":
                System.out.println("応答が完了しました");
                break;
        }
    }
}

// DashScope Java SDK バージョン 2.20.9 以降

import com.alibaba.dashscope.audio.omni.*;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.google.gson.JsonObject;
import javax.sound.sampled.*;
import java.io.ByteArrayOutputStream;
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.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class Main {
    // RealtimePcmPlayer クラス定義開始
    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();
        }

        // 音声チャンクを再生し、再生完了までブロックします。
        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);
            // バッファー内の音声の再生が完了するまで待機します。
            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();
            }
        }
    } // RealtimePcmPlayer クラス定義終了
    // 録音メソッドを追加
    private static void recordAndSend(TargetDataLine line, OmniRealtimeConversation conversation) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        byte[] buffer = new byte[3200];
        AtomicBoolean stopRecording = new AtomicBoolean(false);

        // Enter キーを監視するスレッドを開始します。
        Thread enterKeyListener = new Thread(() -> {
            try {
                System.in.read();
                stopRecording.set(true);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        enterKeyListener.start();

        // 録音ループ
        while (!stopRecording.get()) {
            int count = line.read(buffer, 0, buffer.length);
            if (count > 0) {
                out.write(buffer, 0, count);
            }
        }

        // 録音されたデータを送信します。
        byte[] audioData = out.toByteArray();
        String audioB64 = Base64.getEncoder().encodeToString(audioData);
        conversation.appendAudio(audioB64);
        out.close();
    }

    public static void main(String[] args) throws InterruptedException, LineUnavailableException {
        OmniRealtimeParam param = OmniRealtimeParam.builder()
                .model("qwen3.5-omni-plus-realtime")
                // シンガポールと北京の API キーは異なります。API キーの取得方法については、https://www.alibabacloud.com/help/ja/model-studio/get-api-key をご参照ください。
                // 環境変数を設定していない場合は、次の行を Model Studio の API キーに置き換えてください：.apikey("sk-xxx")
                .apikey(System.getenv("DASHSCOPE_API_KEY"))
                // 以下はシンガポールリージョンの URL です。北京リージョンのモデルを使用する場合は、URL を wss://dashscope.aliyuncs.com/api-ws/v1/realtime に置き換えてください。
                .url("wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime")
                .build();
        AtomicReference<CountDownLatch> responseDoneLatch = new AtomicReference<>(null);
        responseDoneLatch.set(new CountDownLatch(1));

        RealtimePcmPlayer audioPlayer = new RealtimePcmPlayer(24000);
        final AtomicReference<OmniRealtimeConversation> conversationRef = new AtomicReference<>(null);
        OmniRealtimeConversation conversation = new OmniRealtimeConversation(param, new OmniRealtimeCallback() {
            @Override
            public void onOpen() {
                System.out.println("接続が確立されました");
            }
            @Override
            public void onEvent(JsonObject message) {
                String type = message.get("type").getAsString();
                switch(type) {
                    case "session.created":
                        System.out.println("セッションを開始しました： " + message.get("session").getAsJsonObject().get("id").getAsString());
                        break;
                    case "conversation.item.input_audio_transcription.completed":
                        System.out.println("質問： " + message.get("transcript").getAsString());
                        break;
                    case "response.audio_transcript.delta":
                        System.out.println("LLM の応答の差分を取得しました： " + 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("======応答完了======");
                        if (conversationRef.get() != null) {
                            System.out.println("[メトリック] 応答： " + conversationRef.get().getResponseId() +
                                    "、最初のテキスト遅延： " + conversationRef.get().getFirstTextDelay() +
                                    " ms、最初の音声遅延： " + conversationRef.get().getFirstAudioDelay() + " ms");
                        }
                        responseDoneLatch.get().countDown();
                        break;
                    default:
                        break;
                }
            }
            @Override
            public void onClose(int code, String reason) {
                System.out.println("接続が終了しました（コード： " + code + "、理由： " + reason + "）");
            }
        });
        conversationRef.set(conversation);
        try {
            conversation.connect();
        } catch (NoApiKeyException e) {
            throw new RuntimeException(e);
        }
        OmniRealtimeConfig config = OmniRealtimeConfig.builder()
                .modalities(Arrays.asList(OmniRealtimeModality.AUDIO, OmniRealtimeModality.TEXT))
                .voice("Cherry")
                .enableTurnDetection(false)
                // モデルの役割を設定します。
                .parameters(new HashMap<String, Object>() {{
                    put("instructions","あなたは個人アシスタントの「小雲」です。ユーザーの質問に正確かつ親切に回答し、常に親切な態度で応答してください。");
                }})
                .build();
        conversation.updateSession(config);

        // マイク録音機能を追加します。
        AudioFormat format = new AudioFormat(16000, 16, 1, true, false);
        DataLine.Info info = new DataLine.Info(TargetDataLine.class, format);

        if (!AudioSystem.isLineSupported(info)) {
            System.out.println("ラインがサポートされていません");
            return;
        }

        TargetDataLine line = null;
        try {
            line = (TargetDataLine) AudioSystem.getLine(info);
            line.open(format);
            line.start();

            while (true) {
                System.out.println("Enter キーを押して録音を開始します…");
                System.in.read();
                System.out.println("録音を開始しました。話しかけてください… Enter キーを再度押して録音を停止し、送信します。");
                recordAndSend(line, conversation);
                conversation.commit();
                conversation.createResponse(null, null);
                // 次の待機のためにラッチをリセットします。
                responseDoneLatch.set(new CountDownLatch(1));
            }
        } catch (LineUnavailableException | IOException 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
  
  または
  
  pip install pyaudio

  pip install pyaudio の使用を推奨します。インストールに失敗した場合は、まず OS 用の portaudio 依存関係をインストールしてください。

  CentOS

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

  Windows

  pip install pyaudio

  インストール後、pip を使用して WebSocket 関連の依存関係をインストールします。

  pip install websockets==15.0.1

• クライアントの作成

  ローカルディレクトリに omni_realtime_client.py という新しい Python ファイルを作成し、以下のコードをコピーして貼り付けます。

  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"
      MANUAL = "manual"
  
  class OmniRealtimeClient:
  
      def __init__(
              self,
              base_url,
              api_key: str,
              model: str = "",
              voice: str = "Ethan",
              instructions: str = "あなたは親切なアシスタントです。",
              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 {}
  
          # 現在の応答ステータス
          self._current_response_id = None
          self._current_item_id = None
          self._is_responding = False
          # 入力／出力文字起こしの表示ステータス
          self._print_input_transcript = True
          self._output_transcript_buffer = ""
  
      async def connect(self) -> None:
          """リアルタイム API との WebSocket 接続を確立します。"""
          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_config = {
              "modalities": ["text", "audio"],
              "voice": self.voice,
              "instructions": self.instructions,
              "input_audio_format": "pcm",
              "output_audio_format": "pcm",
              "input_audio_transcription": {
                  "model": "gummy-realtime-v1"
              }
          }
  
          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)
          else:
              raise ValueError(f"無効なターン検出モード：{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:
          """セッション構成を更新します。"""
          event = {
              "type": "session.update",
              "session": config
          }
          await self.send_event(event)
  
      async def stream_audio(self, audio_chunk: bytes) -> None:
          """API に生音声データをストリーミングします。"""
          # 16 ビット、16 kHz、モノラル PCM のみサポートされます。
          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:
          """音声バッファーをコミットして処理をトリガーします。"""
          event = {
              "type": "input_audio_buffer.commit"
          }
          await self.send_event(event)
  
      async def append_image(self, image_chunk: bytes) -> None:
          """画像データを画像バッファーに追加します。
          画像データはローカルファイルまたはリアルタイムビデオストリームから取得できます。
          注意：
              - 画像フォーマットは JPG または JPEG のみです。推奨解像度は 480p または 720p です。最大サポート解像度は 1080p です。
              - 単一の画像は 500 KB を超えてはいけません。
              - 送信前に画像データを Base64 エンコードする必要があります。
              - 1 秒あたり最大 1 フレームの頻度で画像を送信することを推奨します。
              - 画像データを送信する前に、少なくとも 1 回は音声データを送信する必要があります。
          """
          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:
          """API に応答の生成をリクエストします（手動モードでのみ必要です）。"""
          event = {
              "type": "response.create"
          }
          await self.send_event(event)
  
      async def cancel_response(self) -> None:
          """現在の応答をキャンセルします。"""
          event = {
              "type": "response.cancel"
          }
          await self.send_event(event)
  
      async def handle_interruption(self):
          """現在の応答に対するユーザーの中断を処理します。"""
          if not self._is_responding:
              return
          # 1. 現在の応答をキャンセルします。
          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(" エラー： ", 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("音声の開始を検出しました")
                      if self._is_responding:
                          print("中断を処理しています")
                          await self.handle_interruption()
                  elif event_type == "input_audio_buffer.speech_stopped":
                      print("音声の終了を検出しました")
                  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.completed":
                      transcript = event.get("transcript", "")
                      print(f"ユーザー： {transcript}")
                      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"LLM： {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(" 接続が閉じられました")
          except Exception as e:
              print(" メッセージ処理中のエラー： ", str(e))
      async def close(self) -> None:
          """WebSocket 接続を閉じます。"""
          if self.ws:
              await self.ws.close()

• 対話モードの選択

  • VAD モード（音声区間検出、音声の開始／終了を自動検出）

    リアルタイム API が音声タイミングを自動検出し、応答します。

  • 手動モード（プッシュ・トゥ・トーク、リリース・トゥ・センド）

    クライアントが音声タイミングを制御します。ユーザーの発話終了後に、クライアントがサーバーにメッセージを送信します。

  VAD モード

  omni_realtime_client.py と同じディレクトリに、別の Python ファイル vad_mode.py を作成し、以下のコードをコピーして貼り付けます。

  vad_mode.py

  # -- coding: utf-8 --
  import os, asyncio, pyaudio, queue, threading
  from omni_realtime_client import OmniRealtimeClient, TurnDetectionMode
  
  # 音声プレーヤークラス（中断処理対応）
  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()
  
  # マイクからの録音および送信
  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("録音を開始しました。話しかけてください…")
      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(
          # 以下は国際（シンガポール）リージョンの base_url です。中国（北京）リージョンの base_url は wss://dashscope.aliyuncs.com/api-ws/v1/realtime です。
          base_url="wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime",
          api_key=os.environ.get("DASHSCOPE_API_KEY"),
          model="qwen3.5-omni-plus-realtime",
          voice="Cherry",
          instructions="あなたはユーモアと機知に富んだアシスタント「小雲」です。",
          turn_detection_mode=TurnDetectionMode.SERVER_VAD,
          on_text_delta=lambda t: print(f"\nアシスタント： {t}", end="", flush=True),
          on_audio_delta=player.add_audio,
      )
  
      await client.connect()
      print("接続に成功しました。リアルタイム会話を開始します…")
  
      # 並列実行
      await asyncio.gather(client.handle_messages(), record_and_send(client))
  
  if __name__ == "__main__":
      try:
          asyncio.run(main())
      except KeyboardInterrupt:
          print("\nプログラムが終了しました。")

  vad_mode.py を実行して、マイク経由でリアルタイムモデルとリアルタイム会話を行います。システムが音声の開始／終了を検出し、手動介入なしで自動的にデータをサーバーに送信します。

  手動モード

  omni_realtime_client.py と同じディレクトリに、別の Python ファイル 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:
      """リアルタイム音声プレーヤークラス"""
  
      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 バイト（16 ビット）
          self.audio_queue = queue.Queue()
          self.is_playing = False
          self.play_thread = None
          self.pyaudio_instance = None
          self.stream = None
          self._lock = threading.Lock()  # 同期アクセスのためのロック
          self._last_data_time = time.time()  # 最後にデータを受信した時刻を記録
          self._response_done = False  # 応答完了を示すフラグ
          self._waiting_for_response = False  # サーバーからの応答待ちを示すフラグ
          # 最後に音声ストリームに書き込まれたデータの時刻と、直近の音声チャンクの持続時間を記録し、より正確な再生終了検出を行う
          self._last_play_time = time.time()
          self._last_chunk_duration = 0.0
  
      def start(self):
          """音声プレーヤーを開始します"""
          with self._lock:
              if self.is_playing:
                  return
  
              self.is_playing = True
  
              try:
                  self.pyaudio_instance = pyaudio.PyAudio()
  
                  # 音声出力ストリームを作成します
                  self.stream = self.pyaudio_instance.open(
                      format=pyaudio.paInt16,  # 16 ビット
                      channels=self.channels,
                      rate=self.sample_rate,
                      output=True,
                      frames_per_buffer=1024
                  )
  
                  # 再生スレッドを開始します
                  self.play_thread = threading.Thread(target=self._play_audio)
                  self.play_thread.daemon = True
                  self.play_thread.start()
  
                  print("音声プレーヤーを開始しました")
              except Exception as e:
                  print(f"音声プレーヤーの開始に失敗しました：{e}")
                  self._cleanup_resources()
                  raise
  
      def stop(self):
          """音声プレーヤーを停止します"""
          with self._lock:
              if not self.is_playing:
                  return
  
              self.is_playing = False
  
          # キューをクリアします
          while not self.audio_queue.empty():
              try:
                  self.audio_queue.get_nowait()
              except queue.Empty:
                  break
  
          # 再生スレッドの終了を待機します（デッドロックを避けるためロック外で待機）
          if self.play_thread and self.play_thread.is_alive():
              self.play_thread.join(timeout=2.0)
  
          # ロックを再取得してリソースをクリーンアップ
          with self._lock:
              self._cleanup_resources()
  
          print("音声プレーヤーを停止しました")
  
      def _cleanup_resources(self):
          """音声リソースをクリーンアップします（ロック内で呼び出す必要があります）"""
          try:
              # 音声ストリームを閉じます
              if self.stream:
                  if not self.stream.is_stopped():
                      self.stream.stop_stream()
                  self.stream.close()
                  self.stream = None
          except Exception as e:
              print(f"音声ストリームの閉じ方に失敗しました：{e}")
  
          try:
              if self.pyaudio_instance:
                  self.pyaudio_instance.terminate()
                  self.pyaudio_instance = None
          except Exception as e:
              print(f"PyAudio インスタンスの終了に失敗しました：{e}")
  
      def add_audio_data(self, audio_data):
          """再生キューに音声データを追加します"""
          if self.is_playing and audio_data            with self._lock:
                  self._last_data_time = time.time()  # 最後にデータを受信した時刻を更新
                  self._waiting_for_response = False  # データ受信中なので、応答待ちではない
  
      def stop_receiving_data(self):
          """これ以上新しい音声データを受信しないことを示します"""
          with self._lock:
              self._response_done = True
              self._waiting_for_response = False  # 応答完了なので、応答待ちではない
  
      def prepare_for_next_turn(self):
          """次の会話ターンのためにプレーヤーの状態をリセットします。"""
          with self._lock:
              self._response_done = False
              self._last_data_time = time.time()
              self._last_play_time = time.time()
              self._last_chunk_duration = 0.0
              self._waiting_for_response = True  # 次の応答を待機開始
  
          # 前のターンの残りの音声データをクリア
          while not self.audio_queue.empty():
              try:
                  self.audio_queue.get_nowait()
              except queue.Empty:
                  break
  
      def is_finished_playing(self):
          """すべての音声データが再生されたかどうかを確認します"""
          with self._lock:
              queue_size = self.audio_queue.qsize()
              time_since_last_data = time.time() - self._last_data_time
              time_since_last_play = time.time() - self._last_play_time
  
              # ---------------------- スマートな終了検出 ----------------------
              # 1. 優先：サーバーが完了を通知し、再生キューが空の場合。
              #    直近の音声チャンクの再生が完了するまで待機（チャンク持続時間＋0.1 秒の許容誤差）。
              if self._response_done and queue_size == 0:
                  min_wait = max(self._last_chunk_duration + 0.1, 0.5)  # 少なくとも 0.5 秒待機
                  if time_since_last_play >= min_wait:
                      return True
  
              # 2. フォールバック：新しいデータを一定時間受信しなかった場合、かつ再生キューが空の場合。
              #    このロジックは、サーバーが明示的に `response.done` を送信しない場合の保険です。
              if not self._waiting_for_response and queue_size == 0 and time_since_last_data > 1.0:
                  print("\n（しばらく新しい音声を受信しなかったため、再生が完了したと見なします）")
                  return True
  
              return False
  
      def _play_audio(self):
          """音声データを再生するワーカースレッド"""
          while True:
              # 停止すべきか確認
              with self._lock:
                  if not self.is_playing:
                      break
                  stream_ref = self.stream  # ストリームへの参照を取得
  
              try:
                  # キューから音声データを取得（タイムアウト 0.1 秒）
                  audio_data = self.audio_queue.get(timeout=0.1)
  
                  # ステータスとストリームの有効性を再度確認
                  with self._lock:
                      if self.is_playing and stream_ref and not stream_ref.is_stopped():
                          try:
                              # 音声データを再生
                              stream_ref.write(audio_data)
                              # 最新の再生情報を更新
                              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"音声ストリームへの書き込みに失敗しました：{e}")
                              break
  
                  # このデータブロックの処理を完了としてマーク
                  self.audio_queue.task_done()
  
              except queue.Empty:
                  # キューが空の場合は継続待機
                  continue
              except Exception as e:
                  print(f"音声再生中にエラーが発生しました：{e}")
                  break
  
  
  class MicrophoneRecorder:
      """リアルタイムマイクレコーダー"""
  
      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):
          """録音ワーカースレッド"""
          # _is_recording が True の間、音声ストリームからデータを継続的に読み取ります
          while self._is_recording:
              try:
                  # バッファオーバーフローによるクラッシュを避けるために exception_on_overflow=False を使用
                  data = self.stream.read(self.chunk_size, exception_on_overflow=False)
                  self.frames.append(data)
              except (IOError, OSError) as e:
                  # ストリームが閉じられた場合、読み取りでエラーが発生することがあります
                  print(f"録音ストリームからの読み取りエラー、おそらく閉じられています：{e}")
                  break
  
      def start(self):
          """録音を開始します"""
          if self._is_recording:
              print("すでに録音中です。")
              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("マイク録音を開始しました…")
          except Exception as e:
              print(f"マイクの開始に失敗しました：{e}")
              self._is_recording = False
              self._cleanup()
              raise
  
      def stop(self):
          """録音を停止し、音声データを返します"""
          if not self._is_recording:
              return None
  
          self._is_recording = False
  
          # 録音スレッドが安全に終了するまで待機
          if self._record_thread:
              self._record_thread.join(timeout=1.0)
  
          self._cleanup()
  
          print("マイク録音を停止しました。")
          return b''.join(self.frames)
  
      def _cleanup(self):
          """PyAudio リソースを安全にクリーンアップします"""
          if self.stream:
              try:
                  if self.stream.is_active():
                      self.stream.stop_stream()
                  self.stream.close()
              except Exception as e:
                  print(f"音声ストリームの閉じ方に失敗しました：{e}")
  
          if self.pyaudio_instance:
              try:
                  self.pyaudio_instance.terminate()
              except Exception as e:
                  print(f"PyAudio インスタンスの終了に失敗しました：{e}")
  
          self.stream = None
          self.pyaudio_instance = None
  
  
  async def interactive_test():
      """
      対話式テストスクリプト：各ターンで音声および画像を送信するマルチターン会話が可能です。
      """
      # ------------------- 1. 初期化および接続（1 回のみ） -------------------
      # シンガポールと北京の API キーは異なります。API キーの取得方法については、https://www.alibabacloud.com/help/ja/model-studio/get-api-key をご参照ください。
      api_key = os.environ.get("DASHSCOPE_API_KEY")
      if not api_key:
          print("DASHSCOPE_API_KEY 環境変数を設定してください。")
          return
  
      print("--- リアルタイムマルチモーダル音声／映像チャットクライアント ---")
      print("音声プレーヤーおよびクライアントを初期化しています…")
  
      audio_player = AudioPlayer()
      audio_player.start()
  
      def on_audio_received(audio_data):
          audio_player.add_audio_data(audio_data)
  
      def on_response_done(event):
          print("\n（応答終了マーカーを受信しました）")
          audio_player.stop_receiving_data()
  
      realtime_client = OmniRealtimeClient(
          # 以下はシンガポールリージョンの base_url です。北京リージョンの base_url は wss://dashscope.aliyuncs.com/api-ws/v1/realtime です。
          base_url="wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime",
          api_key=api_key,
          model="qwen3.5-omni-plus-realtime",
          voice="Ethan",
          instructions="あなたは個人アシスタントの「小雲」です。ユーザーの質問に正確かつ親切に回答し、常に親切な態度で応答してください。", # モデルの役割を設定
          on_text_delta=lambda text: print(f"アシスタントの返答：{text}", end="", flush=True),
          on_audio_delta=on_audio_received,
          turn_detection_mode=TurnDetectionMode.MANUAL,
          extra_event_handlers={"response.done": on_response_done}
      )
  
      message_handler_task = None
      try:
          await realtime_client.connect()
          print("サーバーに接続しました。いつでも 'q' または 'quit' を入力して終了できます。")
          message_handler_task = asyncio.create_task(realtime_client.handle_messages())
          await asyncio.sleep(0.5)
  
          turn_counter = 1
          # ------------------- 2. マルチターン会話ループ -------------------
          while True:
              print(f"\n--- ターン {turn_counter} ---")
              audio_player.prepare_for_next_turn()
  
              recorded_audio = None
              image_paths = []
  
              # --- ユーザー入力の取得：マイクから録音 ---
              loop = asyncio.get_event_loop()
              recorder = MicrophoneRecorder(sample_rate=16000)  # 16k サンプルレートは音声認識に推奨
  
              print("録音を開始する準備ができました。Enter キーを押して録音を開始します（または 'q' を入力して終了）…")
              user_input = await loop.run_in_executor(None, input)
              if user_input.strip().lower() in ['q', 'quit']:
                  print("ユーザーが終了を要求しました…")
                  return
  
              try:
                  recorder.start()
              except Exception:
                  print("録音を開始できませんでした。マイクの権限およびデバイスを確認してください。このターンはスキップします。")
                  continue
  
              print("録音中… 再度 Enter キーを押して停止します。")
              await loop.run_in_executor(None, input)
  
              recorded_audio = recorder.stop()
  
              if not recorded_audio or len(recorded_audio) == 0:
                  print("有効な音声が録音されませんでした。このターンをもう一度やり直してください。")
                  continue
  
              # --- 画像入力の取得（任意） ---
              # 以下の画像入力機能はコメントアウトされており、一時的に無効化されています。有効化するには、下記のコードをアンコメントしてください。
              # print("\n各行に [画像ファイル] の絶対パスを入力してください（任意）。終了したら 's' を入力するか、Enter キーを押してリクエストを送信します。")
              # while True:
              #     path = input("画像パス： ").strip()
              #     if path.lower() == 's' or path == '':
              #         break
              #     if path.lower() in ['q', 'quit']:
              #         print("ユーザーが終了を要求しました…")
              #         return
              #
              #     if not os.path.isabs(path):
              #         print("エラー：絶対パスを入力してください。")
              #         continue
              #     if not os.path.exists(path):
              #         print(f"エラー：ファイルが見つかりません -> {path}")
              #         continue
              #     image_paths.append(path)
              #     print(f"画像を追加しました：{os.path.basename(path)}")
  
              # --- 3. データの送信および応答の取得 ---
              print("\n--- 入力の確認 ---")
              print(f"処理する音声：1（マイクから）、画像：{len(image_paths)}")
              print("------------------")
  
              # 3.1 録音された音声を送信
              try:
                  print(f"マイク録音を送信中 ({len(recorded_audio)} バイト)")
                  await realtime_client.stream_audio(recorded_audio)
                  await asyncio.sleep(0.1)
              except Exception as e:
                  print(f"マイク録音の送信に失敗しました：{e}")
                  continue
  
              # 3.2 すべての画像ファイルを送信
              # 以下の画像送信コードはコメントアウトされており、一時的に無効化されています。
              # for i in range(0, len(image_paths)):
              #     path = image_paths[i]
              #     try:
              #         with open(path, "rb") as f:
              #             data = f.read()
              #         print(f"画像 {i+1} を送信中：{os.path.basename(path)} ({len(data)} バイト)")
              #         await realtime_client.append_image(data)
              #         await asyncio.sleep(0.1)
              #     except Exception as e:
              #         print(f"画像 {os.path.basename(path)} の送信に失敗しました：{e}")
  
              # 3.3 送信およびサーバー応答のリクエスト
              print("すべての入力を送信し、サーバー応答をリクエストしています…")
              await realtime_client.commit_audio_buffer()
              await realtime_client.create_response()
  
              print("サーバー応答音声の待機および再生中…")
              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"\n待機時間がタイムアウトしました（{max_wait_time} 秒）。次のターンに移ります。")
                      break
                  await asyncio.sleep(0.2)
  
              print("\nこのターンの音声再生が完了しました！")
              turn_counter += 1
  
      except (asyncio.CancelledError, KeyboardInterrupt):
          print("\nプログラムが中断されました。")
      except Exception as e:
          print(f"予期せぬエラーが発生しました：{e}")
      finally:
          # ------------------- 4. リソースのクリーンアップ -------------------
          print("\n接続を閉じ、リソースをクリーンアップしています…")
          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("接続を閉じました。")
  
          audio_player.stop()
          print("プログラムが終了しました。")
  
  
  if __name__ == "__main__":
      try:
          asyncio.run(interactive_test())
      except KeyboardInterrupt:
          print("\nユーザーによりプログラムが強制終了されました。")

  manual_mode.py を実行します。Enter キーを押して話しかけ、再度 Enter キーを押してモデルの音声応答を受信します。

VAD モード

session.update イベントにおいて、session.turn_detection を "server_vad" に設定することで、VAD モードを有効化します。このモードでは、サーバーが音声の開始および終了を自動検出し、それに応じて応答します。音声通話などのユースケースに適しています。

インタラクションフローは以下のとおりです：

1. サーバーが音声の開始を検出し、input_audio_buffer.speech_started イベントを送信します。

2. クライアントは任意のタイミングで、input_audio_buffer.append イベントおよび input_image_buffer.append イベントを送信し、オーディオおよびイメージをバッファーに追加できます。

   input_image_buffer.append イベントを送信する前に、少なくとも 1 回の input_audio_buffer.append イベントを送信する必要があります。

3. サーバーが音声の終了を検出し、input_audio_buffer.speech_stopped イベントを送信します。

4. サーバーがオーディオバッファーを確定するために、input_audio_buffer.committed イベントを送信します。

5. サーバーが、バッファーから生成されたユーザーのメッセージ項目を含む conversation.item.created イベントを送信します。

手動モード

session.update イベントにおいて、session.turn_detection を null に設定することで、手動モードを有効化します。このモードでは、クライアントが明示的に input_audio_buffer.commit イベントおよび response.create イベントを送信し、サーバーからの応答を要求します。プッシュ・トゥ・トーク（PTT）シナリオ（例：チャットアプリケーションにおける音声メッセージ送信）に適しています。

インタラクションフローは以下のとおりです：

1. クライアントは任意のタイミングで、input_audio_buffer.append イベントおよび input_image_buffer.append イベントを送信し、オーディオおよびイメージをバッファーに追加できます。

   input_image_buffer.append イベントを送信する前に、少なくとも 1 回の input_audio_buffer.append イベントを送信する必要があります。

2. クライアントが input_audio_buffer.commit イベントを送信し、音声およびイメージのバッファーを確定します。これにより、現在のターンにおけるすべてのユーザー入力（音声およびイメージ）が送信済みであることをサーバーに通知します。

3. サーバーが input_audio_buffer.committed イベントで応答します。

4. クライアントが response.create イベントを送信し、モデルの出力を待機します。

5. サーバーが conversation.item.created イベントで応答します。

qwen3.5-omni-realtime

qwen3-omni-flash-realtime-2025-12-01

qwen3-omni-flash-realtime、qwen3-omni-flash-realtime-2025-09-15

Qwen-Omni-Turbo-Realtime