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

- 什么是流式文本语音合成？ TTS，Text-to-Speech，文本转语音，将一段文本转换成对应的语音，在人机对话、智能助手等多种场景下应用。流式文本语义合成重点在于流式，该场景要求文本不是一次性给到TTS模型，而是一段一段的递交，TTS模型实时一段一段的生成并返回。这在实时场景下很有用，比如LLM流式输出的text，转换成语音，实现数字人等。

本文根据阿里云的TTS服务，实现一个LLM输出流式转换成语音的示例。阿里云链接，官方的python sdk示例很简单，这里给个可以直接使用的示例。服务开通、环境安装、key获取等准备工作参考官网。

以下脚本实现将LLM流式返回的text存放至Queue，再从queue流式取数据发送给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
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"Odd-length audio bytes ({len(audio)}), truncating last byte")
        audio = audio[:-1]
    try:
        audio_queue.put_nowait({"audio": audio, "end": False})
    except Exception:
        logger.exception("Failed to put audio data into queue")

def on_error(message, *args):
    err = RuntimeError(f"NLS Error: {message}, args={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")
        error_holder.append(e)

def on_close(*args):
    logger.info(f"--------------tts stream closed: {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(
    voice="aixia",
    aformat="wav",
    sample_rate=24000,
    volume=50,
    speech_rate=0,
    pitch_rate=0,
)

def tts_stream(
        text_queue: str,
):

    QUEUE_BATCH_SIZE = 400

    # === Worker Thread ===
    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] = []  # 只存 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

            # 合并所有 bytes
            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
        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]

    except Exception as e:
        logger.error(f"tts_stream failed: {e}")
        raise
    finally:
        done_event.set()

# 整体文本，将被切分以模拟LLM流式输出
long_text = "北京，简称 “京”，是中华人民共和国的首都。北京地处华北大平原北部，与天津毗连，其余三面被河北环绕，地势西北高、东南低，属暖温带季风气候。它总面积 16410 平方千米，2024 年末常住人口 2183.2 万人。北京是世界著名古都，有 3000 多年建城史和 870 年建都史，为五朝古都，留存有故宫、长城、颐和园等 7 项世界文化遗产。作为现代化国际大都市，北京是全国的政治中心、文化中心、国际交往中心和科技创新中心。其经济发达，以新一代信息技术、科技服务业等高精尖产业和现代服务业为主，2024 年地区生产总值 49843.1 亿元。北京还是中国的教育中心，拥有北京大学、清华大学等知名学府。同时，它也是国际性综合交通枢纽，交通十分便捷。"

# 模拟LLM流式返回text，压入队列
def producer(long_text):
    for text in split_cn_keep_punc(long_text):
        text_queue.put(text)
        time.sleep(0.1)
    text_queue.put("end")

# 协程进行，不阻塞发送
threading.Thread(target=producer, args=(long_text,), daemon=True).start()

tts_stream(text_queue=text_queue)

print("tts_audio_list len---------->", len(tts_audio_list))
with open("./tts_audio_array_list.pkl", "wb") as f:
    pickle.dump(tts_audio_list, f)

以下脚本将保存的音频数据转化为wav文件：

import numpy as np
import moviepy.editor as mpy
import pickle
import numpy as np
import cv2
from moviepy.audio.AudioClip import AudioArrayClip

with open("./tts_audio_array_list.pkl", "rb") as f:
    audio_chunks = pickle.load(f)  # 直接反序列化为原列表
all_audio_bytes = b''.join(audio_chunks)
int16_arr = np.frombuffer(all_audio_bytes, dtype=np.int16)
audio_float = int16_arr.astype(np.float32) / 32767
print(len(audio_float))
import soundfile as sf
sf.write("debug_audio_.wav", audio_float, 24000, subtype="FLOAT")

该TTS返回的语音在起始时有轻微噪声。