my_voice_assistant_v2/app/pipeline/sentence_chunker.py
Dieter Schlüter c9d8552ec7 feat(tts): Aussprache-Normalisierung vor Piper (Ordinalia/Einheiten/Abk./Lexikon)
In-Process-Layer ausgebaut statt neuer Lib/CLI. Leitprinzip: nicht duplizieren,
was espeak-ng schon kann (Kardinal-/Dezimalzahlen bleiben unangetastet) -- nur die
belegten Luecken fuellen.

- german_numbers.py: deutsche Ordinalzahlen 1.-31. (attributiv/adverbial)
- tts_normalizer.py: Ordinalia (Datum '1. Mai'->'erster Mai', Folgen '1. 2. 3.'->
  'erstens, zweitens, ...'), Einheiten nach Zahl (kg/km/km-h/...), Abkuerzungen
  (Dr./z.B./usw.), optionales YAML-Lexikon (config/pronunciation.<lang>.yaml).
  Provider-abhaengige Stufen auto|full|light|off (TTS_NORMALIZE_LEVEL): piper=full,
  Cloud=light (laesst Zahlen/Abk. fuer das Cloud-Modell in Ruhe).
- spoken_response_adapter.py: nummerierte Listen -> Ordinalwoerter statt Loeschen.
- sentence_chunker.py: trennt nicht mehr nach Ziffer+Punkt, Einzelbuchstabe+Punkt
  ('z. B.', Initialen) oder bekannten Abkuerzungen -> behebt das Streaming-Symptom
  ('1.' wurde als eigener 'Satz' zu 'eins').
- orchestrator/dependencies: normalize_level durchgereicht (auto: piper->full).
- Tests: tests/test_tts_normalizer.py + Chunker-Faelle (85 gruen).
- Doku: BEDIENUNGSANLEITUNG (Aussprache verbessern), .env.example.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-17 23:01:41 +02:00

75 lines
2.7 KiB
Python

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