diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 6bd02eb87..d234e2f04 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -248,6 +248,44 @@ class ElevenLabsHttpTTSSettings(TTSSettings): ) +def _select_alignment( + msg: Mapping[str, Any], + *, + normalized_key: str, + alignment_key: str, + prefer_normalized: bool, +) -> Mapping[str, Any] | None: + """Pick the alignment field to use from a TTS message, with fallback. + + ElevenLabs returns two alignment fields per chunk: + + - ``normalized_key`` (``normalizedAlignment`` for WebSocket, + ``normalized_alignment`` for HTTP): the post-normalized form of what was + spoken - pronunciation-dictionary substitutions, text normalization, or + romanization of non-Latin scripts (e.g., Chinese rendered as pinyin). + - ``alignment_key`` (``alignment``): the original input characters. + + Prefer ``normalized`` only when a pronunciation dictionary is configured - + that's the case where ``alignment`` has overlapping restarts that produce + duplicated/garbled words (issue #4316). Otherwise prefer ``alignment`` so + the LLM context preserves the original input rather than the normalized + form. Fall back to the other field if the preferred one is missing or + null - the API schema marks both as nullable. + + Args: + msg: TTS response message from ElevenLabs. + normalized_key: Key for the normalized-alignment field on this transport. + alignment_key: Key for the original-alignment field on this transport. + prefer_normalized: True iff the caller is using pronunciation dictionaries. + + Returns: + The chosen alignment dict, or ``None`` if both fields are absent/null. + """ + if prefer_normalized: + return msg.get(normalized_key) or msg.get(alignment_key) + return msg.get(alignment_key) or msg.get(normalized_key) + + def _strip_utterance_leading_spaces( alignment: Mapping[str, Any], keys: tuple[str, str, str], should_strip: bool ) -> Mapping[str, Any]: @@ -829,13 +867,15 @@ class ElevenLabsTTSService(WebsocketTTSService): frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=received_ctx_id) await self.append_to_audio_context(received_ctx_id, frame) - if msg.get("normalizedAlignment"): - # Use normalizedAlignment (what was actually spoken) rather than - # alignment (the input text), so word timestamps stay accurate - # when a pronunciation dictionary or text normalization rewrites - # the input. + raw_alignment = _select_alignment( + msg, + normalized_key="normalizedAlignment", + alignment_key="alignment", + prefer_normalized=bool(self._pronunciation_dictionary_locators), + ) + if raw_alignment: alignment = _strip_utterance_leading_spaces( - msg["normalizedAlignment"], + raw_alignment, ("chars", "charStartTimesMs", "charDurationsMs"), received_ctx_id not in self._alignment_started_context_ids, ) @@ -1353,13 +1393,15 @@ class ElevenLabsHttpTTSService(TTSService): audio, self.sample_rate, 1, context_id=context_id ) - # Process alignment if present. Use normalized_alignment - # (what was actually spoken) so word timestamps stay - # accurate when a pronunciation dictionary or text - # normalization rewrites the input. - if data and data.get("normalized_alignment"): + raw_alignment = data and _select_alignment( + data, + normalized_key="normalized_alignment", + alignment_key="alignment", + prefer_normalized=bool(self._pronunciation_dictionary_locators), + ) + if raw_alignment: alignment = _strip_utterance_leading_spaces( - data["normalized_alignment"], + raw_alignment, ( "characters", "character_start_times_seconds", diff --git a/tests/test_elevenlabs_tts.py b/tests/test_elevenlabs_tts.py index 1dafc4eff..89bc94d12 100644 --- a/tests/test_elevenlabs_tts.py +++ b/tests/test_elevenlabs_tts.py @@ -9,6 +9,7 @@ from typing import Any from pipecat.services.elevenlabs.tts import ( + _select_alignment, _strip_utterance_leading_spaces, calculate_word_times, ) @@ -90,3 +91,112 @@ def test_elevenlabs_alignment_strips_only_utterance_leading_spaces(): assert first["chars"] == list("Hello") assert subsequent["chars"] == list(" world") + + +def test_select_alignment_default_prefers_alignment(): + msg = { + "alignment": _chunk("Hello"), + "normalizedAlignment": _chunk(" Hello"), + } + selected = _select_alignment( + msg, + normalized_key="normalizedAlignment", + alignment_key="alignment", + prefer_normalized=False, + ) + assert selected is not None + assert selected["chars"] == list("Hello") + + +def test_select_alignment_dictionary_mode_prefers_normalized(): + msg = { + "alignment": _chunk("Hello"), + "normalizedAlignment": _chunk(" Hello"), + } + selected = _select_alignment( + msg, + normalized_key="normalizedAlignment", + alignment_key="alignment", + prefer_normalized=True, + ) + assert selected is not None + assert selected["chars"] == list(" Hello") + + +def test_select_alignment_falls_back_when_preferred_missing(): + msg_default = {"normalizedAlignment": _chunk(" Hello")} + selected = _select_alignment( + msg_default, + normalized_key="normalizedAlignment", + alignment_key="alignment", + prefer_normalized=False, + ) + assert selected is not None + assert selected["chars"] == list(" Hello") + + msg_dict = {"alignment": _chunk("Hello")} + selected = _select_alignment( + msg_dict, + normalized_key="normalizedAlignment", + alignment_key="alignment", + prefer_normalized=True, + ) + assert selected is not None + assert selected["chars"] == list("Hello") + + +def test_select_alignment_falls_back_when_preferred_null(): + msg = {"alignment": None, "normalizedAlignment": _chunk(" Hello")} + selected = _select_alignment( + msg, + normalized_key="normalizedAlignment", + alignment_key="alignment", + prefer_normalized=False, + ) + assert selected is not None + assert selected["chars"] == list(" Hello") + + +def test_select_alignment_returns_none_when_both_missing(): + assert ( + _select_alignment( + {}, + normalized_key="normalizedAlignment", + alignment_key="alignment", + prefer_normalized=False, + ) + is None + ) + assert ( + _select_alignment( + {"alignment": None, "normalizedAlignment": None}, + normalized_key="normalizedAlignment", + alignment_key="alignment", + prefer_normalized=True, + ) + is None + ) + + +def test_select_alignment_works_with_http_field_names(): + msg = { + "alignment": {"characters": list("Hi")}, + "normalized_alignment": {"characters": list(" Hi")}, + } + selected = _select_alignment( + msg, + normalized_key="normalized_alignment", + alignment_key="alignment", + prefer_normalized=False, + ) + assert selected is not None + assert selected["characters"] == list("Hi") + + selected = _select_alignment( + msg, + normalized_key="normalized_alignment", + alignment_key="alignment", + prefer_normalized=True, + ) + assert selected is not None + assert selected["characters"] == list(" Hi")