跳到主要内容

WebSocket ​流式​（TTS）

WebSocket TTS API ​让​您​发送​完整​文本，​并​在​单个​连接​上​流式​接收生成​的​音频，​从而​在​合成​完成​之前​即可​开始​播放。

wss://api.shisa.ai/ws/tts/realtime
输入完整文本，流式输出音频

尽管​路​径​名​如此，​这不​是像实时​ A​SR ​那样​的​输入​流式​ A​PI。​您​发送​一​个​完整​的​ tts.speak ​文本​请​求;​服务会​以​二​进制帧​或​ base64 JSON​ 块​的​形式​将​生成​的​音频​流式​返回。

如果​需要​一​次​性返​回​整​个​音频​文件​的​单次​请​求​/响应，​请​改用​ POST /tts ​端点。

要求​

  • 一​个​被​允许​使用​所​选​音色​的​ A​PI 密钥​（WebSocket TTS ​的​配额​和​速率​限制​在​ shisa/tts ​服务​下​管理）。
  • 来自​ GET /tts/voices ​的​音色​ UUID​——​参见音色​目录。
  • 一​个​能够​设置​ Authorization ​标头​的​服务器​端 WebSocket ​客户端。​请​将​ A​PI ​密钥​保存在​服务​器端。

在​握手​期间，​使用​标准​的​ bearer ​令牌​进行​认证​:

Authorization: Bearer YOUR_API_KEY

工作​原理​

  1. 使用​ Authorization ​标头​打​开​ WebSocket。
  2. 发送​一​条​包含​必需​的​ voice_id ​和​ format（以及​可选​的​ sample_rate、temperature ​和​ audio_transport）​的​ session.update。
  3. 等​待 session.created。
  4. 发送​一​个​包含​完整​文本​的​ tts.speak ​请​求。
  5. 读取 tts.audio.start、​音频帧、tts.audio.done ​和​ tts.usage。
  6. 依次​发送​更​多​ tts.speak ​请​求，​或​关闭​套​接字。
一次只能进行一个合成

每​个​会话​一​次​只​能​有​一​个​进行​中​的​合成。​在​一​个​合成​仍​在​进行​时​发送​第二​个​ tts.speak 会​返​回 synthesis_in_progress ​错误。​请​在​发送​下​一​个​请求​之前​等​待 tts.usage。

客户​端​消息​

session.update​

在​合成​之前​发送​一​条 session.update:

{
  "type": "session.update",
  "id": "cfg_0001",
  "session": {
    "voice_id": "61ba1141-60aa-4bc3-a3b3-be1ec20700b3",
    "format": "mp3",
    "temperature": 0.7,
    "audio_transport": "binary"
  }
}
字段是否​必​需备​注
id可​选客户​端关联 ​ID，​在​配置​错误时​回显。
voice_id必​需来自​ GET /tts/voices ​的​公开​音色​ UU​ID。
format必​需输出音频格式​——mp3、wav、ogg、pcm ​或​ flac。​允许​的​值取​决于​所​选​音色。
sample_rate可​选省略​或​设为​ 0 ​则​使用​后​端默​认值​（24000 Hz）。
temperature可​选部分​音色​接受​的、​用于​控制​语​音​变化​的​提供​方​专属​参数。​使用​默​认值​时​请​省​略;​显式​的​ 0.0 会​作为​实值​发送。
audio_transport可​选binary（默认）​或​ base64_json。​参见音频​传输。

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,
    "temperature": 0.7,
    "model": "kokoro"
  }
}

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": "kokoro",
  "audio_transport": "binary"
}

音频​传输​

音​频​字节​的​到达​方式​取决于​您​在​ session.update ​中​设置​的​ audio_transport:

  • binary（默认）​——生成​的​音频​以​原始​二​进制 WebSocket​ 帧​的​形式​到达。​请​按​顺序连接​它们。
  • base64_json——​音频​以​ tts.audio.delta JSON ​消息​的​形式​到​达，​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.speak 会​创建​一​条​用量​记​录​:

{
  "type": "tts.usage",
  "request_id": "synthesis-request-id",
  "session_id": "router-session-request-id",
  "client_id": "utt_0001",
  "model": "kokoro",
  "usage": {
    "input_chars": 24,
    "input_bytes": 72,
    "audio_bytes": 12345,
    "status": "completed",
    "final": true
  }
}

被​拒绝​的​请求​（例如​错误​的​ voice_id ​或​ format）​会​返​回 tts.error，​并且不​会创建​用量​记录。

最​小​ 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 帧是事件。在 tts.usage 处结束。
        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.usage":
                print(json.dumps(event, ensure_ascii=False), file=sys.stderr)
                break
            if event_type in {"tts.error", "router.error", "error"}:
                raise RuntimeError(event)

    sys.stdout.buffer.write(audio)


if __name__ == "__main__":
    asyncio.run(main())

后续​步​骤​