Python中同步Text to Speech与Speech to Text的实现问题求助

我用pyttsx3实现文本转语音（TTS），Vosk实现离线语音转文本（STT），将两者放在独立线程中运行时，同一设备上TTS播放的语音无法被STT识别，但用其他设备播放该语音时可以正常识别。相关代码如下：

from vosk import Model, KaldiRecognizer
import pyaudio
import pyttsx3
import threading
import time

def text_to_speech():
    my_text_to_speech = pyttsx3.init()
    my_text_var = "start 0xFF1234"    
    
    for x in range(10): 
        my_text_to_speech.say(my_text_var)
        my_text_to_speech.runAndWait()

def start_speech_to_text(): 
    model = Model(r"C:\vosk-model-small-en-us-0.15")
    recognizer = KaldiRecognizer(model, 16000)
    mic = pyaudio.PyAudio()

    stream = mic.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=8192)
    stream.start_stream()
   
    for x in range(20): 
        data = stream.read(4096)
        if recognizer.AcceptWaveform(data):
            text = recognizer.Result()
            print(f"' {text[14:-3]} '")

#-- Create the two threads 
thread1_STT = threading.Thread(target=start_speech_to_text)
thread2_TTS = threading.Thread(target=text_to_speech)

#-- start threads
thread1_STT.start()
thread2_TTS.start()

#-- wait for threads to finish
thread1_STT.join()
thread2_TTS.join()

问题原因 #

1. 音频捕获限制：默认情况下，麦克风仅捕获外界声音，无法捕获系统自身播放的TTS音频，需开启系统音频回路（如Windows的立体声混音）才能实现内部音频捕获。
2. 线程时序不匹配：STT与TTS线程无同步机制，可能出现TTS已开始播放，但STT未完成初始化；或STT循环提前结束，未覆盖TTS的全部播放时长。
3. 播放与录音不同步：pyttsx3的runAndWait()为阻塞式播放，但线程循环次数未匹配播放时长，导致部分语音未被STT捕获。

解决方案 #

1. 开启系统音频回路 #

• Windows：在声音设置中启用「立体声混音」设备，将系统输出路由到麦克风输入。
• Linux/macOS：使用PulseAudio或Core Audio的音频回路功能，配置系统输出作为麦克风输入源。

2. 同步线程时序 #

使用threading.Event确保STT完成初始化后，TTS才开始播放，避免时序差导致的捕获遗漏。

3. 调整代码逻辑 #

修改STT循环为按时间运行，确保覆盖TTS播放时长；同时枚举音频设备，指定立体声混音作为输入源。

修改后的代码示例 #

from vosk import Model, KaldiRecognizer
import pyaudio
import pyttsx3
import threading
import time

def text_to_speech(start_event):
    start_event.wait()  # 等待STT准备就绪后再播放
    tts_engine = pyttsx3.init()
    target_text = "start 0xFF1234"    
    
    for _ in range(10): 
        tts_engine.say(target_text)
        tts_engine.runAndWait()
        time.sleep(0.1)  # 给STT留足处理时间

def start_speech_to_text(start_event): 
    model = Model(r"C:\vosk-model-small-en-us-0.15")
    recognizer = KaldiRecognizer(model, 16000)
    pa = pyaudio.PyAudio()

    # 查找立体声混音设备（Windows）或音频回路设备（Linux/macOS）
    input_device_idx = None
    for idx in range(pa.get_device_count()):
        dev_info = pa.get_device_info_by_index(idx)
        if "立体声混音" in dev_info["name"] or "Stereo Mix" in dev_info["name"]:
            input_device_idx = idx
            print(f"已选择音频输入设备：{dev_info['name']}")
            break

    # 配置音频流
    stream_config = {
        "format": pyaudio.paInt16,
        "channels": 1,
        "rate": 16000,
        "input": True,
        "frames_per_buffer": 8192
    }
    if input_device_idx is not None:
        stream_config["input_device_index"] = input_device_idx

    stream = pa.open(**stream_config)
    stream.start_stream()
   
    start_event.set()  # 通知TTS可以开始播放
    print("STT初始化完成，等待TTS播放...")

    # 让STT运行20秒，覆盖TTS的播放时长
    start_time = time.time()
    while time.time() - start_time < 20:
        audio_data = stream.read(4096)
        if recognizer.AcceptWaveform(audio_data):
            result = recognizer.Result()
            print(f"识别结果：'{result[14:-3]}'")
    
    # 清理资源
    stream.stop_stream()
    stream.close()
    pa.terminate()

# 创建同步事件
sync_event = threading.Event()

# 启动线程：先启动STT，确保它准备好再启动TTS
stt_thread = threading.Thread(target=start_speech_to_text, args=(sync_event,))
tts_thread = threading.Thread(target=text_to_speech, args=(sync_event,))

stt_thread.start()
tts_thread.start()

# 等待线程结束
stt_thread.join()
tts_thread.join()

关键修改说明 #

• 使用threading.Event实现线程同步，避免STT未就绪时TTS已开始播放。
• 自动查找音频回路设备，确保STT能捕获系统播放的TTS语音。
• 将STT的循环改为按时间运行，确保覆盖TTS的全部播放周期。
• 在TTS每次播放后添加短暂延迟，避免播放过于密集导致STT处理不及时。

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