Adolfo Reyna
65 lines
2.6 KiB
Python
65 lines
2.6 KiB
Python
"""Strip formatting that only makes sense on a screen, before it is spoken.
|
|
|
|
`VOICE_STYLE` asks Claude not to emit markdown, and mostly he doesn't — but
|
|
asking is probabilistic and hearing "asterisk asterisk aloud asterisk asterisk"
|
|
once is enough. This is the deterministic half.
|
|
|
|
Pipecat's `MarkdownTextFilter` does most of it and runs *after* sentence
|
|
aggregation, so formatting split across streaming chunks is already reassembled
|
|
by the time it sees the text. What it leaves behind, measured:
|
|
|
|
'Use ~~strike~~' -> 'Use ~~strike~~' still spoken as tildes
|
|
'- first bullet' -> '- first bullet' spoken as "dash"
|
|
'the sample_rate is' -> 'the sample_rate is' spoken as "underscore"
|
|
'Multiply 3 * 4' -> 'Multiply 3 4' the "times" is lost
|
|
|
|
so this subclass handles those four and tidies the spacing.
|
|
"""
|
|
|
|
import re
|
|
|
|
from pipecat.utils.text.markdown_text_filter import MarkdownTextFilter
|
|
|
|
# Run before the markdown filter, which would otherwise eat the asterisk.
|
|
_TIMES = re.compile(r"(?<=\d)\s*\*\s*(?=\d)")
|
|
|
|
_STRIKETHROUGH = re.compile(r"~~(.+?)~~")
|
|
# Only at the start of a line, so a hyphenated word is untouched.
|
|
_LIST_MARKER = re.compile(r"^[ \t]*[-*•]\s+", re.MULTILINE)
|
|
# Identifiers read better as words: "sample_rate" -> "sample rate".
|
|
_UNDERSCORE_WORD = re.compile(r"(?<=\w)_(?=\w)")
|
|
_EXTRA_SPACE = re.compile(r"[ \t]{2,}")
|
|
_CONTROL_TAGS = re.compile(r"\[(COMPLETE|NEEDS_DEEP|STATUS:[^\]]+)\]", re.IGNORECASE)
|
|
_VOICE_TAG = re.compile(r"\[Voice:\s*([a-zA-Z0-9_\-]+)\]", re.IGNORECASE)
|
|
|
|
|
|
class SpokenTextFilter(MarkdownTextFilter):
|
|
"""Markdown filtering, plus voice tag parsing and leftovers that matter when read aloud."""
|
|
|
|
def __init__(self, voice_manager=None, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self._voice_manager = voice_manager
|
|
|
|
async def filter(self, text: str) -> str:
|
|
if self._voice_manager:
|
|
self._voice_manager.sync_voice()
|
|
|
|
# Intercept and set active voice on [Voice:VoiceName] tags
|
|
match = _VOICE_TAG.search(text)
|
|
if match:
|
|
new_voice = match.group(1)
|
|
text = _VOICE_TAG.sub("", text)
|
|
if self._voice_manager:
|
|
try:
|
|
self._voice_manager.set_voice(new_voice)
|
|
except Exception:
|
|
pass
|
|
|
|
text = _TIMES.sub(" times ", text)
|
|
text = await super().filter(text)
|
|
text = _CONTROL_TAGS.sub("", text)
|
|
text = _STRIKETHROUGH.sub(r"\1", text)
|
|
text = _LIST_MARKER.sub("", text)
|
|
text = _UNDERSCORE_WORD.sub(" ", text)
|
|
return _EXTRA_SPACE.sub(" ", text)
|