my_voice_assistant_v3/app/pipeline/sentence_chunker.py

import re

# Satzende: . ! ? … gefolgt von Whitespace (oder Stringende beim flush).
_SENTENCE_END = re.compile(r"[.!?…]+(?=\s)")

# Abkürzungen, nach denen NICHT getrennt werden darf (sonst zerschneidet der
# Streaming-Chunker mitten in "z. | B." und die Normalisierung greift nicht mehr).
_ABBREVS = (
    "z.b.", "z. b.", "d.h.", "d. h.", "u.a.", "u. a.", "bzw.", "ca.", "usw.",
    "etc.", "dr.", "prof.", "nr.", "str.", "evtl.", "inkl.", "ggf.", "max.",
    "min.", "vgl.", "sog.", "u.ä.", "o.ä.", "bspw.",
)
_ABBR_END = re.compile(
    r"(?:^|[\s(„\"'])(" + "|".join(re.escape(a) for a in _ABBREVS) + r")$",
    re.IGNORECASE,
)


class SentenceChunker:
    """Inkrementelle Satzsegmentierung fuer gestreamte LLM-Token.

    `feed(delta)` liefert die seit dem letzten Aufruf fertig gewordenen Saetze,
    `flush()` den verbleibenden Rest (z. B. der letzte Satz ohne abschliessendes
    Leerzeichen). Damit kann pro Satz schon TTS erzeugt werden, waehrend das LLM
    noch weiterschreibt.

    Es wird NICHT getrennt, wenn der Punkt zu einer Ordinal-/Datumszahl
    ("1. Mai") oder einer bekannten Abkuerzung ("z. B.") gehoert.
    """

    def __init__(self):
        self._buffer = ""

    def _is_real_end(self, match) -> bool:
        start, end = match.start(), match.end()
        punct = match.group()
        dot_only = set(punct) <= {".", "…"}
        if dot_only and start > 0:
            prev = self._buffer[start - 1]
            # Ziffer + Punkt ("1.") = Ordinal-/Listenmarker, kein Satzende.
            if prev.isdigit():
                return False
            # Einzelner Buchstabe + Punkt ("z. B.", Initialen "A.") -> kein Satzende.
            if prev.isalpha():
                before = self._buffer[start - 2] if start >= 2 else ""
                if before == "" or not before.isalpha():
                    return False
        # Bekannte (mehrbuchstabige) Abkuerzung vor dem Punkt -> kein Satzende.
        if _ABBR_END.search(self._buffer[:end]):
            return False
        return True

    def feed(self, text: str) -> list[str]:
        self._buffer += text
        sentences: list[str] = []
        search_start = 0
        while True:
            match = _SENTENCE_END.search(self._buffer, search_start)
            if not match:
                break
            end = match.end()
            if not self._is_real_end(match):
                search_start = end  # diese Stelle nicht trennen, weitersuchen
                continue
            sentence = self._buffer[:end].strip()
            self._buffer = self._buffer[end:]
            search_start = 0
            if sentence:
                sentences.append(sentence)
        return sentences

    def flush(self) -> str:
        rest = self._buffer.strip()
        self._buffer = ""
        return rest