Alibaba Cloud Model Studio:リアルタイム音声合成 - Qwen

音声合成 - Qwen と比較して、Qwen リアルタイム音声合成は次の特徴をサポートしています:

• ストリーミングテキスト入力

  大規模言語モデル (LLM) のストリーミング出力とシームレスに統合します。テキストが生成されると同時に音声が合成されるため、インタラクティブな音声アプリケーションのリアルタイム性能が向上します。

• 双方向通信

  WebSocket プロトコルを使用して、ストリーミングテキスト入力とオーディオ出力を実現します。このメソッドは、複数の接続を確立するオーバーヘッドを回避し、レイテンシーを大幅に削減します。

サポートされるモデル

サポートされるモデルは Qwen3-TTS Realtime です。

Qwen3-TTS Realtime は 17 種類の 音声 を提供し、複数の言語や方言の合成をサポートし、出力オーディオのフォーマット、サンプルレート、話速、音量、ピッチ、ビットレートをカスタマイズできます。

Qwen-TTS Realtime は 7 種類の 音声 のみを提供し、中国語と英語のみをサポートし、出力オーディオのフォーマット、サンプルレート、話速、音量、ピッチ、ビットレートのカスタマイズはできません。

アクセス方法

Qwen リアルタイム音声合成 API は WebSocket プロトコルに基づいています。Java または Python を使用する場合、DashScope SDK を使用して WebSocket の詳細な処理を回避できます。また、任意の言語の WebSocket ライブラリを使用して接続することもできます:

• エンドポイント URL

  中国本土 (北京): wss://dashscope.aliyuncs.com/api-ws/v1/realtime

  国際 (シンガポール): wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime

• クエリパラメーター

  クエリパラメーターは `model` です。アクセスしたいモデルの名前を指定する必要があります。詳細については、「サポートされるモデル」をご参照ください。

• ヘッダー

  認証には Bearer トークンを使用します: `Authorization: Bearer DASHSCOPE_API_KEY`

  `DASHSCOPE_API_KEY` は Alibaba Cloud Model Studio から取得した API キーです。

次のコードを使用して、Qwen-TTS Realtime API との WebSocket 接続を確立できます。

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

# シンガポールリージョンと北京リージョンでは API キーが異なります。API キーを取得するには、https://www.alibabacloud.com/help/en/model-studio/get-api-key を参照してください
# 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください: API_KEY="sk-xxx"
API_KEY=os.getenv("DASHSCOPE_API_KEY")
# 以下はシンガポールリージョンの URL です。北京リージョンのモデルを使用する場合は、URL を次のように置き換える必要があります: wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime
API_URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime"

headers = [
    "Authorization: Bearer " + API_KEY
]

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

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

ws.run_forever()

{
    "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"
    }
}

使用開始

コードを実行する前に、API キーを取得して設定する必要があります。

Python のバージョンは 3.10 以降である必要があります。

次のステップに従って、Realtime API のリアルタイム音声合成機能をすばやくテストします。

1. ランタイム環境の準備

   オペレーティングシステムに基づいて pyaudio をインストールできます。

   macOS

   brew install portaudio && pip install pyaudio

   Debian/Ubuntu

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

   CentOS

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

   Windows

   pip install pyaudio

   インストールが完了したら、pip を使用して WebSocket 関連の依存関係をインストールできます:

   pip install websocket-client==1.8.0 websockets

2. クライアントの作成

   ローカルに tts_realtime_client.py という名前の Python ファイルを作成し、次のコードをファイルにコピーできます:

   tts_realtime_client.py

   # -- coding: utf-8 --
   
   import asyncio
   import websockets
   import json
   import base64
   import time
   from typing import Optional, Callable, Dict, Any
   from enum import Enum
   
   
   class SessionMode(Enum):
       SERVER_COMMIT = "server_commit"
       COMMIT = "commit"
   
   
   class TTSRealtimeClient:
       """
       TTS Realtime API と対話するためのクライアント。
   
       このクラスは、TTS Realtime API への接続、テキストデータの送信、オーディオ出力の取得、WebSocket 接続の管理を行うメソッドを提供します。
   
       Attributes:
           base_url (str):
               Realtime API のベース URL。
           api_key (str):
               ID 検証用の API キー。
           voice (str):
               音声合成のためにサーバーが使用する音声。
           mode (SessionMode):
               セッションモード。有効な値: `server_commit` と `commit`。
           audio_callback (Callable[[bytes], None]):
               オーディオデータを受信するコールバック関数。
           language_type(str)
               合成音声の言語。有効な値: Chinese, English, German, Italian, Portuguese, Spanish, Japanese, Korean, French, Russian, 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
   
   
       async def connect(self) -> None:
           """TTS Realtime 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_" + str(int(time.time() * 1000))
           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 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
                       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 complete")
                   elif event_type == "response.done":
                       self._is_responding = False
                       self._current_response_id = None
                       self._current_item_id = None
                       print("Response complete")
                   elif event_type == "session.finished":
                       print("Session finished")
   
           except websockets.exceptions.ConnectionClosed:
               print("Connection closed")
           except Exception as e:
               print("Error processing message: ", str(e))
   
   
       async def close(self) -> None:
           """WebSocket 接続を閉じます。"""
           if self.ws:
               await self.ws.close()

3. 音声合成モードの選択

   Realtime API は次の 2 つのモードをサポートしています:

   • server_commit モード

     このモードでは、クライアントはテキストを送信するだけです。サーバーはテキストをどのようにセグメント化し、いつ合成するかをインテリジェントに決定します。このモードは、GPS ナビゲーションなど、合成のリズムを手動で制御する必要がない低レイテンシーのシナリオに適しています。

   • commit モード

     このモードでは、クライアントはまずテキストをバッファーに追加し、次にサーバーをアクティブにトリガーして指定されたテキストを合成します。このモードは、ニュース放送など、文の区切りや間を詳細に制御する必要があるシナリオに適しています。

   server_commit モード

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

   server_commit.py

   import os
   import asyncio
   import logging
   import wave
   from tts_realtime_client import TTSRealtimeClient, SessionMode
   import pyaudio
   
   # QwenTTS サービス構成
   # 以下はシンガポールリージョンの URL です。北京リージョンのモデルを使用する場合は、URL を次のように置き換える必要があります: wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime
   URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime"
   # シンガポールリージョンと北京リージョンでは API キーが異なります。API キーを取得するには、https://www.alibabacloud.com/help/en/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's Model Studio is a one-stop platform for large model development and application building.",
           "Both developers and business personnel can be deeply involved in the design and construction of large model applications.", 
           "You can develop a large model application in 5 minutes through simple interface operations,",
           "or train an exclusive model in a few hours, so you can focus more on application innovation.",
       ]
   
       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",
           language_type="Chinese", # このパラメーターをテキストと同じ言語に設定すると、正しい発音と自然なトーンが得られます。
           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  # テキスト送信が完了するのを待ちます
   
       # すべてのオーディオデータが受信されるように、追加の期間待機します
       await asyncio.sleep(5)
   
       # 接続を閉じ、コンシューマータスクをキャンセルします
       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 を実行すると、Realtime API によってリアルタイムで生成された音声を聞くことができます。

   commit モード

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

   commit.py

   import os
   import asyncio
   import logging
   import wave
   from tts_realtime_client import TTSRealtimeClient, SessionMode
   import pyaudio
   
   # QwenTTS サービス構成
   # 以下はシンガポールリージョンの URL です。北京リージョンのモデルを使用する場合は、URL を次のように置き換える必要があります: wss://dashscope.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime
   URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-tts-flash-realtime"
   # シンガポールリージョンと北京リージョンでは API キーが異なります。API キーを取得するには、https://www.alibabacloud.com/help/en/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 _user_input_loop(client: TTSRealtimeClient):
       """継続的にユーザー入力を取得し、テキストを送信します。ユーザーが空のテキストを入力すると、コミットイベントを送信して現在のセッションを終了します。"""
       print("テキストを入力してください (Enter を直接押すとコミットイベントが送信され、現在のセッションが終了します。Ctrl+C または Ctrl+D を押すとプログラムが終了します):")
       
       while True:
           try:
               user_text = input("> ")
               if not user_text:  # ユーザー入力が空です
                   # 空の入力は会話の終了として扱われます: バッファーをコミット -> セッションを終了 -> ループを中断
                   logging.info("Empty input, sending commit event and finishing the current session")
                   await client.commit_text_buffer()
                   # セッションの早期終了によるオーディオの損失を防ぐため、サーバーがコミットを処理するのを短時間待ちます
                   await asyncio.sleep(0.3)
                   await client.finish_session()
                   break  # ユーザー入力ループを直接終了します。再度 Enter を押す必要はありません
               else:
                   logging.info(f"Sending text: {user_text}")
                   await client.append_text(user_text)
                   
           except EOFError:  # ユーザーが Ctrl+D を押しました
               break
           except KeyboardInterrupt:  # ユーザーが Ctrl+C を押しました
               break
       
       # セッションを終了します
       logging.info("Finishing 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",
           language_type="Chinese",  # このパラメーターをテキストと同じ言語に設定すると、正しい発音と自然なトーンが得られます。
           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 asyncio.sleep(5)
   
       # 接続を閉じ、コンシューマータスクをキャンセルします
       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() 

   commit.py を実行できます。合成するテキストを複数回入力できます。テキストを入力せずに Enter キーを押すと、スピーカーから Realtime API が返した音声が聞こえます。

インタラクションフロー