text_segment_map

Segment-level mapping between original and TTS-transformed text.

pipecat.utils.context.text_segment_map.strip_markup(text: str) → str

Remove XML/SSML-like markup from a word-timestamp fragment.

Syntax-based, not tag-name based: treats anything between ‘<’ and ‘>’ as markup and preserves text outside it. An unclosed ‘<’ swallows the rest of text, matching how a raw word-timestamp token can arrive mid-tag (see _iter_clean_chars()).

For a complete text (not a possibly-truncated fragment), use strip_complete_markup() instead – swallowing the rest of the string past a lone ‘<’ is only correct for a genuinely truncated tag; in a complete text a lone ‘<’ is real content (e.g. "5 < 10" or "<3").

Used by TextSegmentMap._classify_hop()’s markup-stripped matching, where the incoming word may be a fragment of a still-open tag.

pipecat.utils.context.text_segment_map.strip_complete_markup(text: str) → str

Remove well-formed ‘<…>’ markup from a complete, static text.

Unlike strip_markup(), only strips matched ‘<…>’ pairs – a lone ‘<’ with no later ‘>’ is left in place as real content rather than swallowing the rest of text, since there is no streamed fragment here that could be mid-tag. Mirrors the tag-stripping regex in pipecat.utils.text.transforms._alnum_utils.normalize().

Used by TextSegment.is_transformed and by WordCompletionTracker to default user_facing_text to a tag-free string.

class pipecat.utils.context.text_segment_map.TextSegment(original: str, tts: str, original_start: int, original_end: int)

Bases: object

Immutable aligned chunk between original and TTS text.

Parameters:

• original – Chunk of the user-facing / LLM text.

• tts – Corresponding chunk in the TTS-transformed text.

• original_start – Byte offset in original_text where this chunk begins.

• original_end – Byte offset in original_text where this chunk ends.

original: str

tts: str

original_start: int

original_end: int

property is_transformed: bool

True when this segment cannot be tracked by proportional char advancement.

This holds when:

• alphanumeric content differs between original and TTS sides;

• a replacement changed word count / tokenization;

• the TTS side contains markup, even if the spoken alphanumeric content is the same as the original.

The markup check is syntax-based and tag-name independent. For example, <phoneme ...>Siobhan</phoneme> is transformed because the TTS segment has raw markup around the original word, so the raw segment cursor can move while the original/LLM cursors must remain held.

property tts_alnum_count: int

Number of alphanumeric characters in the spoken TTS content.

property original_alnum_count: int

Number of alphanumeric characters in the original side of this segment.

class pipecat.utils.context.text_segment_map.TextSegmentMap(tts_text: str, original_text: str, llm_text: str | None = None)

Bases: object

Maps cursor positions across three parallel texts as TTS words stream in.

The three texts describe the same utterance at different stages:

• tts_text: what was sent to the TTS service (may carry SSML markup and text transforms, e.g. "forty two dollars and fifty cents").

• original_text: the user-facing string (no markup/transforms, e.g. "$42.50").

• llm_text: the LLM-produced string, which may add delimiters (e.g. <card>$42.50</card>); defaults to original_text.

Built once by diffing tts_text against original_text into aligned TextSegment chunks. A single cursor drives everything – raw_pos, the position reached in tts_text as words are consumed. The user_facing_pos and llm_pos cursors are derived from it:

• Across an unchanged segment they advance proportionally, char for char.

• Across a transformed segment (alnum content, tokenization, or markup differs) they are held until the segment’s entire raw text is consumed, then jump to the end of its original span in one step – the transform is atomic, so there is no meaningful mid-segment original position.

Callers drive the map word-by-word: word_belongs_current_segment() asks whether a raw word-timestamp token plausibly continues the remaining TTS text, and advance_word() consumes it. Both match the token against the segment’s remaining raw text directly – literally, then (as stateless fallbacks, recomputed fresh each call) with the word’s own trailing punctuation stripped, with both sides case/accent-folded, or with markup stripped from both sides – so a token that adds punctuation or changes case/diacritics the source text doesn’t have, or is a fragment of a still- open SSML tag (e.g. an attribute-only word from a multi-attribute tag some TTS providers split across several word-timestamp events), needs no special handling from the caller. See _classify_hop() for the full ordered list of fallbacks.

Example:

# "$42.50" was expanded to "forty two dollars and fifty cents"
smap = TextSegmentMap(
    "Your balance is forty two dollars and fifty cents",
    "Your balance is $42.50",
)
for word in ["Your", "balance", "is"]:
    smap.advance_word(word)   # unchanged segment
for word in ["forty", "two", "dollars", "and", "fifty"]:
    smap.advance_word(word)   # transformed segment, cursors held
smap.advance_word("cents")    # segment completes, cursors jump
assert smap.last_completed_segment.original == "$42.50"
assert not smap.in_transformed_segment

__init__(tts_text: str, original_text: str, llm_text: str | None = None)

Initialize the segment map.

Parameters:

• tts_text – Post-transform text sent to TTS.

• original_text – User-facing pre-transform text.

• llm_text – LLM-produced text, which may have surrounding tags. Defaults to original_text when not provided.

advance_word(word: str) → None

Match word against the remaining TTS text and advance cursors.

Parameters:

word – Raw TTS word-timestamp token. May be a fragment of a tag, a spoken word, or a mix – the matching is purely textual, no tag parsing is required from callers.

word_belongs_current_segment(word: str) → bool

Return True if word plausibly continues the remaining TTS text.

A non-mutating dry run of the same matching advance_word() uses. Used to detect when a TTS provider silently dropped a word-timestamp event: if the incoming word does not match, the caller should force-complete this slot and route the word to the next.

property user_facing_pos: int

Current byte offset in the original user-facing text.

property llm_pos: int

Current byte offset in the LLM text.

property raw_pos: int

Current global byte offset into tts_text.

property last_overflow: str | None

Raw suffix of the last advance_word() call that overflowed.

None unless that call’s word ran past the end of tts_text (no segments left to carry the remainder into). Always a suffix of the word passed to that call – the consumed prefix is word[: len(word) - len(last_overflow)].

property is_complete: bool

True once every segment’s alphanumeric content has been accounted for.

Not simply “cursor past the last segment”: a frame whose remaining content is entirely punctuation/markup (zero alphanumeric chars) is already complete even if its raw text hasn’t been walked yet – with one exception. Trailing punctuation separated from the preceding word by whitespace (e.g. the ? in the French "Comment ça va ?") is emitted by the TTS as its own word-timestamp token, so the frame is not complete until that token arrives (see _pending_separated_punctuation()). Punctuation attached directly to the last word ("you?") was already consumed with it, and trailing markup (closing tags) never arrives as its own token, so neither holds completion open.

property in_transformed_segment: bool

True when the cursor is on a transformed segment that is not complete.

property last_completed_segment: TextSegment | None

The segment completed by the last advance_word() call.

reset() → None

Reset all cursor and consumption state.