Qwen-TTS リアルタイム音声合成の WebSocket API

​

サービスエンドポイント

​

リクエストヘッダー

​

インタラクションフロー

• ServerCommit モード：サーバーがテキストセグメンテーションと合成タイミングを自動的に決定します。
• Commit モード：クライアントが各テキストセグメントを送信するタイミングを制御します。合成をトリガーするには、input_text_buffer.commit を明示的に呼び出す必要があります。

​

モードの詳細

• ServerCommit モードでは、input_text_buffer.append を呼び出してテキストをバッファリングします。システムは合成タイミングを自動的に決定します。
• ServerCommit モードで input_text_buffer.commit を呼び出すと、現在のバッファが即座に合成されます。その後、セッションは ServerCommit モードに戻ります。
• Commit モードでは、input_text_buffer.append のみでは合成はトリガーされません。合成を開始するには、input_text_buffer.commit を呼び出す必要があります。

1. 接続：クライアントが WebSocket 接続を開きます。サーバーは準備が完了すると session.created を返します。
2. セッション設定：クライアントは session.update を送信して、音声、フォーマット、モードを設定します。
3. テキスト入力：クライアントは input_text_buffer.append を送信して、テキストをバッファに追加します。
4. 合成トリガー：
   • ServerCommit モードでは、合成が自動的に開始されます。クライアントは input_text_buffer.commit を呼び出して、即座に合成を強制できます。
   • Commit モードでは、input_text_buffer.commit のみが合成をトリガーします。
5. 音声生成：サーバーは response.created を送信し、response.audio.done が送信されるまで、response.audio.delta イベントとして base64 エンコードされた音声をストリーミングします。
6. セッションの終了：クライアントは session.finish を送信してリソースを解放します。サーバーは session.finished を返し、接続を閉じます。

{
    "event_id": "event_xxx",
    "type": "session.created",
    "session": {
        "object": "realtime.session",
        "mode": "server_commit",
        "model": "qwen3-tts-flash-realtime",
        "voice": "Cherry",
        "response_format": "pcm",
        "sample_rate": 24000,
        "id": "sess_xxx"
    }
}

​

コード例

<dependencies>
    <dependency>
        <groupId>org.java-websocket</groupId>
        <artifactId>Java-WebSocket</artifactId>
        <version>1.5.3</version>
    </dependency>
    <dependency>
        <groupId>com.google.code.gson</groupId>
        <artifactId>gson</artifactId>
        <version>2.10.1</version>
    </dependency>
</dependencies>

1. クライアントの作成
   • Java
   • Python

   TTSRealtimeClient.java という名前の Java ファイルを作成し、以下のコードをファイルにコピーします:

   Copy

   import com.google.gson.Gson;
   import com.google.gson.JsonObject;
   import org.java_websocket.client.WebSocketClient;
   import org.java_websocket.handshake.ServerHandshake;
   
   import java.net.URI;
   import java.util.Base64;
   import java.util.HashMap;
   import java.util.Map;
   import java.util.concurrent.CountDownLatch;
   import java.util.function.Consumer;
   
   /**
    * TTS リアルタイム API とやり取りするためのクライアント。
    *
    * このクラスは、TTS リアルタイム API への接続、テキストデータの送信、オーディオ出力の取得、WebSocket 接続の管理を行うメソッドを提供します。
    */
   public class TTSRealtimeClient {
   
       public enum SessionMode {
           SERVER_COMMIT("server_commit"),
           COMMIT("commit");
           private final String value;
           SessionMode(String value) { this.value = value; }
           public String getValue() { return value; }
       }
   
       /**
        * オーディオコールバックインターフェイス
        */
       public interface AudioCallback {
           void onAudio(byte[] audioData);
       }
   
       private final String baseUrl;
       private final String apiKey;
       private final String voice;
       private final SessionMode mode;
       private final String languageType;
       private final AudioCallback audioCallback;
       private final Gson gson = new Gson();
   
       private WebSocketClient ws;
       private CountDownLatch responseDoneLatch;
       private CountDownLatch sessionFinishedLatch;
   
       public TTSRealtimeClient(String baseUrl, String apiKey, String voice,
                                SessionMode mode, AudioCallback audioCallback,
                                String languageType) {
           this.baseUrl = baseUrl;
           this.apiKey = apiKey;
           this.voice = voice;
           this.mode = mode;
           this.audioCallback = audioCallback;
           this.languageType = languageType;
       }
   
       public TTSRealtimeClient(String baseUrl, String apiKey, String voice,
                                SessionMode mode, AudioCallback audioCallback) {
           this(baseUrl, apiKey, voice, mode, audioCallback, "Auto");
       }
   
       /**
        * TTS リアルタイム API への WebSocket 接続を確立します。
        */
       public void connect() throws Exception {
           Map<String, String> headers = new HashMap<>();
           headers.put("Authorization", "Bearer " + apiKey);
   
           responseDoneLatch = new CountDownLatch(0);
           sessionFinishedLatch = new CountDownLatch(1);
   
           ws = new WebSocketClient(new URI(baseUrl), headers) {
               @Override
               public void onOpen(ServerHandshake handshake) {
                   System.out.println("WebSocket connection established");
                   // デフォルトのセッション構成を送信します
                   JsonObject session = new JsonObject();
                   session.addProperty("mode", mode.getValue());
                   session.addProperty("voice", TTSRealtimeClient.this.voice);
                   session.addProperty("language_type", languageType);
                   session.addProperty("response_format", "pcm");
                   session.addProperty("sample_rate", 24000);
                   updateSession(session);
               }
   
               @Override
               public void onMessage(String message) {
                   JsonObject event = gson.fromJson(message, JsonObject.class);
                   String eventType = event.has("type") ? event.get("type").getAsString() : "";
   
                   if (!"response.audio.delta".equals(eventType)) {
                       System.out.println("Received event: " + eventType);
                   }
   
                   switch (eventType) {
                       case "error":
                           System.err.println("Error: " + event.get("error"));
                           break;
                       case "session.created":
                           System.out.println("Session created, ID: " +
                               event.getAsJsonObject("session").get("id").getAsString());
                           break;
                       case "session.updated":
                           System.out.println("Session updated, ID: " +
                               event.getAsJsonObject("session").get("id").getAsString());
                           break;
                       case "input_text_buffer.committed":
                           System.out.println("Text buffer committed, item ID: " + event.get("item_id"));
                           break;
                       case "input_text_buffer.cleared":
                           System.out.println("Text buffer cleared");
                           break;
                       case "response.created":
                           System.out.println("Response created, ID: " +
                               event.getAsJsonObject("response").get("id").getAsString());
                           responseDoneLatch = new CountDownLatch(1);
                           break;
                       case "response.output_item.added":
                           System.out.println("Output item added, ID: " +
                               event.getAsJsonObject("item").get("id").getAsString());
                           break;
                       case "response.audio.delta":
                           if (audioCallback != null) {
                               byte[] audioBytes = Base64.getDecoder().decode(
                                   event.get("delta").getAsString());
                               audioCallback.onAudio(audioBytes);
                           }
                           break;
                       case "response.audio.done":
                           System.out.println("Audio generation completed");
                           break;
                       case "response.done":
                           System.out.println("Response completed");
                           responseDoneLatch.countDown();
                           break;
                       case "session.finished":
                           System.out.println("Session finished");
                           sessionFinishedLatch.countDown();
                           break;
                   }
               }
   
               @Override
               public void onClose(int code, String reason, boolean remote) {
                   System.out.println("Connection closed: " + reason);
               }
   
               @Override
               public void onError(Exception ex) {
                   System.err.println("WebSocket error: " + ex.getMessage());
               }
           };
           ws.connectBlocking();
       }
   
       /**
        * サーバーにイベントを送信します。
        */
       public void sendEvent(JsonObject event) {
           String eventId = "event_" + java.util.UUID.randomUUID().toString();
           event.addProperty("event_id", eventId);
           System.out.println("Sending event: type=" + event.get("type").getAsString()
               + ", event_id=" + eventId);
           ws.send(gson.toJson(event));
       }
   
       /**
        * セッション構成を更新します。
        */
       public void updateSession(JsonObject config) {
           JsonObject event = new JsonObject();
           event.addProperty("type", "session.update");
           event.add("session", config);
           System.out.println("Updating session configuration: " + event);
           sendEvent(event);
       }
   
       /**
        * API にテキストデータを送信します。
        */
       public void appendText(String text) {
           JsonObject event = new JsonObject();
           event.addProperty("type", "input_text_buffer.append");
           event.addProperty("text", text);
           sendEvent(event);
       }
   
       /**
        * テキストバッファーをコミットして処理をトリガーします。
        */
       public void commitTextBuffer() {
           JsonObject event = new JsonObject();
           event.addProperty("type", "input_text_buffer.commit");
           sendEvent(event);
       }
   
       /**
        * テキストバッファーをクリアします。
        */
       public void clearTextBuffer() {
           JsonObject event = new JsonObject();
           event.addProperty("type", "input_text_buffer.clear");
           sendEvent(event);
       }
   
       /**
        * セッションを終了します。
        */
       public void finishSession() {
           JsonObject event = new JsonObject();
           event.addProperty("type", "session.finish");
           sendEvent(event);
       }
   
       /**
        * response.done イベントを待ちます。
        */
       public void waitForResponseDone() throws InterruptedException {
           responseDoneLatch.await();
       }
   
       /**
        * session.finished イベントを待ちます。
        */
       public void waitForSessionFinished() throws InterruptedException {
           sessionFinishedLatch.await();
       }
   
       /**
        * WebSocket 接続をクローズします。
        */
       public void close() {
           if (ws != null) {
               ws.close();
           }
       }
   }
   

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

   Copy

   # -- coding: utf-8 --
   
   import asyncio
   import websockets
   import json
   import base64
   import time
   import uuid
   from typing import Optional, Callable, Dict, Any
   from enum import Enum
   
   class SessionMode(Enum):
       SERVER_COMMIT = "server_commit"
       COMMIT = "commit"
   
   class TTSRealtimeClient:
       """
       TTS リアルタイム API とやり取りするためのクライアント。
   
       このクラスは、TTS リアルタイム API への接続、テキストデータの送信、
       オーディオ出力の受信、WebSocket 接続の管理を行うメソッドを提供します。
   
       Attributes:
           base_url (str):
               リアルタイム API のベース URL。
           api_key (str):
               認証用の API キー。
           voice (str):
               サーバー側音声合成で使用する音声。
           mode (SessionMode):
               セッションモード。server_commit または commit。
           audio_callback (Callable[[bytes], None]):
               オーディオデータを受信するためのコールバック関数。
           language_type(str)
               合成音声の言語。オプション: 中国語、英語、ドイツ語、イタリア語、ポルトガル語、スペイン語、日本語、韓国語、フランス語、ロシア語、Auto
       """
   
       def __init__(
               self,
               base_url: str,
               api_key: str,
               voice: str = "Cherry",
               mode: SessionMode = SessionMode.SERVER_COMMIT,
               audio_callback: Optional[Callable[[bytes], None]] = None,
           language_type: str = "Auto"):
           self.base_url = base_url
           self.api_key = api_key
           self.voice = voice
           self.mode = mode
           self.ws = None
           self.audio_callback = audio_callback
           self.language_type = language_type
   
           # 現在の応答状態
           self._current_response_id = None
           self._current_item_id = None
           self._is_responding = False
           self._response_done_future = None
   
       async def connect(self) -> None:
           """TTS リアルタイム API との WebSocket 接続を確立します。"""
           headers = {
               "Authorization": f"Bearer {self.api_key}"
           }
   
           self.ws = await websockets.connect(self.base_url, additional_headers=headers)
   
           # デフォルトのセッション構成を設定します
           await self.update_session({
               "mode": self.mode.value,
               "voice": self.voice,
               "language_type": self.language_type,
               "response_format": "pcm",
               "sample_rate": 24000
           })
   
       async def send_event(self, event) -> None:
           """サーバーにイベントを送信します。"""
           event['event_id'] = "event_" + uuid.uuid4().hex
           print(f"Sending event: type={event['type']}, event_id={event['event_id']}")
           await self.ws.send(json.dumps(event))
   
       async def update_session(self, config: Dict[str, Any]) -> None:
           """セッション構成を更新します。"""
           event = {
               "type": "session.update",
               "session": config
           }
           print("Updating session configuration: ", event)
           await self.send_event(event)
   
       async def append_text(self, text: str) -> None:
           """API にテキストデータを送信します。"""
           event = {
               "type": "input_text_buffer.append",
               "text": text
           }
           await self.send_event(event)
   
       async def commit_text_buffer(self) -> None:
           """テキストバッファーをコミットして処理をトリガーします。"""
           event = {
               "type": "input_text_buffer.commit"
           }
           await self.send_event(event)
   
       async def clear_text_buffer(self) -> None:
           """テキストバッファーをクリアします。"""
           event = {
               "type": "input_text_buffer.clear"
           }
           await self.send_event(event)
   
       async def finish_session(self) -> None:
           """セッションを終了します。"""
           event = {
               "type": "session.finish"
           }
           await self.send_event(event)
   
       async def wait_for_response_done(self):
           """response.done イベントを待ちます"""
           if self._response_done_future:
               await self._response_done_future
   
       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 != "response.audio.delta":
                       print(f"Received event: {event_type}")
   
                   if event_type == "error":
                       print("Error: ", event.get('error', {}))
                       continue
                   elif event_type == "session.created":
                       print("Session created, ID: ", event.get('session', {}).get('id'))
                   elif event_type == "session.updated":
                       print("Session updated, ID: ", event.get('session', {}).get('id'))
                   elif event_type == "input_text_buffer.committed":
                       print("Text buffer committed, item ID: ", event.get('item_id'))
                   elif event_type == "input_text_buffer.cleared":
                       print("Text buffer cleared")
                   elif event_type == "response.created":
                       self._current_response_id = event.get("response", {}).get("id")
                       self._is_responding = True
                       # response.done を待つための新しい future を作成します
                       self._response_done_future = asyncio.Future()
                       print("Response created, ID: ", self._current_response_id)
                   elif event_type == "response.output_item.added":
                       self._current_item_id = event.get("item", {}).get("id")
                       print("Output item added, ID: ", self._current_item_id)
                   # オーディオデルタを処理します
                   elif event_type == "response.audio.delta" and self.audio_callback:
                       audio_bytes = base64.b64decode(event.get("delta", ""))
                       self.audio_callback(audio_bytes)
                   elif event_type == "response.audio.done":
                       print("Audio generation completed")
                   elif event_type == "response.done":
                       self._is_responding = False
                       self._current_response_id = None
                       self._current_item_id = None
                       # future を完了としてマークします
                       if self._response_done_future and not self._response_done_future.done():
                           self._response_done_future.set_result(True)
                       print("Response completed")
                   elif event_type == "session.finished":
                       print("Session finished")
   
           except websockets.exceptions.ConnectionClosed:
               print("Connection closed")
           except Exception as e:
               print("Error handling messages: ", str(e))
   
       async def close(self) -> None:
           """WebSocket 接続をクローズします。"""
           if self.ws:
               await self.ws.close()
   

2. 音声合成モードの選択 リアルタイム API は 2 つのモードをサポートしています:
   • サーバーコミットモード サーバーがテキスト分割と合成タイミングを自動的に処理します。クライアントはテキストを送信するだけです。低遅延が求められるシナリオ（GPS ナビゲーションなど）に適しています。
   • コミットモード クライアントがバッファーにテキストを追加し、明示的に合成をトリガーします。ニュース放送など、文の分割制御を正確に行う必要があるシナリオに適しています。
   • サーバーコミットモード
   • コミットモード

   • Java
   • Python

   TTSRealtimeClient.java と同じディレクトリに、ServerCommit.java という名前の別の Java ファイルを作成し、以下のコードをファイルにコピーします:

   Copy

   import javax.sound.sampled.*;
   import java.io.*;
   import java.util.ArrayList;
   import java.util.List;
   import java.util.concurrent.ConcurrentLinkedQueue;
   import java.util.concurrent.atomic.AtomicBoolean;
   
   public class ServerCommit {
       // 以下はシンガポールリージョンの構成です。
       private static final String URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime";
       // API キーはシンガポールリージョンと中国 (北京) リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
       // 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: private static final String API_KEY = "sk-xxx";
       private static final String API_KEY = System.getenv("DASHSCOPE_API_KEY");
       private static final int SAMPLE_RATE = 24000;
   
       // オーディオデータキャッシュ
       private static final List<byte[]> audioChunks = new ArrayList<>();
       // リアルタイム再生キュー
       private static final ConcurrentLinkedQueue<byte[]> playbackQueue = new ConcurrentLinkedQueue<>();
       private static final AtomicBoolean playing = new AtomicBoolean(true);
   
       public static void main(String[] args) throws Exception {
           if (API_KEY == null || API_KEY.isEmpty()) {
               throw new IllegalStateException("Please set the DASHSCOPE_API_KEY environment variable");
           }
   
           // オーディオ再生を初期化します
           AudioFormat format = new AudioFormat(SAMPLE_RATE, 16, 1, true, false);
           DataLine.Info info = new DataLine.Info(SourceDataLine.class, format);
           SourceDataLine audioLine = (SourceDataLine) AudioSystem.getLine(info);
           audioLine.open(format);
           audioLine.start();
   
           // 再生スレッドを開始します
           Thread playerThread = new Thread(() -> {
               while (playing.get() || !playbackQueue.isEmpty()) {
                   byte[] chunk = playbackQueue.poll();
                   if (chunk != null) {
                       audioLine.write(chunk, 0, chunk.length);
                   } else {
                       try { Thread.sleep(10); } catch (InterruptedException ignored) {}
                   }
               }
           });
           playerThread.start();
   
           // TTS クライアントを作成します
           TTSRealtimeClient client = new TTSRealtimeClient(
               URL, API_KEY, "Cherry",
               TTSRealtimeClient.SessionMode.SERVER_COMMIT,
               audioData -> {
                   playbackQueue.add(audioData);
                   audioChunks.add(audioData);
                   System.out.println("Received audio data: " + audioData.length + " bytes");
               }
           );
   
           client.connect();
   
           // テキストフラグメントを送信します
           String[] textFragments = {
               "Alibaba Cloud の大規模言語モデルプラットフォームである Model Studio は、大規模言語モデルアプリケーションを開発・構築するためのオールインワンプラットフォームです。",
               "開発者とビジネスユーザーの両方が、大規模言語モデルアプリケーションの設計と開発に深く関与できます。",
               "シンプルなインターフェイスを使用して、5 分で大規模言語モデルアプリケーションを開発できます。",
               "または、数時間で専用モデルをトレーニングでき、アプリケーションの革新により多くのエネルギーを注げます。"
           };
   
           System.out.println("Starting to send text...");
           for (String text : textFragments) {
               System.out.println("Sending fragment: " + text);
               client.appendText(text);
               Thread.sleep(100);
           }
   
           Thread.sleep(1000);
           client.finishSession();
   
           // 応答の完了を待ちます
           client.waitForResponseDone();
           client.waitForSessionFinished();
           client.close();
   
           // 再生の完了を待ちます
           playing.set(false);
           playerThread.join();
           audioLine.drain();
           audioLine.close();
   
           // オーディオファイルを保存します
           saveWav("output.wav");
           System.out.println("Done");
       }
   
       private static void saveWav(String filename) throws IOException {
           if (audioChunks.isEmpty()) {
               System.out.println("No audio data to save");
               return;
           }
           ByteArrayOutputStream bos = new ByteArrayOutputStream();
           for (byte[] chunk : audioChunks) {
               bos.write(chunk);
           }
           byte[] allAudio = bos.toByteArray();
           AudioFormat format = new AudioFormat(SAMPLE_RATE, 16, 1, true, false);
           AudioInputStream ais = new AudioInputStream(
               new ByteArrayInputStream(allAudio), format, allAudio.length / 2);
           new File("outputs").mkdirs();
           AudioSystem.write(ais, AudioFileFormat.Type.WAVE,
               new File("outputs/" + filename));
           System.out.println("Audio saved to: outputs/" + filename);
       }
   }
   

   ServerCommit.java をコンパイルして実行し、リアルタイム API によって生成されたオーディオをリアルタイムで聞きます。

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

   Copy

   import os
   import asyncio
   import logging
   import wave
   from tts_realtime_client import TTSRealtimeClient, SessionMode
   import pyaudio
   
   # QwenTTS サービス構成
   # 以下はシンガポールリージョンの構成です。
   URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime"
   # API キーはシンガポールリージョンと北京リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
   # 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: API_KEY="sk-xxx"
   API_KEY = os.getenv("DASHSCOPE_API_KEY")
   
   if not API_KEY:
       raise ValueError("Please set DASHSCOPE_API_KEY environment variable")
   
   # オーディオデータを収集します
   _audio_chunks = []
   # リアルタイム再生関連
   _AUDIO_SAMPLE_RATE = 24000
   _audio_pyaudio = pyaudio.PyAudio()
   _audio_stream = None  # 実行時にオープンされます
   
   def _audio_callback(audio_bytes: bytes):
       """TTSRealtimeClient オーディオコールバック: リアルタイム再生とキャッシュ"""
       global _audio_stream
       if _audio_stream is not None:
           try:
               _audio_stream.write(audio_bytes)
           except Exception as exc:
               logging.error(f"PyAudio playback error: {exc}")
       _audio_chunks.append(audio_bytes)
       logging.info(f"Received audio chunk: {len(audio_bytes)} bytes")
   
   def _save_audio_to_file(filename: str = "output.wav", sample_rate: int = 24000) -> bool:
       """収集したオーディオデータを WAV ファイルとして保存します"""
       if not _audio_chunks:
           logging.warning("No audio data to save")
           return False
   
       try:
           audio_data = b"".join(_audio_chunks)
           with wave.open(filename, 'wb') as wav_file:
               wav_file.setnchannels(1)  # モノラル
               wav_file.setsampwidth(2)  # 16 ビット
               wav_file.setframerate(sample_rate)
               wav_file.writeframes(audio_data)
           logging.info(f"Audio saved to: {filename}")
           return True
       except Exception as exc:
           logging.error(f"Failed to save audio: {exc}")
           return False
   
   async def _produce_text(client: TTSRealtimeClient):
       """サーバーにテキストフラグメントを送信します"""
       text_fragments = [
           "Alibaba Cloud の大規模言語モデルプラットフォームである Model Studio は、大規模言語モデルアプリケーションを開発・構築するためのオールインワンプラットフォームです。",
           "開発者とビジネスユーザーの両方が、大規模言語モデルアプリケーションの設計と開発に深く関与できます。",
           "シンプルなインターフェイスを使用して、5 分で大規模言語モデルアプリケーションを開発できます。",
           "または、数時間で専用モデルをトレーニングでき、アプリケーションの革新により多くのエネルギーを注げます。",
       ]
   
       logging.info("Sending text fragments…")
       for text in text_fragments:
           logging.info(f"Sending fragment: {text}")
           await client.append_text(text)
           await asyncio.sleep(0.1)  # フラグメント間の短い遅延
   
       # サーバーが内部処理を完了するのを待ってからセッションを終了します
       await asyncio.sleep(1.0)
       await client.finish_session()
   
   async def _run_demo():
       """完全なデモを実行します"""
       global _audio_stream
       # PyAudio 出力ストリームを開きます
       _audio_stream = _audio_pyaudio.open(
           format=pyaudio.paInt16,
           channels=1,
           rate=_AUDIO_SAMPLE_RATE,
           output=True,
           frames_per_buffer=1024
       )
   
       client = TTSRealtimeClient(
           base_url=URL,
           api_key=API_KEY,
           voice="Cherry",
           mode=SessionMode.SERVER_COMMIT,
           audio_callback=_audio_callback
       )
   
       # 接続を確立します
       await client.connect()
   
       # メッセージ処理とテキスト送信を並行して実行します
       consumer_task = asyncio.create_task(client.handle_messages())
       producer_task = asyncio.create_task(_produce_text(client))
   
       await producer_task  # テキスト送信の完了を待ちます
   
       # response.done を待ちます
       await client.wait_for_response_done()
   
       # 接続をクローズし、コンシューマータスクをキャンセルします
       await client.close()
       consumer_task.cancel()
   
       # オーディオストリームをクローズします
       if _audio_stream is not None:
           _audio_stream.stop_stream()
           _audio_stream.close()
       _audio_pyaudio.terminate()
   
       # オーディオデータを保存します
       os.makedirs("outputs", exist_ok=True)
       _save_audio_to_file(os.path.join("outputs", "qwen_tts_output.wav"))
   
   def main():
       """同期エントリポイント"""
       logging.basicConfig(
           level=logging.INFO,
           format='%(asctime)s [%(levelname)s] %(message)s',
           datefmt='%Y-%m-%d %H:%M:%S'
       )
       logging.info("Starting QwenTTS Realtime Client demo…")
       asyncio.run(_run_demo())
   
   if __name__ == "__main__":
       main()
   

   server_commit.py を実行して、リアルタイム API によって生成されたオーディオをリアルタイムで聞きます。

   • Java
   • Python

   TTSRealtimeClient.java と同じディレクトリに、Commit.java という名前の別の Java ファイルを作成し、以下のコードをファイルにコピーします:

   Copy

   import javax.sound.sampled.*;
   import java.io.*;
   import java.util.ArrayList;
   import java.util.List;
   import java.util.Scanner;
   import java.util.concurrent.ConcurrentLinkedQueue;
   import java.util.concurrent.atomic.AtomicBoolean;
   
   public class Commit {
       // 以下はシンガポールリージョンの構成です。
       private static final String URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime";
       // API キーはシンガポールリージョンと中国 (北京) リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
       // 環境変数を設定していない場合は、次の行を中国 Model Studio API キーに置き換えてください: private static final String API_KEY = "sk-xxx";
       private static final String API_KEY = System.getenv("DASHSCOPE_API_KEY");
       private static final int SAMPLE_RATE = 24000;
   
       private static final List<byte[]> audioChunks = new ArrayList<>();
       private static final ConcurrentLinkedQueue<byte[]> playbackQueue = new ConcurrentLinkedQueue<>();
       private static final AtomicBoolean playing = new AtomicBoolean(true);
   
       public static void main(String[] args) throws Exception {
           if (API_KEY == null || API_KEY.isEmpty()) {
               throw new IllegalStateException("Please set the DASHSCOPE_API_KEY environment variable");
           }
   
           // オーディオ再生を初期化します
           AudioFormat format = new AudioFormat(SAMPLE_RATE, 16, 1, true, false);
           DataLine.Info info = new DataLine.Info(SourceDataLine.class, format);
           SourceDataLine audioLine = (SourceDataLine) AudioSystem.getLine(info);
           audioLine.open(format);
           audioLine.start();
   
           // 再生スレッドを開始します
           Thread playerThread = new Thread(() -> {
               while (playing.get() || !playbackQueue.isEmpty()) {
                   byte[] chunk = playbackQueue.poll();
                   if (chunk != null) {
                       audioLine.write(chunk, 0, chunk.length);
                   } else {
                       try { Thread.sleep(10); } catch (InterruptedException ignored) {}
                   }
               }
           });
           playerThread.start();
   
           // TTS クライアントを作成します（コミットモード）
           TTSRealtimeClient client = new TTSRealtimeClient(
               URL, API_KEY, "Cherry",
               TTSRealtimeClient.SessionMode.COMMIT,
               audioData -> {
                   playbackQueue.add(audioData);
                   audioChunks.add(audioData);
                   System.out.println("Received audio data: " + audioData.length + " bytes");
               }
           );
   
           client.connect();
   
           // インタラクティブ入力
           System.out.println("Enter text (press Enter directly to send a commit event and end the session, press Ctrl+D to exit the program):");
           Scanner scanner = new Scanner(System.in);
           while (true) {
               System.out.print("> ");
               if (!scanner.hasNextLine()) {
                   client.finishSession();
                   break;
               }
               String userText = scanner.nextLine();
               if (userText.isEmpty()) {
                   // 空の入力: バッファーをコミットしてセッションを終了します
                   System.out.println("Empty input, sending commit event and ending the session");
                   client.commitTextBuffer();
                   Thread.sleep(300);
                   client.finishSession();
                   break;
               } else {
                   System.out.println("Sending text: " + userText);
                   client.appendText(userText);
               }
           }
           scanner.close();
   
           // 応答の完了を待ちます
           client.waitForResponseDone();
           client.waitForSessionFinished();
           client.close();
   
           // 再生の完了を待ちます
           playing.set(false);
           playerThread.join();
           audioLine.drain();
           audioLine.close();
   
           // オーディオファイルを保存します
           saveWav("output.wav");
           System.out.println("Done");
       }
   
       private static void saveWav(String filename) throws IOException {
           if (audioChunks.isEmpty()) {
               System.out.println("No audio data to save");
               return;
           }
           ByteArrayOutputStream bos = new ByteArrayOutputStream();
           for (byte[] chunk : audioChunks) {
               bos.write(chunk);
           }
           byte[] allAudio = bos.toByteArray();
           AudioFormat format = new AudioFormat(SAMPLE_RATE, 16, 1, true, false);
           AudioInputStream ais = new AudioInputStream(
               new ByteArrayInputStream(allAudio), format, allAudio.length / 2);
           new File("outputs").mkdirs();
           AudioSystem.write(ais, AudioFileFormat.Type.WAVE,
               new File("outputs/" + filename));
           System.out.println("Audio saved to: outputs/" + filename);
       }
   }
   

   Commit.java をコンパイルして実行します。複数回テキストを入力して合成できます。テキストを入力せずに Enter を押すと、リアルタイム API から返されたオーディオがスピーカーから再生されます。

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

   Copy

   import os
   import asyncio
   import logging
   import wave
   from tts_realtime_client import TTSRealtimeClient, SessionMode
   import pyaudio
   
   # QwenTTS サービスの構成
   # 以下はシンガポールリージョンの構成です。
   URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime"
   # API キーはシンガポールリージョンと北京リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
   # 環境変数を構成していない場合は、次の行を中国 Model Studio API キーに置き換えてください: API_KEY="sk-xxx"
   API_KEY = os.getenv("DASHSCOPE_API_KEY")
   
   if not API_KEY:
       raise ValueError("DASHSCOPE_API_KEY 環境変数を設定してください")
   
   # 音声データを収集
   _audio_chunks = []
   _AUDIO_SAMPLE_RATE = 24000
   _audio_pyaudio = pyaudio.PyAudio()
   _audio_stream = None
   
   def _audio_callback(audio_bytes: bytes):
       """TTSRealtimeClient 音声コールバック: リアルタイム再生とキャッシュ"""
       global _audio_stream
       if _audio_stream is not None:
           try:
               _audio_stream.write(audio_bytes)
           except Exception as exc:
               logging.error(f"PyAudio 再生エラー: {exc}")
       _audio_chunks.append(audio_bytes)
       logging.info(f"受信した音声チャンク: {len(audio_bytes)} バイト")
   
   def _save_audio_to_file(filename: str = "output.wav", sample_rate: int = 24000) -> bool:
       """収集した音声データを WAV ファイルとして保存"""
       if not _audio_chunks:
           logging.warning("保存する音声データがありません")
           return False
   
       try:
           audio_data = b"".join(_audio_chunks)
           with wave.open(filename, 'wb') as wav_file:
               wav_file.setnchannels(1)  # モノラル
               wav_file.setsampwidth(2)  # 16 ビット
               wav_file.setframerate(sample_rate)
               wav_file.writeframes(audio_data)
           logging.info(f"音声を保存しました: {filename}")
           return True
       except Exception as exc:
           logging.error(f"音声の保存に失敗しました: {exc}")
           return False
   
   async def _user_input_loop(client: TTSRealtimeClient):
       """ユーザー入力を継続的に取得してテキストを送信します。ユーザーが空のテキストを入力した場合、コミットイベントを送信して現在のセッションを終了します"""
       print("テキストを入力してください (直接 Enter キーを押してコミットイベントを送信し、現在のセッションを終了します。プログラムを終了するには Ctrl+C または Ctrl+D を押してください):")
   
       while True:
           try:
               user_text = input("> ")
               if not user_text:  # ユーザー入力が空
                   # 空の入力は会話の終了として扱われます: コミットバッファー -> セッション終了 -> ループ中断
                   logging.info("空の入力です。コミットイベントを送信して現在のセッションを終了します")
                   await client.commit_text_buffer()
                   # サーバーがコミットを処理する時間を確保し、音声を失う可能性のある早期のセッション終了を防ぎます
                   await asyncio.sleep(0.3)
                   await client.finish_session()
                   break  # ユーザー入力ループを直接終了し、再度 Enter キーを押す必要はありません
               else:
                   logging.info(f"テキストを送信します: {user_text}")
                   await client.append_text(user_text)
   
           except EOFError:  # ユーザーが Ctrl+D を押しました
               break
           except KeyboardInterrupt:  # ユーザーが Ctrl+C を押しました
               break
   
       # セッション終了
       logging.info("セッションを終了しています...")
   async def _run_demo():
       """完全なデモを実行します"""
       global _audio_stream
       # PyAudio 出力ストリームを開く
       _audio_stream = _audio_pyaudio.open(
           format=pyaudio.paInt16,
           channels=1,
           rate=_AUDIO_SAMPLE_RATE,
           output=True,
           frames_per_buffer=1024
       )
   
       client = TTSRealtimeClient(
           base_url=URL,
           api_key=API_KEY,
           voice="Cherry",
           mode=SessionMode.COMMIT,  # COMMIT モードに変更
           audio_callback=_audio_callback
       )
   
       # 接続を確立
       await client.connect()
   
       # メッセージ処理とユーザー入力を並行して実行
       consumer_task = asyncio.create_task(client.handle_messages())
       producer_task = asyncio.create_task(_user_input_loop(client))
   
       await producer_task  # ユーザー入力の完了を待ちます
   
       # 応答完了を待ちます
       await client.wait_for_response_done()
   
       # 接続を閉じてコンシューマータスクをキャンセル
       await client.close()
       consumer_task.cancel()
   
       # 音声ストリームを閉じる
       if _audio_stream is not None:
           _audio_stream.stop_stream()
           _audio_stream.close()
       _audio_pyaudio.terminate()
   
       # 音声データを保存
       os.makedirs("outputs", exist_ok=True)
       _save_audio_to_file(os.path.join("outputs", "qwen_tts_output.wav"))
   
   def main():
       logging.basicConfig(
           level=logging.INFO,
           format='%(asctime)s [%(levelname)s] %(message)s',
           datefmt='%Y-%m-%d %H:%M:%S'
       )
       logging.info("QwenTTS リアルタイムクライアントデモを開始しています…")
       asyncio.run(_run_demo())
   
   if __name__ == "__main__":
       main()
   

   commit.py を実行します。複数回テキストを入力して合成できます。テキストを入力せずに Enter を押すと、リアルタイム API から返されたオーディオがスピーカーから再生されます。