Alibaba Cloud Model Studio:CosyVoice Python SDK

Utilize o DashScope Python SDK para integrar a síntese de fala em tempo real do CosyVoice à sua aplicação por meio dos modos sem streaming, streaming unidirecional ou streaming bidirecional.

Endpoint do service

O endpoint do SDK é determinado pelos parâmetros workspace e region do construtor do Client (correspondem aos endereços abaixo, que incluem o WorkspaceId). Para alternar para outra região, altere os parâmetros de construção do Client para os valores da região desejada.

export DASHSCOPE_API_KEY="your-api-key"
# Substitua {WorkspaceId} pelo ID do workspace da região Beijing.
export DASHSCOPE_HTTP_BASE_URL="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"
dashscope speech-synthesis create -m cosyvoice-v3-plus -t "Hello world" --voice longanyang

Recurso principal (client.audio.tts)

Caminho do recurso: client.audio.tts

Função: síntese de fala baseada em WebSocket, com três formas de chamada: síntese em uma etapa (synthesize), síntese em streaming (stream) e sessão com streaming bidirecional (connect).

Ponto de entrada de chamada

# Síntese em uma etapa: retorna os bytes de áudio completos
audio = client.audio.tts.synthesize(
    text,                      # text é obrigatório (argumento posicional)
    model="cosyvoice-v3-plus", # O modelo ao qual a voz pertence
    voice="longanyang",        # A voz
    format="mp3",              # Formato de codificação de áudio, padrão "mp3"
    sample_rate=22050,         # Taxa de amostragem, padrão 22050
)

# Síntese em streaming: produz bytes de áudio bloco a bloco (grave/reproduza enquanto recebe)
for chunk in client.audio.tts.stream(text, model=..., voice=...):
    ...

# Sessão com streaming bidirecional: envie o texto em segmentos; a conexão é estabelecida automaticamente ao entrar no contexto with
with client.audio.tts.connect(model=..., voice=...) as session:
    session.send_text(text)    # Pode ser chamado várias vezes para anexar texto
    session.stop()             # Notifica o servidor de que a entrada terminou
    for chunk in session.audio_chunks():
        ...

synthesize() - sem streaming

Assinatura do método:

def synthesize(self, text: str, *, model: str = "cosyvoice-v1",
               voice: str = "longxiaochun", format: str = "mp3",
               sample_rate: int = 22050, volume: int = None,
               rate: float = None, pitch: float = None,
               language_hints: list = None, enable_ssml: bool = None,
               parameters: Dict = None, ...) -> bytes

Parâmetros:

Valor de retorno: bytes, todos os dados de áudio.

Descrição: chamada sem streaming, bloqueia e retorna todos os dados de áudio de uma só vez. Adequada para textos curtos sem requisitos estritos de tempo real. synthesize e stream criam a própria sessão a cada chamada; não é necessário reutilizar instâncias.

stream() - com streaming

Assinatura do método:

def stream(self, text: str, *, model: str, voice: str,
           format: str = "mp3", sample_rate: int = 22050, ...) -> Iterator[bytes]

Parâmetros:

Valor de retorno: Iterator[bytes], dados de áudio produzidos bloco a bloco.

Descrição: síntese em streaming que envia o texto completo de uma só vez e produz o áudio bloco a bloco (Iterator[bytes]). Adequada para cenários de síntese de fala de textos curtos que exigem alta capacidade de resposta em tempo real. Para enviar o texto em segmentos, use a sessão com streaming bidirecional (connect).

send_text() - entrada em streaming (método da sessão)

Envia o texto em segmentos em uma sessão com streaming bidirecional (client.audio.tts.connect(...)).

Assinatura do método:

def send_text(self, text: str) -> None

Parâmetros: text é o segmento de texto atual a ser sintetizado. Chame send_text várias vezes para anexar texto; máximo por chamada: 20.000 caracteres, máximo cumulativo: 200.000 caracteres. O envio de texto e o recebimento de áudio devem ocorrer de forma concorrente (use em conjunto com audio_chunks()/events() e stop).

stop() - finalizar entrada em streaming (método da sessão)

Assinatura do método:

def stop(self) -> None

Descrição: Notifica o servidor de que todo o texto foi enviado. Esta chamada não bloqueia a thread atual; a thread receptora deve continuar consumindo audio_chunks()/events() até que o texto restante seja sintetizado e todos os dados de áudio sejam retornados. A falha ao chamar este método pode resultar na não conversão do texto final em fala.

cancel() - cancelar síntese em streaming (método da sessão)

Assinatura do método:

def cancel(self) -> None

Descrição: Cancela a rodada atual da tarefa de síntese de fala em streaming. Após chamar este método, o SDK encerra imediatamente a tarefa atual. Após o cancelamento, a sessão atual não pode mais enviar texto; crie uma nova sessão para continuar a síntese.

Métricas e respostas da sessão (atributos da sessão)

session.first_package_delay
session.usage
session.last_response

• first_package_delay: float, a latência em milissegundos entre o envio do texto e o recebimento do primeiro bloco de dados de áudio. Deve ser lido após a conclusão da síntese.
• usage: dict, informações de uso da tarefa de síntese mais recente.
• last_response: dict, mensagem de resposta da tarefa de síntese mais recente (com as chaves header/payload). O request_id da solicitação mais recente pode ser obtido no header de session.last_response, útil para solução de problemas e correlação de logs.

Parâmetros de solicitação

Os parâmetros a seguir são definidos por meio dos métodos synthesize/stream/connect de client.audio.tts, para controlar o modelo, a voz, o formato e as características do áudio (model, voice, format, sample_rate, volume, rate, pitch, language_hints e enable_ssml são parâmetros nomeados; os demais parâmetros são passados por meio do dicionário parameters).

audio = client.audio.tts.synthesize(
    "Texto a sintetizar",
    model="cosyvoice-v3-plus",
    voice="longanyang",
    parameters={"bit_rate": 128},
)

audio = client.audio.tts.stream(
    "Texto a sintetizar",
    model="cosyvoice-v3-flash",
    voice="your_voice",  # Uma voz do sistema ou clonada que suporta carimbos de data/hora no nível da palavra
    parameters={"word_timestamp_enabled": True},
)

audio = client.audio.tts.synthesize(
    "Texto a sintetizar",
    model="cosyvoice-v3-plus",
    voice="longanyang",
    parameters={
        "enable_aigc_tag": True,
        "aigc_propagator": "your_propagator",
        "aigc_propagate_id": "your_propagate_id"
    },
)

audio = client.audio.tts.synthesize(
    "Texto a sintetizar",
    model="cosyvoice-v3.5-plus",
    voice="cosyvoice-v3.5-plus-vd-announcer-xxxxxx", # Voz
    parameters={
        "hot_fix": {
            "pronunciation": [{"weather": "tian1 qi4"}],
            "replace": [{"today": "gold day"}]
        }
    },
)

audio = client.audio.tts.synthesize(
    "Texto a sintetizar",
    model="cosyvoice-v3-flash",
    voice="your_voice",  # Uma voz clonada de cosyvoice-v3-flash
    parameters={"enable_markdown_filter": True},
)

Sessão e iteração de eventos

A síntese em streaming não usa mais interfaces de callback: o áudio e os eventos do servidor são recebidos por meio dos métodos de iteração da sessão. stream()/synthesize() não exigem gerenciamento de sessão; o ciclo de vida de uma sessão com streaming bidirecional (client.audio.tts.connect(...)) é: entrada no contexto with com conexão estabelecida automaticamente → múltiplas chamadas de send_text() → stop() → leitura de todos os resultados → saída do contexto com fechamento automático da conexão.

audio_chunks() - iterar dados de áudio

Assinatura do método:

def audio_chunks(self) -> Iterator[bytes]

Valor de retorno: Iterator[bytes], fragmentos de dados binários de áudio produzidos bloco a bloco, no formato especificado pelo parâmetro format.

Descrição: Itere na thread receptora para receber os dados de áudio bloco a bloco; durante a síntese, vários blocos são produzidos. Grave os dados em um arquivo ou envie-os a um dispositivo de reprodução. A mesma sessão permite apenas um leitor de events() ou audio_chunks().

events() - iterar eventos do servidor

Assinatura do método:

def events(self) -> Generator[Union[Dict, bytes], None, None]

Valor de retorno: produz simultaneamente eventos de resposta do servidor (dict, contendo header (informações da solicitação) e payload (informações de saída)) e bytes de áudio. O campo payload.output contém o tipo de evento, o texto original e outros detalhes. Consulte campo output nas mensagens de events().

Descrição: Itere as respostas do servidor na thread receptora; elas contêm as informações de saída dos eventos de síntese (tipo de evento, texto original, informações da frase etc.). Acesse diretamente payload.output para obter detalhes.

Tratamento de erros

Quando ocorre um erro durante a síntese, a camada WebSocket lança uma exceção DashScopeError e a conexão é fechada automaticamente após a exceção ser lançada. Use try/except no envio/recebimento para capturar e registrar os erros para solução de problemas:

from dashscope.v2 import DashScopeError

try:
    ...
except DashScopeError as e:
    print(e.code, e.message, e.request_id)

Ciclo de vida da sessão

• A conexão é estabelecida automaticamente na entrada do contexto with (inicialize fluxos de saída de áudio ou abra recursos de arquivo antes disso).
• Após todo o texto ser sintetizado e todos os dados de áudio serem produzidos, a iteração de events()/audio_chunks() termina; depois disso é possível ler métricas de desempenho como first_package_delay.
• A conexão é fechada automaticamente na saída do contexto with (libere dispositivos de reprodução de áudio e outros recursos em seguida).

Campo output nas mensagens de events()

O evento dict produzido pela iteração de events() da sessão contém um campo payload.output com informações sobre o evento de síntese. Utilize este campo para acompanhar o progresso da síntese e recuperar detalhes por frase. A estrutura do campo output é a seguinte:

Exemplo de mensagem:

{
  "header": {
    "task_id": "xxx",
    "event": "result-generated",
    "attributes": {}
  },
  "payload": {
    "output": {
      "type": "sentence-begin",
      "original_text": "How is the weather today?",
      "sentence": {
        "index": 0,
        "words": []
      }
    }
  }
}

Exemplo de análise:

def handle_event(event):
    output = (event.get('payload') or {}).get('output') or {}
    event_type = output.get('type', '')
    original_text = output.get('original_text', '')
    if event_type:
        print(f'Event type: {event_type}, original text: {original_text}')

Exemplos de código

O SDK fornece as interfaces principais da síntese de fala e suporta os seguintes modos de chamada:

• Sem streaming: uma chamada bloqueante que envia o texto completo de uma só vez e retorna o áudio integral diretamente. Mais adequado para síntese de fala de textos curtos.
• Streaming unidirecional: uma chamada não bloqueante que envia o texto completo de uma só vez e recebe os dados iterando os blocos de áudio retornados por stream. Mais adequado para cenários de texto curto que exigem alta capacidade de resposta em tempo real.
• Streaming bidirecional: uma chamada não bloqueante que envia texto em vários segmentos e recebe em tempo real o fluxo de áudio sintetizado incrementalmente por meio da iteração de audio_chunks()/events() da sessão. Mais adequado para cenários de texto longo que exigem alta capacidade de resposta em tempo real.

# coding=utf-8

import os

import dashscope

# As API Keys das regiões Singapore e Beijing são diferentes. Obtenha uma API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# Se a variável de ambiente não estiver configurada, substitua a linha abaixo pela sua API Key do Model Studio: api_key="sk-xxx"
# A configuração abaixo refere-se à região Singapura. Substitua {WorkspaceId} pelo ID real do seu workspace. O valor de region varia conforme a região.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# Modelo
model = "cosyvoice-v3-plus"
# Voz
voice = "longanyang"

# Envie o texto para síntese e obtenha o áudio binário
audio = client.audio.tts.synthesize("How is the weather today?", model=model, voice=voice)

# Salve o áudio em um arquivo local
with open('output.mp3', 'wb') as f:
    f.write(audio)

# coding=utf-8

import os

from datetime import datetime

import dashscope

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# As API Keys das regiões Singapore e Beijing são diferentes. Obtenha uma API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# Se a variável de ambiente não estiver configurada, substitua a linha abaixo pela sua API Key do Model Studio: api_key="sk-xxx"
# A configuração abaixo refere-se à região Singapura. Substitua {WorkspaceId} pelo ID real do seu workspace. O valor de region varia conforme a região.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# Modelo
model = "cosyvoice-v3-plus"
# Voz
voice = "longanyang"

# Envie o texto completo de uma só vez e receba o áudio binário bloco a bloco
with open("output.mp3", "wb") as f:
    for chunk in client.audio.tts.stream("How is the weather today?", model=model, voice=voice):
        print(get_timestamp() + " Comprimento do áudio binário: " + str(len(chunk)))
        f.write(chunk)
print("Síntese de fala concluída, todos os resultados foram recebidos: " + get_timestamp())

# coding=utf-8
#
# Instruções de instalação do pyaudio:
# No macOS, execute os seguintes comandos:
#   brew install portaudio
#   pip install pyaudio
# No Debian/Ubuntu, execute os seguintes comandos:
#   sudo apt-get install python-pyaudio python3-pyaudio
#   ou
#   pip install pyaudio
# No CentOS, execute os seguintes comandos:
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# No Microsoft Windows, execute o seguinte comando:
#   python -m pip install pyaudio

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

import dashscope

from datetime import datetime

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# As API Keys das regiões Singapore e Beijing são diferentes. Obtenha uma API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# Se a variável de ambiente não estiver configurada, substitua a linha abaixo pela sua API Key do Model Studio: api_key="sk-xxx"
# A configuração abaixo refere-se à região Singapura. Substitua {WorkspaceId} pelo ID real do seu workspace. O valor de region varia conforme a região.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# Modelo
model = "cosyvoice-v3-plus"
# Voz
voice = "longanyang"

test_text = [
    "Streaming text-to-speech SDK,",
    "converts input text",
    "into binary audio data.",
    "Compared with non-streaming speech synthesis,",
    "streaming synthesis offers better real-time performance.",
    "Users hear near-synchronous audio output while typing,",
    "greatly improving interaction experience",
    "and reducing wait time.",
    "Ideal for large language model (LLM) integration,",
    "where text is streamed for speech synthesis.",
]

# Crie uma sessão com streaming bidirecional: a conexão é estabelecida automaticamente ao entrar no contexto with
with client.audio.tts.connect(
        model=model,
        voice=voice,
        format="pcm",
        sample_rate=22050,
) as session:
    # Inicialize o player
    player = pyaudio.PyAudio()
    stream = player.open(format=pyaudio.paInt16, channels=1, rate=22050, output=True)

    # Thread receptora: itere os dados de áudio e reproduza-os em tempo real
    def receive_audio():
        for chunk in session.audio_chunks():
            print(get_timestamp() + " Comprimento do áudio binário: " + str(len(chunk)))
            stream.write(chunk)

    with ThreadPoolExecutor(max_workers=1) as pool:
        receiver = pool.submit(receive_audio)
        # Envie o texto para síntese em streaming
        for text in test_text:
            session.send_text(text)
            time.sleep(0.1)
        # Encerre a síntese em streaming
        session.stop()
        receiver.result()

    print("Síntese de fala concluída, todos os resultados foram recebidos: " + get_timestamp())
    # Interrompa o player
    stream.stop_stream()
    stream.close()
    player.terminate()

    # A primeira conexão WebSocket é estabelecida quando o primeiro texto é enviado, portanto o atraso do primeiro pacote inclui o tempo de estabelecimento da conexão
    print('[Métrica] atraso do primeiro pacote: {} ms'.format(session.first_package_delay))