如何在pyttsx3语音播报过程中中断或停止播报？

pyttsx3本身的runAndWait()是阻塞调用，且没有原生的即时停止API，核心原因是它依赖的底层语音引擎（如Windows的SAPI5、macOS的NSSpeechSynthesizer）的控制逻辑需要和主线程分离才能实现中断。以下是经过验证的可行方案：

核心思路 #

将语音播报任务放到独立线程中执行，通过线程间通信触发引擎停止操作，避免阻塞主线程导致无法响应停止指令。

具体实现代码 #

import pyttsx3
import threading
from queue import Queue

class SpeechController:
    def __init__(self):
        self.engine = pyttsx3.init()
        self.stop_queue = Queue()
        self.thread = None

    def _speak_task(self, text):
        self.engine.say(text)
        # 循环处理事件，同时监听停止信号
        while self.engine.isBusy():
            if not self.stop_queue.empty():
                self.engine.stop()
                break
            self.engine.iterate()
        # 清空队列，避免影响下一次播报
        while not self.stop_queue.empty():
            self.stop_queue.get()

    def speak(self, text):
        # 如果已有播报任务，先停止
        if self.thread and self.thread.is_alive():
            self.stop()
        self.stop_queue.empty()
        self.thread = threading.Thread(target=self._speak_task, args=(text,))
        self.thread.start()

    def stop(self):
        if self.engine.isBusy():
            self.stop_queue.put(True)
            self.thread.join(timeout=1)

代码说明 #

• SpeechController类封装了引擎实例、线程管理和停止信号队列
• _speak_task方法在独立线程中执行播报，通过engine.iterate()代替runAndWait()，实现非阻塞的事件循环，同时监听停止队列的信号
• 调用stop()方法时，向队列发送停止信号，触发engine.stop()中断当前播报
• 每次启动新播报前会自动停止正在进行的任务，避免冲突

注意事项 #

• 不同平台的底层引擎对engine.stop()的响应速度略有差异，Windows平台的SAPI5响应最快
• 避免在主线程中直接调用engine.stop()，必须通过线程安全的方式触发，否则可能导致引擎状态异常
• 如果需要频繁启停，建议复用同一个SpeechController实例，不要重复初始化pyttsx3引擎

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