From 8a5135110a0397bc96bf7d2b462fd1783f2bed45 Mon Sep 17 00:00:00 2001 From: Dan Lemon Date: Fri, 24 Jul 2026 00:57:26 +0200 Subject: [PATCH] fix(soniox): align synthesized SRT/VTT cues to real speech timing The previous cue grouping broke every 15 subword tokens or 5s, which produced uniform ~3s cues that split mid-word and bridged silence gaps, so subtitles did not track the actual speech. Cues are now built from whole words and break on sentence-final punctuation, speaker changes, silence gaps >= 700ms, a 84-char budget, or a 7s duration cap, with timestamps taken directly from token timings. Untimestamped translation tokens are excluded from cues so translated text is never mixed into original-language subtitles. --- litellm/llms/soniox/common_utils.py | 162 +++++++++++------- ...niox_audio_transcription_transformation.py | 98 +++++++++++ 2 files changed, 194 insertions(+), 66 deletions(-) diff --git a/litellm/llms/soniox/common_utils.py b/litellm/llms/soniox/common_utils.py index 90be94b8133..b736d2a5d72 100644 --- a/litellm/llms/soniox/common_utils.py +++ b/litellm/llms/soniox/common_utils.py @@ -113,11 +113,13 @@ def render_soniox_tokens(tokens: list[dict[str, Any]]) -> str: # SRT / VTT subtitle rendering # --------------------------------------------------------------------------- -# Maximum number of tokens to group into a single subtitle cue. -_CUE_MAX_TOKENS: Final[int] = 15 +_CUE_MAX_CHARS: Final[int] = 84 -# Maximum duration (in ms) for a single cue before forcing a break. -_CUE_MAX_DURATION_MS: Final[int] = 5000 +_CUE_MAX_DURATION_MS: Final[int] = 7000 + +_CUE_GAP_MS: Final[int] = 700 + +_SENTENCE_END_CHARS = (".", "!", "?", "。", "！", "？") def _format_timestamp_srt(ms: int) -> str: @@ -144,83 +146,111 @@ def _format_timestamp_vtt(ms: int) -> str: return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{millis:03d}" +def _merge_tokens_into_words(tokens: list[dict[str, Any]]) -> list[dict[str, Any]]: + """ + Merge Soniox subword tokens (e.g. ``"Hel"``, ``"lo"``) into whole words. + + A token starts a new word when its text begins with whitespace, when the + previous token's text ends with whitespace, or when the speaker changes. + Each word carries the first/last available timestamps of its tokens. + + Translation tokens (``translation_status == "translation"``) are excluded: + Soniox does not timestamp them, so they cannot be aligned to the audio and + would otherwise mix translated text into original-language cues. + """ + words: list[dict[str, Any]] = [] + for token in tokens: + text = token.get("text", "") + if not isinstance(text, str) or text == "": + continue + if token.get("translation_status") == "translation": + continue + is_continuation = ( + bool(words) + and not text[0].isspace() + and not words[-1]["text"][-1:].isspace() + and token.get("speaker") == words[-1]["speaker"] + ) + if is_continuation: + last = words[-1] + last["text"] += text + if last["start_ms"] is None: + last["start_ms"] = token.get("start_ms") + if token.get("end_ms") is not None: + last["end_ms"] = token.get("end_ms") + else: + words.append( + { + "text": text, + "start_ms": token.get("start_ms"), + "end_ms": token.get("end_ms"), + "speaker": token.get("speaker"), + } + ) + return words + + def _group_tokens_into_cues( tokens: list[dict[str, Any]], ) -> list[dict[str, Any]]: """ - Group Soniox tokens into subtitle cues. + Group Soniox tokens into subtitle cues aligned to the actual speech. Each cue has: - start_ms: int - end_ms: int - text: str - Grouping heuristics: - - A new cue starts when token count exceeds _CUE_MAX_TOKENS. - - A new cue starts when duration exceeds _CUE_MAX_DURATION_MS. - - A new cue starts when the speaker changes (if diarization is on). - - Tokens without timestamps are appended to the current cue. + Cues only ever break at word boundaries (Soniox tokens are subwords, so + tokens are first merged into words). A new cue starts when: + - the speaker changes (if diarization is on), + - a silence gap of at least _CUE_GAP_MS separates two words, so + subtitles never bridge pauses in speech, + - adding the next word would exceed _CUE_MAX_CHARS (~two subtitle + lines), or + - the cue would span more than _CUE_MAX_DURATION_MS. + A cue also ends after sentence-final punctuation, which keeps cue breaks + at natural seams. Cue timestamps come straight from token timestamps; + words without timestamps stay attached to the surrounding cue. """ - cues: Final[list[dict[str, Any]]] = [] - current_tokens: list[str] = [] - current_start: int | None = None - current_end: int | None = None - current_speaker: Any | None = None + words = _merge_tokens_into_words(tokens) + cues: list[dict[str, Any]] = [] + current: list[dict[str, Any]] = [] + + def _cue_start(ws: list[dict[str, Any]]) -> int | None: + return next((w["start_ms"] for w in ws if w["start_ms"] is not None), None) + + def _cue_end(ws: list[dict[str, Any]]) -> int | None: + return next((w["end_ms"] for w in reversed(ws) if w["end_ms"] is not None), _cue_start(ws)) + + def _cue_text(ws: list[dict[str, Any]]) -> str: + return "".join(w["text"] for w in ws).strip() def _flush() -> None: - if current_tokens and current_start is not None: - text: Final = "".join(current_tokens).strip() - if text: - cues.append( - { - "start_ms": current_start, - "end_ms": (current_end if current_end is not None else current_start), - "text": text, - } - ) + text = _cue_text(current) + start = _cue_start(current) + if text and start is not None: + cues.append({"start_ms": start, "end_ms": _cue_end(current), "text": text}) + current.clear() - for token in tokens: - start_ms = token.get("start_ms") - end_ms = token.get("end_ms") - text = token.get("text", "") - speaker = token.get("speaker") - - # Skip tokens with no timestamp data entirely if we have no cue started - if start_ms is None and current_start is None: - continue - - # Speaker change forces a new cue - if speaker is not None and speaker != current_speaker: + for word in words: + if current: + start_ms = word["start_ms"] + cue_start = _cue_start(current) + cue_end = _cue_end(current) + speaker_changed = word["speaker"] is not None and any( + w["speaker"] is not None and w["speaker"] != word["speaker"] for w in current + ) + gap_exceeded = start_ms is not None and cue_end is not None and (start_ms - cue_end) >= _CUE_GAP_MS + chars_exceeded = len(_cue_text(current)) + len(word["text"]) > _CUE_MAX_CHARS + duration_exceeded = ( + start_ms is not None and cue_start is not None and (start_ms - cue_start) >= _CUE_MAX_DURATION_MS + ) + if speaker_changed or gap_exceeded or chars_exceeded or duration_exceeded: + _flush() + current.append(word) + if word["text"].rstrip().endswith(_SENTENCE_END_CHARS): _flush() - current_tokens = [] - current_start = start_ms - current_end = end_ms - current_speaker = speaker - current_tokens.append(text) - continue - - # Duration or token count exceeded -> flush - should_break = False - if ( - len(current_tokens) >= _CUE_MAX_TOKENS - or current_start is not None - and start_ms is not None - and (start_ms - current_start) >= _CUE_MAX_DURATION_MS - ): - should_break = True - - if should_break: - _flush() - current_tokens = [] - current_start = start_ms - current_end = end_ms - current_tokens.append(text) - else: - if current_start is None: - current_start = start_ms - if end_ms is not None: - current_end = end_ms - current_tokens.append(text) _flush() return cues diff --git a/tests/test_litellm/llms/soniox/audio_transcription/test_soniox_audio_transcription_transformation.py b/tests/test_litellm/llms/soniox/audio_transcription/test_soniox_audio_transcription_transformation.py index 7ee816d5d9e..45626945bd8 100644 --- a/tests/test_litellm/llms/soniox/audio_transcription/test_soniox_audio_transcription_transformation.py +++ b/tests/test_litellm/llms/soniox/audio_transcription/test_soniox_audio_transcription_transformation.py @@ -369,6 +369,104 @@ class TestRenderSonioxTokensAsSrt: assert "01:01:01,000" in result +def _subword_tokens(words, start_ms=0, subword_ms=150, inter_word_gap_ms=50): + tokens = [] + t = start_ms + for word in words: + halves = [word[: len(word) // 2], word[len(word) // 2 :]] if len(word) > 3 else [word] + for i, piece in enumerate(halves): + text = (" " + piece) if i == 0 else piece + tokens.append({"text": text, "start_ms": t, "end_ms": t + subword_ms}) + t += subword_ms + t += inter_word_gap_ms + return tokens, t + + +class TestCueGroupingAlignment: + def test_should_split_cue_on_silence_gap_with_exact_timestamps(self): + from litellm.llms.soniox.common_utils import render_soniox_tokens_as_srt + + before, t = _subword_tokens(["hello", "there"]) + after, _ = _subword_tokens(["welcome", "back"], start_ms=t + 5000) + result = render_soniox_tokens_as_srt(before + after) + cues = result.strip().split("\n\n") + assert len(cues) == 2 + assert "00:00:00,000 --> 00:00:00,650" in cues[0] + assert "hello there" in cues[0] + assert "00:00:05,700 --> 00:00:06,350" in cues[1] + assert "welcome back" in cues[1] + + def test_should_not_bridge_pause_shorter_than_old_duration_cap(self): + from litellm.llms.soniox.common_utils import render_soniox_tokens_as_srt + + before, t = _subword_tokens(["first", "part"]) + after, _ = _subword_tokens(["second", "part"], start_ms=t + 3000) + result = render_soniox_tokens_as_srt(before + after) + cues = result.strip().split("\n\n") + assert len(cues) == 2 + assert "first part" in cues[0] + assert "second part" in cues[1] + + def test_should_never_split_mid_word(self): + from litellm.llms.soniox.common_utils import render_soniox_tokens_as_srt + + tokens, _ = _subword_tokens(["hello"] * 20) + result = render_soniox_tokens_as_srt(tokens) + text_lines = [ + line for line in result.split("\n") if line and "-->" not in line and not line.isdigit() + ] + assert len(text_lines) >= 2 + for line in text_lines: + assert set(line.split()) == {"hello"} + + def test_should_split_after_sentence_final_punctuation(self): + from litellm.llms.soniox.common_utils import render_soniox_tokens_as_srt + + tokens, _ = _subword_tokens(["That", "is", "done.", "Next", "topic"]) + result = render_soniox_tokens_as_srt(tokens) + cues = result.strip().split("\n\n") + assert len(cues) == 2 + assert cues[0].endswith("That is done.") + assert cues[1].endswith("Next topic") + + def test_should_split_on_char_budget_at_word_boundary(self): + from litellm.llms.soniox.common_utils import render_soniox_tokens_as_srt + + tokens, _ = _subword_tokens(["wonderful"] * 12) + result = render_soniox_tokens_as_srt(tokens) + text_lines = [ + line for line in result.split("\n") if line and "-->" not in line and not line.isdigit() + ] + assert len(text_lines) >= 2 + for line in text_lines: + assert len(line) <= 84 + assert set(line.split()) == {"wonderful"} + + def test_should_exclude_untimestamped_translation_tokens_from_cues(self): + from litellm.llms.soniox.common_utils import render_soniox_tokens_as_srt + + tokens = [ + {"text": " Good", "start_ms": 0, "end_ms": 200, "translation_status": "original", "language": "en"}, + {"text": " Guten", "translation_status": "translation", "language": "de", "source_language": "en"}, + {"text": " morning.", "start_ms": 250, "end_ms": 600, "translation_status": "original", "language": "en"}, + ] + result = render_soniox_tokens_as_srt(tokens) + assert "Good morning." in result + assert "Guten" not in result + assert "00:00:00,000 --> 00:00:00,600" in result + + def test_should_keep_untimestamped_word_in_cue(self): + from litellm.llms.soniox.common_utils import render_soniox_tokens_as_srt + + tokens = [ + {"text": " uh", "start_ms": None, "end_ms": None}, + {"text": " hello", "start_ms": 100, "end_ms": 500}, + ] + result = render_soniox_tokens_as_srt(tokens) + assert "uh hello" in result + assert "00:00:00,100 --> 00:00:00,500" in result + + class TestRenderSonioxTokensAsVtt: def test_should_render_basic_vtt_with_header(self): from litellm.llms.soniox.common_utils import render_soniox_tokens_as_vtt