Python中PiperTTS结合RealtimeTTS实时TTS无音频输出问题求助

尝试将PiperTTS集成到实时文本转语音流水线，目标是从OpenAI API流式获取文本并通过PiperTTS转换为实时音频。目前文本能被处理，但无任何音频输出。已编写PiperVoiceEngineWrapper类实现流式合成，推测问题出在音频合成逻辑或异步播放环节，但无法定位具体位置。

实现代码 #

import time
import openai
from RealtimeTTS import TextToAudioStream
from piper.voice import PiperVoice
import pyaudio
import numpy as np
import sounddevice as sd

# Setup PiperTTS
voicedir = "pipertts/"  # 模型文件存储目录
model = voicedir + "en_US-norman-medium.onnx"
config = voicedir + "en_en_US_norman_medium_en_US-norman-medium.onnx.json"  # 正确的配置文件名

# PiperVoiceEngineWrapper 定义
class PiperVoiceEngineWrapper:
    def __init__(self):
        try:
            # 加载Piper模型
            voice = PiperVoice.load(model, config)
            self.piper_voice = voice
            self.format = pyaudio.paInt16  # 16位音频格式
            self.channels = 1  # 单声道
            self.rate = voice.config.sample_rate
            self.engine_name = "PiperTTS" 
        except Exception as e:
            print(f"加载Piper模型出错: {e}")
            raise e

    def get_stream_info(self):
        return self.format, self.channels, self.rate

    def synthesize(self, text):
        try:
            print(f"正在合成文本: {text}")
            # 流式合成音频
            for audio_bytes in self.piper_voice.synthesize_stream_raw(text):
                int_data = np.frombuffer(audio_bytes, dtype=np.int16)
                yield int_data
        except Exception as e:
            print(f"合成过程出错: {e}")

    def can_consume_generators(self):
        return True

    def set_muted(self, muted):
        pass  # 按需实现静音逻辑

# 初始化OpenAI客户端（对接Ollama本地服务）
openai_client = openai.OpenAI(base_url='http://localhost:11434/v1', api_key='ollama')

# 初始化语音引擎
voice_engine = PiperVoiceEngineWrapper()

# 用户输入与系统提示
user_input = "Who painted the Mona Lisa?"
system_message = "You are a helpful assistant and keep response as 1 sentence and keep it short."

# 准备对话消息
messages = [{"role": "system", "content": system_message}, {"role": "user", "content": user_input}]

def on_audio_chunk_callback(chunk):
    print(f"收到音频块，长度: {len(chunk)}")

def text_start():
    print("[文本开始]", end="", flush=True)

def text_stop():
    print("[文本结束]", end="", flush=True)

def audio_start():
    print("[音频开始]", end="", flush=True)

def audio_stop():
    print("[音频结束]", end="", flush=True)

# 初始化TextToAudioStream
stream = TextToAudioStream(
    voice_engine,
    on_text_stream_start=text_start, 
    on_text_stream_stop=text_stop, 
    on_audio_stream_start=audio_start, 
    on_audio_stream_stop=audio_stop,
    log_characters=True
)

# 从OpenAI流式获取文本
def streamdata():
    streamed_completion = openai_client.chat.completions.create(
        model="tinyllama",
        messages=messages,
        stream=True
    )

    # 逐个返回文本块
    for chunk in streamed_completion:
        if "content" in chunk.choices[0].delta:
            text_chunk = chunk.choices[0].delta["content"]
            yield text_chunk

def main():
    print("启动中...")
    test = streamdata()
    
    # 喂入文本并异步播放
    try:
        stream.feed(test)
        stream.play_async()

        # 保持主线程存活直到播放结束
        while stream.is_playing():
            time.sleep(0.1)
    except Exception as e:
        print(f"流式播放过程出错: {e}")

if __name__ == "__main__":
    main()

排查与修复方案 #

• 修复音频合成格式：Piper的synthesize_stream_raw直接返回符合要求的字节流，无需转成numpy数组。修改synthesize方法：

  def synthesize(self, text):
      try:
          print(f"正在合成文本: {text}")
          for audio_bytes in self.piper_voice.synthesize_stream_raw(text):
              yield audio_bytes  # 直接返回原始字节
      except Exception as e:
          print(f"合成过程出错: {e}")
  

• 绑定音频块回调：将on_audio_chunk_callback传入TextToAudioStream，确认音频块是否被正确传递：

  stream = TextToAudioStream(
      voice_engine,
      on_text_stream_start=text_start, 
      on_text_stream_stop=text_stop, 
      on_audio_stream_start=audio_start, 
      on_audio_stream_stop=audio_stop,
      on_audio_chunk=on_audio_chunk_callback,  # 添加此行
      log_characters=True
  )
  

• 优化播放线程等待逻辑：用stream.join()替代手动sleep循环，确保主线程等待播放完成：

  stream.play_async()
  stream.join()  # 等待播放线程结束
  

• 验证音频设备：检查默认输出设备是否正常，可手动指定设备：

  # 查看可用设备
  print(sd.query_devices())
  # 指定设备，例如使用索引1的设备
  sd.default.device = 1
  

• 确认模型加载状态：在PiperVoiceEngineWrapper的__init__中打印采样率，验证模型是否正确加载：

  print(f"模型采样率: {self.rate}")
  

内容的提问来源于stack exchange，提问作者Bogle