火山引擎TTS使用體驗

文章目錄

前言
1. 簡介
- 1.1 能力體驗
- 1.2 功能特性
- 1.3 音色列表
- 1.4 收費情況
2. 開啟服務
- 2.1 創建應用
- 2.3 使用服務介紹
3.Websocket接入演示
- 3.1 編寫demo
- 3.2 代碼解釋
- 3.4運行demo
4. 參考鏈接

前言

語音合成TTS（text to Speech）是我覺得后續開發產品所不可或缺的一個功能，因為相比較于過去的GUI 圖形+文字展示，動態形象+語音會更利用人與設備之間的交互。

另外語音溝通更加靈活，因為過去的GUI圖形界面都是預先設計好的，就像APP內的界面。這種更加標準但是不夠靈活，不能滿足所有人的需求和愛好。所以我覺得現在了解下TTS也是非常有必要的。

之前看到小智AI中有提到支持了cosyVoice(阿里的TTS模型)與火山引擎的TTS。然后搜索了兩者的區別發現火山引擎的TTS在高擬真克隆這塊做的比較好。這樣的話就能夠使用該TTS生成各種符合產品形象的聲音。所以暫時先選擇了火山引擎的TTS去研究一下。

1. 簡介

火山引擎的TTS也叫做豆包語音合成大模型，它是依托新一代大模型能力，豆包語音合成模型能夠根據上下文智能預測文本的情緒、語調等信息，并生成超自然、高保真、個性化的語音，以滿足不同用戶的個性化需求。

簡介中很多的內容都是來自于火山引擎的文檔中心，我這里就簡單的介紹下，大家需要詳細了解的可以到以下地址去看看：

https://www.volcengine.com/docs/6561/1257544

1.1 能力體驗

在官方的網站中有個能力體驗的頁面，這個體驗很簡單輸入想要描述的文字，然后選擇配音和一些聲音相關的配置就能生成語音了。

我們最終要實現的功能也是類似的，只是這個是人家已經實現好的功能比較固化，我們想要更靈活一些，所以需要通過代碼去使用這個模型來做一些定制化的開發。
在這里插入圖片描述

1.2 功能特性

下面這個表里面說了很多，說實話有一些對于我們這些剛接觸的人來說并沒有什么概念，例如這里我也只是對部署方案比較感興趣。這里后續有需要的時候大家可以去官方介紹文檔中去查看。

在這里插入圖片描述

1.3 音色列表

使用克隆語音需要用到另外一個模型，不是我們本次所使用的“語音合成大模型”，所以我們并不是說想用什么聲音就用什么聲音，而是要使用官方給出的聲音列表。不過好在可選擇性還是很多的。
這里截圖不全，更詳細的內容可以查看官方文檔
在這里插入圖片描述

1.4 收費情況

我最初以為TTS里面包含了語音復制，什么短文本語音合成啥的呢，結果一看乖乖嘞被分成了4個而且是收費的。

但是我們第一次用的話是免費的，會贈送一定的使用額度，所以大家不要太過于擔心。
在這里插入圖片描述

2. 開啟服務

我們需要申請appid、token、secret_key等用來開啟和使用TTS的服務。

這個就類似于從豆包那里申請個賬號，這個賬號里面包含了我們的身份信息，以及能夠使用哪些模型還有我們的剩余額度，有了這些信息后我們才能真正的去使用大模型語音合成功能。

2.1 創建應用

先根據下方的快速入門創建賬號：

https://www.volcengine.com/docs/6561/163043

點擊“創建應用”來新增應用，填入應用名稱、簡介和所需接入的能力服務
在這里插入圖片描述
我們第一次使用會有個免費額度，所以大家不用太擔心。

創建成功后，能夠在應用管理界面看到我們所創建的應用

獲取token和Secret_key信息
在控制臺界面，我們能夠找到屬于我們的Access Token 和 Secret Key，有了這些信息我們才能去使用該服務。
在這里插入圖片描述

2.3 使用服務介紹

使用的話有兩種方式，分別是API和SDK接入。
核心區別對比
SDK的話目前它只能運行在安卓和IOS操作系統上，應該是集成到APP中，這不方便我們去進行體驗。而且像小智AI這種也是采用的API方式接入的，所以這里我們也使用API的方式。

API接入又分為WebSocket還有Http，基本上工作原理大差不差，都是發送請求，然后接收響應處理。這里我們就以WebSocket為主。

3.Websocket接入演示

Websocket接入演示的功能，需要使用賬號申請部分申請到的 appid和access_token進行調用文本一次性送入，后端邊合成邊返回音頻數據。所以大家一定要先按照上面的步驟獲取對應的token和appid等信息。

接口說明地址為下方的鏈接，詳細的使用方法大概可以進入該鏈接查看：

wss://openspeech.bytedance.com/api/v1/tts/ws_binary

3.1 編寫demo

文檔中心有個demo，我們拿下來直接運行即可，本次演示的代碼來源是tts_websocket_demo.py
在這里插入圖片描述
源碼如下：

#coding=utf-8'''
requires Python 3.6 or laterpip install asyncio
pip install websockets'''import asyncio
import websockets
import uuid
import json
import gzip
import copyMESSAGE_TYPES = {11: "audio-only server response", 12: "frontend server response", 15: "error message from server"}
MESSAGE_TYPE_SPECIFIC_FLAGS = {0: "no sequence number", 1: "sequence number > 0",2: "last message from server (seq < 0)", 3: "sequence number < 0"}
MESSAGE_SERIALIZATION_METHODS = {0: "no serialization", 1: "JSON", 15: "custom type"}
MESSAGE_COMPRESSIONS = {0: "no compression", 1: "gzip", 15: "custom compression method"}appid = "xxx"
token = "xxx"
cluster = "xxx"
voice_type = "xxx"
host = "openspeech.bytedance.com"
api_url = f"wss://{host}/api/v1/tts/ws_binary"# version: b0001 (4 bits)
# header size: b0001 (4 bits)
# message type: b0001 (Full client request) (4bits)
# message type specific flags: b0000 (none) (4bits)
# message serialization method: b0001 (JSON) (4 bits)
# message compression: b0001 (gzip) (4bits)
# reserved data: 0x00 (1 byte)
default_header = bytearray(b'\x11\x10\x11\x00')request_json = {"app": {"appid": appid,"token": "access_token","cluster": cluster},"user": {"uid": "388808087185088"},"audio": {"voice_type": "xxx","encoding": "mp3","speed_ratio": 1.0,"volume_ratio": 1.0,"pitch_ratio": 1.0,},"request": {"reqid": "xxx","text": "字節跳動語音合成。","text_type": "plain","operation": "xxx"}
}async def test_submit():submit_request_json = copy.deepcopy(request_json)submit_request_json["audio"]["voice_type"] = voice_typesubmit_request_json["request"]["reqid"] = str(uuid.uuid4())submit_request_json["request"]["operation"] = "submit"payload_bytes = str.encode(json.dumps(submit_request_json))payload_bytes = gzip.compress(payload_bytes)  # if no compression, comment this linefull_client_request = bytearray(default_header)full_client_request.extend((len(payload_bytes)).to_bytes(4, 'big'))  # payload size(4 bytes)full_client_request.extend(payload_bytes)  # payloadprint("\n------------------------ test 'submit' -------------------------")print("request json: ", submit_request_json)print("\nrequest bytes: ", full_client_request)file_to_save = open("test_submit.mp3", "wb")header = {"Authorization": f"Bearer; {token}"}async with websockets.connect(api_url, extra_headers=header, ping_interval=None) as ws:await ws.send(full_client_request)while True:res = await ws.recv()done = parse_response(res, file_to_save)if done:file_to_save.close()breakprint("\nclosing the connection...")async def test_query():query_request_json = copy.deepcopy(request_json)query_request_json["audio"]["voice_type"] = voice_typequery_request_json["request"]["reqid"] = str(uuid.uuid4())query_request_json["request"]["operation"] = "query"payload_bytes = str.encode(json.dumps(query_request_json))payload_bytes = gzip.compress(payload_bytes)  # if no compression, comment this linefull_client_request = bytearray(default_header)full_client_request.extend((len(payload_bytes)).to_bytes(4, 'big'))  # payload size(4 bytes)full_client_request.extend(payload_bytes)  # payloadprint("\n------------------------ test 'query' -------------------------")print("request json: ", query_request_json)print("\nrequest bytes: ", full_client_request)file_to_save = open("test_query.mp3", "wb")header = {"Authorization": f"Bearer; {token}"}async with websockets.connect(api_url, extra_headers=header, ping_interval=None) as ws:await ws.send(full_client_request)res = await ws.recv()parse_response(res, file_to_save)file_to_save.close()print("\nclosing the connection...")def parse_response(res, file):print("--------------------------- response ---------------------------")# print(f"response raw bytes: {res}")protocol_version = res[0] >> 4header_size = res[0] & 0x0fmessage_type = res[1] >> 4message_type_specific_flags = res[1] & 0x0fserialization_method = res[2] >> 4message_compression = res[2] & 0x0freserved = res[3]header_extensions = res[4:header_size*4]payload = res[header_size*4:]print(f"            Protocol version: {protocol_version:#x} - version {protocol_version}")print(f"                 Header size: {header_size:#x} - {header_size * 4} bytes ")print(f"                Message type: {message_type:#x} - {MESSAGE_TYPES[message_type]}")print(f" Message type specific flags: {message_type_specific_flags:#x} - {MESSAGE_TYPE_SPECIFIC_FLAGS[message_type_specific_flags]}")print(f"Message serialization method: {serialization_method:#x} - {MESSAGE_SERIALIZATION_METHODS[serialization_method]}")print(f"         Message compression: {message_compression:#x} - {MESSAGE_COMPRESSIONS[message_compression]}")print(f"                    Reserved: {reserved:#04x}")if header_size != 1:print(f"           Header extensions: {header_extensions}")if message_type == 0xb:  # audio-only server responseif message_type_specific_flags == 0:  # no sequence number as ACKprint("                Payload size: 0")return Falseelse:sequence_number = int.from_bytes(payload[:4], "big", signed=True)payload_size = int.from_bytes(payload[4:8], "big", signed=False)payload = payload[8:]print(f"             Sequence number: {sequence_number}")print(f"                Payload size: {payload_size} bytes")file.write(payload)if sequence_number < 0:return Trueelse:return Falseelif message_type == 0xf:code = int.from_bytes(payload[:4], "big", signed=False)msg_size = int.from_bytes(payload[4:8], "big", signed=False)error_msg = payload[8:]if message_compression == 1:error_msg = gzip.decompress(error_msg)error_msg = str(error_msg, "utf-8")print(f"          Error message code: {code}")print(f"          Error message size: {msg_size} bytes")print(f"               Error message: {error_msg}")return Trueelif message_type == 0xc:msg_size = int.from_bytes(payload[:4], "big", signed=False)payload = payload[4:]if message_compression == 1:payload = gzip.decompress(payload)print(f"            Frontend message: {payload}")else:print("undefined message type!")return Trueif __name__ == '__main__':loop = asyncio.get_event_loop()loop.run_until_complete(test_submit())loop.run_until_complete(test_query())

填寫token等信息
在運行demo之前，我們需要在下方的這里填寫上之前在火山引擎處申請到的信息，這里大家就理解為自己的賬號密碼就行了
在這里插入圖片描述
appid還有token還有cluster在我們的應用空間那里就能看到，然后有個比較特殊的voice_type需要找到文檔中的音色列表去那里找自己想要合成的聲音類型。

3.2 代碼解釋

導入各基本模塊

import asyncio
import websockets
import uuid
import json
import gzip
import copy

這里的asyncio 是python中的用于實現并發的模塊，里面提供了例如協程、協程鎖等各種用于異步通信的功能。其它的就不用說了看名字就知道干啥的了。

基礎配置填寫

appid = "xxx"
token = "xxx"
cluster = "xxx"
voice_type = "xxx"
host = "openspeech.bytedance.com"
api_url = f"wss://{host}/api/v1/tts/ws_binary"

這個就是前面我們提到的把應用空間了申請到的信息填寫上去
在這里插入圖片描述

請求體

# version: b0001 (4 bits)
# header size: b0001 (4 bits)
# message type: b0001 (Full client request) (4bits)
# message type specific flags: b0000 (none) (4bits)
# message serialization method: b0001 (JSON) (4 bits)
# message compression: b0001 (gzip) (4bits)
# reserved data: 0x00 (1 byte)
default_header = bytearray(b'\x11\x10\x11\x00')request_json = {"app": {"appid": appid,"token": "access_token","cluster": cluster},"user": {"uid": "388808087185088"},"audio": {"voice_type": "xxx","encoding": "mp3","speed_ratio": 1.0,"volume_ratio": 1.0,"pitch_ratio": 1.0,},"request": {"reqid": "xxx","text": "字節跳動語音合成。","text_type": "plain","operation": "xxx"}
}

header是發送請求時的消息頭，tts的通訊協議要求二進制的方式進行傳輸，所以頭這里也是采用的二進制。上面的注釋代表的是其二進制代表的內容。

request_json是我們的請求體，里面需要填充我們要發送的具體信息，后續發送時也會將其轉換為二進制發送，請求體中的參數主要就是這幾個
在這里插入圖片描述
可以通過下方的鏈接去查詢

https://www.volcengine.com/docs/6561/1257584

提交轉換請求

請求
async def test_submit():

功能

向字節跳動的語音合成WebSocket API提交一個文本合成請求

主要操作：

準備提交請求的JSON數據，包括appid、token、cluster等認證信息
設置操作類型為"submit"（提交）
生成唯一的請求ID
使用gzip壓縮請求數據
建立WebSocket連接并發送請求
持續接收服務器返回的音頻數據流，保存到test_submit.mp3文件
處理完所有音頻數據后關閉連接

查詢
async def test_query():

功能

向字節跳動的語音合成WebSocket API發送查詢請求

主要操作：

準備查詢請求的JSON數據，結構與submit類似
設置操作類型為"query"（查詢）
生成唯一的請求ID
使用gzip壓縮請求數據
建立WebSocket連接并發送請求
接收服務器響應（通常是一次性返回）
將響應數據保存到test_query.mp3文件
關閉連接

查詢與請求的主要區別：

test_submit()用于提交語音合成任務并持續接收音頻流
test_query()用于查詢狀態或結果，通常只接收一次響應
test_submit()會處理多個響應消息直到完成
test_query()通常只處理單個響應消息

處理接收到的響應

處理響應
def parse_response(res, file):

解析響應頭部信息

協議版本(protocol_version)

頭部大小(header_size)

消息類型(message_type)

消息特定標志(message_type_specific_flags)

序列化方法(serialization_method)

壓縮方法(message_compression)

保留字段(reserved)

處理不同類型的服務器響應：

錯誤消息響應(message_type=0xf)：

解析錯誤代碼(code)
解析錯誤消息大小(msg_size)
解壓縮并顯示錯誤內容

前端消息響應(message_type=0xc)：

解析消息大小(msg_size)
解壓縮并顯示前端消息

3.4運行demo

注意下載下來的demo好像名稱中有個空格，大家注意修改下名稱。

執行指令

python tts_websocket_demo.py

執行結果
通過打印日志能夠，模型服務返回了對應的響應數據（音頻的原始數據）
在這里插入圖片描述
然后我們就能看到我們的文件夾多了兩個mp3的文件，分別是通過請求得到的和通過查詢得到的。
聽了下是熊二說的“字節跳動語音合成”，這里我設置的語音類型也是熊二的。

此時再去查看我們的模型使用情況，會發現少了一定的額度。
在這里插入圖片描述