Alibaba Cloud Model Studio:SDK Python sintesis suara Qwen-Audio-TTS/CosyVoice

Gunakan DashScope Python SDK untuk mengintegrasikan sintesis suara real-time Qwen-Audio-TTS/CosyVoice ke dalam aplikasi Anda melalui mode non-streaming, streaming satu arah, atau streaming dua arah.

Panduan pengguna: Untuk deskripsi model dan rekomendasi pemilihan, lihat Sintesis suara.

Titik akhir layanan

SDK menggunakan titik akhir Wilayah Beijing secara default. Untuk beralih ke wilayah lain, ubah dashscope.base_websocket_api_url sebelum inisialisasi.

Beralih ke wilayah Singapura:

import dashscope

# Setel di awal kode Anda
dashscope.base_websocket_api_url = 'wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'

SpeechSynthesizer

Jalur paket: dashscope.audio.tts_v2.SpeechSynthesizer

Konstruktor

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() - non-streaming

Signature metode:

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

Parameter:

Nilai kembalian: bytes yang berisi data audio lengkap.

Deskripsi: Pemanggilan blocking ini mengembalikan seluruh data audio sekaligus. Metode ini paling cocok untuk teks pendek yang tidak memerlukan streaming real-time. Inisialisasi ulang instans SpeechSynthesizer sebelum setiap pemanggilan.

streaming_call() - streaming

Signature metode:

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

Parameter:

Deskripsi: Pemanggilan streaming dua arah ini menerima teks dalam segmen dan mengirimkan audio hasil sintesis melalui callback secara real-time. Metode ini ideal untuk integrasi dengan model bahasa besar di mana teks dihasilkan secara progresif. Panggil streaming_complete() setelah mengirimkan seluruh teks.

streaming_complete() - akhiri streaming

Signature metode:

def streaming_complete(self) -> None

Deskripsi: Memberi tahu server bahwa seluruh teks telah dikirim. Memblokir thread saat ini hingga sisa teks disintesis dan seluruh data audio dikembalikan. Tidak memanggil metode ini dapat menyebabkan teks di akhir tidak dikonversi menjadi suara.

streaming_cancel() - batalkan sintesis streaming

Signature metode:

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

Parameter:

Deskripsi: Membatalkan giliran sintesis suara streaming saat ini. Setelah memanggil metode ini, SDK segera mengakhiri tugas saat ini. Anda dapat memulai tugas sintesis baru pada koneksi yang sama tanpa perlu menginisialisasi ulang instans SpeechSynthesizer.

get_last_request_id() - dapatkan ID permintaan

Signature metode:

def get_last_request_id(self) -> str

Nilai kembalian: str yang berisi ID permintaan dari permintaan terbaru. Gunakan ini untuk troubleshooting dan pelacakan.

get_first_package_delay() - dapatkan latensi paket pertama

Signature metode:

def get_first_package_delay(self) -> int

Nilai kembalian: int yang merepresentasikan delay dalam milidetik dari pengiriman teks hingga menerima chunk audio pertama. Panggil ini setelah sintesis selesai.

get_response() - dapatkan pesan respons

Signature metode:

def get_response(self) -> str

Nilai kembalian: str yang berisi pesan respons dalam format JSON dari tugas sintesis terbaru, termasuk status permintaan dan informasi output.

Parameter konstruktor

Parameter berikut ditetapkan melalui konstruktor SpeechSynthesizer untuk mengontrol model, voice, format, dan karakteristik 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",  # Voice sistem atau hasil kloning yang mendukung timestamp tingkat kata
          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",  # Voice hasil kloning dari cosyvoice-v3-flash
          additional_params={"enable_markdown_filter": True}
      )

ResultCallback

Jalur paket: dashscope.audio.tts_v2.ResultCallback

on_open() - koneksi terbentuk

Signature metode:

def on_open(self) -> None

Dipicu saat: Koneksi WebSocket berhasil dibentuk. Gunakan callback ini untuk menginisialisasi aliran output audio atau membuka sumber daya file.

on_event() - menerima respons server

Signature metode:

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

Parameter:

Dipicu saat: Respons server diterima. Pesan berupa string JSON yang berisi output event sintesis (tipe event, teks asli, informasi kalimat). Urai dengan json.loads(message) dan akses payload.output untuk detailnya.

on_complete() - sintesis selesai

Signature metode:

def on_complete(self) -> None

Dipicu saat: Seluruh teks telah disintesis dan seluruh data audio telah dikirimkan melalui on_data. Gunakan callback ini untuk memanggil get_first_package_delay() guna mendapatkan metrik performa.

on_data() - menerima data audio

Signature metode:

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

Parameter:

Dipicu saat: Chunk data audio diterima. Callback ini dipanggil beberapa kali selama sintesis. Gunakan untuk menulis data ke file atau mengirimkannya ke perangkat pemutaran.

on_error() - terjadi error

Signature metode:

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

Parameter:

Dipicu saat: Terjadi error selama sintesis. Koneksi ditutup secara otomatis setelah callback ini dijalankan. Catat error untuk troubleshooting.

on_close() - koneksi ditutup

Signature metode:

def on_close(self) -> None

Dipicu saat: Koneksi WebSocket ditutup (baik normal maupun karena error). Gunakan callback ini untuk melepas sumber daya seperti perangkat pemutaran audio.

Bidang output dalam pesan on_event

Pesan JSON yang diterima oleh callback on_event berisi bidang payload.output dengan informasi event sintesis. Gunakan bidang ini untuk melacak progres sintesis dan mengambil detail per kalimat. Struktur bidang output adalah sebagai berikut:

Contoh pesan:

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

Contoh penguraian:

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'Tipe event: {event_type}, Teks asli: {original_text}')

Contoh kode

SDK mendukung mode sintesis berikut:

• Non-streaming: Pemanggilan blocking yang mengirimkan teks lengkap sekaligus dan mengembalikan seluruh audio langsung. Paling cocok untuk sintesis suara teks pendek.
• Streaming satu arah: Pemanggilan non-blocking yang mengirimkan teks lengkap sekaligus dan mengirimkan data audio (mungkin dalam chunk) melalui fungsi callback. Paling cocok untuk skenario teks pendek yang memerlukan latensi rendah.
• Streaming dua arah: Pemanggilan non-blocking yang mengirimkan teks dalam beberapa segmen dan mengirimkan audio hasil sintesis secara inkremental melalui fungsi callback secara real-time. Paling cocok untuk skenario teks panjang yang memerlukan latensi rendah.

Non-streaming

Teks yang dikirim dalam satu pemanggilan tidak boleh melebihi 20.000 karakter. Melebihi batas ini menyebabkan error.

# coding=utf-8

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

# Kunci API untuk wilayah Singapura dan Beijing berbeda. Dapatkan Kunci API: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# Jika variabel lingkungan tidak dikonfigurasi, ganti baris berikut dengan Kunci API Model Studio Anda: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# Konfigurasi berikut untuk wilayah Singapura. Ganti "{WorkspaceId}" dengan workspace ID Anda yang sebenarnya. Konfigurasi bervariasi berdasarkan wilayah.
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"

# Buat instans SpeechSynthesizer, meneruskan parameter permintaan seperti model dan voice dalam konstruktor
synthesizer = SpeechSynthesizer(model=model, voice=voice)
# Kirim teks untuk sintesis dan dapatkan audio biner
audio = synthesizer.call("How is the weather today?")
# Pengiriman teks pertama memerlukan pembentukan koneksi WebSocket, sehingga latensi paket pertama mencakup waktu pembentukan koneksi
print('[Metric] requestId: {}, latensi paket pertama: {} ms'.format(
    synthesizer.get_last_request_id(),
    synthesizer.get_first_package_delay()))

# Simpan audio ke file lokal
with open('output.mp3', 'wb') as f:
    f.write(audio)

Streaming satu arah

Teks yang dikirim dalam satu pemanggilan tidak boleh melebihi 20.000 karakter. Melebihi batas ini menyebabkan 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

# Kunci API untuk wilayah Singapura dan Beijing berbeda. Dapatkan Kunci API: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# Jika variabel lingkungan tidak dikonfigurasi, ganti baris berikut dengan Kunci API Model Studio Anda: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# Konfigurasi berikut untuk wilayah Singapura. Ganti "{WorkspaceId}" dengan workspace ID Anda yang sebenarnya. Konfigurasi bervariasi berdasarkan wilayah.
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"

# Definisikan antarmuka callback
class Callback(ResultCallback):
    _player = None
    _stream = None

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

    def on_complete(self):
        print("Sintesis suara selesai, seluruh hasil diterima: " + get_timestamp())
        # Setelah tugas selesai (callback on_complete dipicu), Anda dapat memanggil get_first_package_delay untuk mendapatkan latensi
        # Pengiriman teks pertama memerlukan pembentukan koneksi WebSocket, sehingga latensi paket pertama mencakup waktu pembentukan koneksi
        print('[Metric] requestId: {}, latensi paket pertama: {} ms'.format(
            synthesizer.get_last_request_id(),
            synthesizer.get_first_package_delay()))

    def on_error(self, message: str):
        print(f"Error sintesis suara: {message}")

    def on_close(self):
        print("Koneksi ditutup: " + get_timestamp())
        self.file.close()

    def on_event(self, message):
        # Urai event sisi server dan dapatkan informasi output
        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"Tipe event: {event_type}, teks asli: {original_text}")

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

callback = Callback()

# Buat instans SpeechSynthesizer, meneruskan parameter permintaan seperti model dan voice dalam konstruktor
synthesizer = SpeechSynthesizer(
    model=model,
    voice=voice,
    callback=callback,
)

# Kirim teks untuk sintesis dan dapatkan audio biner secara real-time melalui metode callback on_data
synthesizer.call("How is the weather today?")

Streaming dua arah

Teks yang dikirim per pemanggilan tidak boleh melebihi 20.000 karakter. Teks kumulatif tidak boleh melebihi 200.000 karakter.

• Selama input streaming, panggil streaming_call beberapa kali untuk mengirimkan segmen teks secara berurutan. Server secara otomatis melakukan segmentasi kalimat pada teks yang diterima:

  • Kalimat lengkap langsung disintesis
  • Kalimat tidak lengkap dibuffer hingga lengkap

  Saat streaming_complete dipanggil, server memaksa mensintesis seluruh teks yang diterima namun belum diproses (termasuk kalimat tidak lengkap).

• Interval antar segmen teks tidak boleh melebihi 23 detik. Jika tidak, exception "request timeout after 23 seconds" akan muncul.

  Jika tidak ada teks yang akan dikirim, segera panggil streaming_complete untuk mengakhiri tugas.

  PentingSelalu panggil streaming_complete. Tidak melakukannya dapat menyebabkan teks di akhir tidak dikonversi menjadi suara.

  Server menerapkan timeout 23 detik. Nilai ini tidak dapat diubah di sisi klien.

# coding=utf-8
#
# Petunjuk instalasi pyaudio:
# Untuk macOS, jalankan perintah berikut:
#   brew install portaudio
#   pip install pyaudio
# Untuk Debian/Ubuntu, jalankan perintah berikut:
#   sudo apt-get install python-pyaudio python3-pyaudio
#   atau
#   pip install pyaudio
# Untuk CentOS, jalankan perintah berikut:
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# Untuk Microsoft Windows, jalankan perintah berikut:
#   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

# Kunci API untuk wilayah Singapura dan Beijing berbeda. Dapatkan Kunci API: https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# Jika variabel lingkungan tidak dikonfigurasi, ganti baris berikut dengan Kunci API Model Studio Anda: dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# Konfigurasi berikut untuk wilayah Singapura. Ganti "{WorkspaceId}" dengan workspace ID Anda yang sebenarnya. Konfigurasi bervariasi berdasarkan wilayah.
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"

# Definisikan antarmuka callback
class Callback(ResultCallback):
    _player = None
    _stream = None

    def on_open(self):
        print("Koneksi terbentuk: " + 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("Sintesis suara selesai, seluruh hasil diterima: " + get_timestamp())

    def on_error(self, message: str):
        print(f"Error sintesis suara: {message}")

    def on_close(self):
        print("Koneksi ditutup: " + get_timestamp())
        # Hentikan pemutar
        self._stream.stop_stream()
        self._stream.close()
        self._player.terminate()

    def on_event(self, message):
        # Urai event sisi server dan dapatkan informasi output
        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"Tipe event: {event_type}, teks asli: {original_text}")

    def on_data(self, data: bytes) -> None:
        print(get_timestamp() + " Panjang audio biner: " + 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.",
]

# Buat instans SpeechSynthesizer, meneruskan parameter permintaan seperti model dan voice dalam konstruktor
synthesizer = SpeechSynthesizer(
    model=model,
    voice=voice,
    format=AudioFormat.PCM_22050HZ_MONO_16BIT,
    callback=callback,
)

# Kirim teks untuk sintesis streaming. Dapatkan audio biner secara real-time melalui metode callback on_data
for text in test_text:
    synthesizer.streaming_call(text)
    time.sleep(0.1)
# Akhiri sintesis suara streaming
synthesizer.streaming_complete()

# Pengiriman teks pertama memerlukan pembentukan koneksi WebSocket, sehingga latensi paket pertama mencakup waktu pembentukan koneksi
print('[Metric] requestId: {}, latensi paket pertama: {} ms'.format(
    synthesizer.get_last_request_id(),
    synthesizer.get_first_package_delay()))