Python 实时语音输入句子分割：使用 PyAudio 和 VAD

使用 PyAudio 和 VAD 实时分割语音句子

本文将展示如何使用 Python 的 PyAudio 库和 VAD（Voice Activity Detection）算法实时录制音频并进行句子分割。这对于将语音数据转换为文本或进行其他语音处理任务非常有用。

代码示例

import pyaudio
import struct
import numpy as np
from collections import deque

# 定义常量
CHUNK = 1024  # 每一块的大小
FORMAT = pyaudio.paInt16  # 采样格式
CHANNELS = 1  # 声道数
RATE = 16000  # 采样率

# VAD 参数
VAD_STEP = 10  # 每次处理的音频时长（毫秒）
VAD_THRESHOLD = 500  # VAD 判断阈值，单位是方差

# 句子分割参数
SILENCE_THRESHOLD = 0.5  # 停顿阈值，单位是秒

# 初始化 VAD
vad_buffer = deque(maxlen=int(1000 / VAD_STEP))
vad_count = 0
vad_threshold_count = 0
vad_threshold = None

# 初始化句子分割
sentence_buffer = deque(maxlen=10)
current_sentence = ''

# 创建 pyaudio 对象
p = pyaudio.PyAudio()

# 打开音频流
stream = p.open(format=FORMAT,
                channels=CHANNELS,
                rate=RATE,
                input=True,
                frames_per_buffer=CHUNK)


# VAD 算法
def vad(data):
    global vad_count, vad_threshold_count, vad_threshold

    # 计算音量
    volume = np.sqrt(np.mean(np.square(data)))

    # 将音量加入缓冲区
    vad_buffer.append(volume)

    # 统计缓冲区内音量超过均值的数量
    vad_count = sum([1 for v in vad_buffer if v > np.mean(vad_buffer)])

    # 如果超过阈值，则增加阈值统计数量
    if vad_count > vad_threshold_count:
        vad_threshold_count += 1

        # 如果连续 5 个时间段内都超过阈值，则更新阈值为当前均值
        if vad_threshold_count >= 5:
            vad_threshold = np.mean(vad_buffer)
            vad_threshold_count = 0
    else:
        vad_threshold_count = 0

    # 如果阈值已经确定，则判断当前音量是否超过阈值
    if vad_threshold is not None:
        if volume > vad_threshold:
            return True
    return False


# 句子分割算法
def sentence_segmentation(data):
    global current_sentence, sentence_buffer

    # 判断是否有停顿
    if np.max(data) < SILENCE_THRESHOLD:
        # 将当前句子加入缓冲区
        sentence_buffer.append(current_sentence)
        current_sentence = ''
    else:
        # 将音频转换为字符串
        s = struct.pack('%dh' % len(data), *data)
        text = s.decode('utf-16')

        # 将字符串加入当前句子
        current_sentence += text

    # 将缓冲区内的所有句子合并为一个字符串
    return ''.join(list(sentence_buffer))


# 实时录音
while True:
    # 读取音频数据
    data = stream.read(CHUNK)
    data = struct.unpack('%dh' % (len(data) / 2), data)

    # VAD 算法判断是否有语音
    if vad(data):
        # 句子分割算法判断句子
        sentence = sentence_segmentation(data)
        print(sentence)

代码说明

导入必要的库：
- pyaudio 用于音频录制
- struct 用于处理音频数据
- numpy 用于音频处理
- collections.deque 用于创建缓冲区
定义常量：
- CHUNK: 每一块音频数据的字节数
- FORMAT: 音频数据格式
- CHANNELS: 声道数
- RATE: 采样率
VAD 参数：
- VAD_STEP: 每次处理的音频时长（毫秒）
- VAD_THRESHOLD: VAD 判断阈值，单位是方差
句子分割参数：
- SILENCE_THRESHOLD: 停顿阈值，单位是秒
初始化 VAD 和句子分割：
- vad_buffer: 用于存储音量数据的缓冲区
- vad_count: 缓冲区内音量超过均值的数量
- vad_threshold_count: 连续超过阈值的时间段数量
- vad_threshold: VAD 阈值
- sentence_buffer: 用于存储分割出的句子的缓冲区
- current_sentence: 当前正在积累的句子
创建 PyAudio 对象并打开音频流：
- p = pyaudio.PyAudio(): 创建 PyAudio 对象
- stream = p.open(...): 打开音频流，设置格式、声道数、采样率和输入模式
VAD 算法：
- vad(data) 函数：
  - 计算音频数据的音量
  - 将音量加入缓冲区
  - 统计缓冲区内音量超过均值的数量
  - 如果连续超过阈值的时间段数量达到 5，则更新 VAD 阈值
  - 判断当前音量是否超过 VAD 阈值，并返回 True 或 False
句子分割算法：
- sentence_segmentation(data) 函数：
  - 判断是否有停顿
  - 如果有停顿，则将当前句子加入缓冲区
  - 如果没有停顿，则将音频数据转换为字符串并添加到当前句子
  - 返回缓冲区内所有句子的合并字符串
实时录音：
- 使用 while True 循环持续读取音频数据
- 使用 vad(data) 判断是否有语音
- 如果有语音，则使用 sentence_segmentation(data) 进行句子分割
- 打印分割出的句子

注意事项

该代码使用了简单的 VAD 算法，对于复杂的环境可能需要更高级的算法
句子分割的阈值需要根据实际情况进行调整
该代码仅用于演示，实际应用中需要根据具体需求进行修改和完善

进一步学习

希望本文能帮助您了解如何使用 PyAudio 和 VAD 实时分割语音句子。如果您有任何问题或建议，请随时提出。