大模型服务平台百炼：Python SDK

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

import dashscope

text_to_synthesize = [
    '对吧~我就特别喜欢这种超市，',
    '尤其是过年的时候',
    '去逛超市',
    '就会觉得',
    '超级超级开心！',
    '想买好多好多的东西呢！'
]

# The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# If you have not configured the environment variable, replace the following line with your Alibaba Cloud Model Studio API Key: api_key="sk-xxx"
client = dashscope.Client(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # The following is the configuration for the Singapore (ap-southeast-1) region. When calling, replace "{WorkspaceId}" with your actual workspace ID. Configurations differ by region.
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# Receiving thread: iterate over the server events and write the audio to the file incrementally
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 ...')

    # Create a real-time speech synthesis session: entering the with context automatically establishes the connection
    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)
            # Notify the server that the input is complete (in server_commit mode the server decides when to synthesize)
            session.end_session()
            receiver.result()
        print('[Metric] session: {}, first audio delay: {}'.format(
            session.session_id,
            session.last_first_audio_delay,
        ))

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

import dashscope

text_to_synthesize = [
    '这是第一句话。',
    '这是第二句话。',
    '这是第三句话。',
]

# The API Key differs between the Singapore and Beijing regions. Get your API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# If you have not configured the environment variable, replace the following line with your Alibaba Cloud Model Studio API Key: api_key="sk-xxx"
client = dashscope.Client(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # The following is the configuration for the Singapore (ap-southeast-1) region. When calling, replace "{WorkspaceId}" with your actual workspace ID. Configurations differ by region.
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# Receiving thread: iterate over the server events and write the audio to the file incrementally
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 ...')

    # Create a real-time speech synthesis session: entering the with context automatically establishes the connection
    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='commit'
        )
        with ThreadPoolExecutor(max_workers=1) as pool:
            receiver = pool.submit(receive_events, session)
            for text in text_to_synthesize:
                print(f'send text: {text}')
                session.send_event({'type': 'input_text_buffer.append', 'text': text})
                # In commit mode the client triggers the synthesis
                session.send_event({'type': 'input_text_buffer.commit'})
                time.sleep(0.1)
            session.end_session()
            receiver.result()
        print('[Metric] session: {}, first audio delay: {}'.format(
            session.session_id,
            session.last_first_audio_delay,
        ))