Vladimir BryzgalovandClaude Opus 5
5d576dd3c4
Диаризация занимала 79 % времени, а её метку мы не используем: на записи с одним микрофоном голоса не расходятся, роли расставляет LLM по смыслу. Silero VAD решает единственную нужную задачу и делает это вчетверо быстрее. Замер на 205 минутах: x16 -> x60, слов на 1,8 % меньше. Семь записей из восьми в пределах +-4 %, на одной теряется 19 %. Отдельно найдено: max_speech_duration у Silero по умолчанию 20 с, и на таких кусках распознавание теряет текст. Снижение до 6 с даёт 6-8 п.п. полноты бесплатно. Диаризация сохранена настройкой diarize. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
423 lines
22 KiB
Python
423 lines
22 KiB
Python
"""Распознавание речи с разделением по говорящим.
|
||
|
||
Модели загружаются один раз при старте сервиса и живут в памяти: холодная
|
||
загрузка GigaAM занимает секунды, и делать её на каждый запрос недопустимо.
|
||
"""
|
||
import logging
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
import wave
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
|
||
from app.acoustics import segment_acoustics, separation_quality
|
||
from app.text import (apply_replacements, load_replacements,
|
||
normalize_typography, split_sentences)
|
||
from app.turns import Segment, chunk_ranges, merge_turns, speech_stats
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
SAMPLE_RATE = 16000
|
||
# GigaAM падает на кусках длиннее ~200 с, держим запас.
|
||
# Предел зависит от сборки onnxruntime: на macOS реплика в 190 с проходит,
|
||
# на Windows та же падает в свёртке энкодера. Берём с большим запасом, а на
|
||
# случай сюрпризов есть деление куска пополам при ошибке.
|
||
MAX_CHUNK_SEC = 60.0
|
||
# Ниже этого делить бессмысленно: явно не длина виновата.
|
||
MIN_SPLIT_SEC = 2.0
|
||
# Модель отпечатков рассчитана на короткий фрагмент и на длинном падает.
|
||
# Для оценки голоса секунд более чем достаточно.
|
||
EMBED_SEC = 8.0
|
||
# Ниже этого значения голоса практически неразличимы и разметка по говорящим
|
||
# случайна. На записях с одним микрофоном в комнате так бывает часто.
|
||
RELIABLE_SEPARATION = 0.35
|
||
# Промежутки короче этого - обычные паузы в разговоре, туда лезть незачем.
|
||
GAP_MIN_SEC = 3.0
|
||
# Короткий обрывок из промежутка почти всегда шум, принятый за слово:
|
||
# шелест бумаг и стук клавиш на записи со стола дают почти ту же громкость,
|
||
# что и речь. Порог отсекает их, не трогая настоящие фразы.
|
||
MIN_RECOVERED_CHARS = 15
|
||
ASR_MODEL_NAME = "gigaam-v3-e2e-rnnt"
|
||
ASR_MODEL_REL = Path("gigaam")
|
||
SEG_MODEL_REL = Path("sherpa-onnx-pyannote-segmentation-3-0") / "model.onnx"
|
||
EMB_MODEL_REL = Path("nemo_en_titanet_small.onnx")
|
||
VAD_MODEL_REL = Path("silero_vad.onnx")
|
||
# Порог детектора речи. Ниже - подхватывает фон, выше - теряет тихие реплики.
|
||
VAD_THRESHOLD = 0.5
|
||
# Паузу короче этой внутри речи не считаем разрывом: иначе фраза рвётся
|
||
# на каждом вдохе.
|
||
VAD_MIN_SILENCE_SEC = 0.5
|
||
VAD_MIN_SPEECH_SEC = 0.25
|
||
# Предельная длина отрезка речи. Решающая настройка: распознавание теряет
|
||
# текст на длинных кусках, и умолчание Silero в 20 с даёт заметно меньше
|
||
# слов, чем более мелкие отрезки диаризации.
|
||
VAD_MAX_SPEECH_SEC = 6.0
|
||
|
||
|
||
class ModelsMissing(RuntimeError):
|
||
"""Модели не скачаны - сервис не может работать."""
|
||
|
||
|
||
def find_ffmpeg(base_dir: Path) -> str:
|
||
"""Ищет ffmpeg сначала в папке пакета, потом в системе."""
|
||
for candidate in (base_dir / "bin" / "ffmpeg.exe", base_dir / "bin" / "ffmpeg"):
|
||
if candidate.is_file():
|
||
return str(candidate)
|
||
found = shutil.which("ffmpeg")
|
||
if not found:
|
||
raise ModelsMissing(
|
||
"не найден ffmpeg - положите ffmpeg.exe в папку bin рядом с сервисом")
|
||
return found
|
||
|
||
|
||
# Выравнивание громкости - решающий шаг для записей с одним микрофоном на столе.
|
||
# На проверочных разговорах доля второго участника выросла с 1.8 до 24.3 процента,
|
||
# а переключений между репликами - с 21 до 71 процента. Распознанный текст при
|
||
# этом не меняется: сравнение фрагментов дало полностью совпадающие расшифровки.
|
||
# Параметры подобраны замером: f=400:g=3 даёт лишь 8.6 процента вместо 43.2,
|
||
# так что менять их наугад не стоит.
|
||
NORMALIZE_FILTER = "dynaudnorm=f=200:g=5"
|
||
|
||
|
||
def to_wav16k(src: Path, dst: Path, ffmpeg: str, normalize: bool = True) -> float:
|
||
"""Приводит любой аудиофайл к 16 кГц моно WAV. Возвращает длительность."""
|
||
command = [ffmpeg, "-nostdin", "-v", "error", "-y", "-i", str(src), "-vn"]
|
||
if normalize:
|
||
command += ["-af", NORMALIZE_FILTER]
|
||
command += ["-ar", str(SAMPLE_RATE), "-ac", "1", "-c:a", "pcm_s16le", str(dst)]
|
||
subprocess.run(command, check=True, capture_output=True)
|
||
with wave.open(str(dst)) as w:
|
||
return w.getnframes() / w.getframerate()
|
||
|
||
|
||
def read_wav(path: Path) -> np.ndarray:
|
||
with wave.open(str(path)) as w:
|
||
if w.getframerate() != SAMPLE_RATE or w.getnchannels() != 1:
|
||
raise ValueError(f"ожидался {SAMPLE_RATE} Гц моно, получено "
|
||
f"{w.getframerate()} Гц / {w.getnchannels()} кан.")
|
||
raw = w.readframes(w.getnframes())
|
||
return np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0
|
||
|
||
|
||
def gap_ranges(turns: list[dict], duration: float):
|
||
"""Промежутки между репликами, куда распознавание не заглядывало."""
|
||
previous = 0.0
|
||
for turn in turns:
|
||
if turn["start"] - previous >= GAP_MIN_SEC:
|
||
yield previous, turn["start"]
|
||
previous = max(previous, turn["end"])
|
||
if duration - previous >= GAP_MIN_SEC:
|
||
yield previous, duration
|
||
|
||
|
||
class Pipeline:
|
||
"""Диаризация плюс распознавание. Потокобезопасен на уровне одного воркера."""
|
||
|
||
def __init__(self, models_dir: Path, threads: int, replacements_path: Path,
|
||
base_dir: Path):
|
||
self.models_dir = Path(models_dir)
|
||
self.threads = threads
|
||
self.replacements_path = Path(replacements_path)
|
||
self.base_dir = Path(base_dir)
|
||
self._ffmpeg: str | None = None
|
||
self._asr = None
|
||
self._diar_cache: dict[int, object] = {}
|
||
self._embedder = None
|
||
self._replacements: dict[str, str] = {}
|
||
|
||
@property
|
||
def ffmpeg(self) -> str:
|
||
"""Ищется при первом обращении, а не в конструкторе.
|
||
|
||
Иначе отсутствие ffmpeg роняло бы сам импорт модуля - и вместо внятного
|
||
сообщения пользователь получал бы стек при старте, а обновление кода
|
||
не могло бы проверить работоспособность новой версии.
|
||
"""
|
||
if self._ffmpeg is None:
|
||
self._ffmpeg = find_ffmpeg(self.base_dir)
|
||
return self._ffmpeg
|
||
|
||
def warmup(self) -> None:
|
||
"""Грузит модели заранее, чтобы первый запрос не ждал."""
|
||
import onnx_asr
|
||
|
||
missing = [str(p) for p in (self.models_dir / SEG_MODEL_REL,
|
||
self.models_dir / EMB_MODEL_REL,
|
||
self.models_dir / ASR_MODEL_REL / "config.json")
|
||
if not p.is_file()]
|
||
if missing:
|
||
raise ModelsMissing(
|
||
f"не найдены модели: {', '.join(missing)}. Запустите download_models.bat")
|
||
|
||
t0 = time.time()
|
||
# CoreML и DirectML на GigaAM падают на отдельных узлах графа, поэтому явно CPU:
|
||
# на 16-ядерном процессоре этого более чем достаточно.
|
||
self._asr = onnx_asr.load_model(
|
||
ASR_MODEL_NAME, str(self.models_dir / ASR_MODEL_REL),
|
||
providers=["CPUExecutionProvider"],
|
||
sess_options=self._session_options(),
|
||
)
|
||
self._reload_replacements()
|
||
log.info("модели загружены за %.1f с, потоков: %d", time.time() - t0, self.threads)
|
||
|
||
def _session_options(self):
|
||
import onnxruntime as ort
|
||
|
||
opts = ort.SessionOptions()
|
||
opts.intra_op_num_threads = self.threads
|
||
opts.inter_op_num_threads = 1
|
||
return opts
|
||
|
||
def _reload_replacements(self) -> None:
|
||
self._replacements = load_replacements(self.replacements_path)
|
||
|
||
def _diarizer(self, num_speakers: int):
|
||
"""Диаризатор кэшируется по числу говорящих - его создание не бесплатно."""
|
||
import sherpa_onnx
|
||
|
||
if num_speakers in self._diar_cache:
|
||
return self._diar_cache[num_speakers]
|
||
|
||
clustering = (sherpa_onnx.FastClusteringConfig(num_clusters=num_speakers)
|
||
if num_speakers > 0 else sherpa_onnx.FastClusteringConfig(threshold=0.5))
|
||
cfg = sherpa_onnx.OfflineSpeakerDiarizationConfig(
|
||
segmentation=sherpa_onnx.OfflineSpeakerSegmentationModelConfig(
|
||
pyannote=sherpa_onnx.OfflineSpeakerSegmentationPyannoteModelConfig(
|
||
model=str(self.models_dir / SEG_MODEL_REL)),
|
||
num_threads=self.threads),
|
||
embedding=sherpa_onnx.SpeakerEmbeddingExtractorConfig(
|
||
model=str(self.models_dir / EMB_MODEL_REL), num_threads=self.threads),
|
||
clustering=clustering,
|
||
# Подобрано на реальных звонках: отсекает обрывки на перекрывающейся
|
||
# речи, из-за которых в расшифровку лезли огрызки чужих слов.
|
||
min_duration_on=1.0,
|
||
min_duration_off=0.7,
|
||
)
|
||
diar = sherpa_onnx.OfflineSpeakerDiarization(cfg)
|
||
self._diar_cache[num_speakers] = diar
|
||
return diar
|
||
|
||
def _separation_quality(self, samples: np.ndarray, segments: list[Segment]) -> float:
|
||
"""Оценивает, действительно ли голоса разошлись, или разметка случайна."""
|
||
import sherpa_onnx
|
||
|
||
usable = [s for s in segments if s.end - s.start >= 1.0][:120]
|
||
if len(usable) < 4:
|
||
return 0.0
|
||
if self._embedder is None:
|
||
self._embedder = sherpa_onnx.SpeakerEmbeddingExtractor(
|
||
sherpa_onnx.SpeakerEmbeddingExtractorConfig(
|
||
model=str(self.models_dir / EMB_MODEL_REL), num_threads=self.threads))
|
||
vectors, labels = [], []
|
||
for seg in usable:
|
||
# Берём кусок из середины реплики: там речь устойчивее, чем на краях,
|
||
# а длинный фрагмент модель отпечатков просто не переваривает.
|
||
middle = (seg.start + seg.end) / 2
|
||
half = min(EMBED_SEC, seg.end - seg.start) / 2
|
||
piece = samples[int((middle - half) * SAMPLE_RATE):int((middle + half) * SAMPLE_RATE)]
|
||
if len(piece) < SAMPLE_RATE // 2:
|
||
continue
|
||
stream = self._embedder.create_stream()
|
||
stream.accept_waveform(SAMPLE_RATE, piece)
|
||
stream.input_finished()
|
||
vectors.append(np.array(self._embedder.compute(stream)))
|
||
labels.append(seg.speaker)
|
||
if len(vectors) < 4:
|
||
return 0.0
|
||
return separation_quality(np.array(vectors), np.array(labels))
|
||
|
||
def _recognize_safely(self, audio: np.ndarray, depth: int = 0) -> str:
|
||
"""Распознаёт кусок, при ошибке деля его пополам.
|
||
|
||
Пределы длины у разных сборок onnxruntime разные и заранее неизвестны,
|
||
поэтому вместо подбора константы просто уменьшаем кусок и пробуем снова.
|
||
"""
|
||
try:
|
||
return self._asr.recognize(audio, sample_rate=SAMPLE_RATE)
|
||
except Exception:
|
||
seconds = len(audio) / SAMPLE_RATE
|
||
if seconds <= MIN_SPLIT_SEC or depth >= 5:
|
||
log.warning("не удалось распознать кусок в %.1f с, пропускаю", seconds)
|
||
return ""
|
||
log.info("кусок в %.1f с не прошёл, делю пополам", seconds)
|
||
middle = len(audio) // 2
|
||
left = self._recognize_safely(audio[:middle], depth + 1)
|
||
right = self._recognize_safely(audio[middle:], depth + 1)
|
||
return " ".join(part for part in (left, right) if part)
|
||
|
||
def _vad_segments(self, samples: np.ndarray) -> list[Segment]:
|
||
"""Границы речи без разделения по говорящим.
|
||
|
||
Диаризация занимает около 79 % времени обработки, а её метку мы всё
|
||
равно не используем: на записи с одним микрофоном голоса не расходятся,
|
||
и роли расставляет LLM по смыслу. Детектор речи решает единственную
|
||
нужную здесь задачу - где речь есть - и делает это на порядок быстрее.
|
||
|
||
Говорящий у всех отрезков один и тот же (0): это честнее, чем выдавать
|
||
номер, за которым ничего не стоит.
|
||
"""
|
||
import sherpa_onnx
|
||
|
||
model = self.models_dir / VAD_MODEL_REL
|
||
if not model.is_file():
|
||
raise ModelsMissing(f"нет модели детектора речи: {model}")
|
||
|
||
config = sherpa_onnx.VadModelConfig()
|
||
config.silero_vad.model = str(model)
|
||
config.silero_vad.threshold = VAD_THRESHOLD
|
||
config.silero_vad.min_silence_duration = VAD_MIN_SILENCE_SEC
|
||
config.silero_vad.min_speech_duration = VAD_MIN_SPEECH_SEC
|
||
config.silero_vad.max_speech_duration = VAD_MAX_SPEECH_SEC
|
||
config.sample_rate = SAMPLE_RATE
|
||
config.num_threads = self.threads
|
||
|
||
window = 512
|
||
detector = sherpa_onnx.VoiceActivityDetector(config, buffer_size_in_seconds=60)
|
||
out: list[Segment] = []
|
||
for offset in range(0, len(samples), window):
|
||
detector.accept_waveform(samples[offset:offset + window])
|
||
while not detector.empty():
|
||
segment = detector.front
|
||
start = segment.start / SAMPLE_RATE
|
||
out.append(Segment(start=start,
|
||
end=start + len(segment.samples) / SAMPLE_RATE,
|
||
speaker=0))
|
||
detector.pop()
|
||
detector.flush()
|
||
while not detector.empty():
|
||
segment = detector.front
|
||
start = segment.start / SAMPLE_RATE
|
||
out.append(Segment(start=start,
|
||
end=start + len(segment.samples) / SAMPLE_RATE,
|
||
speaker=0))
|
||
detector.pop()
|
||
return out
|
||
|
||
def _recover_gaps(self, samples: np.ndarray, turns: list[dict],
|
||
duration: float) -> list[dict]:
|
||
"""Распознаёт то, что детектор речи не отметил как речь.
|
||
|
||
Диаризация задаёт не только «кто говорит», но и границы того, что
|
||
вообще попадает в распознавание. Модель pyannote обучена разделять
|
||
говорящих, а не искать речь, и на тихой записи со стола часть фраз
|
||
в эти границы не попадает - они пропадают молча. Замер на реальной
|
||
записи: 4 промежутка из 35 содержали речь, около 2 % слов.
|
||
|
||
Кто произнёс найденное, неизвестно: диаризация этих участков не
|
||
видела. Поэтому speaker = 0, а роль потом расставит LLM по смыслу.
|
||
"""
|
||
out: list[dict] = []
|
||
for start, end in gap_ranges(turns, duration):
|
||
pieces = []
|
||
for chunk_start, chunk_stop in chunk_ranges(start, end, MAX_CHUNK_SEC):
|
||
audio = samples[int(chunk_start * SAMPLE_RATE):int(chunk_stop * SAMPLE_RATE)]
|
||
if len(audio) < SAMPLE_RATE * 0.2:
|
||
continue
|
||
pieces.append(self._recognize_safely(audio))
|
||
text = " ".join(p.strip() for p in pieces if p and p.strip())
|
||
if len(text) < MIN_RECOVERED_CHARS:
|
||
continue
|
||
text = normalize_typography(apply_replacements(text, self._replacements))
|
||
audio = samples[int(start * SAMPLE_RATE):int(end * SAMPLE_RATE)]
|
||
out.append({
|
||
"speaker": 0,
|
||
"start": round(start, 2),
|
||
"end": round(end, 2),
|
||
"text": text,
|
||
"acoustics": segment_acoustics(audio),
|
||
# Признак второго прохода: говорящий неизвестен, и потребителю
|
||
# стоит знать, что это не результат диаризации.
|
||
"recovered": True,
|
||
})
|
||
return out
|
||
|
||
def transcribe(self, wav_path: Path, num_speakers: int = 2,
|
||
recover_gaps: bool = True, diarize: bool = False) -> dict:
|
||
"""Полный проход: диаризация, распознавание реплик, постобработка.
|
||
|
||
Модели грузятся при первой задаче, а не при создании: воркеров несколько,
|
||
и держать копии моделей у тех, кому работа ещё не досталась, незачем.
|
||
"""
|
||
if self._asr is None:
|
||
self.warmup()
|
||
self._reload_replacements()
|
||
|
||
samples = read_wav(wav_path)
|
||
duration = len(samples) / SAMPLE_RATE
|
||
|
||
t0 = time.time()
|
||
if diarize:
|
||
raw = self._diarizer(num_speakers).process(samples).sort_by_start_time()
|
||
segments = [Segment(start=s.start, end=s.end, speaker=s.speaker) for s in raw]
|
||
try:
|
||
quality = self._separation_quality(samples, segments)
|
||
except Exception as exc: # noqa: BLE001 - оценка вспомогательная
|
||
# Метрика не должна ронять задачу: расшифровка нужна и без неё.
|
||
log.warning("не удалось оценить разделение говорящих: %s", exc)
|
||
quality = 0.0
|
||
else:
|
||
segments = self._vad_segments(samples)
|
||
# Разделения не было, и притворяться, что оно удалось, нельзя.
|
||
quality = 0.0
|
||
t_diar = time.time() - t0
|
||
|
||
t0 = time.time()
|
||
turns_out = []
|
||
for turn in merge_turns(segments, gap=0.8):
|
||
pieces = []
|
||
for start, stop in chunk_ranges(turn.start, turn.end, MAX_CHUNK_SEC):
|
||
audio = samples[int(start * SAMPLE_RATE):int(stop * SAMPLE_RATE)]
|
||
if len(audio) < SAMPLE_RATE * 0.2:
|
||
continue
|
||
pieces.append(self._recognize_safely(audio))
|
||
text = " ".join(p.strip() for p in pieces if p and p.strip())
|
||
if not text:
|
||
continue
|
||
text = normalize_typography(apply_replacements(text, self._replacements))
|
||
audio = samples[int(turn.start * SAMPLE_RATE):int(turn.end * SAMPLE_RATE)]
|
||
turns_out.append({
|
||
"speaker": turn.speaker + 1,
|
||
"start": round(turn.start, 2),
|
||
"end": round(turn.end, 2),
|
||
"text": text,
|
||
# Подсказка для LLM: у говорящего ближе к микрофону громкость
|
||
# и доля высоких выше. Когда разделение по голосу ненадёжно,
|
||
# это единственный объективный признак, кто есть кто.
|
||
"acoustics": segment_acoustics(audio),
|
||
})
|
||
t_asr = time.time() - t0
|
||
|
||
t0 = time.time()
|
||
recovered = (self._recover_gaps(samples, turns_out, duration)
|
||
if recover_gaps else [])
|
||
t_recovery = time.time() - t0
|
||
if recovered:
|
||
turns_out = sorted(turns_out + recovered, key=lambda t: t["start"])
|
||
log.info("во втором проходе найдено реплик: %d", len(recovered))
|
||
|
||
return {
|
||
"duration_sec": round(duration, 1),
|
||
"turns": turns_out,
|
||
# Готовый материал для разметки ролей на стороне Talkscore: делить
|
||
# реплики самостоятельно ей незачем, пунктуацию ставит ASR.
|
||
"sentences": split_sentences(turns_out),
|
||
"stats": {
|
||
**speech_stats(segments),
|
||
"diarized": diarize,
|
||
"separation_quality": quality,
|
||
"speakers_reliable": diarize and quality >= RELIABLE_SEPARATION,
|
||
"silence_sec": round(max(0.0, duration - sum(s.end - s.start for s in segments)), 1),
|
||
"turns_count": len(turns_out),
|
||
"recovered_turns": len(recovered),
|
||
"recovered_sec": round(sum(t["end"] - t["start"] for t in recovered), 1),
|
||
},
|
||
"timing": {
|
||
"diarization_sec": round(t_diar, 1),
|
||
"asr_sec": round(t_asr, 1),
|
||
"recovery_sec": round(t_recovery, 1),
|
||
"realtime_factor": round(duration / max(t_diar + t_asr, 0.001), 1),
|
||
},
|
||
}
|