从LLM到TTS，阿里云的流式文本语音合成实践

流式文本语音合成简介

TTS（Text-to-Speech），即文本转语音技术，能够将文本信息转化为相应的语音信号，广泛应用于人机交互、智能助手等多个领域。流式文本语音合成的核心在于其“流式”特性，这意味着文本不是一次性全部提供给TTS模型，而是分段逐步输入，TTS模型则实时地逐段生成并返回语音。这种模式特别适用于实时场景，例如将LLM（大型语言模型）流式输出的文本即时转换为语音，用于驱动数字人等应用。

阿里云TTS服务的应用实例

本文基于阿里云提供的TTS服务，展示如何将LLM的流式输出转换为语音的一个具体实例。阿里云提供了详细的Python SDK使用指南，本文将在此基础上提供一个可直接运行的代码示例。关于服务开通、环境配置以及API密钥的获取等前期准备步骤，请参阅阿里云官方网站的相关文档。

以下代码展示了如何将LLM流式返回的文本存储到队列中，然后从队列中流式读取数据并发送给TTS服务，接收TTS返回的音频片段进行拼接，最后保存为WAV文件：

import json
import queue
import threading
import time
import pickle
import nls
import numpy as np
import time
from log import logger
from utils.rds import Rds
import re

tts_audio_list = []
audio_queue: queue.Queue[dict] = queue.Queue()  # 用于存储TTS返回的音频块
text_queue = queue.Queue()  # 用于存储发送给TTS的文本块（模拟LLM流式输出）
done_event = threading.Event()
error_holder: list = []

def split_cn_keep_punc(text: str):
    # 匹配中文常用的句子结束标点
    pattern = r'[^，。！？；：、]+[，。！？；：、]?'
    matches = re.findall(pattern, text)
    # 去除完全空白的结果
    return [m.strip() for m in matches if m.strip()]

# === 回调函数 ===
def on_data(audio: bytes, *args):  # 将TTS服务返回的音频放入队列
    if not audio:
        return
    if len(audio) % 2 != 0:
        logger.warning(f"奇数长度的音频字节 ({len(audio)})，截断最后一个字节")
        audio = audio[:-1]
    try:
        audio_queue.put_nowait({"audio": audio, "end": False})
    except Exception:
        logger.exception("无法将音频数据放入队列")

def on_error(message, *args):
    err = RuntimeError(f"NLS 错误: {message}, 参数={args}")
    logger.error(err)
    error_holder.append(err)
    done_event.set()

def on_completed(message, *args):
    try:
        status = json.loads(message).get("header", {}).get("status", 0)
        if True:
            return
        done_event.set()
    except Exception as e:
        logger.exception("on_completed 错误")
        error_holder.append(e)

def on_close(*args):
    logger.info(f"--------------TTS 流关闭: {args}----------------------------------")
    done_event.set()

sdk = nls.NlsStreamInputTtsSynthesizer(
    url="wss://nls-gateway-cn-beijing.aliyuncs.com/ws/v1",
    token=Rds.get(),  # 获取token的方法请参见阿里云文档
    appkey="xxx",
    on_data=on_data,
    on_error=on_error,
    on_close=on_close,
    on_completed=on_completed,
    callback_args=[],
)
sdk.startStreamInputTts()

Queue

参数设置如下：

• voice="aixia"
• aformat="wav"
• sample_rate=24000
• volume=50
• speech_rate=0
• pitch_rate=0

定义了一个用于文本转语音流式处理的函数tts_stream，该函数接受一个字符串队列作为输入参数：

def tts_stream(text_queue: str):

每个批次处理的最大文本量被设定为400个字符：

QUEUE_BATCH_SIZE = 400

工作线程逻辑

下面定义了一个名为tts_worker的工作线程，用于处理从队列中获取的文本数据并将其转换为语音流。

def tts_worker():
    try:
        while True:
            if text_queue.qsize() >= 1:
                text = text_queue.get_nowait()
                if text == "end":
                    break
                else:
                    sdk.sendStreamInputTts(text)  # 流式发送
            else:
                time.sleep(0.1)
        sdk.stopStreamInputTts()  # 结束发送
    except Exception as e:
        logger.exception("TTS worker failed")
        error_holder.append(e)
    finally:
        done_event.set()

创建并启动工作线程：

worker_thread = threading.Thread(target=tts_worker, daemon=True, args=())
worker_thread.start()

初始化结束标志：

__end = False

主循环逻辑

在主循环中，程序会持续检查是否已完成或音频队列是否为空，同时处理可能出现的错误：

try:
    while not done_event.is_set() or not audio_queue.empty():
        print("#########", audio_queue.qsize())
        if error_holder:
            raise error_holder[0]
        
        batch_audio: list[bytes] = []  # 存储字节数据
        force_flush = done_event.is_set()
        
        try:
            while len(batch_audio) < (float('inf') if force_flush else QUEUE_BATCH_SIZE):
                payload = audio_queue.get_nowait()
                batch_audio.append(payload["audio"])
                __end = payload["end"]
                if __end:
                    force_flush = True
        except queue.Empty:
            pass
        
        if not batch_audio and not force_flush:
            time.sleep(0.1)
            continue
        
        all_audio_bytes = b''.join(batch_audio)
        print("all_audio_bytes len------->", len(all_audio_bytes), len(batch_audio), audio_queue.qsize())
        
        if len(all_audio_bytes) == 0:
            length = 960
            audio_float = np.zeros(length, dtype=np.float32)
        else:
            int16_arr = np.frombuffer(all_audio_bytes, dtype=np.int16)
            audio_float = int16_arr.astype(np.float32) / 32767
        
        if __end and len(audio_float) < 960:
            pad = np.zeros(960 - len(audio_float), dtype=np.float32)
            audio_float = np.concatenate([audio_float, pad])
        
        tts_audio_list.append(audio_float)
        
        if __end:
            break
        
        time.sleep(0.001)

等待工作线程完成：

worker_thread.join(timeout=3.0)
if worker_thread.is_alive():
    logger.warning("TTS worker thread did not exit cleanly")

处理任何可能发生的异常：

if error_holder:
    raise error_holder[0]

确保所有资源正确释放：

finally:
    done_event.set()

整体文本将被切分，以模拟大型语言模型的流式输出过程。

“京”是北京的简称，作为中华人民共和国的首都，它位于华北大平原的北部，紧邻天津，其他三面则被河北省所环抱。北京的地势呈现出西北高、东南低的特点，属于暖温带季风气候区。该市的总面积达到16410平方千米，根据2024年末的数据，常住人口数量为2183.2万人。

北京不仅是一座历史悠久的城市，拥有超过3000年的建城历史和870年的建都历史，还曾是五个朝代的首都。这座城市保留了故宫、长城、颐和园等七处世界级的文化遗产。作为现代化的国际大都市，北京在中国乃至全球都扮演着重要的角色，它是国家的政治、文化、国际交流以及科技创新的中心。

北京的经济发展水平较高，主要产业集中在新一代信息技术、科技服务等高端领域和现代服务业。据2024年的统计，该地区的生产总值达到了49843.1亿元。此外，北京是中国的教育中心，汇集了北京大学、清华大学等多所享誉国内外的高等学府。与此同时，北京也是一个重要的国际交通枢纽，交通网络极为发达。