即時工作流程支援您按照我們提供的標準協議來接入文本轉語音(TTS)模型。
自研TTS標準介面
您需要實現一個公網可訪問的HTTP流式Data Transmission Service,按照我們定義的輸入輸出規範,封裝好您的文本轉語音模型,即可將您的自研TTS添加到工作流程中。
首先您需要在編排管理主控台中的TTS 文字轉語音節點中,配置以下參數:
名稱 | 類型 | 必填 | 描述 | 樣本值 |
請求服務地址 | String | 是 | 自研模型HTTPS介面地址 | https://www.abc.com |
Token | String | 否 | 服務校正令牌 | AUJH-pfnTNMPBm6iWXcJAcWsrscb5KYaLitQhHBLKrI |
採樣率 | Integer | 是 | 採樣率,單位Hz,可選項:
| 48000 |
介面只支援S16LE格式的單聲道音頻資料,在發送前需重採樣到介面支援的格式。
在即時工作流程運行期間,我們會按照以下格式,組裝表單資料,通過Post請求來訪問您配置的自研文本轉語音模型HTTPS地址,輸入參數如下:
名稱 | 類型 | 必填 | 描述 | 樣本值 |
Text | String | 是 | 語音文本 | 你好 |
VoiceId | String | 否 | 音色 | yourVoiceId |
SampleRate | Integer | 是 | 採樣率,單位Hz | 48000 |
Token | String | 否 | 服務校正令牌 | AUJH-pfnTNMPBm6iWXcJAcWsrscb5KYaLitQhHBLKrI |
ExtendData | String | 是 | 提交自訂TTS拓展資訊欄位,會包含執行個體Id,以及啟動執行個體時您傳入的UserData業務欄位資料。 | {'InstanceId':'68e00b6640e*****3e943332fee7','ChannelId':'123','SentenceId':'3',UserData':'{"aaaa":"bbbb"}'} |
| String | 是 | 執行個體ID資訊 | 68e00b6640e*****3e943332fee7 |
| String | 是 | 房間ID資訊 | 123 |
| Int | 是 | 問答ID資訊 說明 對於使用者的同一次提問,智能體回答會使用相同的ID資訊。 | 3 |
| String | 否 | 說話情緒:
說明 不傳入該參數時,表示這次語音合成不含情緒。 | happy |
| String | 否 | 啟動執行個體時您傳入的UserData業務欄位資料 | {"aaaa":"bbbb"} |
您需要將產生的對應音色以及採樣率的音頻源流PCM通過HTTP流式響應,分段發送給我們的服務,我們將即時地推送到後續節點當中。
自訂TTS Server
Python
自訂TTS Server程式碼範例如下:
from aiohttp import web
async def stream_audio(request):
data = await request.json()
text = data.get('Text', "")
token = data.get('Token', None)
sample_rate = data.get('SampleRate', 48000)
extend_data = data.get('ExtendData', "")
print(f"text:{text}, token:{token}, sample_rate:{sample_rate}, extend_data:{extend_data}")
# TODO:校正Token是否合法
response = web.StreamResponse(
status=200,
reason='OK',
headers={'Content-Type': 'audio/mpeg'}
)
# 開始響應
await response.prepare(request)
# generate_tts_data 是一個協程,用於產生音頻資料
async for chunk in generate_tts_data(text, sample_rate):
await response.write(chunk)
# 完成響應
await response.write_eof()
return response
async def generate_tts_data(text: str, sample_rate: int):
# TODO: 調用TTS服務,產生對應音頻對應採樣率的音頻資料
# 樣本:從檔案讀取音頻資料
file_path = '/your_dir/sample.pcm'
with open(file_path, 'rb') as f:
while True:
chunk = f.read(4096) # 每次讀取4KB的資料
if not chunk:
break
yield chunk
app = web.Application()
app.add_routes([web.post('/stream-audio', stream_audio)])
if __name__ == '__main__':
web.run_app(app)