fix(elevenlabs): use alignment by default, normalizedAlignment only with pronunciation dicts

PR #4344 unconditionally switched to normalizedAlignment to fix garbled
words with pronunciation dictionaries (#4316). But normalizedAlignment
returns the post-normalized form of what was spoken - including
romanization of non-Latin scripts (Chinese rendered as pinyin), which
ends up in the LLM context and degrades subsequent turns.

Gate the switch on pronunciation_dictionary_locators being configured.
Adds a _select_alignment helper with preferred-with-fallback (both
fields are nullable per the API schema), used by both the WebSocket
and HTTP services. Tests cover dictionary mode, default mode, fallback
when preferred is missing or null, and HTTP field-name variants.
This commit is contained in:
Mark Backman
2026-05-05 14:32:29 -04:00
parent 735cd09c7e
commit 61a81ed87b
2 changed files with 164 additions and 12 deletions

View File

@@ -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",

View File

@@ -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")