多线程语音助手执行指令后停止语音识别的问题求助

我正在开发个人语音助手，想用多线程实现在后台执行其他任务的同时监听并识别语音。现在遇到的问题是：程序识别到语音关键词并执行对应任务后，语音识别功能会停止工作，但其他功能还能正常运行。

原实现代码 #

import random
import time
import datetime
import speech_recognition as sr
import pyttsx3
from threading import Thread
import pyaudio

engine = pyttsx3.init()
engine.setProperty("rate", 165)
voices = engine.getProperty("voices")
engine.setProperty("voice", voices[1].id)

threadlist = []
        
def listening():
    while 2 == 2:
        try:
            with sr.Microphone() as source:
                r = sr.Recognizer()
                print("start listening")
                audio_data = r.listen(source)
                print("Recognizing...")
                global sentence
                text = r.recognize_google(audio_data, language="us-US")
                sentence = text.lower()
                print(sentence)
                if "time" in sentence.split():
                    print("said time")
                    engine.say(f"Right now it is {current_hour} hours and {current_minute} minutes")
                    engine.runAndWait()
                elif "thank" in sentence.split():
                    if "you" in sentence.split():
                        engine.say("You are welcome, sir")
                        engine.runAndWait()
        except:
            print("Failed to recognize the sound")
            text = ""               


def get_datetime():
    while 2 == 2:
        global current_hour
        global current_minute
        global current_day
        print("getting datetime")
        current_datetime = datetime.datetime.now()

        current_hour = current_datetime.strftime("%H")
        current_minute = current_datetime.strftime("%M") 
        current_day = current_datetime.strftime("%u") 
        time.sleep(30)

def test():
    i = 0
    while i < 50:
        i = i + 1
        time.sleep(2)
        print(i)

threadlist.append(Thread(target=get_datetime))
threadlist.append(Thread(target=listening))
threadlist.append(Thread(target=test))

for t in threadlist:
    t.start()
for t in threadlist:
    t.join()

问题原因 #

1. TTS阻塞监听线程：engine.runAndWait()会直接阻塞当前的监听线程，而且pyttsx3的引擎不是线程安全的，执行TTS时会抢占麦克风资源，导致后续无法继续监听语音。
2. Recognizer重复初始化：每次循环都创建新的sr.Recognizer()实例，频繁重置识别器会破坏持续监听的状态，引发异常导致监听停止。
3. 异常捕获太宽泛：原代码用except:捕获所有异常，会掩盖一些关键错误（比如语音识别服务请求失败），导致无法定位真正的停止原因。

修复方案 #

• 把TTS任务放到单独线程执行，避免阻塞监听流程；给TTS引擎加锁，保证线程安全。
• 只初始化一次Recognizer和Microphone，提前校准环境噪音，提升识别稳定性。
• 细化异常捕获，区分不同类型的错误，方便排查问题。

修改后的代码 #

import time
import datetime
import speech_recognition as sr
import pyttsx3
from threading import Thread, Lock

# 初始化TTS引擎并添加线程锁
engine = pyttsx3.init()
engine.setProperty("rate", 165)
voices = engine.getProperty("voices")
engine.setProperty("voice", voices[1].id)
engine_lock = Lock()

# 全局时间变量
current_hour = ""
current_minute = ""
current_day = ""

def speak(text):
    """封装TTS函数，加锁确保线程安全"""
    with engine_lock:
        engine.say(text)
        engine.runAndWait()

def listening():
    # 只初始化一次识别器和麦克风
    r = sr.Recognizer()
    with sr.Microphone() as source:
        # 校准环境噪音，仅执行一次
        r.adjust_for_ambient_noise(source, duration=1)
        print("开始持续监听...")
        while True:
            try:
                print("等待语音输入...")
                audio_data = r.listen(source)
                print("正在识别语音...")
                text = r.recognize_google(audio_data, language="us-US")
                sentence = text.lower()
                print(f"识别结果: {sentence}")
                
                # 将TTS任务放入独立线程，不阻塞监听
                if "time" in sentence.split():
                    print("触发时间查询指令")
                    Thread(target=speak, args=(f"现在是{current_hour}点{current_minute}分",)).start()
                elif "thank" in sentence.split() and "you" in sentence.split():
                    print("触发感谢回复指令")
                    Thread(target=speak, args=("不客气，先生",)).start()
                    
            except sr.UnknownValueError:
                print("无法识别当前语音，请重新输入")
            except sr.RequestError as e:
                print(f"语音识别服务请求失败: {e}")
            except Exception as e:
                print(f"监听过程出错: {e}")
            # 短暂休眠降低CPU占用
            time.sleep(0.1)

def get_datetime():
    global current_hour, current_minute, current_day
    while True:
        current_datetime = datetime.datetime.now()
        current_hour = current_datetime.strftime("%H")
        current_minute = current_datetime.strftime("%M")
        current_day = current_datetime.strftime("%u")
        print(f"已更新系统时间: {current_hour}:{current_minute}")
        time.sleep(30)

def test():
    i = 0
    while i < 50:
        i += 1
        time.sleep(2)
        print(f"测试计数: {i}")

# 创建线程并设置为守护线程（主线程退出时自动结束）
threads = [
    Thread(target=get_datetime, daemon=True),
    Thread(target=listening, daemon=True),
    Thread(target=test, daemon=True)
]

# 启动所有线程
for t in threads:
    t.start()

# 主线程等待所有线程完成
for t in threads:
    t.join()

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