当前位置： 首页 > news >正文

火山引擎TTS使用体验

news 2025/6/20 15:15:08

文章目录

  • 前言
  • 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):
  1. 解析响应头部信息

  • 协议版本(protocol_version)

  • 头部大小(header_size)

  • 消息类型(message_type)

  • 消息特定标志(message_type_specific_flags)

  • 序列化方法(serialization_method)

  • 压缩方法(message_compression)

  • 保留字段(reserved)

  1. 处理不同类型的服务器响应：

错误消息响应(message_type=0xf)：

  • 解析错误代码(code)
  • 解析错误消息大小(msg_size)
  • 解压缩并显示错误内容

前端消息响应(message_type=0xc)：

  • 解析消息大小(msg_size)
  • 解压缩并显示前端消息

3.4运行demo

注意下载下来的demo好像名称中有个空格，大家注意修改下名称。

执行指令

python tts_websocket_demo.py

执行结果
通过打印日志能够，模型服务返回了对应的响应数据（音频的原始数据）
在这里插入图片描述
然后我们就能看到我们的文件夹多了两个mp3的文件，分别是通过请求得到的和通过查询得到的。
听了下是熊二说的“字节跳动语音合成”，这里我设置的语音类型也是熊二的。
在这里插入图片描述
此时再去查看我们的模型使用情况，会发现少了一定的额度。
在这里插入图片描述

4. 参考链接

豆包语音合成大模型官网

语音技术开发参考 - 豆包官方的

http://www.lryc.cn/news/572446.html

相关文章：

  • 类与对象(中)（详解）
  • 多卡解决报错torch.distributed.elastic.multiprocessing.errors.ChildFailedError的问题
  • API 接口：程序世界的通用语言与交互基因
  • 【音视频】PJSIP库——示例简介、C++类说明
  • 深度学习——激活函数
  • # python正则表达式——实战学习+理论
  • 跟踪大型语言模型的思想：对语言之间共享；提前规划；cot
  • RK3588调试之旅：adbd服务配置全攻略
  • stm32之使用中断控制led灯
  • 新生活的开启：从 Trae AI 离开后的三个月
  • linux操作命令（最常用）
  • 打破物理桎梏：CAN-ETH网关如何用UDP封装重构工业网络边界
  • 大模型更重要关注工艺
  • 目标检测之YOLOV11自定义数据使用OBB训练与验证
  • Neo4j常用语法-path
  • JS红宝书笔记 8.3 继承
  • 煤矿井下Modbus转Profibus网关的传感器与PLC互联解决方案
  • 机器学习×第十二卷：回归树与剪枝策略——她剪去多余的分支，只保留想靠近你的那一层
  • 运维人员常用网站列表
  • 【unitrix】 3.2 位取反运算(not.rs)
  • 【数字人开发】Unity+百度智能云平台实现长短文本个性化语音生成功能
  • 吃透 Golang 基础：Goroutine
  • golang excel导出时需要显示刷新
  • Set_path_margin 命令介绍
  • C++中所有数据类型
  • Jenkins通过Pipeline流水线方式编译Java项目
  • Docker+Jenkins+git实现Golang项目自动部署
  • springboot 打的可执行jar包单独更新某个jar包
  • JMeter 高阶玩法：分布式压测的技术核心技术要点
  • 【K8S】详解NodePort 和 ClusterIP