Merge pull request #4354 from pipecat-ai/filipi/includes_inter_frame_spaces
feat(tts): add includes_inter_frame_spaces flag to word-timestamp API - follow-up
This commit is contained in:
2
changelog/4330.changed.md
Normal file
2
changelog/4330.changed.md
Normal file
@@ -0,0 +1,2 @@
|
||||
- Added `includes_inter_frame_spaces` parameter to `TTSService.add_word_timestamps` and `_add_word_timestamps` (default `None`). When `True`, downstream consumers will not inject additional spaces between tokens; `None` leaves each frame's own default unchanged.
|
||||
- `InworldTTSService` now passes `includes_inter_frame_spaces=True` when reporting word timestamps, since Inworld tokens already include inter-word spacing.
|
||||
@@ -103,7 +103,7 @@ piper = [ "piper-tts>=1.3.0,<2", "requests>=2.32.5,<3" ]
|
||||
qwen = []
|
||||
resembleai = [ "pipecat-ai[websockets-base]" ]
|
||||
rime = [ "pipecat-ai[websockets-base]" ]
|
||||
runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<1", "pipecat-ai-small-webrtc-prebuilt>=2.4.0"]
|
||||
runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<1", "pipecat-ai-small-webrtc-prebuilt>=2.5.0"]
|
||||
sagemaker = ["aws_sdk_sagemaker_runtime_http2; python_version>='3.12'"]
|
||||
sambanova = []
|
||||
sarvam = [ "sarvamai==0.1.28", "pipecat-ai[websockets-base]" ]
|
||||
|
||||
@@ -410,7 +410,9 @@ class InworldHttpTTSService(TTSService):
|
||||
word_times, chunk_end_time = self._calculate_word_times(timestamp_info)
|
||||
if word_times:
|
||||
self._current_run_had_timestamps = True
|
||||
await self.add_word_timestamps(word_times, context_id)
|
||||
await self.add_word_timestamps(
|
||||
word_times, context_id, includes_inter_frame_spaces=True
|
||||
)
|
||||
# Track the maximum end time across all chunks
|
||||
utterance_duration = max(utterance_duration, chunk_end_time)
|
||||
|
||||
@@ -447,7 +449,9 @@ class InworldHttpTTSService(TTSService):
|
||||
word_times, chunk_end_time = self._calculate_word_times(timestamp_info)
|
||||
if word_times:
|
||||
self._current_run_had_timestamps = True
|
||||
await self.add_word_timestamps(word_times, context_id)
|
||||
await self.add_word_timestamps(
|
||||
word_times, context_id, includes_inter_frame_spaces=True
|
||||
)
|
||||
utterance_duration = chunk_end_time
|
||||
|
||||
audio_data = base64.b64decode(response_data["audioContent"])
|
||||
@@ -1013,7 +1017,9 @@ class InworldTTSService(WebsocketTTSService):
|
||||
if word_times:
|
||||
if ctx_id:
|
||||
self._contexts_with_timestamps.add(ctx_id)
|
||||
await self.add_word_timestamps(word_times, ctx_id)
|
||||
await self.add_word_timestamps(
|
||||
word_times, ctx_id, includes_inter_frame_spaces=True
|
||||
)
|
||||
|
||||
# Handle flush completion, which indicates the end of a generation
|
||||
if "flushCompleted" in result:
|
||||
|
||||
@@ -96,6 +96,7 @@ class _WordTimestampEntry:
|
||||
word: str
|
||||
timestamp: float
|
||||
context_id: str
|
||||
includes_inter_frame_spaces: bool | None = None
|
||||
|
||||
|
||||
class TTSService(AIService):
|
||||
@@ -1070,8 +1071,10 @@ class TTSService(AIService):
|
||||
if self._initial_word_times:
|
||||
cached = self._initial_word_times.copy()
|
||||
self._initial_word_times = []
|
||||
for word, timestamp_seconds, ctx_id in cached:
|
||||
await self._add_word_timestamps([(word, timestamp_seconds)], ctx_id)
|
||||
for word, timestamp_seconds, ctx_id, ifs in cached:
|
||||
await self._add_word_timestamps(
|
||||
[(word, timestamp_seconds)], ctx_id, includes_inter_frame_spaces=ifs
|
||||
)
|
||||
|
||||
async def reset_word_timestamps(self):
|
||||
"""Reset word timestamp tracking."""
|
||||
@@ -1081,7 +1084,10 @@ class TTSService(AIService):
|
||||
self._initial_word_times = []
|
||||
|
||||
async def add_word_timestamps(
|
||||
self, word_times: list[tuple[str, float]], context_id: str | None = None
|
||||
self,
|
||||
word_times: list[tuple[str, float]],
|
||||
context_id: str | None = None,
|
||||
includes_inter_frame_spaces: bool | None = None,
|
||||
):
|
||||
"""Add word timestamps for processing.
|
||||
|
||||
@@ -1093,6 +1099,10 @@ class TTSService(AIService):
|
||||
Args:
|
||||
word_times: List of (word, timestamp) tuples where timestamp is in seconds.
|
||||
context_id: Unique identifier for the TTS context.
|
||||
includes_inter_frame_spaces: When True, the tokens already embed inter-word
|
||||
spacing (spaces and punctuation are part of the token text). Downstream
|
||||
consumers must not inject additional spaces between tokens. None leaves
|
||||
the frame's own default unchanged.
|
||||
"""
|
||||
if context_id and self.audio_context_available(context_id):
|
||||
for word, timestamp in word_times:
|
||||
@@ -1102,13 +1112,21 @@ class TTSService(AIService):
|
||||
word=word,
|
||||
timestamp=timestamp,
|
||||
context_id=context_id,
|
||||
includes_inter_frame_spaces=includes_inter_frame_spaces,
|
||||
),
|
||||
)
|
||||
else:
|
||||
await self._add_word_timestamps(word_times=word_times, context_id=context_id)
|
||||
await self._add_word_timestamps(
|
||||
word_times=word_times,
|
||||
context_id=context_id,
|
||||
includes_inter_frame_spaces=includes_inter_frame_spaces,
|
||||
)
|
||||
|
||||
async def _add_word_timestamps(
|
||||
self, word_times: list[tuple[str, float]], context_id: str | None = None
|
||||
self,
|
||||
word_times: list[tuple[str, float]],
|
||||
context_id: str | None = None,
|
||||
includes_inter_frame_spaces: bool | None = None,
|
||||
):
|
||||
"""Process word timestamps directly, building and pushing TTSTextFrames inline.
|
||||
|
||||
@@ -1124,11 +1142,13 @@ class TTSService(AIService):
|
||||
ts_ns = seconds_to_nanoseconds(timestamp)
|
||||
if self._initial_word_timestamp == -1:
|
||||
# Cache until we have audio and can compute PTS.
|
||||
self._initial_word_times.append((word, timestamp, context_id))
|
||||
self._initial_word_times.append(
|
||||
(word, timestamp, context_id, includes_inter_frame_spaces)
|
||||
)
|
||||
else:
|
||||
# Assumption: word-by-word text frames don't include spaces, so
|
||||
# we can rely on the default includes_inter_frame_spaces=False
|
||||
frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD)
|
||||
if includes_inter_frame_spaces is not None:
|
||||
frame.includes_inter_frame_spaces = includes_inter_frame_spaces
|
||||
frame.pts = self._initial_word_timestamp + ts_ns
|
||||
frame.context_id = context_id
|
||||
if context_id in self._tts_contexts:
|
||||
@@ -1331,7 +1351,9 @@ class TTSService(AIService):
|
||||
# Route word timestamps through _add_word_timestamps so they are
|
||||
# processed in playback order alongside audio frames.
|
||||
await self._add_word_timestamps(
|
||||
[(frame.word, frame.timestamp)], frame.context_id
|
||||
[(frame.word, frame.timestamp)],
|
||||
frame.context_id,
|
||||
includes_inter_frame_spaces=frame.includes_inter_frame_spaces,
|
||||
)
|
||||
continue
|
||||
elif isinstance(frame, TTSAudioRawFrame):
|
||||
|
||||
@@ -208,6 +208,103 @@ class MockWebSocketPauseTTSService(TTSService):
|
||||
yield
|
||||
|
||||
|
||||
class _MockWordTimestampHttpTTSService(TTSService):
|
||||
"""HTTP-style TTS: yields audio synchronously, calls add_word_timestamps first.
|
||||
|
||||
``word_times`` pins the exact tokens and their timestamps. When omitted the
|
||||
service splits the input text on spaces, assigning 0.1 s gaps.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
includes_inter_frame_spaces: bool = False,
|
||||
word_times: list[tuple[str, float]] | None = None,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(
|
||||
push_start_frame=True,
|
||||
push_stop_frames=True,
|
||||
push_text_frames=False,
|
||||
sample_rate=_SAMPLE_RATE,
|
||||
**kwargs,
|
||||
)
|
||||
self._includes_inter_frame_spaces = includes_inter_frame_spaces
|
||||
self._word_times = word_times
|
||||
|
||||
def can_generate_metrics(self) -> bool:
|
||||
return False
|
||||
|
||||
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
|
||||
word_times = self._word_times or [(w, i * 0.1) for i, w in enumerate(text.split())]
|
||||
await self.add_word_timestamps(
|
||||
word_times,
|
||||
context_id=context_id,
|
||||
includes_inter_frame_spaces=self._includes_inter_frame_spaces,
|
||||
)
|
||||
yield TTSAudioRawFrame(
|
||||
audio=_FAKE_AUDIO,
|
||||
sample_rate=_SAMPLE_RATE,
|
||||
num_channels=1,
|
||||
context_id=context_id,
|
||||
)
|
||||
|
||||
|
||||
class _MockWordTimestampWSTTSService(TTSService):
|
||||
"""WebSocket-style TTS: delivers audio asynchronously via the audio context.
|
||||
|
||||
Word timestamps are enqueued as ``_WordTimestampEntry`` items (audio context
|
||||
already exists at call time) and processed by ``_handle_audio_context`` in
|
||||
playback order.
|
||||
|
||||
``word_times`` pins the exact tokens and their timestamps. When omitted the
|
||||
service splits the input text on spaces, assigning 0.1 s gaps.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
includes_inter_frame_spaces: bool = False,
|
||||
word_times: list[tuple[str, float]] | None = None,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(
|
||||
push_start_frame=True,
|
||||
push_text_frames=False,
|
||||
pause_frame_processing=False,
|
||||
sample_rate=_SAMPLE_RATE,
|
||||
**kwargs,
|
||||
)
|
||||
self._includes_inter_frame_spaces = includes_inter_frame_spaces
|
||||
self._word_times = word_times
|
||||
|
||||
def can_generate_metrics(self) -> bool:
|
||||
return False
|
||||
|
||||
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
|
||||
async def _deliver():
|
||||
await asyncio.sleep(0.01)
|
||||
word_times = self._word_times or [(w, i * 0.1) for i, w in enumerate(text.split())]
|
||||
await self.add_word_timestamps(
|
||||
word_times,
|
||||
context_id=context_id,
|
||||
includes_inter_frame_spaces=self._includes_inter_frame_spaces,
|
||||
)
|
||||
await self.append_to_audio_context(
|
||||
context_id,
|
||||
TTSAudioRawFrame(
|
||||
audio=_FAKE_AUDIO,
|
||||
sample_rate=_SAMPLE_RATE,
|
||||
num_channels=1,
|
||||
context_id=context_id,
|
||||
),
|
||||
)
|
||||
await self.append_to_audio_context(context_id, TTSStoppedFrame(context_id=context_id))
|
||||
await self.remove_audio_context(context_id)
|
||||
|
||||
self.create_task(_deliver(), name=f"mock_ws_word_deliver_{context_id}")
|
||||
if False:
|
||||
yield
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Assertion helper
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -406,5 +503,159 @@ async def test_http_push_text_llm_response_end_after_tts_text():
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_http_word_timestamps_verbatim_tokens():
|
||||
"""HTTP path: text, PTS order, flag, and text-before-audio are all verified.
|
||||
|
||||
Word timestamps arrive in the audio context queue before the audio frame.
|
||||
_handle_audio_context caches them, then flushes when the first audio frame
|
||||
arrives (start_word_timestamps), so TTSTextFrames must be emitted before
|
||||
the TTSAudioRawFrame in the downstream sequence.
|
||||
"""
|
||||
word_times = [("hello", 0.0), ("world", 0.2)]
|
||||
tts = _MockWordTimestampHttpTTSService(
|
||||
includes_inter_frame_spaces=True,
|
||||
word_times=word_times,
|
||||
)
|
||||
frames_received = await run_test(
|
||||
tts,
|
||||
frames_to_send=[TTSSpeakFrame(text="hello world", append_to_context=False)],
|
||||
)
|
||||
down = frames_received[0]
|
||||
tts_text_frames = [f for f in down if isinstance(f, TTSTextFrame)]
|
||||
audio_frames = [f for f in down if isinstance(f, TTSAudioRawFrame)]
|
||||
|
||||
assert [f.text for f in tts_text_frames] == ["hello", "world"]
|
||||
assert all(f.includes_inter_frame_spaces is True for f in tts_text_frames)
|
||||
|
||||
pts_values = [f.pts for f in tts_text_frames]
|
||||
assert pts_values == sorted(pts_values) and len(set(pts_values)) == len(pts_values), (
|
||||
f"PTS values must be strictly increasing, got {pts_values}"
|
||||
)
|
||||
|
||||
# TTSTextFrames must precede the audio frame (they are flushed from cache
|
||||
# at the moment the first audio chunk sets the timestamp baseline).
|
||||
last_text_idx = max(down.index(f) for f in tts_text_frames)
|
||||
first_audio_idx = down.index(audio_frames[0])
|
||||
assert last_text_idx < first_audio_idx, (
|
||||
"TTSTextFrames must appear before TTSAudioRawFrame in the downstream sequence"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_http_word_timestamps_punctuation_tokens():
|
||||
"""Verbatim punctuation tokens are preserved with flag=True; default flag is False.
|
||||
|
||||
Models the Inworld API scenario: the TTS returns tokens exactly as sent.
|
||||
Space placement rule:
|
||||
- word-follows-word: space is the leading char of the next word (e.g. " world")
|
||||
- word-follows-punctuation: space is the trailing char of the punctuation token
|
||||
(e.g. "! "), so the following word token carries no leading space.
|
||||
The flag must reach every frame and the text must not be modified.
|
||||
Also acts as a regression guard that flag=False is the default.
|
||||
"""
|
||||
verbatim_tokens = [
|
||||
("hello", 0.0),
|
||||
(" world", 0.15),
|
||||
("! ", 0.3),
|
||||
("How", 0.45),
|
||||
(" are", 0.6),
|
||||
(" you", 0.75),
|
||||
("?", 0.9),
|
||||
]
|
||||
expected_texts = ["hello", " world", "! ", "How", " are", " you", "?"]
|
||||
|
||||
# With flag=True: all tokens verbatim, all frames carry the flag.
|
||||
tts_ifs = _MockWordTimestampHttpTTSService(
|
||||
includes_inter_frame_spaces=True,
|
||||
word_times=verbatim_tokens,
|
||||
)
|
||||
frames_ifs = await run_test(
|
||||
tts_ifs,
|
||||
frames_to_send=[TTSSpeakFrame(text="hello world! How are you?", append_to_context=False)],
|
||||
)
|
||||
text_frames_ifs = [f for f in frames_ifs[0] if isinstance(f, TTSTextFrame)]
|
||||
assert [f.text for f in text_frames_ifs] == expected_texts, (
|
||||
"Verbatim tokens must not be modified"
|
||||
)
|
||||
assert all(f.includes_inter_frame_spaces is True for f in text_frames_ifs)
|
||||
|
||||
# With flag=False (default): same tokens, flag must be False on every frame.
|
||||
tts_plain = _MockWordTimestampHttpTTSService(
|
||||
word_times=verbatim_tokens,
|
||||
)
|
||||
frames_plain = await run_test(
|
||||
tts_plain,
|
||||
frames_to_send=[TTSSpeakFrame(text="hello world! How are you?", append_to_context=False)],
|
||||
)
|
||||
text_frames_plain = [f for f in frames_plain[0] if isinstance(f, TTSTextFrame)]
|
||||
assert [f.text for f in text_frames_plain] == expected_texts
|
||||
assert all(f.includes_inter_frame_spaces is False for f in text_frames_plain)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_websocket_word_timestamps_verbatim_tokens():
|
||||
"""WebSocket path: _WordTimestampEntry carries verbatim text, PTS, and flag.
|
||||
|
||||
Unlike the HTTP path the word timestamps are sent asynchronously from a
|
||||
background task. They arrive before the audio frame and are cached until
|
||||
start_word_timestamps() fires, so the same text-before-audio ordering
|
||||
property must hold.
|
||||
"""
|
||||
word_times = [("hello", 0.0), ("world", 0.2)]
|
||||
tts = _MockWordTimestampWSTTSService(
|
||||
includes_inter_frame_spaces=True,
|
||||
word_times=word_times,
|
||||
)
|
||||
frames_received = await run_test(
|
||||
tts,
|
||||
frames_to_send=[TTSSpeakFrame(text="hello world", append_to_context=False)],
|
||||
)
|
||||
down = frames_received[0]
|
||||
tts_text_frames = [f for f in down if isinstance(f, TTSTextFrame)]
|
||||
audio_frames = [f for f in down if isinstance(f, TTSAudioRawFrame)]
|
||||
|
||||
assert [f.text for f in tts_text_frames] == ["hello", "world"]
|
||||
assert all(f.includes_inter_frame_spaces is True for f in tts_text_frames)
|
||||
|
||||
pts_values = [f.pts for f in tts_text_frames]
|
||||
assert pts_values == sorted(pts_values) and len(set(pts_values)) == len(pts_values), (
|
||||
f"PTS values must be strictly increasing, got {pts_values}"
|
||||
)
|
||||
|
||||
last_text_idx = max(down.index(f) for f in tts_text_frames)
|
||||
first_audio_idx = down.index(audio_frames[0])
|
||||
assert last_text_idx < first_audio_idx, (
|
||||
"TTSTextFrames must appear before TTSAudioRawFrame in the downstream sequence"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_websocket_word_timestamps_punctuation_tokens():
|
||||
"""WebSocket path: verbatim punctuation tokens reach TTSTextFrame unchanged."""
|
||||
verbatim_tokens = [
|
||||
("hello", 0.0),
|
||||
(" world", 0.15),
|
||||
("! ", 0.3),
|
||||
("How", 0.45),
|
||||
(" are", 0.6),
|
||||
(" you", 0.75),
|
||||
("?", 0.9),
|
||||
]
|
||||
tts = _MockWordTimestampWSTTSService(
|
||||
includes_inter_frame_spaces=True,
|
||||
word_times=verbatim_tokens,
|
||||
)
|
||||
frames_received = await run_test(
|
||||
tts,
|
||||
frames_to_send=[TTSSpeakFrame(text="hello world! How are you?", append_to_context=False)],
|
||||
)
|
||||
text_frames = [f for f in frames_received[0] if isinstance(f, TTSTextFrame)]
|
||||
assert [f.text for f in text_frames] == ["hello", " world", "! ", "How", " are", " you", "?"], (
|
||||
"Verbatim tokens must not be modified"
|
||||
)
|
||||
assert all(f.includes_inter_frame_spaces is True for f in text_frames)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
8
uv.lock
generated
8
uv.lock
generated
@@ -4511,7 +4511,7 @@ requires-dist = [
|
||||
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'ultravox'" },
|
||||
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'websocket'" },
|
||||
{ name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'xai'" },
|
||||
{ name = "pipecat-ai-small-webrtc-prebuilt", marker = "extra == 'runner'", specifier = ">=2.4.0" },
|
||||
{ name = "pipecat-ai-small-webrtc-prebuilt", marker = "extra == 'runner'", specifier = ">=2.5.0" },
|
||||
{ name = "piper-tts", marker = "extra == 'piper'", specifier = ">=1.3.0,<2" },
|
||||
{ name = "protobuf", specifier = ">=6.31.1,<7" },
|
||||
{ name = "pvkoala", marker = "extra == 'koala'", specifier = "~=2.0.3" },
|
||||
@@ -4573,14 +4573,14 @@ docs = [
|
||||
|
||||
[[package]]
|
||||
name = "pipecat-ai-small-webrtc-prebuilt"
|
||||
version = "2.4.0"
|
||||
version = "2.5.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "fastapi", extra = ["all"] },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/ba/02/1e6e90f084ebb1fc954f37661c4614219e4c9fec3d305c8abe5141707b0c/pipecat_ai_small_webrtc_prebuilt-2.4.0.tar.gz", hash = "sha256:c5eddca4e061afb7c5f98cf52ccb85511978a8c834447f6c6d662029e02950c4", size = 472449, upload-time = "2026-03-13T14:17:08.164Z" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/2c/4f/40bfc9fc1a13f9b1f2657e292c51ff3e3516c530ca722effdcf342d465d9/pipecat_ai_small_webrtc_prebuilt-2.5.0.tar.gz", hash = "sha256:51481506b7b5dff10eff0357ff929cba504a5198c3393697178d2be9895ad9e6", size = 474299, upload-time = "2026-04-22T18:05:16.494Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/25/77/8f6f67142a153943fff31530d51dcf7a2374c39dfa9aba6ef163bf0c622f/pipecat_ai_small_webrtc_prebuilt-2.4.0-py3-none-any.whl", hash = "sha256:9e9a3aa24231b1bf4101a6a2b42c4164a186c0c3d3e49bd51f77280eaa402d12", size = 472792, upload-time = "2026-03-13T14:17:06.556Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/34/58/1a2e10c1fb7b44e47558cb6c0954e24a60f98afe912fe55c74fdee66f080/pipecat_ai_small_webrtc_prebuilt-2.5.0-py3-none-any.whl", hash = "sha256:23b1eee95662a0072d9ee5128b8567108eda10d5a54ad71f279730afbb678bfe", size = 474308, upload-time = "2026-04-22T18:05:14.552Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
Reference in New Issue
Block a user