Alibaba Cloud Model Studio:Qwen-Audio-TTS/CosyVoice speech synthesis Python SDK

Intégrez la synthèse vocale en temps réel Qwen-Audio-TTS/CosyVoice à votre application via le SDK Python DashScope, en utilisant les modes non diffusés (non-streaming), unidirectionnel ou bidirectionnel.

Guide d'utilisation : Pour des descriptions des modèles et des recommandations de sélection, consultez la rubrique Synthèse vocale.

Endpoint du service

Le SDK utilise par défaut l'endpoint de la région Pékin. Pour basculer vers une autre région, modifiez dashscope.base_websocket_api_url avant l'initialisation.

Basculer vers la région Singapour :

import dashscope

# Set at the beginning of your code
dashscope.base_websocket_api_url = 'wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'

SpeechSynthesizer

Chemin du package : dashscope.audio.tts_v2.SpeechSynthesizer

Constructeur

SpeechSynthesizer(
    model: str,
    voice: str,
    format: AudioFormat = AudioFormat.MP3_22050HZ_MONO_256KBPS,
    volume: int = 50,
    speech_rate: float = 1.0,
    pitch_rate: float = 1.0,
    callback: ResultCallback = None)

call() - mode non diffusé (non-streaming)

Signature de la méthode :

def call(self, text: str) -> bytes

Paramètres :

Valeur de retour : bytes contenant les données audio complètes.

Description : Cet appel bloquant renvoie l'intégralité des données audio en une seule fois. Il est idéal pour les textes courts ne nécessitant pas de diffusion en temps réel. Réinitialisez l'instance SpeechSynthesizer avant chaque appel.

streaming_call() - mode diffusé (streaming)

Signature de la méthode :

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

Paramètres :

Description : Cet appel en diffusion bidirectionnelle accepte le texte par segments et fournit l'audio synthétisé en temps réel via des callbacks. Il est parfaitement adapté à l'intégration avec des grands modèles de langage où le texte est généré progressivement. Appelez streaming_complete() après avoir envoyé tout le texte.

streaming_complete() - fin de la diffusion

Signature de la méthode :

def streaming_complete(self) -> None

Description : Notifie au serveur que tout le texte a été envoyé. Bloque le thread actuel jusqu'à ce que le texte restant soit synthétisé et que toutes les données audio soient renvoyées. L'omission de cet appel peut entraîner la non-conversion en parole du texte final.

streaming_cancel() - annulation de la synthèse en diffusion

Signature de la méthode :

def streaming_cancel(self, complete_timeout_millis: int = 10000) -> None

Paramètres :

Description : Annule la session de synthèse vocale en diffusion actuelle. Après cet appel, le SDK termine immédiatement la tâche en cours. Vous pouvez démarrer une nouvelle tâche de synthèse sur la même connexion sans réinitialiser l'instance SpeechSynthesizer.

get_last_request_id() - obtention de l'ID de requête

Signature de la méthode :

def get_last_request_id(self) -> str

Valeur de retour : str contenant l'ID de la requête la plus récente. Utilisez-le pour le dépannage et le suivi.

get_first_package_delay() - obtention de la latence du premier paquet

Signature de la méthode :

def get_first_package_delay(self) -> int

Valeur de retour : int représentant le délai en millisecondes entre l'envoi du texte et la réception du premier fragment audio. Appelez cette méthode après la fin de la synthèse.

get_response() - obtention du message de réponse

Signature de la méthode :

def get_response(self) -> str

Valeur de retour : str contenant le message de réponse au format JSON de la dernière tâche de synthèse, incluant le statut de la requête et les informations de sortie.

Paramètres du constructeur

Les paramètres suivants sont définis via le constructeur SpeechSynthesizer pour contrôler le modèle, la voix, le format et les caractéristiques audio.

synthesizer = SpeechSynthesizer(
              model="qwen-audio-3.0-tts-flash",
              voice="longanhuan_v3.6",
              additional_params={"bit_rate": 128}
          )

synthesizer = SpeechSynthesizer(
              model="cosyvoice-v3-flash",
              voice="your_voice",  # Une voix système ou clonée prenant en charge les horodatages au niveau des mots
              additional_params={"word_timestamp_enabled": True}
          )

synthesizer = SpeechSynthesizer(
              model="qwen-audio-3.0-tts-flash",
              voice="longanhuan_v3.6",
              additional_params={
                  "enable_aigc_tag": True,
                  "aigc_propagator": "your_propagator",
                  "aigc_propagate_id": "your_propagate_id"
              }
          )

synthesizer = SpeechSynthesizer(
        model="cosyvoice-v3.5-plus",
        voice="cosyvoice-v3.5-plus-vd-announcer-xxxxxx", # Voice
        hot_fix={
            "pronunciation": [{"weather": "tian1 qi4"}],
            "replace": [{"today": "gold day"}]
        }
    )

synthesizer = SpeechSynthesizer(
              model="cosyvoice-v3-flash",
              voice="your_voice",  # Une voix clonée de cosyvoice-v3-flash
              additional_params={"enable_markdown_filter": True}
          )

ResultCallback

Chemin du package : dashscope.audio.tts_v2.ResultCallback

on_open() - connexion établie

Signature de la méthode :

def on_open(self) -> None

Déclenchement : lorsque la connexion WebSocket est correctement établie. Utilisez ce callback pour initialiser les flux de sortie audio ou ouvrir des ressources de fichiers.

on_event() - réception de la réponse du serveur

Signature de la méthode :

def on_event(self, message: str) -> None

Paramètres :

Déclenchement : à la réception d'une réponse du serveur. Le message est une chaîne JSON contenant la sortie de l'événement de synthèse (type d'événement, texte original, informations sur la phrase). Analysez-le avec json.loads(message) et accédez à payload.output pour obtenir les détails.

on_complete() - synthèse terminée

Signature de la méthode :

def on_complete(self) -> None

Déclenchement : lorsque tout le texte a été synthétisé et que toutes les données audio ont été transmises via on_data. Utilisez ce callback pour appeler get_first_package_delay() afin d'obtenir les métriques de performance.

on_data() - réception des données audio

Signature de la méthode :

def on_data(self, data: bytes) -> None

Paramètres :

Déclenchement : à la réception d'un fragment de données audio. Ce callback est invoqué plusieurs fois pendant la synthèse. Utilisez-le pour écrire les données dans un fichier ou les transmettre à un dispositif de lecture.

on_error() - erreur survenue

Signature de la méthode :

def on_error(self, message: str) -> None

Paramètres :

Déclenchement : lorsqu'une erreur survient pendant la synthèse. La connexion se ferme automatiquement après l'exécution de ce callback. Enregistrez l'erreur pour le dépannage.

on_close() - connexion fermée

Signature de la méthode :

def on_close(self) -> None

Déclenchement : lorsque la connexion WebSocket se ferme (normalement ou en raison d'une erreur). Utilisez ce callback pour libérer des ressources telles que les dispositifs de lecture audio.

champ output dans les messages on_event

Le message JSON reçu par le callback on_event contient un champ payload.output avec les informations de l'événement de synthèse. Utilisez ce champ pour suivre la progression de la synthèse et récupérer les détails par phrase. La structure du champ output est la suivante :

Exemple de message :

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

Exemple d'analyse :

import json

def on_event(self, message):
    data = json.loads(message)
    output = data.get('payload', {}).get('output', {})
    event_type = output.get('type', '')
    original_text = output.get('original_text', '')
    if event_type:
        print(f'Event type: {event_type}, Original text: {original_text}')

Exemples de code

Le SDK prend en charge les modes de synthèse suivants :

• Non streaming : appel bloquant qui envoie le texte complet en une seule fois et renvoie directement l'audio intégral. Idéal pour la synthèse vocale de textes courts.
• Streaming unidirectionnel : appel non bloquant qui envoie le texte complet en une seule fois et transmet les données audio (potentiellement par fragments) via une fonction de callback. Adapté aux scénarios de textes courts nécessitant une faible latence.
• Streaming bidirectionnel : appel non bloquant qui envoie le texte en plusieurs segments et transmet l'audio synthétisé de manière incrémentielle via une fonction de callback en temps réel. Conçu pour les scénarios de textes longs nécessitant une faible latence.

Non streaming

Le texte envoyé en un seul appel ne doit pas dépasser 20 000 caractères. Le dépassement de cette limite entraîne une erreur.

# coding=utf-8

import dashscope
from dashscope.audio.tts_v2 import *
import os

# The API Keys for Singapore and Beijing regions are different. Get API Key: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# If environment variable is not configured, replace the following line with your Model Studio API Key: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# The following configuration is for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
dashscope.base_websocket_api_url='wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'

# Model
model = "qwen-audio-3.0-tts-flash"
# Voice
voice = "longanhuan_v3.6"

# Instantiate SpeechSynthesizer, passing request parameters such as model and voice in the constructor
synthesizer = SpeechSynthesizer(model=model, voice=voice)
# Send text for synthesis and get binary audio
audio = synthesizer.call("How is the weather today?")
# The first text submission requires establishing a WebSocket connection, so the first packet latency includes the connection setup time
print('[Metric] requestId: {}, first packet latency: {} ms'.format(
    synthesizer.get_last_request_id(),
    synthesizer.get_first_package_delay()))

# Save audio to local file
with open('output.mp3', 'wb') as f:
    f.write(audio)

Streaming unidirectionnel

Le texte envoyé en un seul appel ne doit pas dépasser 20 000 caractères. Le dépassement de cette limite entraîne une erreur.

# coding=utf-8

import os
import json
import dashscope
from dashscope.audio.tts_v2 import *

from datetime import datetime

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

# The API Keys for Singapore and Beijing regions are different. Get API Key: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# If environment variable is not configured, replace the following line with your Model Studio API Key: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# The following configuration is for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
dashscope.base_websocket_api_url='wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'

# Model
model = "qwen-audio-3.0-tts-flash"
# Voice
voice = "longanhuan_v3.6"

# Define callback interface
class Callback(ResultCallback):
    _player = None
    _stream = None

    def on_open(self):
        self.file = open("output.mp3", "wb")
        print("Connection established: " + get_timestamp())

    def on_complete(self):
        print("Speech synthesis completed, all results received: " + get_timestamp())
        # After the task completes (on_complete callback triggered), you can call get_first_package_delay to get the latency
        # The first text submission requires establishing a WebSocket connection, so the first packet latency includes the connection setup time
        print('[Metric] requestId: {}, first packet latency: {} ms'.format(
            synthesizer.get_last_request_id(),
            synthesizer.get_first_package_delay()))

    def on_error(self, message: str):
        print(f"Speech synthesis error: {message}")

    def on_close(self):
        print("Connection closed: " + get_timestamp())
        self.file.close()

    def on_event(self, message):
        # Parse server-side events and get output information
        data = json.loads(message)
        output = data.get('payload', {}).get('output', {})
        event_type = output.get('type', '')
        original_text = output.get('original_text', '')
        if event_type:
            print(f"Event type: {event_type}, original text: {original_text}")

    def on_data(self, data: bytes) -> None:
        print(get_timestamp() + " Binary audio length: " + str(len(data)))
        self.file.write(data)

callback = Callback()

# Instantiate SpeechSynthesizer, passing request parameters such as model and voice in the constructor
synthesizer = SpeechSynthesizer(
    model=model,
    voice=voice,
    callback=callback,
)

# Send text for synthesis and get binary audio in real time via the on_data callback method
synthesizer.call("How is the weather today?")

Streaming bidirectionnel

Le texte envoyé par appel ne doit pas dépasser 20 000 caractères. Le texte cumulé ne doit pas dépasser 200 000 caractères.

• Pendant l'entrée en streaming, appelez streaming_call plusieurs fois pour soumettre les segments de texte séquentiellement. Le serveur effectue automatiquement la segmentation des phrases sur le texte reçu :

  • Les phrases complètes sont synthétisées immédiatement
  • Les phrases incomplètes sont mises en mémoire tampon jusqu'à leur achèvement

  Lorsque streaming_complete est appelé, le serveur force la synthèse de tout le texte reçu mais non traité (y compris les phrases incomplètes).

• L'intervalle entre les segments de texte ne doit pas dépasser 23 secondes. Sinon, une exception « request timeout after 23 seconds » est levée.

  S'il n'y a pas de texte à envoyer, appelez streaming_complete rapidement pour terminer la tâche.

  ImportantAppelez toujours streaming_complete. Ne pas le faire peut empêcher la conversion du texte final en parole.

  Le serveur applique un délai d'expiration de 23 secondes. Cette valeur ne peut pas être modifiée côté client.

# coding=utf-8
#
# pyaudio installation instructions:
# For macOS, run the following commands:
#   brew install portaudio
#   pip install pyaudio
# For Debian/Ubuntu, run the following commands:
#   sudo apt-get install python-pyaudio python3-pyaudio
#   or
#   pip install pyaudio
# For CentOS, run the following commands:
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# For Microsoft Windows, run the following command:
#   python -m pip install pyaudio

import os
import time
import pyaudio
import json
import dashscope
from dashscope.api_entities.dashscope_response import SpeechSynthesisResponse
from dashscope.audio.tts_v2 import *

from datetime import datetime

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

# The API Keys for Singapore and Beijing regions are different. Get API Key: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# If environment variable is not configured, replace the following line with your Model Studio API Key: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# The following configuration is for the Singapore region. Replace "{WorkspaceId}" with your actual workspace ID. The configuration varies by region.
dashscope.base_websocket_api_url='wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'

# Model
model = "qwen-audio-3.0-tts-flash"
# Voice
voice = "longanhuan_v3.6"

# Define callback interface
class Callback(ResultCallback):
    _player = None
    _stream = None

    def on_open(self):
        print("Connection established: " + get_timestamp())
        self._player = pyaudio.PyAudio()
        self._stream = self._player.open(
            format=pyaudio.paInt16, channels=1, rate=22050, output=True
        )

    def on_complete(self):
        print("Speech synthesis completed, all results received: " + get_timestamp())

    def on_error(self, message: str):
        print(f"Speech synthesis error: {message}")

    def on_close(self):
        print("Connection closed: " + get_timestamp())
        # Stop the player
        self._stream.stop_stream()
        self._stream.close()
        self._player.terminate()

    def on_event(self, message):
        # Parse server-side events and get output information
        data = json.loads(message)
        output = data.get('payload', {}).get('output', {})
        event_type = output.get('type', '')
        original_text = output.get('original_text', '')
        if event_type:
            print(f"Event type: {event_type}, original text: {original_text}")

    def on_data(self, data: bytes) -> None:
        print(get_timestamp() + " Binary audio length: " + str(len(data)))
        self._stream.write(data)

callback = Callback()

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.",
]

# Instantiate SpeechSynthesizer, passing request parameters such as model and voice in the constructor
synthesizer = SpeechSynthesizer(
    model=model,
    voice=voice,
    format=AudioFormat.PCM_22050HZ_MONO_16BIT,
    callback=callback,
)

# Send text for streaming synthesis. Get binary audio in real time via the on_data callback method
for text in test_text:
    synthesizer.streaming_call(text)
    time.sleep(0.1)
# End streaming speech synthesis
synthesizer.streaming_complete()

# The first text submission requires establishing a WebSocket connection, so the first packet latency includes the connection setup time
print('[Metric] requestId: {}, first packet latency: {} ms'.format(
    synthesizer.get_last_request_id(),
    synthesizer.get_first_package_delay()))