Alibaba Cloud Model Studio:SDK de Python de CosyVoice

Singapore

wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference

Reemplace {WorkspaceId} con su ID de espacio de trabajo real.

China (Beijing)

wss://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api-ws/v1/inference

Reemplace {WorkspaceId} con su ID de espacio de trabajo real.

Sin streaming

El texto enviado en una sola llamada no debe exceder los 20,000 caracteres. Superar este límite provoca un error.

# 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/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 = "cosyvoice-v3-plus"
# Voice
voice = "longanyang"

# 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)

export DASHSCOPE_API_KEY="your-api-key"
# Replace {WorkspaceId} with your Beijing workspace ID.
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

Streaming unidireccional

El texto enviado en una sola llamada no debe exceder los 20,000 caracteres. Superar este límite provoca un error.

# 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/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 = "cosyvoice-v3-plus"
# Voice
voice = "longanyang"

# 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 bidireccional

El texto enviado por llamada no debe exceder los 20,000 caracteres. El texto acumulado no debe exceder los 200,000 caracteres.

• Durante la entrada en streaming, llame a streaming_call varias veces para enviar segmentos de texto en secuencia. El servidor realiza automáticamente la segmentación de oraciones en el texto recibido:

  • Las oraciones completas se sintetizan inmediatamente
  • Las oraciones incompletas se almacenan en búfer hasta completarse

  Cuando se llama a streaming_complete, el servidor fuerza la síntesis de todo el texto recibido pero no procesado (incluidas las oraciones incompletas).

• El intervalo entre segmentos de texto no debe exceder los 23 segundos. De lo contrario, se genera una excepción de "tiempo de espera de solicitud agotado después de 23 segundos".

  Si no hay texto para enviar, llame a streaming_complete rápidamente para finalizar la tarea.

  ImportanteLlame siempre a streaming_complete. No hacerlo puede provocar que el texto final no se convierta en voz.

  El servidor aplica un tiempo de espera de 23 segundos. Este valor no se puede modificar en el lado del cliente.

# 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/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 = "cosyvoice-v3-plus"
# Voice
voice = "longanyang"

# 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()))