TTS标准接口

实时工作流支持您按照我们提供的标准协议来接入文本转语音(TTS)模型。

自研TTS标准接口

您需要实现一个公网可访问的HTTP流式数据传输服务,按照我们定义的输入输出规范,封装好您的文本转语音模型,即可将您的自研TTS添加到工作流中。

  1. 首先您需要在编排管理控制台中的TTS 文字转语音节点中,配置以下参数:

名称

类型

必填

描述

示例值

请求服务地址

String

自研模型HTTPS接口地址

https://www.abc.com

Token

String

服务校验令牌

AUJH-pfnTNMPBm6iWXcJAcWsrscb5KYaLitQhHBLKrI

码率

Integer

仅支持48000

48000

  1. 在实时工作流运行期间,我们会按照以下格式,组装表单数据,通过Post请求来访问您配置的自研文本转语音模型HTTPS地址,输入参数如下:

名称

类型

必填

描述

示例值

Text

String

语音文本

你好

SampleRate

Integer

码率

48000

Token

String

服务校验令牌

AUJH-pfnTNMPBm6iWXcJAcWsrscb5KYaLitQhHBLKrI

ExtendData

String

提交自定义TTS拓展信息字段,会包含实例Id,以及启动实例时您传入的UserData业务字段数据。

{'InstanceId':'68e00b6640e*****3e943332fee7','ChannelId':'123','SentenceId':'3',UserData':'{"aaaa":"bbbb"}'}

  • InstanceId

String

实例ID信息

68e00b6640e*****3e943332fee7

  • ChannelId

String

房间ID信息

123

  • SentenceId

Int

问答ID信息

说明

对于用户的同一次提问,智能体回答会使用相同的ID信息。

3

  • UserData

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)

相关文档

通过控制台创建实时工作流模板