WebSocket 流式(TTS)
WebSocket TTS API 让您发送完整文本,并在单个连接上流式接收生成的音频,从而在合成完成之前即可开始播放。
wss://api.shisa.ai/ws/tts/realtime
尽管路径名如此,这不是像实时 ASR 那样的输入流式 API。您发送一个完整的 tts.speak 文本请求;服务会以二进制帧或 base64 JSON 块的形式将生成的音频流式返回。
如果需要一次性返回整个音频文件的单次请求/响应,请改用 POST /tts 端点。
要求
- 一个被允许使用所选音色的 API 密钥(WebSocket TTS 的配额和速率限制在
shisa/tts服务下管理)。 - 来自
GET /tts/voices的音色 UUID——参见音色目录。 - 一个能够设置
Authorization标头的服务器端 WebSocket 客户端。请将 API 密钥保存在服务器端。
在握手期间,使用标准的 bearer 令牌进行认证:
Authorization: Bearer YOUR_API_KEY
工作原理
- 使用
Authorization标头打开 WebSocket。 - 发送一条包含必需的
voice_id和format(以及可选的sample_rate、temperature和audio_transport)的session.update。 - 等待
session.created。 - 发送一个包含完整文本的
tts.speak请求。 - 读取
tts.audio.start、音频帧、tts.audio.done和tts.usage。 - 依次发送更多
tts.speak请求,或关闭套接字。
每个会话一次只能有一个进行中的合成。在一个合成仍在进行时发送第二个 tts.speak 会返回 synthesis_in_progress 错误。对于已在后端开始的尝试,请在收到 tts.usage 后再发送下一个请求;对于到达后端之前被拒绝的请求,请等待作为终止事件的 tts.error。如果代码为 backend_request_failed,还应读取紧随其后的 tts.usage。
客户端消息
session.update
在合成之前发送一条 session.update:
{
"type": "session.update",
"id": "cfg_0001",
"session": {
"voice_id": "61ba1141-60aa-4bc3-a3b3-be1ec20700b3",
"format": "mp3",
"audio_transport": "binary"
}
}
| 字段 | 是否必需 | 备注 |
|---|---|---|
id | 可选 | 客户端关联 ID,在配置错误时回显。 |
voice_id | 必需 | 来自 GET /tts/voices 的公开音色 UUID。 |
format | 必需 | 输出音频格式——mp3、wav、ogg、pcm 或 flac。该格式必须受所选音色支持,并且可在对应提供商上流式传输。 |
sample_rate | 可选 | 省略或设为0则使用后端默认值(24000 Hz)。仅使用所选音色支持的可配置值。Qwen 后端音色使用ogg时不接受覆盖值。 |
temperature | 可选 | 提供商专属的语音变化控制,目前由 Qwen 后端音色接受。其他音色以及使用默认值时请省略;显式的0.0会作为实值发送。 |
audio_transport | 可选 | binary(默认)或 base64_json。参见音频传输。 |
音色目录中的streaming: true是必要条件,但不表示formats中的每个值都可流式传输。由于此端点始终流式输出,不受支持的提供商/格式组合会返回tts.error,其code为"unsupported_streaming_format"。
tts.speak
发送要合成的完整文本:
{
"type": "tts.speak",
"id": "utt_0001",
"text": "こんにちは。WebSocket TTS のテストです。"
}
| 字段 | 是否必需 | 备注 |
|---|---|---|
id | 可选 | 客户端关联 ID,在此请求的 TTS 事件中回显。 |
text | 必需 | 要合成的完整文本。默认上限为 5000 个字符。 |
服务事件
session.created
会话已配置完成,可以进行 tts.speak:
{
"type": "session.created",
"session": {
"id": "router-session-request-id",
"service": "shisa/tts",
"voice_id": "61ba1141-60aa-4bc3-a3b3-be1ec20700b3",
"format": "mp3",
"sample_rate": 24000,
"model": "speech-2.8-hd"
}
}
tts.audio.start
标记一个被接受的合成请求,并描述您将收到的音频:
{
"type": "tts.audio.start",
"id": "utt_0001",
"request_id": "synthesis-request-id",
"format": "mp3",
"content_type": "audio/mpeg",
"sample_rate": 24000,
"voice_id": "61ba1141-60aa-4bc3-a3b3-be1ec20700b3",
"model": "speech-2.8-hd",
"audio_transport": "binary"
}
音频传输
音频字节的到达方式取决于您在 session.update 中设置的 audio_transport:
binary(默认)——生成的音频以原始二进制 WebSocket 帧的形式到达。请按顺序连接它们。base64_json——音频以tts.audio.deltaJSON 消息的形式到达,base64 编码的字节位于audio中,并带有递增的seq:
{
"type": "tts.audio.delta",
"id": "utt_0001",
"request_id": "synthesis-request-id",
"seq": 1,
"audio": "<base64-audio-bytes>"
}
tts.audio.done
此请求的合成已完成:
{
"type": "tts.audio.done",
"id": "utt_0001",
"request_id": "synthesis-request-id",
"audio_bytes": 12345,
"chunks": 4
}
tts.usage
已在后端开始的合成尝试的最终用量。成功时,该事件会在 tts.audio.done 之后发送:
{
"type": "tts.usage",
"request_id": "synthesis-request-id",
"session_id": "router-session-request-id",
"client_id": "utt_0001",
"model": "speech-2.8-hd",
"usage": {
"input_chars": 24,
"input_bytes": 72,
"audio_bytes": 12345,
"status": "completed",
"final": true
}
}
如果后端请求失败,路由器会先发送代码为 backend_request_failed 的 tts.error,随后发送 tts.usage。未返回音频时,usage.status 为 backend_error;已返回部分音频时,则为 partial_backend_error。
对于到达后端的尝试(包括失败的尝试),路由器会创建用量记录;只要连接仍可写,也会发送 tts.usage。在输入验证、访问控制、速率限制或后端选择阶段被拒绝的请求只返回 tts.error,不会创建用量记录。因此,并非每个 tts.error 后都会跟随 tts.usage。
tts.error 与 router.error
在 WebSocket 升级之前发生的失败会返回结构化 HTTP 错误,而不是 WebSocket 事件。常见代码包括 invalid_websocket_upgrade、auth_token_required、invalid_token_format、invalid_token、legacy_api_key_rejected、ws_origin_denied、unknown_service、ws_connection_limit_exceeded、ws_connection_quota_unavailable 和 websocket_draining。
TTS 配置、发声和后端错误采用以下格式。若客户端关联 ID 可用,id 会将其回显:
{
"type": "tts.error",
"id": "utt_0001",
"code": "backend_request_failed",
"message": "Backend TTS request failed"
}
| 代码 | 含义 |
|---|---|
invalid_message, invalid_id, id_too_long, invalid_session | 已识别消息、关联 ID 或路由器会话状态无效。 |
session_already_configured | 此连接上的 session.update 已成功处理。 |
missing_voice_id, unknown_voice_id, tts_service_unavailable, service_access_denied | 音色选择、服务可用性或访问权限无效。 |
unsupported_audio_format, unsupported_sample_rate, unsupported_audio_transport, unsupported_streaming_format | 所选音色或后端不支持请求的输出设置。 |
session_not_configured, empty_text, tts_text_too_long, synthesis_in_progress | tts.speak 的状态或文本无效。 |
rate_limited, service_rate_limited | 超出 API 密钥或 TTS 服务的速率限制。 |
rate_limiter_unavailable, service_rate_limiter_unavailable | 速率限制服务暂时不可用。 |
backend_unavailable | 当前没有可用的 TTS 后端。后端请求尚未开始,因此不会创建用量记录。 |
backend_request_failed | 合成已在后端开始但失败。此错误之后会发送最终的 tts.usage。 |
tts_control_frame_too_large | JSON 控制消息超过配置的字节上限。若读取上限先检测到该问题,连接会以代码 1009 关闭。 |
在消息类型得到处理之前检测到的控制错误使用 type: "router.error",代码包括 invalid_json、missing_message_type 和 unknown_message_type。客户端发送二进制帧会使连接以 WebSocket 代码 1003 关闭;控制帧过大时则以代码 1009 关闭。
最小 Python 客户端
安装依赖:
python -m pip install websockets
设置密钥并运行:
export SHISA_API_KEY="shsk:..."
export TTS_WS_URL="wss://api.shisa.ai/ws/tts/realtime"
export TTS_VOICE_ID="61ba1141-60aa-4bc3-a3b3-be1ec20700b3"
python websocket_tts.py > output.mp3
websocket_tts.py:
#!/usr/bin/env python3
import asyncio
import json
import os
import sys
import websockets
async def main() -> None:
url = os.environ.get("TTS_WS_URL", "wss://api.shisa.ai/ws/tts/realtime")
api_key = os.environ["SHISA_API_KEY"]
voice_id = os.environ["TTS_VOICE_ID"]
text = os.environ.get("TTS_TEXT", "こんにちは。WebSocket TTS のテストです。")
audio_format = os.environ.get("TTS_FORMAT", "mp3")
headers = [("Authorization", f"Bearer {api_key}")]
audio = bytearray()
async with websockets.connect(url, additional_headers=headers, max_size=None) as ws:
await ws.send(
json.dumps(
{
"type": "session.update",
"id": "cfg_0001",
"session": {
"voice_id": voice_id,
"format": audio_format,
"audio_transport": "binary",
},
}
)
)
# 在发话之前等待 session.created。
while True:
event = json.loads(await ws.recv())
if event.get("type") == "session.created":
break
if event.get("type") in {"tts.error", "router.error", "error"}:
raise RuntimeError(event)
await ws.send(json.dumps({"type": "tts.speak", "id": "utt_0001", "text": text}))
# 二进制帧是音频,JSON 帧是事件。
backend_error = None
while True:
msg = await ws.recv()
if isinstance(msg, bytes):
audio.extend(msg)
continue
event = json.loads(msg)
event_type = event.get("type")
if event_type == "tts.error":
if event.get("code") == "backend_request_failed":
# 后端失败后会发送最终用量。
backend_error = event
continue
raise RuntimeError(event)
if event_type == "tts.usage":
print(json.dumps(event, ensure_ascii=False), file=sys.stderr)
if backend_error is not None:
raise RuntimeError(backend_error)
break
if event_type in {"router.error", "error"}:
raise RuntimeError(event)
sys.stdout.buffer.write(audio)
if __name__ == "__main__":
asyncio.run(main())
后续步骤
- 使用
POST /tts端点通过单次调用返回整个音频文件。 - 在音色目录中浏览可用音色。
- 在定价中了解用量如何计费。