diff --git a/bridge/server.py b/bridge/server.py index 3ab173b..f1cc9be 100644 --- a/bridge/server.py +++ b/bridge/server.py @@ -51,8 +51,10 @@ from flask import Flask, request, jsonify, Response, stream_with_context try: # package-relative when imported as ``bridge.server`` from bridge.text_utils import split_sentences + from bridge.stt_filter import filter_speech_segments except ImportError: # script-relative when run as ``bridge/server.py`` from text_utils import split_sentences + from stt_filter import filter_speech_segments app = Flask(__name__) @@ -183,7 +185,19 @@ def transcribe(wav_bytes: bytes) -> dict: x_new = np.linspace(0.0, 1.0, num=n_out, endpoint=False) audio = np.interp(x_new, x_old, audio).astype(np.float32) segments, info = _whisper.transcribe(audio, beam_size=1) - text = "".join(seg.text for seg in segments).strip() + # Speech gate: drop non-speech / hallucinated segments so a brief loud sound + # or background noise (mic blip with no real speech) does not become a + # transcript and make the bot reply to nothing. Mirrors the desktop + # listener's policy, driven by the same config thresholds. + no_speech_threshold = getattr(_cfg, "whisper_no_speech_threshold", 0.5) + min_confidence = getattr(_cfg, "whisper_min_confidence", 0.3) + kept = filter_speech_segments( + segments, + no_speech_threshold=no_speech_threshold, + min_confidence=min_confidence, + log=lambda m: print(f"[bridge] {m}", flush=True), + ) + text = "".join(seg.text for seg in kept).strip() return {"text": text, "language": getattr(info, "language", None)} diff --git a/bridge/stt_filter.py b/bridge/stt_filter.py new file mode 100644 index 0000000..7e36ba1 --- /dev/null +++ b/bridge/stt_filter.py @@ -0,0 +1,79 @@ +"""Speech gate for the Discord STT path. + +Whisper will transcribe, and frequently *hallucinate*, on non-speech audio: +silence, background noise, or a brief loud blip (a cough, a key clack, a mic +pop) that momentarily opens the voice gate without anyone actually speaking. +Left unfiltered those produce phantom transcripts ("MBC 뉴스", "감사합니다", ...) +and the assistant ends up replying to noise. + +This mirrors the desktop listener's ``_filter_noisy_segments`` policy +(``src/jarvis/listening/listener.py``) so both entry points apply identical +rules, both driven by the same config thresholds: + + 1. Hard ``no_speech_prob`` cutoff (``whisper_no_speech_threshold``): Whisper's + own "this segment is not speech" probability. Checked first and + independently of confidence, because Whisper can be *confident* about a + hallucinated phrase on pure noise. + 2. ``avg_logprob`` confidence floor (``whisper_min_confidence``): drops + low-quality decodes that survive the no-speech check. + +A segment must pass both to count as real human speech. +""" + +from __future__ import annotations + +from typing import Callable, Optional + + +def is_non_speech(no_speech_prob: float, threshold: float) -> bool: + """True when Whisper flags a segment as non-speech (``>= threshold``).""" + return no_speech_prob >= threshold + + +def segment_confidence(seg) -> Optional[float]: + """Map a Whisper segment to a 0..1 confidence. + + Prefers ``avg_logprob`` (mapped to 0..1 the same way the desktop listener + does), falling back to ``1 - no_speech_prob`` when the log-prob is absent. + Returns ``None`` when neither signal is available so the caller keeps the + segment rather than dropping it on missing metadata. + """ + avg = getattr(seg, "avg_logprob", None) + if avg is not None: + return min(1.0, max(0.0, avg + 1.0)) + nsp = getattr(seg, "no_speech_prob", None) + if nsp is not None: + return 1.0 - nsp + return None + + +def filter_speech_segments( + segments, + *, + no_speech_threshold: float = 0.5, + min_confidence: float = 0.3, + log: Optional[Callable[[str], None]] = None, +) -> list: + """Keep only the segments that look like real human speech, in order. + + ``log(msg)``, if given, is called with a short reason for each dropped + segment (used by the bridge to surface why a noisy turn produced no reply). + """ + kept = [] + for seg in segments: + nsp = getattr(seg, "no_speech_prob", None) + if nsp is not None and is_non_speech(nsp, no_speech_threshold): + if log: + log(f"segment dropped (no_speech_prob={nsp:.2f}): {_preview(seg)}") + continue + conf = segment_confidence(seg) + if conf is not None and conf < min_confidence: + if log: + log(f"segment dropped (confidence={conf:.2f}): {_preview(seg)}") + continue + kept.append(seg) + return kept + + +def _preview(seg) -> str: + return repr(getattr(seg, "text", "").strip()[:50]) diff --git a/tests/test_bridge_stt_filter.py b/tests/test_bridge_stt_filter.py new file mode 100644 index 0000000..b99e613 --- /dev/null +++ b/tests/test_bridge_stt_filter.py @@ -0,0 +1,102 @@ +"""Unit tests for the bridge STT speech gate. + +The gate decides whether a Whisper segment is real human speech or just noise / +a brief loud blip that Whisper hallucinated text from. Only speech should reach +the reply engine, so a noisy mic that momentarily opens without anyone speaking +produces no transcript and no reply. Thresholds are config-driven, so the tests +pass explicit references rather than hardcoding the production defaults. +""" + +import pytest + +from bridge.stt_filter import ( + filter_speech_segments, + is_non_speech, + segment_confidence, +) + + +class Seg: + """Minimal stand-in for a faster-whisper segment.""" + + def __init__(self, text, no_speech_prob=0.0, avg_logprob=0.0): + self.text = text + self.no_speech_prob = no_speech_prob + self.avg_logprob = avg_logprob + + +@pytest.mark.unit +def test_real_speech_is_kept(): + seg = Seg("오늘 일정 알려줘", no_speech_prob=0.02, avg_logprob=-0.2) + assert filter_speech_segments( + [seg], no_speech_threshold=0.5, min_confidence=0.3 + ) == [seg] + + +@pytest.mark.unit +def test_noise_with_high_no_speech_prob_is_dropped(): + # A mic blip Whisper hallucinated "감사합니다" from: not speech. + seg = Seg("감사합니다", no_speech_prob=0.92, avg_logprob=0.5) + assert filter_speech_segments( + [seg], no_speech_threshold=0.5, min_confidence=0.3 + ) == [] + + +@pytest.mark.unit +def test_no_speech_cutoff_runs_before_the_confidence_check(): + # Confident hallucination: high avg_logprob but also high no_speech_prob. + # The no-speech cutoff must catch it regardless of confidence. + seg = Seg("MBC 뉴스", no_speech_prob=0.8, avg_logprob=0.9) + assert filter_speech_segments( + [seg], no_speech_threshold=0.5, min_confidence=0.3 + ) == [] + + +@pytest.mark.unit +def test_low_confidence_decode_is_dropped(): + # avg_logprob -0.8 -> confidence 0.2, below the 0.3 floor. + seg = Seg("어버버", no_speech_prob=0.1, avg_logprob=-0.8) + assert filter_speech_segments( + [seg], no_speech_threshold=0.5, min_confidence=0.3 + ) == [] + + +@pytest.mark.unit +def test_order_preserved_dropping_only_non_speech(): + a = Seg("진짜 말한 문장", no_speech_prob=0.05, avg_logprob=-0.1) + noise = Seg("음", no_speech_prob=0.95, avg_logprob=0.4) + b = Seg("두 번째 문장", no_speech_prob=0.05, avg_logprob=-0.1) + kept = filter_speech_segments( + [a, noise, b], no_speech_threshold=0.5, min_confidence=0.3 + ) + assert kept == [a, b] + + +@pytest.mark.unit +def test_segments_missing_metadata_are_kept(): + # No no_speech_prob / avg_logprob -> we can't prove it's noise, so keep it. + class Bare: + text = "메타데이터 없는 문장" + + seg = Bare() + assert filter_speech_segments( + [seg], no_speech_threshold=0.5, min_confidence=0.3 + ) == [seg] + + +@pytest.mark.unit +def test_is_non_speech_uses_an_inclusive_threshold(): + assert is_non_speech(0.5, 0.5) is True + assert is_non_speech(0.49, 0.5) is False + + +@pytest.mark.unit +def test_segment_confidence_prefers_avg_logprob(): + assert segment_confidence(Seg("x", avg_logprob=-0.2)) == pytest.approx(0.8) + # Falls back to 1 - no_speech_prob when avg_logprob is absent. + + class NoLogprob: + text = "x" + no_speech_prob = 0.25 + + assert segment_confidence(NoLogprob()) == pytest.approx(0.75)