Python SDK - Alibaba Cloud Model Studio: Central de Documentação do Alibaba Cloud

Guia do usuário: Para apresentações dos modelos e recomendações de seleção, consulte Síntese de fala em tempo real – Qwen ou Síntese de fala – Qwen.

Pré-requisitos

Requer o DashScope Python SDK 1.25.11 ou posterior.

Primeiros passos

Server commit mode

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

qwen_tts_realtime: QwenTtsRealtime = None
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!'
]

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

    # API keys differ between the Singapore and Beijing regions. Get an API key: https://www.alibabacloud.com/help/en/model-studio/get-api-key
    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):
        self.complete_event = threading.Event()
        self.file = open('result_24k.pcm', 'wb')

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

    def on_close(self, close_status_code, close_msg) -> None:
        self.file.close()
        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')
            if 'session.finished' == type:
                print('session finished')
                self.complete_event.set()
        except Exception as e:
            print('[Error] {}'.format(e))
            return

    def wait_for_finished(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, 
        # Singapore region
        url='wss://dashscope.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 = 'server_commit'        
    )
    for text_chunk in text_to_synthesize:
        print(f'send text: {text_chunk}')
        qwen_tts_realtime.append_text(text_chunk)
        time.sleep(0.1)
    qwen_tts_realtime.finish()
    callback.wait_for_finished()
    print('[Metric] session: {}, first audio delay: {}'.format(
                    qwen_tts_realtime.get_session_id(), 
                    qwen_tts_realtime.get_first_audio_delay(),
                    ))

Commit mode

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

    # API keys differ between the Singapore and Beijing regions. Get an API key: https://www.alibabacloud.com/help/en/model-studio/get-api-key
    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,
        # Singapore region
        url='wss://dashscope.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(),
                    ))

Server commit mode

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

qwen_tts_realtime: QwenTtsRealtime = None
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!'
]

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

    # API keys differ between the Singapore and Beijing regions. Get an API key: https://www.alibabacloud.com/help/en/model-studio/get-api-key
    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):
        self.complete_event = threading.Event()
        self.file = open('result_24k.pcm', 'wb')

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

    def on_close(self, close_status_code, close_msg) -> None:
        self.file.close()
        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')
            if 'session.finished' == type:
                print('session finished')
                self.complete_event.set()
        except Exception as e:
            print('[Error] {}'.format(e))
            return

    def wait_for_finished(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,
        # Singapore region
        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 = 'server_commit'        
    )
    for text_chunk in text_to_synthesize:
        print(f'send text: {text_chunk}')
        qwen_tts_realtime.append_text(text_chunk)
        time.sleep(0.1)
    qwen_tts_realtime.finish()
    callback.wait_for_finished()
    print('[Metric] session: {}, first audio delay: {}'.format(
                    qwen_tts_realtime.get_session_id(), 
                    qwen_tts_realtime.get_first_audio_delay(),
                    ))

Commit mode

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

    # API keys differ between the Singapore and Beijing regions. Get an API key: https://www.alibabacloud.com/help/en/model-studio/get-api-key
    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, 
        # Singapore region
        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(),
                    ))

Acesse o GitHub para baixar mais códigos de exemplo.

Parâmetros de solicitação

Defina estes parâmetros no construtor QwenTtsRealtime:

Parâmetro	Tipo	Obrigatório	Descrição
model	str	Sim	Nome do modelo (consulte Modelos compatíveis).
url	str	Sim	China (Pequim): `wss://dashscope.aliyuncs.com/api-ws/v1/realtime` Singapura: `wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime`

Configure estes parâmetros usando update_session:

Parâmetro	Tipo	Obrigatório	Descrição
voice	str	Sim	Voz usada na síntese de fala. Para mais informações, consulte Vozes compatíveis. Há suporte a vozes do sistema e personalizadas: Vozes do sistema: Disponíveis apenas para as séries de modelos Qwen3-TTS-Instruct-Flash-Realtime, Qwen3-TTS-Flash-Realtime e Qwen-TTS-Realtime. Para amostras de voz, consulte Vozes compatíveis. Vozes personalizadas Vozes personalizadas pelo recurso Clonagem de Voz (Qwen): Disponíveis apenas para a série de modelos Qwen3-TTS-VC-Realtime. Vozes personalizadas pelo recurso Design de Voz (Qwen): Disponíveis apenas para a série de modelos Qwen3-TTS-VD-Realtime.
language_type	str	Não	Define o idioma do áudio sintetizado. O valor padrão é `Auto`. `Auto`: Use este valor quando o idioma do texto for incerto ou houver vários idiomas. O modelo ajusta automaticamente a pronúncia para diferentes segmentos de idioma no texto, mas não garante precisão total. Idioma específico: Use esta opção para textos em um único idioma. Especificar um idioma melhora significativamente a qualidade da síntese e geralmente produz resultados melhores que `Auto`. Os valores válidos incluem: `Chinese` `English` `German` `Italian` `Portuguese` `Spanish` `Japanese` `Korean` `French` `Russian`
mode	str	Não	Modo de interação. Valores válidos: `server_commit` (padrão): O servidor determina automaticamente quando sintetizar a fala, equilibrando latência e qualidade. Recomendado para a maioria dos cenários. `commit`: O cliente aciona manualmente a síntese. Este modo oferece a menor latência, mas exige que você gerencie a integridade das frases por conta própria.
format	str	Não	Formato da saída de áudio do modelo. Formatos compatíveis: `pcm` (padrão) `wav` `mp3` `opus` Qwen-TTS-Realtime (consulte Modelos compatíveis) aceita apenas `pcm`.
sample_rate	int	Não	Taxa de amostragem da saída de áudio do modelo, em Hz. Taxas de amostragem compatíveis: 8000 16000 24000 (padrão) 48000 Qwen-TTS-Realtime (consulte Modelos compatíveis) aceita apenas 24000.
speech_rate	float	Não	Velocidade da fala do áudio. O valor 1,0 representa a velocidade normal. Valores menores que 1,0 tornam a fala mais lenta; valores maiores que 1,0 a tornam mais rápida. Valor padrão: 1,0. Intervalo válido: [0,5, 2,0]. Qwen-TTS-Realtime (consulte Modelos compatíveis) não aceita este parâmetro.
volume	int	Não	Volume do áudio. Valor padrão: 50. Intervalo válido: [0, 100]. Qwen-TTS-Realtime (consulte Modelos compatíveis) não aceita este parâmetro.
pitch_rate	float	Não	Tom do áudio sintetizado. Valor padrão: 1,0. Intervalo válido: [0,5, 2,0]. Qwen-TTS-Realtime (consulte Modelos compatíveis) não aceita este parâmetro.
bit_rate	int	Não	Define a taxa de bits do áudio em kbps. Uma taxa de bits maior resulta em melhor qualidade de áudio e tamanho de arquivo maior. Este parâmetro está disponível apenas quando o formato de áudio (`response_format`) está definido como `opus`. Valor padrão: 128. Intervalo válido: [6, 510]. Qwen-TTS-Realtime (consulte Modelos compatíveis) não aceita este parâmetro.
instructions	str	Não	Define as instruções. Para mais informações, consulte Síntese de fala em tempo real - Qwen. Valor padrão: None. O parâmetro permanece inativo se não for definido. Limite de comprimento: Não pode exceder 1600 tokens. Idiomas aceitos: Apenas chinês e inglês. Escopo: Disponível apenas para a série de modelos Qwen3-TTS-Instruct-Flash-Realtime.
optimize_instructions	bool	Não	Define se as `instructions` devem ser otimizadas para melhorar a naturalidade e a expressividade da síntese de fala. Valor padrão: False Comportamento: Quando definido como True, o sistema aprimora e reescreve semanticamente as `instructions` para gerar instruções internas otimizadas para síntese de fala. Cenários: Recomendado para cenários que exigem expressão vocal de alta qualidade e granularidade fina. Dependência: Depende da definição do parâmetro `instructions`. Se `instructions` estiver vazio, este parâmetro não terá efeito. Escopo: Disponível apenas para a série de modelos Qwen3-TTS-Instruct-Flash-Realtime.

Principais interfaces

Classe QwenTtsRealtime

Importação: from dashscope.audio.qwen_tts_realtime import QwenTtsRealtime

Assinatura do método	Eventos de resposta do servidor (entregues via callback)	Descrição
`def connect(self) -> None`	session.created Sessão criada session.updated Configuração da sessão atualizada	Conecta ao servidor.
`def update_session(self, voice: str, response_format: AudioFormat = AudioFormat. PCM_24000HZ_MONO_16BIT, mode: str = 'server_commit', language_type : str = "Chinese", **kwargs) -> None`	session.updated Configuração da sessão atualizada	Atualiza as configurações padrão da sessão. Para detalhes dos parâmetros, consulte Parâmetros de solicitação. Após a conexão, o servidor retorna configurações padrão de entrada e saída. Chame este método imediatamente após conectar para atualizar esses padrões. O servidor valida os parâmetros ao receber um evento session.update. Se a validação falhar, ele retorna um erro; caso contrário, atualiza a configuração da sessão.
`def append_text(self, text: str) -> None`	None	Adiciona um trecho de texto ao buffer de entrada na nuvem (armazenamento temporário para texto antes do envio). No modo server_commit, o servidor decide quando enviar e sintetizar o texto no buffer. No modo commit, o cliente deve acionar a síntese chamando commit.
`def clear_appended_text(self, ) -> None`	input_text_buffer.cleared Limpa o texto recebido pelo servidor	Exclui todo o texto no buffer da nuvem.
`def commit(self, ) -> None`	input_text_buffer.committed Envia o texto e aciona a síntese de fala response.output_item.added Novo conteúdo de saída adicionado response.content_part.added Novo conteúdo adicionado à mensagem do assistente response.audio.delta Áudio incremental gerado pelo modelo response.audio.done Geração de áudio concluída response.content_part.done Streaming de conteúdo de áudio para mensagem do assistente concluído response.output_item.done Streaming de todo o item de saída para mensagem do assistente concluído response.done Resposta concluída	Envia todo o texto no buffer da nuvem e sintetiza imediatamente. Retorna um erro se o buffer estiver vazio. No modo server_commit, os clientes não precisam enviar este evento. O servidor envia o buffer automaticamente. No modo commit, os clientes devem chamar commit para acionar a síntese de fala.
`def finish(self, ) -> None`	session.finished Resposta concluída	Encerra a tarefa.
`def close(self, ) -> None`	None	Fecha a conexão.
`def get_session_id(self) -> str`	None	Obtém o ID da sessão da tarefa atual.
`def get_last_response_id(self) -> str`	None	Obtém o ID da resposta mais recente.
`def get_first_audio_delay(self)`	None	Obtém o atraso até a chegada do primeiro pacote de áudio.

Interface de callback (QwenTtsRealtimeCallback)

O servidor envia respostas e dados por meio de callbacks. Implemente métodos de callback para lidar com eles.

Importação: from dashscope.audio.qwen_tts_realtime import QwenTtsRealtimeCallback

Método	Parâmetros	Valor de retorno	Descrição
`def on_open(self) -> None`	None	None	Chamado quando a conexão com o servidor é estabelecida.
`def on_event(self, message: str) -> None`	message: Evento de resposta do servidor.	None	Inclui respostas de chamadas de API, texto gerado pelo modelo e áudio. Consulte Eventos do servidor.
`def on_close(self, close_status_code, close_msg) -> None`	close_status_code: Código de status de fechamento do WebSocket. close_msg: Mensagem de fechamento do WebSocket.	None	Chamado após o servidor fechar a conexão.