From e2db164f0b282f14f9ddb7e1f384bb82ec692061 Mon Sep 17 00:00:00 2001 From: Vladimir Bryzgalov Date: Sat, 15 Aug 2026 23:21:56 +0500 Subject: [PATCH] =?UTF-8?q?=D0=9F=D1=80=D0=B8=D0=B7=D0=BD=D0=B0=D0=BA?= =?UTF-8?q?=D0=B8=20=D0=B4=D0=BB=D1=8F=20=D1=80=D0=B0=D1=81=D1=81=D1=82?= =?UTF-8?q?=D0=B0=D0=BD=D0=BE=D0=B2=D0=BA=D0=B8=20=D1=80=D0=BE=D0=BB=D0=B5?= =?UTF-8?q?=D0=B9=20=D0=B8=20=D0=BE=D1=86=D0=B5=D0=BD=D0=BA=D0=B0=20=D0=BD?= =?UTF-8?q?=D0=B0=D0=B4=D1=91=D0=B6=D0=BD=D0=BE=D1=81=D1=82=D0=B8=20=D1=80?= =?UTF-8?q?=D0=B0=D0=B7=D0=B4=D0=B5=D0=BB=D0=B5=D0=BD=D0=B8=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit На записях с микрофоном на столе голоса участников для модели почти неразличимы: перебор четырёх моделей отпечатков и смена алгоритма кластеризации баланс улучшают, но роли всё равно скачут. Поэтому сервис теперь отдаёт то, на что можно опереться: акустику каждой реплики (громкость, доля высоких, центроид - они связаны с расстоянием до микрофона) и метрику separation_quality с флагом speakers_reliable. Ниже 0.35 разметка по говорящим случайна, и роли должна определять LLM по смыслу реплик. Co-Authored-By: Claude Opus 5 (1M context) --- app/acoustics.py | 55 +++++++++++++++++++++++++++++++++++++++ app/pipeline.py | 34 ++++++++++++++++++++++++ app/version.py | 2 +- tests/test_concurrency.py | 50 +++++++++++++++++++++++++++++++++++ 4 files changed, 140 insertions(+), 1 deletion(-) create mode 100644 app/acoustics.py diff --git a/app/acoustics.py b/app/acoustics.py new file mode 100644 index 0000000..88b9498 --- /dev/null +++ b/app/acoustics.py @@ -0,0 +1,55 @@ +"""Акустические признаки реплики. + +На записи с одним микрофоном на столе участники сидят на разном расстоянии, +и это слышно: дальний голос глуше и тише. Тембровые модели такую разницу +почти не улавливают, поэтому признаки считаются отдельно и отдаются наружу - +по ним LLM может уточнить, кто где, когда разделение по голосу ненадёжно. +""" +import numpy as np + +__all__ = ["segment_acoustics", "separation_quality"] + +SAMPLE_RATE = 16000 + + +def segment_acoustics(samples: np.ndarray) -> dict: + """Считает признаки, связанные с расстоянием до микрофона.""" + if len(samples) < SAMPLE_RATE // 4: + return {} + + rms = float(np.sqrt((samples ** 2).mean()) + 1e-9) + spectrum = np.abs(np.fft.rfft(samples * np.hanning(len(samples)))) + freqs = np.fft.rfftfreq(len(samples), 1 / SAMPLE_RATE) + total = float(spectrum.sum()) + 1e-9 + + return { + # Громкость: ближний к микрофону обычно громче + "loudness_db": round(20 * float(np.log10(rms)), 1), + # Доля высоких частот: у дальнего голоса верх съедается расстоянием + "hf_ratio": round(float(spectrum[(freqs > 3000) & (freqs < 7000)].sum()) / total, 3), + # Центр тяжести спектра, тот же смысл другим числом + "centroid_hz": round(float((spectrum * freqs).sum()) / total), + } + + +def separation_quality(embeddings: np.ndarray, labels: np.ndarray) -> float: + """Насколько уверенно голоса разошлись по кластерам. + + Единица - участники различаются чётко, около нуля - модель фактически + не видит разницы и разметка по говорящим случайна. На записях с одним + микрофоном в комнате значение обычно низкое. + """ + if len(embeddings) < 4 or len(set(labels.tolist())) < 2: + return 0.0 + + norm = embeddings / (np.linalg.norm(embeddings, axis=1, keepdims=True) + 1e-9) + distances = 1 - norm @ norm.T + scores = [] + for i in range(len(norm)): + own = distances[i][(labels == labels[i])] + other = distances[i][(labels != labels[i])] + own = own[own > 0] + if len(own) and len(other): + a, b = float(own.mean()), float(other.mean()) + scores.append((b - a) / max(a, b)) + return round(float(np.mean(scores)), 3) if scores else 0.0 diff --git a/app/pipeline.py b/app/pipeline.py index 60b21c4..49e63dc 100644 --- a/app/pipeline.py +++ b/app/pipeline.py @@ -12,6 +12,7 @@ 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 from app.turns import Segment, chunk_ranges, merge_turns, speech_stats @@ -20,6 +21,9 @@ log = logging.getLogger(__name__) SAMPLE_RATE = 16000 # GigaAM падает на кусках длиннее ~200 с, держим запас. MAX_CHUNK_SEC = 150.0 +# Ниже этого значения голоса практически неразличимы и разметка по говорящим +# случайна. На записях с одним микрофоном в комнате так бывает часто. +RELIABLE_SEPARATION = 0.35 ASR_MODEL_NAME = "gigaam-v3-e2e-rnnt" ASR_MODEL_REL = Path("gigaam") SEG_MODEL_REL = Path("sherpa-onnx-pyannote-segmentation-3-0") / "model.onnx" @@ -74,6 +78,7 @@ class Pipeline: self._ffmpeg: str | None = None self._asr = None self._diar_cache: dict[int, object] = {} + self._embedder = None self._replacements: dict[str, str] = {} @property @@ -148,6 +153,27 @@ class Pipeline: 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 = [] + for seg in usable: + stream = self._embedder.create_stream() + stream.accept_waveform(SAMPLE_RATE, + samples[int(seg.start * SAMPLE_RATE):int(seg.end * SAMPLE_RATE)]) + stream.input_finished() + vectors.append(np.array(self._embedder.compute(stream))) + return separation_quality(np.array(vectors), + np.array([s.speaker for s in usable])) + def transcribe(self, wav_path: Path, num_speakers: int = 2) -> dict: """Полный проход: диаризация, распознавание реплик, постобработка. @@ -164,6 +190,7 @@ class Pipeline: t0 = time.time() 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] + quality = self._separation_quality(samples, segments) t_diar = time.time() - t0 t0 = time.time() @@ -179,11 +206,16 @@ class Pipeline: 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 @@ -192,6 +224,8 @@ class Pipeline: "turns": turns_out, "stats": { **speech_stats(segments), + "separation_quality": quality, + "speakers_reliable": 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), }, diff --git a/app/version.py b/app/version.py index 6a9beea..3d26edf 100644 --- a/app/version.py +++ b/app/version.py @@ -1 +1 @@ -__version__ = "0.4.0" +__version__ = "0.4.1" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index a74326f..b37a033 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -158,3 +158,53 @@ class TestBenchmarkEndpoint: with TestClient(main.app) as c: assert c.post("/v1/benchmark", files={"file": ("a.wav", b"x")}).status_code == 401 + + +class TestSeparationQuality: + """Метрика нужна, чтобы было видно, когда разметке по голосам верить нельзя.""" + + def test_distinct_voices_score_high(self): + import numpy as np + + from app.acoustics import separation_quality + + emb = np.vstack([np.random.RandomState(0).randn(8, 64), + np.random.RandomState(1).randn(8, 64) + 6]) + labels = np.array([0] * 8 + [1] * 8) + assert separation_quality(emb, labels) > 0.4 + + def test_indistinguishable_voices_score_low(self): + import numpy as np + + from app.acoustics import separation_quality + + emb = np.random.RandomState(2).randn(16, 64) + labels = np.array([0] * 8 + [1] * 8) + assert separation_quality(emb, labels) < 0.3 + + def test_too_few_segments_returns_zero(self): + import numpy as np + + from app.acoustics import separation_quality + + assert separation_quality(np.random.randn(2, 64), np.array([0, 1])) == 0.0 + + def test_acoustics_reflect_high_frequency_content(self): + import numpy as np + + from app.acoustics import segment_acoustics + + sr = 16000 + t = np.arange(sr) / sr + low = np.sin(2 * np.pi * 300 * t).astype(np.float32) + high = np.sin(2 * np.pi * 5000 * t).astype(np.float32) + assert segment_acoustics(high)["hf_ratio"] > segment_acoustics(low)["hf_ratio"] + + def test_acoustics_reflect_loudness(self): + import numpy as np + + from app.acoustics import segment_acoustics + + loud = (np.random.RandomState(0).randn(16000) * 0.3).astype(np.float32) + quiet = loud * 0.1 + assert segment_acoustics(loud)["loudness_db"] > segment_acoustics(quiet)["loudness_db"]