如何修改实时语音识别脚本实现麦克风自动启停与转录确认？

以下是修改后的代码，实现了「监听→用户停止说话后关闭麦克风→转录文本+执行操作→恢复监听」的完整流程，同时保留了用户验证转录结果的交互环节：

import io
from pydub import AudioSegment
import speech_recognition as sr
import whisper
import queue
import tempfile
import os
import threading
import click
import torch
import numpy as np

@click.command()
@click.option("--model", default="base", help="Model to use", type=click.Choice(["tiny","base", "small","medium","large"]))
@click.option("--english", default=False, help="是否使用英文模型",is_flag=True, type=bool)
@click.option("--verbose", default=False, help="是否输出详细日志", is_flag=True,type=bool)
@click.option("--energy", default=300, help="麦克风检测的能量阈值", type=int)
@click.option("--dynamic_energy", default=False,is_flag=True, help="是否启用动态能量阈值", type=bool)
@click.option("--pause", default=0.8, help="判定说话结束的停顿时间", type=float)
@click.option("--save_file",default=False, help="是否保存录音文件", is_flag=True,type=bool)

def main(model, english,verbose, energy, pause,dynamic_energy,save_file):
    temp_dir = tempfile.mkdtemp() if save_file else None
    # large模型没有英文专用版本
    if model != "large" and english:
        model = model + ".en"
    audio_model = whisper.load_model(model)
    audio_queue = queue.Queue()
    result_queue = queue.Queue()
    # 线程安全的监听激活信号，初始为激活状态
    listen_active = threading.Event()
    listen_active.set()

    def run_recording_loop():
        while True:
            # 等待监听激活信号
            listen_active.wait()
            record_audio(audio_queue, energy, pause, dynamic_energy, save_file, temp_dir)
            # 录音完成后，暂时关闭监听信号
            listen_active.clear()

    # 启动录音循环线程
    threading.Thread(target=run_recording_loop, daemon=True).start()
    # 启动转录线程
    threading.Thread(target=transcribe_forever,
                     args=(audio_queue, result_queue, audio_model, english, verbose, save_file), daemon=True).start()

    while True:
        result = result_queue.get()
        print(result)
        # 执行自定义操作示例：询问用户转录是否正确
        user_input = input("转录内容是否正确？(y/n): ").strip().lower()
        if user_input == 'y':
            print("执行后续操作...")
            # 这里添加你的自定义操作逻辑
        # 重新激活监听，开启麦克风
        print("重新开启监听，请说话...")
        listen_active.set()


def record_audio(audio_queue, energy, pause, dynamic_energy, save_file, temp_dir):
    r = sr.Recognizer()
    r.energy_threshold = energy
    r.pause_threshold = pause
    r.dynamic_energy_threshold = dynamic_energy

    # 单次录音完成后自动关闭麦克风上下文
    with sr.Microphone(sample_rate=16000) as source:
        print("正在监听，请说话...")
        # 监听直到用户停止说话（触发pause_threshold）
        audio = r.listen(source)
        if save_file:
            data = io.BytesIO(audio.get_wav_data())
            audio_clip = AudioSegment.from_file(data)
            filename = os.path.join(temp_dir, f"temp_{os.urandom(4).hex()}.wav")
            audio_clip.export(filename, format="wav")
            audio_data = filename
        else:
            torch_audio = torch.from_numpy(np.frombuffer(audio.get_raw_data(), np.int16).flatten().astype(np.float32) / 32768.0)
            audio_data = torch_audio

        audio_queue.put_nowait(audio_data)


def transcribe_forever(audio_queue, result_queue, audio_model, english, verbose, save_file):
    while True:
        audio_data = audio_queue.get()
        if english:
            result = audio_model.transcribe(audio_data, language='english')
        else:
            result = audio_model.transcribe(audio_data)

        if not verbose:
            predicted_text = result["text"]
            result_queue.put_nowait(f"你说的是: {predicted_text}")
        else:
            result_queue.put_nowait(result)

        if save_file:
            os.remove(audio_data)


if __name__ == "__main__":
    main()

关键改动说明 #

• 监听控制信号：使用threading.Event()作为线程安全的启停标志，替代原有的无限循环录音，实现麦克风的可控启停。
• 单次录音逻辑：修改record_audio函数，每次仅完成一次录音后就退出麦克风上下文（自动关闭麦克风），等待下一次激活信号。
• 主循环交互：在获取转录结果后，加入用户验证环节，执行自定义操作后再重新激活监听信号，恢复麦克风监听。
• 线程优化：将录音线程改为循环等待激活的模式，配合守护线程确保程序退出时线程自动终止。

内容的提问来源于stack exchange，提问作者Tahseen Abbas hyder