Qwen-TTS realtime Python SDK

import base64
import os
import threading
import dashscope
from dashscope.audio.qwen_tts_realtime import *


qwen_tts_realtime: QwenTtsRealtime = None
text_to_synthesize = [
  'This is the first sentence.',
  'This is the second sentence.',
  'This is the third sentence.',
]

DO_VIDEO_TEST = False

def init_dashscope_api_key():
  """
    Set your DashScope API key. More information:
    https://github.com/aliyun/alibabacloud-bailian-speech-demo/blob/master/PREREQUISITES.md
  """

  if 'DASHSCOPE_API_KEY' in os.environ:
    dashscope.api_key = os.environ[
      'DASHSCOPE_API_KEY']  # Load API key from environment variable DASHSCOPE_API_KEY
  else:
    dashscope.api_key = 'your-dashscope-api-key'  # Set API key manually



class MyCallback(QwenTtsRealtimeCallback):
  def __init__(self):
    super().__init__()
    self.response_counter = 0
    self.complete_event = threading.Event()
    self.file = open(f'result_{self.response_counter}_24k.pcm', 'wb')

  def reset_event(self):
    self.response_counter += 1
    self.file = open(f'result_{self.response_counter}_24k.pcm', 'wb')
    self.complete_event = threading.Event()

  def on_open(self) -> None:
    print('connection opened, init player')

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

  def on_event(self, response: str) -> None:
    try:
      global qwen_tts_realtime
      type = response['type']
      if 'session.created' == type:
        print('start session: {}'.format(response['session']['id']))
      if 'response.audio.delta' == type:
        recv_audio_b64 = response['delta']
        self.file.write(base64.b64decode(recv_audio_b64))
      if 'response.done' == type:
        print(f'response {qwen_tts_realtime.get_last_response_id()} done')
        self.complete_event.set()
        self.file.close()
      if 'session.finished' == type:
        print('session finished')
        self.complete_event.set()
    except Exception as e:
      print('[Error] {}'.format(e))
      return

  def wait_for_response_done(self):
    self.complete_event.wait()


if __name__  == '__main__':
  init_dashscope_api_key()

  print('Initializing ...')

  callback = MyCallback()

  qwen_tts_realtime = QwenTtsRealtime(
    # To use instruction control, replace the model with qwen3-tts-instruct-flash-realtime
    model='qwen3-tts-flash-realtime',
    callback=callback,
    url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime'
    )

  qwen_tts_realtime.connect()
  qwen_tts_realtime.update_session(
    voice = 'Cherry',
    response_format = AudioFormat.PCM_24000HZ_MONO_16BIT,
    # To use instruction control, uncomment the following lines and replace the model with qwen3-tts-instruct-flash-realtime
    # instructions='Speak quickly with a rising intonation, suitable for introducing fashion products.',
    # optimize_instructions=True,
    mode = 'commit'
  )
  print(f'send text: {text_to_synthesize[0]}')
  qwen_tts_realtime.append_text(text_to_synthesize[0])
  qwen_tts_realtime.commit()
  callback.wait_for_response_done()
  callback.reset_event()

  print(f'send text: {text_to_synthesize[1]}')
  qwen_tts_realtime.append_text(text_to_synthesize[1])
  qwen_tts_realtime.commit()
  callback.wait_for_response_done()
  callback.reset_event()

  print(f'send text: {text_to_synthesize[2]}')
  qwen_tts_realtime.append_text(text_to_synthesize[2])
  qwen_tts_realtime.commit()
  callback.wait_for_response_done()

  qwen_tts_realtime.finish()
  print('[Metric] session: {}, first audio delay: {}'.format(
          qwen_tts_realtime.get_session_id(),
          qwen_tts_realtime.get_first_audio_delay(),
          ))