基于whisper流式语音识别

为了实现持续监听麦克风并在检测到声音时进行转录，我们可以将流的监听时间设置为无限长。通过使用一个音量门限来检测是否有声音，然后进行转录。

安装依赖

确保安装必要的库：

pip install torch torchaudio openai-whisper sounddevice numpy

代码实现

import torch
import whisper
import sounddevice as sd
import numpy as np

# 加载预训练的 Whisper 模型
model = whisper.load_model("base")

# 设置模型为评估模式
model.eval()

# 定义流式解码函数
def stream_decode(audio_buffer, sample_rate=16000):
    audio_tensor = torch.tensor(audio_buffer).float()
    result = model.transcribe(audio_tensor, fp16=False)
    return result['text']

# 音频缓冲区和其他参数
buffer_size = 16000  # 每个音频块的大小（1秒）
audio_buffer = np.zeros(buffer_size * 10, dtype=np.float32)  # 预留10秒缓冲区
buffer_offset = 0
silence_threshold = 0.01  # 声音门限

# 麦克风回调函数
def callback(indata, frames, time, status):
    global audio_buffer, buffer_offset

    if status:
        print(status, flush=True)

    # 计算当前音频块的音量
    volume_norm = np.linalg.norm(indata) * 10
    if volume_norm > silence_threshold:
        # 将新音频数据复制到缓冲区
        audio_buffer[buffer_offset:buffer_offset+frames] = indata[:, 0]
        buffer_offset += frames

        # 当缓冲区达到或超过设定的大小时进行处理
        if buffer_offset >= buffer_size:
            text = stream_decode(audio_buffer[:buffer_size])
            print(f"Transcription: {text}", flush=True)

            # 移动缓冲区的数据
            audio_buffer = np.roll(audio_buffer, -buffer_size)
            buffer_offset -= buffer_size
    else:
        # 如果检测到的音量低于门限，将缓冲区位置重置
        buffer_offset = 0

# 启动麦克风流
def start_streaming():
    stream = sd.InputStream(callback=callback, channels=1, samplerate=16000, blocksize=buffer_size)
    with stream:
        print("Listening...")
        while True:
            sd.sleep(1000)  # 继续监听

# 开始流式解码
start_streaming()

代码说明

依赖加载：

流式解码函数 stream_decode：

音频缓冲区和参数设置：

麦克风回调函数 callback：

启动麦克风流 start_streaming：

这个示例代码展示了如何实现持续监听麦克风，并在检测到声音时进行实时转录。你可以根据实际需求进一步优化音量门限和缓冲区处理逻辑，以提高性能和准确性。

总结