本文档介绍如何通过火山引擎边缘大模型网关，使用有状态、基于事件的 Realtime API 来调用您自部署的语音合成（TTS）模型。

概述

Realtime API 通过 WebSocket 协议进行通信，适用于需要低延迟、实时流式传输的场景。它允许客户端与服务端建立持久连接，并在此连接上进行双向的事件驱动通信，从而实现将文本数据实时合成为音频流。

建连参数

与 Realtime API 建立 WebSocket 连接时，需要指定以下参数：

• URL：wss://ai-gateway.vei.volces.com/v1/realtime
• 查询参数：?model=<your_model_name>
• 请求头 (Header)：Authorization: Bearer $YOUR_API_KEY

快速开始

本节将引导您完成一次完整的 API 调用，从建立连接到接收合成的音频。

准备工作

在开始之前，请确保您已获取以下信息：

• 模型调用名称 (<your_model_name>)：在 大模型管理 页面获取。
• API 密钥 ($YOUR_API_KEY)：在 网关访问密钥 页面获取。

  说明

  请确保该密钥已绑定您的自部署 TTS 模型。关于如何绑定自部署 TTS 模型，请参见 调用自部署模型。

步骤 1：编写调用代码

以下是一个完整的 Python 示例，展示了如何连接到 Realtime API、发送文本并接收音频流。

1. 创建一个名为 requirements.txt 的文件，并填入以下依赖：

   numpy
   soundfile
   scipy
   websockets==12.0
   openai
   

2. 创建一个名为 tts_client.py 的文件，并将以下示例代码粘贴进去。

   说明

   • 注意将代码中的<your_model_name>和$YOUR_API_KEY替换为实际值。
   • 以下示例代码仅用于测试，请勿用于生产环境。

   import asyncio
   import base64
   import json
   
   import numpy as np
   from scipy.signal import resample
   import websockets
   
   import logging
   
   logger = logging.getLogger('my_logger')
   
   # 设置日志记录器的级别为DEBUG
   logger.setLevel(logging.INFO)
   # 创建一个控制台处理器并设置其级别为DEBUG
   console_handler = logging.StreamHandler()
   console_handler.setLevel(logging.DEBUG)
   
   # 创建一个格式化器并将其添加到处理器
   formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
   console_handler.setFormatter(formatter)
   
   # 将处理器添加到日志记录器
   logger.addHandler(console_handler)
   
   
   def resample_audio(audio_data, original_sample_rate, target_sample_rate):
       number_of_samples = round(len(audio_data) * float(target_sample_rate) / original_sample_rate)
       resampled_audio = resample(audio_data, number_of_samples)
       return resampled_audio.astype(np.int16)
   
   
   async def send_text(client, text: str):
       for t in text:
           await asyncio.sleep(0.05)
           event = {
               "type": "input_text.append",
               "delta": t
           }
           await client.send(json.dumps(event))
       event = {
           "type": "input_text.done"
       }
       await client.send(json.dumps(event))
   
   
   # 定义一个函数来写入音频数据
   def write_audio_data(stream, data):
       stream.write(data)
   
   
   async def receive_messages(client, file_path="response_audio.pcm"):
       audio_list = bytearray()
       while not client.closed:
           message = await client.recv()
           if message is None:
               print("===None Message===")
               continue
           event = json.loads(message)
           message_type = event.get("type")
           if message_type == "response.audio.delta":
               audio_bytes = base64.b64decode(event["delta"])
               audio_list.extend(audio_bytes)
               del event['delta']
               print(event)
               continue
   
           print(event)
   
           with open(file_path, 'wb') as ff:
               ff.write(audio_list)
   
           if message_type == "response.audio.done":
               break
   
           continue
   
   
   def get_session_update_msg():
       config = {
           "voice": "your_voice",
           "output_audio_format": "pcm",
           "output_audio_sample_rate": 24000,
           "enable_subtitle": True,
       }
       event = {
           "type": "tts_session.update",
           "session": config
       }
       return json.dumps(event)
   
   
   async def with_openai():
       key = "$YOUR_API_KEY"
       ws_url = "wss://ai-gateway.vei.volces.com/v1/realtime?model=<your_model_name>"
   
       headers = {
           "Authorization": f"Bearer {key}",
       }
       async with websockets.connect(ws_url, ping_interval=None, logger=logger, extra_headers=headers) as client:
           session_msg = get_session_update_msg()
           await client.send(session_msg)
           await asyncio.gather(send_text(client, "你好呀"), receive_messages(client))
   
   
   if __name__ == "__main__":
       asyncio.run(with_openai())
   

步骤 2：安装依赖并运行

1. 打开终端，运行以下命令安装所需的 Python 库：

   pip install -r requirements.txt
   

2. 执行脚本：

   python3 tts_client.py
   

运行成功后，您将在项目目录下看到一个名为 response_audio.pcm 的文件，其中包含了合成的音频数据。

API 参考

事件列表

API 调用流程

Realtime API 的交互是基于事件的。客户端和服务端通过交换一系列 JSON 格式的事件来完成一次完整的语音合成任务。

客户端事件详解

会话初始化 - tts_session.update

此事件用于在会话开始时配置语音合成的参数。必须在 WebSocket 连接建立后立即发送，且仅能发送一次。
参数：

• session(object)
  包含语音合成的各项配置参数。各配置参数的值需要符合自部署 TTS 模型的要求。
  • voice(string)
    指定音色。
  • output_audio_format(string)
    输出音频的编码格式，例如 pcm。
  • output_audio_sample_rate(integer)
    输出音频的采样率，例如 16000 或 24000 (Hz)。
  • output_audio_speed_rate(float)
    语速。
  • output_audio_volume(float)
    音量。
  • output_audio_pitch_rate(float)
    音调。
  • output_audio_channel(integer)
    音频通道数，例如 1 (单声道)。
  • enable_subtitle(bool)
    是否启用字幕功能。设置为 True 时，服务端将返回 response.audio_subtitle.delta 事件。
  • extra_data(object)
    用于传递额外参数。此对象会以 Body 的形式透传给下游模型。
  • extra_header(object)

    注意

    extra_header仅在自部署模型使用 HTTPS 协议时才会生效。

    用于传递额外请求头。此对象会以 Header 的形式透传给下游模型。

  说明

  不同的 TTS 模型可能支持一些标准参数之外的自定义功能。您可以通过 extra_data 和 extra_header 字段将这些特定参数传递给后端的模型服务。

示例：

{
    "event_id": "event_123",
    "type": "tts_session.update",
    "session": {
        "voice": "xxx",
        "output_audio_format": "pcm",
        "output_audio_sample_rate": 16000,
        "output_audio_speed_rate": 0.0,
        "output_audio_volume": 1,
        "output_audio_pitch_rate": 0,
        "output_audio_channel": 1,
        "enable_subtitle": true,
        "extra_data": {
          "key1": "value1"
        } // map
        "extra_header": { // map[string]string
          "key1": "value1"
        }
    }
}

文字输入 - input_text.append

此事件用于向服务端流式发送待合成的文本。您可以将长文本拆分为多个片段，并通过此事件连续发送。
参数：

• delta(string)
  本次发送的文本片段。

示例：

{
    "event_id": "event_345",
    "type": "input_text.append",
    "delta": "你好"
}

输入完成 - input_text.done

当所有文本片段都通过 input_text.append 发送完毕后，发送此事件来通知服务端文本流结束。
参数：无
示例：

{
    "event_id": "event_346",
    "type": "input_text.done"
}

服务端事件详解

配置确认 - tts_session.updated

在收到客户端发送的 tts_session.update 事件后，服务端会返回此事件，以确认会话配置已成功应用。
参数：

• session(object)
  服务端最终应用的会话配置。

示例：

{
    "event_id": "event_18B756FcWVhWyjCn8gn3u",
    "type": "tts_session.updated",
    "session": {
        "voice": "zh_male_shaonianzixin_moon_bigtts",
        "output_audio_format": "pcm",
        "output_audio_sample_rate": 24000,
        "output_audio_channel": 1,
        "enable_subtitle": true
    }
}

增量音频输出 - response.audio.delta

服务端通过此事件实时返回合成的音频数据。音频数据经过 Base64 编码。
参数：

• item_id(string)
  唯一标识符，用于关联同一合成任务的音频流和字幕。
• delta(string)
  Base64 编码的二进制音频数据片段。

示例：

{
    "event_id": "event_wQp5vyn4VloVTOcrQS6bo",
    "type": "response.audio.delta",
    "item_id": "item_ezC2f0aiYLtsTILllMIXe",
    "delta": "xxxxxx"
}

额外的响应元信息 - response.trace_info.added

服务端通过此事件返回从自部署模型响应头中获取的追踪信息。本事件仅适用于用户的自部署模型为 HTTP/HTTPS 协议接入。
参数：

• data(string)
  来源于每次调用用户的自部署模型的 response header X-Biz-Trace-Info。在每次调用用户的自部署模型并识别到该 Header 后输出。如果该次调用自部署模型没有返回这个 Header，则不会输出该事件。

示例：

{
    "event_id": "event_wQp5vyn4VloVTOcrQS6bo",
    "type": "response.trace_info.added",
    "item_id": "item_ezC2f0aiYLtsTILllMIXe",
    "data": "xxxxxx"
}

输出完成 - response.audio.done

当本轮语音合成任务的所有音频数据都已发送完毕时，服务端会发送此事件。
参数：

• item_id(string)
  关联的合成任务标识符。

示例：

{
    "event_id": "event_zh366ItZEWdcnSOiTPR2N",
    "type": "response.audio.done",
    "item_id": "item_ezC2f0aiYLtsTILllMIXe"
}

字幕输出 - response.audio_subtitle.delta

如果 tts_session.update 事件中 enable_subtitle 设置为 true，服务端会通过此事件返回与音频流同步的字幕信息。
参数：

• item_id(string)
  关联的合成任务标识符。
• subtitles(object)
  包含字幕的详细信息。
  • text(string)
    当前片段对应的完整文本。
  • words: array
    包含每个词语及其出现时间的对象数组。
    • start(float)
      词语开始的时间（秒）。
    • end(float)
      词语结束的时间（秒）。
    • word(string)
      具体的词语。

示例：

{
    "event_id": "event_2122",
    "type": "response.audio_subtitle.delta",
    "item_id": "msg_003",
    "subtitles": {
        "text": "你好",
        "words": [{
          "start": 0.0,
          "end": 1.319999933242798,
          "word": "你"
        },{
        "start": 1.319999933242798,
          "end": 2.6,
          "word": "好"
        }],
    ...
  }
}