Anti-hallucination: engine flags + transcript cleanup pass

src/scripto/core/cleanup.py

@@ -0,0 +1,103 @@
+"""Transcript hallucination cleanup — engine-agnostic safety net.
+
+Whisper's failure mode in silence/noise is degenerate repetition: a real
+sentence trails off into ``1, 2, 3, 3, 4, 4, ...`` and the next 30-second
+window is pure ``9, 9, 9, ...`` (decoding derails inside a window, and
+conditioning on previous text carries the loop forward). The engines get
+robustness flags too, but whatever still slips through is scrubbed here,
+between transcription and output.
+
+Rules, deliberately conservative:
+- A trailing run of tokens with almost no variety is cut off the segment.
+- A segment that is degenerate wall-to-wall (or CJK char-spam) is dropped.
+- Runs of *identical* consecutive segments are capped at two — the
+  "Thank you. / Thank you. / Thank you." silence filler.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+
+from ..engines.base import Segment
+
+logger = logging.getLogger(__name__)
+
+_TOKEN_RE = re.compile(r"[^\W_]+", re.UNICODE)
+
+# Loop detection: a window of tokens with almost no vocabulary variety.
+# Real speech runs ~10-12 distinct words per 12 tokens; a repetition loop
+# (even an "ascending" one like 1,2,3,3,4,4,…) stays at or below 5.
+MIN_LOOP_TOKENS = 10
+LOOP_WINDOW = 12
+LOOP_WINDOW_UNIQUE_MAX = 5
+# Spaceless spam (CJK "啊啊啊…"): long text drawn from a tiny alphabet.
+MIN_SPAM_CHARS = 30
+SPAM_UNIQUE_CHARS = 4
+# Identical consecutive segments: keep the first, drop from this count on.
+MAX_IDENTICAL_RUN = 2
+
+
+def trim_repetition_tail(text: str) -> str:
+    """Cut the trailing repetition loop off ``text``; '' if it is all loop.
+
+    Walks windows of ``LOOP_WINDOW`` tokens back from the end while their
+    vocabulary stays tiny; the leftmost such window marks where the loop
+    begins, and the cut lands on that token — right at the boundary between
+    real speech (fresh vocabulary) and the loop (recycled vocabulary).
+    """
+    matches = list(_TOKEN_RE.finditer(text))
+    count = len(matches)
+    if count < MIN_LOOP_TOKENS:
+        return text
+    tokens = [m.group().lower() for m in matches]
+    window = min(LOOP_WINDOW, count)
+
+    def loopy(i: int) -> bool:
+        return len(set(tokens[i:i + window])) <= LOOP_WINDOW_UNIQUE_MAX
+
+    start = count - window
+    if not loopy(start):
+        return text
+    while start > 0 and loopy(start - 1):
+        start -= 1
+    if count - start < MIN_LOOP_TOKENS:
+        return text
+    if start == 0:
+        return ""
+    return text[: matches[start].start()].rstrip(" ,.;:、，。").rstrip()
+
+
+def is_spam(text: str) -> bool:
+    """Long text drawn from a tiny character set (spaceless CJK loops)."""
+    chars = [c for c in text if not c.isspace() and _TOKEN_RE.match(c)]
+    return len(chars) >= MIN_SPAM_CHARS and len(set(chars)) <= SPAM_UNIQUE_CHARS
+
+
+def clean_segments(segments: list[Segment]) -> tuple[list[Segment], int]:
+    """(cleaned segments, number of segments dropped or trimmed)."""
+    cleaned: list[Segment] = []
+    touched = 0
+    identical_run = 0
+    for seg in segments:
+        text = seg.text.strip()
+        if not text:
+            continue
+        trimmed = "" if is_spam(text) else trim_repetition_tail(text)
+        if not trimmed:
+            touched += 1
+            continue
+        if cleaned and trimmed == cleaned[-1].text:
+            identical_run += 1
+            if identical_run >= MAX_IDENTICAL_RUN:
+                touched += 1
+                continue
+        else:
+            identical_run = 0
+        if trimmed != text:
+            touched += 1
+        cleaned.append(Segment(start=seg.start, end=seg.end, text=trimmed))
+    if touched:
+        logger.info("hallucination cleanup: %d segment(s) trimmed or dropped",
+                    touched)
+    return cleaned, touched

src/scripto/core/pipeline.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import logging
 import queue
 import threading
@@ -25,6 +26,7 @@
 from pathlib import Path
 from typing import Protocol
 
+from . import cleanup
 from ..engines.base import TranscribeEngine
 from ..engines.models import WhisperModelSpec
 from ..media import access, ffmpeg
@@ -252,6 +254,8 @@ def _transcribe_loop(
                     self._engine.load(self._s.model)
                     loaded = True
                 result = self._transcribe(wav, job, stop)
+                cleaned, _ = cleanup.clean_segments(result.segments)
+                result = dataclasses.replace(result, segments=cleaned)
                 job.language = result.language or self._s.language
                 target = out.output_path(
                     job.source, language=job.language, fmt=self._s.fmt,

src/scripto/engines/fw_engine.py

@@ -72,7 +72,15 @@ def transcribe(
         if stop_check is not None and stop_check():
             raise OperationStopped()
 
-        segment_iter, info = self._model.transcribe(str(audio_path), language=language)
+        # vad_filter skips non-speech (where hallucination loops are born);
+        # not conditioning on previous text stops a derailed 30s window from
+        # infecting the next one. Both also make silence-heavy files faster.
+        segment_iter, info = self._model.transcribe(
+            str(audio_path),
+            language=language,
+            vad_filter=True,
+            condition_on_previous_text=False,
+        )
         segments: list[Segment] = []
         total = float(getattr(info, "duration", 0.0) or 0.0)
         for seg in segment_iter:

src/scripto/engines/mlx_engine.py

@@ -64,7 +64,13 @@ def transcribe(
 
         import mlx_whisper
 
-        kwargs: dict = {"path_or_hf_repo": self._repo, "verbose": None}
+        kwargs: dict = {
+            "path_or_hf_repo": self._repo,
+            "verbose": None,
+            # A derailed 30s window must not infect the next one — this is
+            # how "9, 9, 9…" loops spread through silence-heavy recordings.
+            "condition_on_previous_text": False,
+        }
         if language:
             kwargs["language"] = language
         try:

tests/test_cleanup.py

@@ -0,0 +1,63 @@
+"""Hallucination cleanup: repetition tails, spam segments, identical runs.
+
+The nasty inputs are taken from a real lecture transcript where whisper
+derailed in silence ("…assignment 1, 2, 3, 3, 3, 4, 4, …" followed by a
+full 30-second window of "9, 9, 9, …").
+"""
+
+from scripto.core.cleanup import clean_segments, is_spam, trim_repetition_tail
+from scripto.engines.base import Segment
+
+REAL_SPEECH = (
+    "resources that you can use for your assignment. And so these days in "
+    "every subject talks about what about AI use policy. I would like you "
+    "to do programming yourself. Don't use help of AI to do assignment"
+)
+LOOP_TAIL = " 1, 2, 3, 3, 3, 4, 4, 4, 4, 4, 5, 5, 5, 5, 6, 6, 7, 7, " + \
+    ", ".join(["8"] * 40) + ", " + ", ".join(["9"] * 30)
+PURE_LOOP = ", ".join(["9"] * 111)
+
+
+def seg(text: str, start: float = 0.0) -> Segment:
+    return Segment(start=start, end=start + 30.0, text=text)
+
+
+def test_trailing_loop_is_cut_but_speech_kept():
+    cleaned = trim_repetition_tail(REAL_SPEECH + LOOP_TAIL)
+    assert cleaned.startswith("resources that you can use")
+    assert "programming yourself" in cleaned
+    assert "9" not in cleaned and "8, 8" not in cleaned
+
+
+def test_pure_loop_segment_is_dropped():
+    segments, touched = clean_segments([seg(REAL_SPEECH + LOOP_TAIL), seg(PURE_LOOP)])
+    assert len(segments) == 1
+    assert touched == 2
+    assert segments[0].text.startswith("resources")
+
+
+def test_normal_speech_is_untouched():
+    texts = [
+        "One, two, three, only a few.",           # legit short enumeration
+        "So each tutorial has 30 students approximately 30 28 to 30.",
+        "So, so my best consulting time would be just after the class.",
+    ]
+    segments, touched = clean_segments([seg(t) for t in texts])
+    assert [s.text for s in segments] == texts
+    assert touched == 0
+
+
+def test_identical_runs_are_capped_at_two():
+    segments, touched = clean_segments(
+        [seg("Thank you.", start=i * 30) for i in range(5)]
+        + [seg("Good afternoon.", start=200)]
+    )
+    assert [s.text for s in segments] == ["Thank you.", "Thank you.", "Good afternoon."]
+    assert touched == 3
+
+
+def test_cjk_spam_is_dropped():
+    assert is_spam("啊" * 40)
+    segments, touched = clean_segments([seg("啊" * 40), seg("正常的中文字幕内容")])
+    assert [s.text for s in segments] == ["正常的中文字幕内容"]
+    assert touched == 1