Alibaba Cloud Model Studio:リアルタイム音声合成 — CosyVoice

Python

# coding=utf-8

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

# シンガポールリージョンと北京リージョンでは API キーが異なります。API キーの取得方法：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください：dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 以下はシンガポールリージョン用の URL です。北京リージョンのモデルを使用する場合は、URL を次のように置き換えてください：wss://dashscope.aliyuncs.com/api-ws/v1/inference
dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

# モデル
# モデルのバージョンごとに、対応する音声が必要です：
# cosyvoice-v3-flash/cosyvoice-v3-plus：longanyang などの音声を使用します。
# cosyvoice-v2：longxiaochun_v2 などの音声を使用します。
# 各音声は異なる言語をサポートしています。日本語や韓国語などの中国語以外の言語を合成する場合は、対応する言語をサポートする音声を選択してください。詳細については、CosyVoice 音声一覧をご参照ください。
model = "cosyvoice-v3-flash"
# 音声
voice = "longanyang"

# SpeechSynthesizer をインスタンス化し、コンストラクターで model および voice などのパラメーターを渡します。
synthesizer = SpeechSynthesizer(model=model, voice=voice)
# 合成するテキストを送信し、バイナリ形式のオーディオを取得します。
audio = synthesizer.call("今日の天気はどうですか？")
# 最初にテキストを送信すると WebSocket 接続が確立されます。そのため、最初のパケット遅延には接続確立時間が含まれます。
print('[Metric] requestId: {} , first packet delay: {} milliseconds'.format(
    synthesizer.get_last_request_id(),
    synthesizer.get_first_package_delay()))

# オーディオをローカルに保存します。
with open('output.mp3', 'wb') as f:
    f.write(audio)

Java

import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesisParam;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesizer;
import com.alibaba.dashscope.utils.Constants;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;

public class Main {
    // モデル
    // モデルのバージョンごとに、対応する音声が必要です：
    // cosyvoice-v3-flash/cosyvoice-v3-plus：longanyang などの音声を使用します。
    // cosyvoice-v2：longxiaochun_v2 などの音声を使用します。
    // 各音声は異なる言語をサポートしています。日本語や韓国語などの中国語以外の言語を合成する場合は、対応する言語をサポートする音声を選択してください。詳細については、CosyVoice 音声一覧をご参照ください。
    private static String model = "cosyvoice-v3-flash";
    // 音声
    private static String voice = "longanyang";

    public static void streamAudioDataToSpeaker() {
        // リクエストパラメーター
        SpeechSynthesisParam param =
                SpeechSynthesisParam.builder()
                        // シンガポールリージョンと北京リージョンでは API キーが異なります。API キーの取得方法：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
                        // 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください： .apiKey("sk-xxx")
                        .apiKey(System.getenv("DASHSCOPE_API_KEY"))
                        .model(model) // モデル
                        .voice(voice) // 音声
                        .build();

        // 同期モード：コールバックを無効化（第2パラメーターを null に設定）。
        SpeechSynthesizer synthesizer = new SpeechSynthesizer(param, null);
        ByteBuffer audio = null;
        try {
            // オーディオが返されるまでブロックします。
            audio = synthesizer.call("今日の天気はどうですか？");
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            // タスク終了時に WebSocket 接続を閉じます。
            synthesizer.getDuplexApi().close(1000, "bye");
        }
        if (audio != null) {
            // オーディオデータをローカルファイル "output.mp3" に保存します。
            File file = new File("output.mp3");
            // 最初にテキストを送信すると WebSocket 接続が確立されます。そのため、最初のパケット遅延には接続確立時間が含まれます。
            System.out.println(
                    "[Metric] requestId: "
                            + synthesizer.getLastRequestId()
                            + ", first packet delay (milliseconds): "
                            + synthesizer.getFirstPackageDelay());
            try (FileOutputStream fos = new FileOutputStream(file)) {
                fos.write(audio.array());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public static void main(String[] args) {
        // 以下はシンガポールリージョン用の URL です。北京リージョンのモデルを使用する場合は、URL を次のように置き換えてください： wss://dashscope.aliyuncs.com/api-ws/v1/inference
        Constants.baseWebsocketApiUrl = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference";
        streamAudioDataToSpeaker();
        System.exit(0);
    }
}

Python

Python のサンプルを実行する前に、pip を使用してサードパーティ製のオーディオ再生ライブラリをインストールしてください。

# coding=utf-8
# pyaudio のインストール手順：
# APPLE Mac OS X
#   brew install portaudio
#   pip install pyaudio
# Debian/Ubuntu
#   sudo apt-get install python-pyaudio python3-pyaudio
#   または
#   pip install pyaudio
# CentOS
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# Microsoft Windows
#   python -m pip install pyaudio

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


from http import HTTPStatus
from dashscope import Generation

# シンガポールリージョンおよび中国（北京）リージョンの API キーは異なります。API キーの取得方法については、https://www.alibabacloud.com/help/zh/model-studio/get-api-key をご参照ください。
# 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください： dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 以下はシンガポールリージョン用の URL です。中国（北京）リージョンのモデルを使用する場合は、URL を次のように置き換えてください： wss://dashscope.aliyuncs.com/api-ws/v1/inference
dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

# モデルのバージョンごとに、対応する音声バージョンが必要です：
# cosyvoice-v3-flash/cosyvoice-v3-plus：longanyang などの音声を使用します。
# cosyvoice-v2：longxiaochun_v2 などの音声を使用します。
# 各音声は異なる言語をサポートしています。日本語や韓国語などの中国語以外の言語を合成する場合は、対応する言語をサポートする音声を選択してください。詳細については、CosyVoice 音声一覧をご参照ください。
model = "cosyvoice-v3-flash"
voice = "longanyang"


class Callback(ResultCallback):
    _player = None
    _stream = None

    def on_open(self):
        print("websocket is open.")
        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 task complete successfully.")

    def on_error(self, message: str):
        print(f"speech synthesis task failed, {message}")

    def on_close(self):
        print("websocket is closed.")
        # stop player
        self._stream.stop_stream()
        self._stream.close()
        self._player.terminate()

    def on_event(self, message):
        print(f"recv speech synthsis message {message}")

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


def synthesizer_with_llm():
    callback = Callback()
    synthesizer = SpeechSynthesizer(
        model=model,
        voice=voice,
        format=AudioFormat.PCM_22050HZ_MONO_16BIT,
        callback=callback,
    )

    messages = [{"role": "user", "content": "自己紹介をお願いします"}]
    responses = Generation.call(
        model="qwen-turbo",
        messages=messages,
        result_format="message",  # 結果フォーマットを 'message' に設定
        stream=True,  # ストリーム出力を有効化
        incremental_output=True,  # 増分出力を有効化 
    )
    for response in responses:
        if response.status_code == HTTPStatus.OK:
            print(response.output.choices[0]["message"]["content"], end="")
            synthesizer.streaming_call(response.output.choices[0]["message"]["content"])
        else:
            print(
                "Request id: %s, Status code: %s, error code: %s, error message: %s"
                % (
                    response.request_id,
                    response.status_code,
                    response.code,
                    response.message,
                )
            )
    synthesizer.streaming_complete()
    print('requestId: ', synthesizer.get_last_request_id())


if __name__ == "__main__":
    synthesizer_with_llm()

Java