使用Python SDK實現通義千問即時語音合成-大模型服務平台百鍊-阿里雲 - Alibaba Cloud Model Studio

使用者指南：關於模型介紹和選型建議請參見即時語音合成或非即時語音合成。

前期準備

DashScope Python SDK 版本需要不低於1.25.11。

快速開始

server commit模式

import os
import base64
import threading
import time
import dashscope
from dashscope.audio.qwen_tts_realtime import *


qwen_tts_realtime: QwenTtsRealtime = None
text_to_synthesize = [
    '對吧~我就特別喜歡這種超市，',
    '尤其是過年的時候',
    '去逛超市',
    '就會覺得',
    '超級超級開心！',
    '想買好多好多的東西呢！'
]

DO_VIDEO_TEST = False

def init_dashscope_api_key():
    """
        Set your DashScope API-key. More information:
        https://github.com/aliyun/alibabacloud-bailian-speech-demo/blob/master/PREREQUISITES.md
    """

    # 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    if 'DASHSCOPE_API_KEY' in os.environ:
        dashscope.api_key = os.environ[
            'DASHSCOPE_API_KEY']  # load API-key from environment variable DASHSCOPE_API_KEY
    else:
        dashscope.api_key = 'your-dashscope-api-key'  # set API-key manually



class MyCallback(QwenTtsRealtimeCallback):
    def __init__(self):
        self.complete_event = threading.Event()
        self.file = open('result_24k.pcm', 'wb')

    def on_open(self) -> None:
        print('connection opened, init player')

    def on_close(self, close_status_code, close_msg) -> None:
        self.file.close()
        print('connection closed with code: {}, msg: {}, destroy player'.format(close_status_code, close_msg))

    def on_event(self, response: str) -> None:
        try:
            global qwen_tts_realtime
            type = response['type']
            if 'session.created' == type:
                print('start session: {}'.format(response['session']['id']))
            if 'response.audio.delta' == type:
                recv_audio_b64 = response['delta']
                self.file.write(base64.b64decode(recv_audio_b64))
            if 'response.done' == type:
                print(f'response {qwen_tts_realtime.get_last_response_id()} done')
            if 'session.finished' == type:
                print('session finished')
                self.complete_event.set()
        except Exception as e:
            print('[Error] {}'.format(e))
            return

    def wait_for_finished(self):
        self.complete_event.wait()


if __name__  == '__main__':
    init_dashscope_api_key()

    print('Initializing ...')

    callback = MyCallback()

    qwen_tts_realtime = QwenTtsRealtime(
        # 如需使用指令控制功能，請將model替換為qwen3-tts-instruct-flash-realtime
        model='qwen3-tts-flash-realtime',
        callback=callback,
        # 以下為新加坡地區的配置。
        url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime'
        )

    qwen_tts_realtime.connect()
    qwen_tts_realtime.update_session(
        voice = 'Cherry',
        response_format = AudioFormat.PCM_24000HZ_MONO_16BIT,
        # 如需使用指令控制功能，請取消下方注釋，並將model替換為qwen3-tts-instruct-flash-realtime
        # instructions='語速較快，帶有明顯的上揚語調，適合介紹時尚產品。',
        # optimize_instructions=True,
        mode = 'server_commit'        
    )
    for text_chunk in text_to_synthesize:
        print(f'send text: {text_chunk}')
        qwen_tts_realtime.append_text(text_chunk)
        time.sleep(0.1)
    qwen_tts_realtime.finish()
    callback.wait_for_finished()
    print('[Metric] session: {}, first audio delay: {}'.format(
                    qwen_tts_realtime.get_session_id(), 
                    qwen_tts_realtime.get_first_audio_delay(),
                    ))

commit模式

import base64
import os
import threading
import dashscope
from dashscope.audio.qwen_tts_realtime import *


qwen_tts_realtime: QwenTtsRealtime = None
text_to_synthesize = [
    '這是第一句話。',
    '這是第二句話。',
    '這是第三句話。',
]

DO_VIDEO_TEST = False

def init_dashscope_api_key():
    """
        Set your DashScope API-key. More information:
        https://github.com/aliyun/alibabacloud-bailian-speech-demo/blob/master/PREREQUISITES.md
    """

    # 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    if 'DASHSCOPE_API_KEY' in os.environ:
        dashscope.api_key = os.environ[
            'DASHSCOPE_API_KEY']  # load API-key from environment variable DASHSCOPE_API_KEY
    else:
        dashscope.api_key = 'your-dashscope-api-key'  # set API-key manually



class MyCallback(QwenTtsRealtimeCallback):
    def __init__(self):
        super().__init__()
        self.response_counter = 0
        self.complete_event = threading.Event()
        self.file = open(f'result_{self.response_counter}_24k.pcm', 'wb')

    def reset_event(self):
        self.response_counter += 1
        self.file = open(f'result_{self.response_counter}_24k.pcm', 'wb')
        self.complete_event = threading.Event()

    def on_open(self) -> None:
        print('connection opened, init player')

    def on_close(self, close_status_code, close_msg) -> None:
        print('connection closed with code: {}, msg: {}, destroy player'.format(close_status_code, close_msg))

    def on_event(self, response: str) -> None:
        try:
            global qwen_tts_realtime
            type = response['type']
            if 'session.created' == type:
                print('start session: {}'.format(response['session']['id']))
            if 'response.audio.delta' == type:
                recv_audio_b64 = response['delta']
                self.file.write(base64.b64decode(recv_audio_b64))
            if 'response.done' == type:
                print(f'response {qwen_tts_realtime.get_last_response_id()} done')
                self.complete_event.set()
                self.file.close()
            if 'session.finished' == type:
                print('session finished')
                self.complete_event.set()
        except Exception as e:
            print('[Error] {}'.format(e))
            return

    def wait_for_response_done(self):
        self.complete_event.wait()


if __name__  == '__main__':
    init_dashscope_api_key()

    print('Initializing ...')

    callback = MyCallback()

    qwen_tts_realtime = QwenTtsRealtime(
        # 如需使用指令控制功能，請將model替換為qwen3-tts-instruct-flash-realtime
        model='qwen3-tts-flash-realtime',
        callback=callback, 
        # 以下為新加坡地區的配置。
        url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime'
        )

    qwen_tts_realtime.connect()
    qwen_tts_realtime.update_session(
        voice = 'Cherry',
        response_format = AudioFormat.PCM_24000HZ_MONO_16BIT,
        # 如需使用指令控制功能，請取消下方注釋，並將model替換為qwen3-tts-instruct-flash-realtime
        # instructions='語速較快，帶有明顯的上揚語調，適合介紹時尚產品。',
        # optimize_instructions=True,
        mode = 'commit'        
    )
    print(f'send text: {text_to_synthesize[0]}')
    qwen_tts_realtime.append_text(text_to_synthesize[0])
    qwen_tts_realtime.commit()
    callback.wait_for_response_done()
    callback.reset_event()
    
    print(f'send text: {text_to_synthesize[1]}')
    qwen_tts_realtime.append_text(text_to_synthesize[1])
    qwen_tts_realtime.commit()
    callback.wait_for_response_done()
    callback.reset_event()

    print(f'send text: {text_to_synthesize[2]}')
    qwen_tts_realtime.append_text(text_to_synthesize[2])
    qwen_tts_realtime.commit()
    callback.wait_for_response_done()
    
    qwen_tts_realtime.finish()
    print('[Metric] session: {}, first audio delay: {}'.format(
                    qwen_tts_realtime.get_session_id(), 
                    qwen_tts_realtime.get_first_audio_delay(),
                    ))

訪問github下載更多範例程式碼。

請求參數

下述請求參數可以通過QwenTtsRealtime的構造方法進行設定。

參數	類型	是否必須	說明
model	str	是	模型名稱。參見支援的模型。
url	str	是	華北2（北京）地區：`wss://dashscope.aliyuncs.com/api-ws/v1/realtime` 新加坡地區：`wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime`

下述請求參數可以通過update_session介面配置。

參數	類型	是否必須	說明
voice	str	是	語音合成所使用的音色。參見支援的音色。支援系統音色和專屬音色：系統音色：僅限千問3-TTS-Instruct-Flash-Realtime、千問3-TTS-Flash-Realtime和千問-TTS-Realtime系列模型。音色效果請參見：支援的音色。專屬音色聲音複刻（Qwen）功能定製的音色：僅限千問3-TTS-VC-Realtime系列模型聲音設計（Qwen）功能定製的音色：僅限千問3-TTS-VD-Realtime系列模型
language_type	str	否	指定合成音訊語種，預設為 `Auto`。 `Auto`：適用無法確定文本的語種或文本包含多種語言的情境，模型會自動為文本中的不同語言片段匹配各自的發音，但無法保證發音完全精準。指定語種：適用於文本為單一語種的情境，此時指定為具體語種，能顯著提升合成品質，效果通常優於 `Auto`。可選值包括： `Chinese` `English` `German` `Italian` `Portuguese` `Spanish` `Japanese` `Korean` `French` `Russian`
mode	str	否	互動模式，可選值： `server_commit`（預設）：服務端自動判斷合成時機，平衡延遲與品質，推薦大多數情境使用 `commit`：用戶端手動觸發合成，延遲最低，但需自行管理句子完整性
format	str	否	模型輸出音訊格式。支援的格式： `pcm`（預設） `wav` `mp3` `opus` 千問-TTS-Realtime（參見支援的模型）僅支援`pcm`。
sample_rate	int	否	模型輸出音訊採樣率（Hz）。支援的採樣率： 8000 16000 24000（預設） 48000 千問-TTS-Realtime（參見支援的模型）僅支援24000。
speech_rate	float	否	音訊語速。1.0為正常語速，小於1.0為慢速，大於1.0為快速。預設值：1.0。取值範圍：[0.5, 2.0]。千問-TTS-Realtime（參見支援的模型）不支援該參數。
volume	int	否	音訊音量。預設值：50。取值範圍：[0, 100]。千問-TTS-Realtime（參見支援的模型）不支援該參數。
pitch_rate	float	否	合成音訊語調。預設值：1.0。取值範圍：[0.5, 2.0]。千問-TTS-Realtime（參見支援的模型）不支援該參數。
bit_rate	int	否	指定音訊碼率（kbps）。碼率越大，音質越好，音頻檔案體積越大。僅在音頻格式（`response_format`）為`opus`時可用。預設值：128。取值範圍：[6, 510]。千問-TTS-Realtime（參見支援的模型）不支援該參數。
instructions	str	否	設定指令，參見指令控制。預設值：無預設值，不設定不生效。長度限制：長度不得超過 1600 Token。支援語言：僅支援中文和英文。適用範圍：該功能僅適用於千問3-TTS-Instruct-Flash-Realtime系列模型。
optimize_instructions	bool	否	是否對 `instructions` 進行最佳化，以提升語音合成的自然度和表現力。預設值：False 行為說明：當設定為 True 時，系統將對 `instructions` 的內容進行語義增強與重寫，產生更適合語音合成的內部指令。適用情境：推薦在追求高品質、精細化語音表達的情境下開啟。依賴關係：此參數依賴於 `instructions` 參數被設定。如果 `instructions` 為空白，此參數不生效。適用範圍：該功能僅適用於千問3-TTS-Instruct-Flash-Realtime系列模型。

關鍵介面

QwenTtsRealtime類

QwenTtsRealtime通過from dashscope.audio.qwen_tts_realtime import QwenTtsRealtime方法引入。

方法簽名	服務端響應事件（通過回調下發）	說明
`def connect(self) -> None`	session.created 會話已建立 session.updated 會話配置已更新	和服務端建立串連。
`def update_session(self, voice: str, response_format: AudioFormat = AudioFormat. PCM_24000HZ_MONO_16BIT, mode: str = 'server_commit', language_type : str = "Chinese", **kwargs) -> None`	session.updated 會話配置已更新	更新本次會話互動的預設配置。參數配置請參考《請求參數》章節。在您建立連結，服務端會及時返回用於此會話的預設輸出輸入配置。如果您需要更新預設會話配置，我們也推薦您總是在建立連結後即刻調用此介面。服務端在收到session.update事件後，會進行參數校正，如果參數不合法則返回錯誤，否則更新服務端側的會話配置。
`def append_text(self, text: str) -> None`	無	將文本片段追加到雲端輸入文本緩衝區。緩衝區是你可以寫入並稍後提交的臨時儲存。 "server_commit"模式下，伺服器決定何時提交併合成文本緩衝區中的文本。 "commit"模式下，用戶端需要主動通過commit觸發語音合成。
`def clear_appended_text(self, ) -> None`	input_text_buffer.cleared 清空服務端收到的文本	刪除當前雲端緩衝區的文本。
`def commit(self, ) -> None`	input_text_buffer.committed 提交文本並觸發語音合成 response.output_item.added 響應時有新的輸出內容 response.content_part.added 新的輸出內容添加到assistant message 項 response.audio.delta 模型增量產生的音頻 response.audio.done 完成音頻產生 response.content_part.done Assistant message 的音頻內容流式輸出完成 response.output_item.done Assistant message 的整個輸出項串流完成 response.done 響應完成	提交之前通過append添加到雲端緩衝區的文本，並立刻合成所有文本。如果輸入的文本緩衝區為空白將產生錯誤。 "server_commit"模式下，用戶端不需要發送此事件，伺服器會自動認可文本緩衝區。 "commit"模式下，用戶端必須通過commit觸發語音合成。
`def finish(self, ) -> None`	session.finished 響應完成	終止任務。
`def close(self, ) -> None`	無	關閉串連。
`def get_session_id(self) -> str`	無	擷取當前任務的session_id。
`def get_last_response_id(self) -> str`	無	擷取最近一次response的response_id。
`def get_first_audio_delay(self)`	無	擷取首包音頻延遲。

回調介面（QwenTtsRealtimeCallback）

服務端會通過回調的方式，將服務端響應事件和資料返回給用戶端。您需要實現回調方法，處理服務端返回的資訊或者資料。

通過from dashscope.audio.qwen_tts_realtime import QwenTtsRealtimeCallback引入。

方法	參數	傳回值	描述
`def on_open(self) -> None`	無	無	當和服務端建立串連完成後，該方法立刻被回調。
`def on_event(self, message: str) -> None`	message：服務端響應事件。	無	包括對介面調用的回複響應和模型產生的文本和音頻。具體可以參考：服務端事件
`def on_close(self, close_status_code, close_msg) -> None`	close_status_code：關閉WebSocket的狀態代碼。 close_msg：關閉WebSocket的關閉資訊。	無	當服務已經關閉串連後進行回調。