Python SDK

import os
import base64
import time
from concurrent.futures import ThreadPoolExecutor

import dashscope

text_to_synthesize = [
    'Right? I love supermarkets like this.',
    'Especially during Chinese New Year,',
    'I go shopping at supermarkets.',
    'And I feel',
    'absolutely thrilled!',
    'I want to buy so many things!'
]

# API キーは、シンガポールリージョンと北京リージョンで異なります。API キーの取得: https://www.alibabacloud.com/help/model-studio/get-api-key
# 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください: api_key="sk-xxx"
client = dashscope.Client(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 以下はシンガポール (ap-southeast-1) リージョンの設定です。呼び出す際に、"{WorkspaceId}" を実際のワークスペース ID に置き換えてください。設定はリージョンによって異なります。
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# 受信スレッド：サーバーイベントを反復処理し、音声をファイルに増分書き込みします
def receive_events(session):
    with open('result_24k.pcm', 'wb') as f:
        for event in session.events():
            event_type = event.get('type')
            if event_type == 'session.created':
                print('start session: {}'.format(event['session']['id']))
            elif event_type == 'response.audio.delta':
                f.write(base64.b64decode(event['delta']))
            elif event_type == 'response.done':
                print(f'response {session.last_response_id} done')
            elif event_type == 'session.finished':
                print('session finished')

if __name__  == '__main__':
    print('Initializing ...')

    # リアルタイム音声合成セッションを作成します：with コンテキストに入ると自動的に接続が確立されます
    with client.audio.realtime.connect(
            model='qwen3-tts-flash-realtime',
    ) as session:
        session.update_session(
            output_modalities=['audio'],
            voice='Cherry',
            response_format='pcm',
            sample_rate=24000,
            mode='server_commit'
        )
        with ThreadPoolExecutor(max_workers=1) as pool:
            receiver = pool.submit(receive_events, session)
            for text_chunk in text_to_synthesize:
                print(f'send text: {text_chunk}')
                session.send_event({'type': 'input_text_buffer.append', 'text': text_chunk})
                time.sleep(0.1)
            # サーバーに入力完了を通知します (server_commit モードでは、サーバーが合成タイミングを決定します)
            session.end_session()
            receiver.result()
        print('[Metric] session: {}, first audio delay: {}'.format(
            session.session_id,
            session.last_first_audio_delay,
        ))