Реализация эндпоинтинга речи (End-of-Speech Detection)

Реализация эндпоинтинга речи (End-of-Speech Detection)

Endpointing — определение момента, когда пользователь закончил говорить и система должна обрабатывать его запрос. Критически важно для голосовых ботов: слишком быстрый отклик перебивает пользователя, слишком медленный создаёт неловкие паузы.

Алгоритм endpointing

import collections
import time
from enum import Enum

class SpeechState(Enum):
    SILENCE = 0
    SPEECH = 1

class EndpointDetector:
    def __init__(
        self,
        vad,
        sample_rate: int = 16000,
        frame_ms: int = 30,
        silence_threshold_ms: int = 700,  # пауза для завершения
        min_speech_ms: int = 300,          # минимальная длина высказывания
    ):
        self.vad = vad
        self.sample_rate = sample_rate
        self.frame_bytes = int(sample_rate * frame_ms / 1000) * 2
        self.silence_frames_needed = silence_threshold_ms // frame_ms
        self.min_speech_frames = min_speech_ms // frame_ms

        self.state = SpeechState.SILENCE
        self.silence_counter = 0
        self.speech_buffer = bytearray()
        self.speech_frame_count = 0

    def process_frame(self, frame: bytes) -> tuple[bool, bytes | None]:
        """
        Returns: (endpoint_detected, speech_audio_or_none)
        """
        is_speech = self.vad.is_speech(frame, self.sample_rate)

        if is_speech:
            self.state = SpeechState.SPEECH
            self.silence_counter = 0
            self.speech_buffer.extend(frame)
            self.speech_frame_count += 1
        else:
            if self.state == SpeechState.SPEECH:
                self.silence_counter += 1
                self.speech_buffer.extend(frame)  # включаем финальную тишину

                if self.silence_counter >= self.silence_frames_needed:
                    if self.speech_frame_count >= self.min_speech_frames:
                        audio = bytes(self.speech_buffer)
                        self._reset()
                        return True, audio
                    else:
                        self._reset()

        return False, None

    def _reset(self):
        self.state = SpeechState.SILENCE
        self.silence_counter = 0
        self.speech_buffer = bytearray()
        self.speech_frame_count = 0

Адаптивный endpointing

В реальных диалогах нужен контекстный эндпоинтинг: при ответе на открытый вопрос ждём дольше, при подтверждении — меньше:

# Разные пороги для разных типов запросов
THRESHOLDS = {
    "open_question": 1200,   # мс тишины
    "yes_no": 500,
    "command": 600,
    "default": 700,
}

Практические параметры

Для телефонного голосового бота оптимальные параметры: тишина 600–800 мс, минимальная речь 200 мс. Для диктовки: тишина 1500–2000 мс.

Сроки: реализация базового endpointing — 2–3 дня. Адаптивный с ML-предсказанием — 1 неделя.