wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactored all 30+ TTS service implementations to support context tracking

Browse Source
This commit is contained in:
filipi87
2026-02-10 11:28:08 -03:00
parent 19cd242261
commit a47d7f98ee
27 changed files with 404 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
src/pipecat/services/asyncai/tts.py
View File

@@ -9,7 +9,6 @@
import asyncio
import base64
import json
import uuid
from typing import AsyncGenerator, Optional
import aiohttp
@@ -148,7 +147,6 @@ class AsyncAITTSService(AudioContextTTSService):
self._receive_task = None
self._keepalive_task = None
self._started = False
self._context_id = None
def can_generate_metrics(self) -> bool:
@@ -265,7 +263,6 @@ class AsyncAITTSService(AudioContextTTSService):
finally:
self._websocket = None
self._context_id = None
self._started = False
await self._call_event_handler("on_disconnected")
def _get_websocket(self):
@@ -289,8 +286,6 @@ class AsyncAITTSService(AudioContextTTSService):
direction: The direction to push the frame.
"""
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
self._started = False
async def _receive_messages(self):
async for message in self._get_websocket():
@@ -323,7 +318,7 @@ class AsyncAITTSService(AudioContextTTSService):
if msg.get("audio"):
await self.stop_ttfb_metrics()
audio = base64.b64decode(msg["audio"])
frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=received_ctx_id)
await self.append_to_audio_context(received_ctx_id, frame)
async def _keepalive_task_handler(self):
@@ -365,14 +360,14 @@ class AsyncAITTSService(AudioContextTTSService):
except Exception as e:
logger.error(f"Error closing context on interruption: {e}")
self._context_id = None
self._started = False
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Async API websocket endpoint.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -384,28 +379,21 @@ class AsyncAITTSService(AudioContextTTSService):
await self._connect()
try:
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._started = True
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
if not self._context_id:
self._context_id = str(uuid.uuid4())
if not self.audio_context_available(self._context_id):
await self.create_audio_context(self._context_id)
if not self._context_id:
self._context_id = context_id
if not self.audio_context_available(self._context_id):
await self.create_audio_context(self._context_id)
msg = self._build_msg(text=text, force=True, context_id=self._context_id)
await self._get_websocket().send(msg)
await self.start_tts_usage_metrics(text)
else:
if self._websocket and self._context_id:
msg = self._build_msg(text=text, force=True, context_id=self._context_id)
await self._get_websocket().send(msg)
msg = self._build_msg(text=text, force=True, context_id=self._context_id)
await self._get_websocket().send(msg)
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
self._started = False
yield TTSStoppedFrame(context_id=context_id)
return
yield None
except Exception as e:
@@ -510,11 +498,12 @@ class AsyncAIHttpTTSService(TTSService):
self._settings["output_format"]["sample_rate"] = self.sample_rate
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Async's HTTP streaming API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -531,7 +520,7 @@ class AsyncAIHttpTTSService(TTSService):
"output_format": self._settings["output_format"],
"language": self._settings["language"],
}
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
headers = {
"version": self._api_version,
"x-api-key": self._api_key,
@@ -560,6 +549,7 @@ class AsyncAIHttpTTSService(TTSService):
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield frame
@@ -568,4 +558,4 @@ class AsyncAIHttpTTSService(TTSService):
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

11
src/pipecat/services/aws/tts.py
View File

@@ -253,11 +253,12 @@ class AWSPollyTTSService(TTSService):
return ssml
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using AWS Polly.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -298,7 +299,7 @@ class AWSPollyTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
CHUNK_SIZE = self.chunk_size
@@ -306,16 +307,16 @@ class AWSPollyTTSService(TTSService):
chunk = audio_data[i : i + CHUNK_SIZE]
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1, context_id=context_id)
yield frame
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
except (BotoCoreError, ClientError) as error:
error_message = f"AWS Polly TTS error: {str(error)}"
yield ErrorFrame(error=error_message)
finally:
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
class PollyTTSService(AWSPollyTTSService):

38
src/pipecat/services/azure/tts.py
View File

@@ -277,7 +277,6 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
self._audio_queue = asyncio.Queue()
self._word_boundary_queue = asyncio.Queue()
self._word_processor_task = None
self._started = False
self._first_chunk = True
self._cumulative_audio_offset: float = 0.0 # Cumulative audio duration in seconds
self._current_sentence_base_offset: float = 0.0 # Base offset for current sentence
@@ -287,6 +286,9 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
)
self._last_word: Optional[str] = None # Track last word for punctuation merging
self._last_timestamp: Optional[float] = None # Track last timestamp
self._current_context_id: Optional[str] = (
None # Track current context_id for word timestamps
)
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -478,7 +480,10 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
while True:
try:
word, timestamp_seconds = await self._word_boundary_queue.get()
await self.add_word_timestamps([(word, timestamp_seconds)])
if self._current_context_id:
await self.add_word_timestamps(
[(word, timestamp_seconds)], self._current_context_id
)
self._word_boundary_queue.task_done()
except asyncio.CancelledError:
break
@@ -536,12 +541,11 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
self._reset_state()
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("Reset", 0)])
if isinstance(frame, TTSStoppedFrame) and self._current_context_id:
await self.add_word_timestamps([("Reset", 0)], self._current_context_id)
def _reset_state(self):
"""Reset TTS state between turns."""
self._started = False
self._first_chunk = True
self._cumulative_audio_offset = 0.0
self._current_sentence_base_offset = 0.0
@@ -549,6 +553,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
self._current_sentence_max_word_offset = 0.0
self._last_word = None
self._last_timestamp = None
self._current_context_id = None
async def flush_audio(self):
"""Flush any pending audio data."""
@@ -592,11 +597,12 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
break
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Azure's streaming synthesis.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing synthesized speech data.
@@ -615,11 +621,10 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
return
try:
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._started = True
self._first_chunk = True
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
self._first_chunk = True
self._current_context_id = context_id
# Capture base offset BEFORE starting synthesis to avoid race conditions
# Word boundary callbacks will use this value
@@ -647,6 +652,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
audio=chunk,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield frame
@@ -662,7 +668,7 @@ class AzureTTSService(WordTTSService, AzureBaseTTSService):
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
self._reset_state()
return
@@ -738,11 +744,12 @@ class AzureHttpTTSService(TTSService, AzureBaseTTSService):
)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Azure's HTTP synthesis API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the complete synthesized speech.
@@ -758,14 +765,15 @@ class AzureHttpTTSService(TTSService, AzureBaseTTSService):
if result.reason == ResultReason.SynthesizingAudioCompleted:
await self.start_tts_usage_metrics(text)
await self.stop_ttfb_metrics()
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
# Azure always sends a 44-byte header. Strip it off.
yield TTSAudioRawFrame(
audio=result.audio_data[44:],
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
elif result.reason == ResultReason.Canceled:
cancellation_details = result.cancellation_details
logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}")

9
src/pipecat/services/camb/tts.py
View File

@@ -259,11 +259,12 @@ class CambTTSService(TTSService):
self._sample_rate = MODEL_SAMPLE_RATES.get(self.model_name, 22050)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Camb.ai's TTS API.
Args:
text: The text to synthesize into speech (max 3000 characters).
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -292,7 +293,7 @@ class CambTTSService(TTSService):
tts_kwargs["user_instructions"] = self._settings["user_instructions"]
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
assert self._client is not None, "Camb.ai TTS service not initialized"
@@ -312,6 +313,7 @@ class CambTTSService(TTSService):
audio=aligned_audio,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
# Yield any remaining complete samples
@@ -322,9 +324,10 @@ class CambTTSService(TTSService):
audio=aligned_audio,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
except Exception as e:
yield ErrorFrame(error=f"Camb.ai TTS error: {e}")
finally:
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

30
src/pipecat/services/cartesia/tts.py
View File

@@ -8,7 +8,6 @@
import base64
import json
import uuid
import warnings
from enum import Enum
from typing import AsyncGenerator, List, Literal, Optional
@@ -554,16 +553,17 @@ class CartesiaTTSService(AudioContextWordTTSService):
msg = json.loads(message)
if not msg or not self.audio_context_available(msg["context_id"]):
continue
ctx_id = msg["context_id"]
if msg["type"] == "done":
await self.stop_ttfb_metrics()
await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)])
await self.remove_audio_context(msg["context_id"])
await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id)
await self.remove_audio_context(ctx_id)
elif msg["type"] == "timestamps":
# Process the timestamps based on language before adding them
processed_timestamps = self._process_word_timestamps_for_language(
msg["word_timestamps"]["words"], msg["word_timestamps"]["start"]
)
await self.add_word_timestamps(processed_timestamps)
await self.add_word_timestamps(processed_timestamps, ctx_id)
elif msg["type"] == "chunk":
await self.stop_ttfb_metrics()
await self.start_word_timestamps()
@@ -571,10 +571,11 @@ class CartesiaTTSService(AudioContextWordTTSService):
audio=base64.b64decode(msg["data"]),
sample_rate=self.sample_rate,
num_channels=1,
context_id=ctx_id,
)
await self.append_to_audio_context(msg["context_id"], frame)
await self.append_to_audio_context(ctx_id, frame)
elif msg["type"] == "error":
await self.push_frame(TTSStoppedFrame())
await self.push_frame(TTSStoppedFrame(context_id=ctx_id))
await self.stop_all_metrics()
await self.push_error(error_msg=f"Error: {msg}")
self._context_id = None
@@ -590,11 +591,12 @@ class CartesiaTTSService(AudioContextWordTTSService):
await self._connect_websocket()
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Cartesia's streaming API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -607,8 +609,8 @@ class CartesiaTTSService(AudioContextWordTTSService):
if not self._context_id:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._context_id = str(uuid.uuid4())
yield TTSStartedFrame(context_id=context_id)
self._context_id = context_id
await self.create_audio_context(self._context_id)
msg = self._build_msg(text=text)
@@ -618,7 +620,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return
@@ -761,11 +763,12 @@ class CartesiaHttpTTSService(TTSService):
await self._client.close()
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Cartesia's HTTP API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -808,7 +811,7 @@ class CartesiaHttpTTSService(TTSService):
if self._settings["pronunciation_dict_id"]:
payload["pronunciation_dict_id"] = self._settings["pronunciation_dict_id"]
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
session = await self._client._get_session()
@@ -834,6 +837,7 @@ class CartesiaHttpTTSService(TTSService):
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield frame
@@ -842,4 +846,4 @@ class CartesiaHttpTTSService(TTSService):
yield ErrorFrame(error=f"Unknown error occurred: {e}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

19
src/pipecat/services/deepgram/tts.py
View File

@@ -97,6 +97,7 @@ class DeepgramTTSService(WebsocketTTSService):
self.set_voice(voice)
self._receive_task = None
self._context_id: Optional[str] = None
def can_generate_metrics(self) -> bool:
"""Check if the service can generate metrics.
@@ -211,6 +212,7 @@ class DeepgramTTSService(WebsocketTTSService):
logger.error(f"{self} exception: {e}")
await self.push_error(ErrorFrame(error=f"{self} error: {e}"))
finally:
self._context_id = None
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -242,7 +244,7 @@ class DeepgramTTSService(WebsocketTTSService):
if isinstance(message, bytes):
# Binary message contains audio data
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(message, self.sample_rate, 1)
frame = TTSAudioRawFrame(message, self.sample_rate, 1, context_id=self._context_id)
await self.push_frame(frame)
elif isinstance(message, str):
# Text message contains metadata or control messages
@@ -283,11 +285,12 @@ class DeepgramTTSService(WebsocketTTSService):
logger.error(f"{self} error sending Flush message: {e}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Deepgram's WebSocket TTS API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech, plus start/stop frames.
@@ -302,7 +305,9 @@ class DeepgramTTSService(WebsocketTTSService):
await self.start_ttfb_metrics()
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
# Store context_id for use in _receive_messages
self._context_id = context_id
# Send text message to Deepgram
# Note: We don't send Flush here - that should only be sent when the
@@ -366,11 +371,12 @@ class DeepgramHttpTTSService(TTSService):
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Deepgram's TTS API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech, plus start/stop frames.
@@ -404,7 +410,7 @@ class DeepgramHttpTTSService(TTSService):
raise Exception(f"HTTP {response.status}: {error_text}")
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
CHUNK_SIZE = self.chunk_size
@@ -419,9 +425,10 @@ class DeepgramHttpTTSService(TTSService):
audio=chunk,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
except Exception as e:
yield ErrorFrame(f"Error getting audio: {str(e)}")

92
src/pipecat/services/elevenlabs/tts.py
View File

@@ -13,7 +13,6 @@ with support for streaming audio, word timestamps, and voice customization.
import asyncio
import base64
import json
import uuid
from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple, Union
import aiohttp
@@ -337,9 +336,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
self._voice_settings = self._set_voice_settings()
self._pronunciation_dictionary_locators = params.pronunciation_dictionary_locators
# Indicates if we have sent TTSStartedFrame. It will reset to False when
# there's an interruption or TTSStoppedFrame.
self._started = False
self._cumulative_time = 0
# Track partial words that span across alignment chunks
self._partial_word = ""
@@ -426,7 +422,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
self._context_id = None
self._started = False
async def start(self, frame: StartFrame):
"""Start the ElevenLabs TTS service.
@@ -473,9 +468,8 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
"""
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
self._started = False
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("Reset", 0)])
await self.add_word_timestamps([("Reset", 0)], self._context_id)
async def _connect(self):
await super()._connect()
@@ -557,7 +551,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._started = False
self._context_id = None
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -587,7 +580,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
self._context_id = None
self._started = False
self._partial_word = ""
self._partial_word_start_time = 0.0
@@ -624,7 +616,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await self.start_word_timestamps()
audio = base64.b64decode(msg["audio"])
frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
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("alignment"):
@@ -639,7 +631,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
)
if word_times:
await self.add_word_timestamps(word_times)
await self.add_word_timestamps(word_times, received_ctx_id)
# Calculate the actual end time of this audio chunk
char_start_times_ms = alignment.get("charStartTimesMs", [])
@@ -689,11 +681,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await self._websocket.send(json.dumps(msg))
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using ElevenLabs' streaming WebSocket API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -705,41 +698,37 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await self._connect()
try:
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._started = True
self._cumulative_time = 0
self._partial_word = ""
self._partial_word_start_time = 0.0
# If a context ID does not exist, create a new one and
# register it. If an ID exists, that means the Pipeline
# doesn't allow user interruptions, so continue using the
# current ID. When interruptions are allowed, user speech
# results in an interruption, which resets the context ID.
if not self._context_id:
self._context_id = str(uuid.uuid4())
if not self.audio_context_available(self._context_id):
await self.create_audio_context(self._context_id)
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
self._cumulative_time = 0
self._partial_word = ""
self._partial_word_start_time = 0.0
# If a context ID does not exist, use the provided one.
# If an ID exists, that means the Pipeline doesn't allow
# user interruptions, so continue using the current ID.
# When interruptions are allowed, user speech results in
# an interruption, which resets the context ID.
if not self._context_id:
self._context_id = context_id
if not self.audio_context_available(self._context_id):
await self.create_audio_context(self._context_id)
# Initialize context with voice settings and pronunciation dictionaries
msg = {"text": " ", "context_id": self._context_id}
if self._voice_settings:
msg["voice_settings"] = self._voice_settings
if self._pronunciation_dictionary_locators:
msg["pronunciation_dictionary_locators"] = [
locator.model_dump()
for locator in self._pronunciation_dictionary_locators
]
await self._websocket.send(json.dumps(msg))
logger.trace(f"Created new context {self._context_id}")
# Initialize context with voice settings and pronunciation dictionaries
msg = {"text": " ", "context_id": self._context_id}
if self._voice_settings:
msg["voice_settings"] = self._voice_settings
if self._pronunciation_dictionary_locators:
msg["pronunciation_dictionary_locators"] = [
locator.model_dump() for locator in self._pronunciation_dictionary_locators
]
await self._websocket.send(json.dumps(msg))
logger.trace(f"Created new context {self._context_id}")
await self._send_text(text)
await self.start_tts_usage_metrics(text)
except Exception as e:
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
yield ErrorFrame(error=f"Unknown error occurred: {e}")
self._started = False
return
yield None
except Exception as e:
@@ -840,7 +829,6 @@ class ElevenLabsHttpTTSService(WordTTSService):
# Track cumulative time to properly sequence word timestamps across utterances
self._cumulative_time = 0
self._started = False
# Store previous text for context within a turn
self._previous_text = ""
@@ -879,7 +867,6 @@ class ElevenLabsHttpTTSService(WordTTSService):
def _reset_state(self):
"""Reset internal state variables."""
self._cumulative_time = 0
self._started = False
self._previous_text = ""
self._partial_word = ""
self._partial_word_start_time = 0.0
@@ -976,7 +963,7 @@ class ElevenLabsHttpTTSService(WordTTSService):
return word_times
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using ElevenLabs streaming API with timestamps.
Makes a request to the ElevenLabs API to generate audio and timing data.
@@ -985,6 +972,7 @@ class ElevenLabsHttpTTSService(WordTTSService):
Args:
text: Text to convert to speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio and control frames containing the synthesized speech.
@@ -1048,11 +1036,9 @@ class ElevenLabsHttpTTSService(WordTTSService):
await self.start_tts_usage_metrics(text)
# Start TTS sequence if not already started
if not self._started:
await self.start_word_timestamps()
yield TTSStartedFrame()
self._started = True
# Start TTS sequence
await self.start_word_timestamps()
yield TTSStartedFrame(context_id=context_id)
# Track the duration of this utterance based on the last character's end time
utterance_duration = 0
@@ -1069,7 +1055,9 @@ class ElevenLabsHttpTTSService(WordTTSService):
if data and "audio_base64" in data:
await self.stop_ttfb_metrics()
audio = base64.b64decode(data["audio_base64"])
yield TTSAudioRawFrame(audio, self.sample_rate, 1)
yield TTSAudioRawFrame(
audio, self.sample_rate, 1, context_id=context_id
)
# Process alignment if present
if data and "alignment" in data:
@@ -1085,7 +1073,7 @@ class ElevenLabsHttpTTSService(WordTTSService):
# Calculate word timestamps
word_times = self.calculate_word_times(alignment)
if word_times:
await self.add_word_timestamps(word_times)
await self.add_word_timestamps(word_times, context_id)
except json.JSONDecodeError as e:
logger.warning(f"Failed to parse JSON from stream: {e}")
continue
@@ -1097,7 +1085,7 @@ class ElevenLabsHttpTTSService(WordTTSService):
# since this is the end of the utterance
if self._partial_word:
final_word_time = [(self._partial_word, self._partial_word_start_time)]
await self.add_word_timestamps(final_word_time)
await self.add_word_timestamps(final_word_time, context_id)
self._partial_word = ""
self._partial_word_start_time = 0.0

9
src/pipecat/services/fish/tts.py
View File

@@ -135,7 +135,6 @@ class FishAudioTTSService(InterruptibleTTSService):
self._websocket = None
self._receive_task = None
self._request_id = None
self._started = False
self._settings = {
"sample_rate": 0,
@@ -249,7 +248,6 @@ class FishAudioTTSService(InterruptibleTTSService):
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._request_id = None
self._started = False
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -291,11 +289,12 @@ class FishAudioTTSService(InterruptibleTTSService):
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Fish Audio's streaming API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames and control frames for the synthesized speech.
@@ -308,7 +307,7 @@ class FishAudioTTSService(InterruptibleTTSService):
if not self._request_id:
await self.start_ttfb_metrics()
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
self._request_id = str(uuid.uuid4())
# Send the text
@@ -325,7 +324,7 @@ class FishAudioTTSService(InterruptibleTTSService):
await self._get_websocket().send(ormsgpack.packb(flush_message))
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()

33
src/pipecat/services/google/tts.py
View File

@@ -682,11 +682,12 @@ class GoogleHttpTTSService(TTSService):
return ssml
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Google's HTTP TTS API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -731,7 +732,7 @@ class GoogleHttpTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
# Skip the first 44 bytes to remove the WAV header
audio_content = response.audio_content[44:]
@@ -743,10 +744,10 @@ class GoogleHttpTTSService(TTSService):
if not chunk:
break
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1, context_id=context_id)
yield frame
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
except Exception as e:
error_message = f"TTS generation error: {str(e)}"
@@ -828,6 +829,7 @@ class GoogleBaseTTSService(TTSService):
self,
streaming_config: texttospeech_v1.StreamingSynthesizeConfig,
text: str,
context_id: str,
prompt: Optional[str] = None,
) -> AsyncGenerator[Frame, None]:
"""Shared streaming synthesis logic.
@@ -835,6 +837,7 @@ class GoogleBaseTTSService(TTSService):
Args:
streaming_config: The streaming configuration.
text: The text to synthesize.
context_id: Unique identifier for this TTS context.
prompt: Optional prompt for style instructions (Gemini only).
Yields:
@@ -856,7 +859,7 @@ class GoogleBaseTTSService(TTSService):
streaming_responses = await self._client.streaming_synthesize(request_generator())
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
audio_buffer = b""
first_chunk_for_ttfb = False
@@ -876,12 +879,12 @@ class GoogleBaseTTSService(TTSService):
while len(audio_buffer) >= CHUNK_SIZE:
piece = audio_buffer[:CHUNK_SIZE]
audio_buffer = audio_buffer[CHUNK_SIZE:]
yield TTSAudioRawFrame(piece, self.sample_rate, 1)
yield TTSAudioRawFrame(piece, self.sample_rate, 1, context_id=context_id)
if audio_buffer:
yield TTSAudioRawFrame(audio_buffer, self.sample_rate, 1)
yield TTSAudioRawFrame(audio_buffer, self.sample_rate, 1, context_id=context_id)
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
class GoogleTTSService(GoogleBaseTTSService):
@@ -976,11 +979,12 @@ class GoogleTTSService(GoogleBaseTTSService):
await super()._update_settings(settings)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate streaming speech from text using Google's streaming API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech as it's generated.
@@ -1014,7 +1018,7 @@ class GoogleTTSService(GoogleBaseTTSService):
)
# Use base class streaming logic
async for frame in self._stream_tts(streaming_config, text):
async for frame in self._stream_tts(streaming_config, text, context_id):
yield frame
except Exception as e:
@@ -1213,11 +1217,12 @@ class GeminiTTSService(GoogleBaseTTSService):
await super()._update_settings(settings)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate streaming speech from text using Gemini TTS models.
Args:
text: The text to synthesize into speech. Can include markup tags
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames. Can include markup tags
like [sigh], [laughing], [whispering] for expressive control.
Yields:
@@ -1267,7 +1272,9 @@ class GeminiTTSService(GoogleBaseTTSService):
)
# Use base class streaming logic with prompt support
async for frame in self._stream_tts(streaming_config, text, self._settings["prompt"]):
async for frame in self._stream_tts(
streaming_config, text, context_id, self._settings["prompt"]
):
yield frame
except Exception as e:

15
src/pipecat/services/gradium/tts.py
View File

@@ -95,6 +95,7 @@ class GradiumTTSService(InterruptibleWordTTSService):
# State tracking
self._receive_task = None
self._current_context_id: Optional[str] = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -261,11 +262,15 @@ class GradiumTTSService(InterruptibleWordTTSService):
audio=base64.b64decode(msg["audio"]),
sample_rate=self.sample_rate,
num_channels=1,
context_id=self._current_context_id,
)
await self.push_frame(frame)
elif msg["type"] == "text":
await self.add_word_timestamps([(msg["text"], msg["start_s"])])
if self._current_context_id:
await self.add_word_timestamps(
[(msg["text"], msg["start_s"])], self._current_context_id
)
elif msg["type"] == "end_of_stream":
await self.push_frame(TTSStoppedFrame())
await self.stop_all_metrics()
@@ -285,11 +290,12 @@ class GradiumTTSService(InterruptibleWordTTSService):
await super().push_frame(frame, direction)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Gradium's streaming API.
Args:
text: The text to convert to speech.
context_id: Unique identifier for this TTS context.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -302,14 +308,15 @@ class GradiumTTSService(InterruptibleWordTTSService):
await self._connect()
try:
yield TTSStartedFrame()
self._current_context_id = context_id
yield TTSStartedFrame(context_id=context_id)
msg = self._build_msg(text=text)
await self._get_websocket().send(json.dumps(msg))
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return

9
src/pipecat/services/groq/tts.py
View File

@@ -112,11 +112,12 @@ class GroqTTSService(TTSService):
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Groq's TTS API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech data.
@@ -124,7 +125,7 @@ class GroqTTSService(TTSService):
logger.debug(f"{self}: Generating TTS [{text}]")
measuring_ttfb = True
await self.start_ttfb_metrics()
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
try:
response = await self._client.audio.speech.create(
@@ -144,8 +145,8 @@ class GroqTTSService(TTSService):
frame_rate = w.getframerate()
num_frames = w.getnframes()
bytes = w.readframes(num_frames)
yield TTSAudioRawFrame(bytes, frame_rate, channels)
yield TTSAudioRawFrame(bytes, frame_rate, channels, context_id=context_id)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

8
src/pipecat/services/hathora/tts.py
View File

@@ -115,11 +115,12 @@ class HathoraTTSService(TTSService):
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Run text-to-speech synthesis on the provided text.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -142,7 +143,7 @@ class HathoraTTSService(TTSService):
for option in self._settings["config"]
]
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
async with aiohttp.ClientSession() as session:
async with session.post(
@@ -161,6 +162,7 @@ class HathoraTTSService(TTSService):
audio=pcm_audio,
sample_rate=self.sample_rate,
num_channels=num_channels,
context_id=context_id,
)
yield frame
@@ -170,4 +172,4 @@ class HathoraTTSService(TTSService):
finally:
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

11
src/pipecat/services/hume/tts.py
View File

@@ -209,11 +209,12 @@ class HumeTTSService(WordTTSService):
await super().update_setting(key, value)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Hume TTS with word timestamps.
Args:
text: The text to be synthesized.
context_id: Unique identifier for this TTS context.
Returns:
An async generator that yields `Frame` objects, including
@@ -245,7 +246,7 @@ class HumeTTSService(WordTTSService):
# Start TTS sequence if not already started
if not self._started:
await self.start_word_timestamps()
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
self._started = True
try:
@@ -281,6 +282,7 @@ class HumeTTSService(WordTTSService):
audio=self._audio_bytes,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield frame
self._audio_bytes = b""
@@ -297,7 +299,9 @@ class HumeTTSService(WordTTSService):
utterance_duration = max(utterance_duration, word_end_time)
# Add word timestamp
await self.add_word_timestamps([(timestamp.text, word_start_time)])
await self.add_word_timestamps(
[(timestamp.text, word_start_time)], context_id
)
# Flush any remaining audio bytes
if self._audio_bytes:
@@ -305,6 +309,7 @@ class HumeTTSService(WordTTSService):
audio=self._audio_bytes,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield frame

79
src/pipecat/services/inworld/tts.py
View File

@@ -16,7 +16,6 @@ Inworld’s text-to-speech (TTS) models offer ultra-realistic, context-aware spe
import asyncio
import base64
import json
import uuid
from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple
import aiohttp
@@ -125,7 +124,6 @@ class InworldHttpTTSService(WordTTSService):
if params.speaking_rate is not None:
self._settings["audioConfig"]["speakingRate"] = params.speaking_rate
self._started = False
self._cumulative_time = 0.0
self.set_voice(voice_id)
@@ -173,7 +171,6 @@ class InworldHttpTTSService(WordTTSService):
"""
await super().push_frame(frame, direction)
if isinstance(frame, (InterruptionFrame, TTSStoppedFrame)):
self._started = False
self._cumulative_time = 0.0
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("Reset", 0)])
@@ -214,11 +211,12 @@ class InworldHttpTTSService(WordTTSService):
return (word_times, chunk_end_time)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate TTS audio for the given text.
Args:
text: The text to generate TTS audio for.
context_id: Unique identifier for this TTS context.
Returns:
An asynchronous generator of frames.
@@ -246,10 +244,8 @@ class InworldHttpTTSService(WordTTSService):
try:
await self.start_ttfb_metrics()
if not self._started:
await self.start_word_timestamps()
yield TTSStartedFrame()
self._started = True
await self.start_word_timestamps()
yield TTSStartedFrame(context_id=context_id)
async with self._session.post(
self._base_url, json=payload, headers=headers
@@ -261,10 +257,10 @@ class InworldHttpTTSService(WordTTSService):
return
if self._streaming:
async for frame in self._process_streaming_response(response):
async for frame in self._process_streaming_response(response, context_id):
yield frame
else:
async for frame in self._process_non_streaming_response(response):
async for frame in self._process_non_streaming_response(response, context_id):
yield frame
await self.start_tts_usage_metrics(text)
@@ -276,12 +272,13 @@ class InworldHttpTTSService(WordTTSService):
await self.stop_all_metrics()
async def _process_streaming_response(
self, response: aiohttp.ClientResponse
self, response: aiohttp.ClientResponse, context_id: str
) -> AsyncGenerator[Frame, None]:
"""Process a streaming response from the Inworld API.
Args:
response: The response from the Inworld API.
context_id: Unique identifier for this TTS context.
Yields:
An asynchronous generator of frames.
@@ -306,7 +303,7 @@ class InworldHttpTTSService(WordTTSService):
if "result" in chunk_data and "audioContent" in chunk_data["result"]:
await self.stop_ttfb_metrics()
async for frame in self._process_audio_chunk(
base64.b64decode(chunk_data["result"]["audioContent"])
base64.b64decode(chunk_data["result"]["audioContent"]), context_id
):
yield frame
@@ -314,7 +311,7 @@ class InworldHttpTTSService(WordTTSService):
timestamp_info = chunk_data["result"]["timestampInfo"]
word_times, chunk_end_time = self._calculate_word_times(timestamp_info)
if word_times:
await self.add_word_timestamps(word_times)
await self.add_word_timestamps(word_times, context_id)
# Track the maximum end time across all chunks
utterance_duration = max(utterance_duration, chunk_end_time)
@@ -327,12 +324,13 @@ class InworldHttpTTSService(WordTTSService):
self._cumulative_time += utterance_duration
async def _process_non_streaming_response(
self, response: aiohttp.ClientResponse
self, response: aiohttp.ClientResponse, context_id: str
) -> AsyncGenerator[Frame, None]:
"""Process a non-streaming response from the Inworld API.
Args:
response: The response from the Inworld API.
context_id: Unique identifier for this TTS context.
Returns:
An asynchronous generator of frames.
@@ -349,7 +347,7 @@ class InworldHttpTTSService(WordTTSService):
timestamp_info = response_data["timestampInfo"]
word_times, chunk_end_time = self._calculate_word_times(timestamp_info)
if word_times:
await self.add_word_timestamps(word_times)
await self.add_word_timestamps(word_times, context_id)
utterance_duration = chunk_end_time
audio_data = base64.b64decode(response_data["audioContent"])
@@ -363,20 +361,21 @@ class InworldHttpTTSService(WordTTSService):
if chunk:
await self.stop_ttfb_metrics()
yield TTSAudioRawFrame(
audio=chunk,
sample_rate=self.sample_rate,
num_channels=1,
audio=chunk, sample_rate=self.sample_rate, num_channels=1, context_id=context_id
)
# After processing all audio, add the utterance duration to cumulative time
if utterance_duration > 0:
self._cumulative_time += utterance_duration
async def _process_audio_chunk(self, audio_chunk: bytes) -> AsyncGenerator[Frame, None]:
async def _process_audio_chunk(
self, audio_chunk: bytes, context_id: str
) -> AsyncGenerator[Frame, None]:
"""Process an audio chunk from the Inworld API.
Args:
audio_chunk: The audio chunk to process.
context_id: Unique identifier for this TTS context.
Returns:
An asynchronous generator of frames.
@@ -394,6 +393,7 @@ class InworldHttpTTSService(WordTTSService):
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
@@ -500,7 +500,6 @@ class InworldTTSService(AudioContextWordTTSService):
self._receive_task = None
self._keepalive_task = None
self._context_id = None
self._started = False
# Track cumulative time across generations for monotonic timestamps within a turn.
# When auto_mode is enabled, the server controls generations and timestamps reset
@@ -569,7 +568,6 @@ class InworldTTSService(AudioContextWordTTSService):
"""
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
self._started = False
logger.trace(
f"{self}: Resetting timestamp tracking due to {type(frame).__name__} - "
f"cumulative_time was {self._cumulative_time}"
@@ -637,7 +635,6 @@ class InworldTTSService(AudioContextWordTTSService):
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
self._context_id = None
self._started = False
self._cumulative_time = 0.0
self._generation_end_time = 0.0
logger.trace(f"{self}: Interruption handled, context reset to None")
@@ -726,7 +723,6 @@ class InworldTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._started = False
self._context_id = None
self._websocket = None
self._cumulative_time = 0.0
@@ -801,7 +797,7 @@ class InworldTTSService(AudioContextWordTTSService):
audio = base64.b64decode(audio_b64)
if len(audio) > 44 and audio.startswith(b"RIFF"):
audio = audio[44:]
frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=ctx_id)
if ctx_id:
await self.append_to_audio_context(ctx_id, frame)
@@ -811,7 +807,7 @@ class InworldTTSService(AudioContextWordTTSService):
if timestamp_info:
word_times = self._calculate_word_times(timestamp_info)
if word_times:
await self.add_word_timestamps(word_times)
await self.add_word_timestamps(word_times, ctx_id)
# Handle context created confirmation
if "contextCreated" in result:
@@ -832,10 +828,9 @@ class InworldTTSService(AudioContextWordTTSService):
# Only reset if this is our current context
if ctx_id == self._context_id:
self._context_id = None
self._started = False
if ctx_id and self.audio_context_available(ctx_id):
await self.remove_audio_context(ctx_id)
await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)])
await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], ctx_id)
async def _keepalive_task_handler(self):
"""Send periodic keepalive messages to maintain WebSocket connection."""
@@ -917,11 +912,12 @@ class InworldTTSService(AudioContextWordTTSService):
await self.send_with_retry(json.dumps(msg), self._report_error)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate TTS audio for the given text using the Inworld WebSocket TTS service.
Args:
text: The text to generate TTS audio for.
context_id: Unique identifier for this TTS context.
Returns:
An asynchronous generator of frames.
@@ -933,28 +929,25 @@ class InworldTTSService(AudioContextWordTTSService):
await self._connect()
try:
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._started = True
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
if not self._context_id:
self._context_id = str(uuid.uuid4())
logger.trace(f"{self}: Creating new context {self._context_id}")
await self.create_audio_context(self._context_id)
await self._send_context(self._context_id)
elif not self.audio_context_available(self._context_id):
# Context exists on server but local tracking was removed
logger.trace(f"{self}: Recreating local audio context {self._context_id}")
await self.create_audio_context(self._context_id)
if not self._context_id:
self._context_id = context_id
logger.trace(f"{self}: Creating new context {self._context_id}")
await self.create_audio_context(self._context_id)
await self._send_context(self._context_id)
elif not self.audio_context_available(self._context_id):
# Context exists on server but local tracking was removed
logger.trace(f"{self}: Recreating local audio context {self._context_id}")
await self.create_audio_context(self._context_id)
await self._send_text(self._context_id, text)
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
self._started = False
yield TTSStoppedFrame(context_id=context_id)
return
yield None
except Exception as e:

12
src/pipecat/services/kokoro/tts.py
View File

@@ -143,13 +143,14 @@ class KokoroTTSService(TTSService):
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Synthesize speech from text using kokoro-onnx.
Uses the async streaming API to generate audio frames.
Args:
text: The text to synthesize.
context_id: Unique identifier for this TTS context.
"""
logger.debug(f"{self}: Generating TTS [{text}]")
@@ -157,7 +158,7 @@ class KokoroTTSService(TTSService):
try:
await self.start_ttfb_metrics()
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
stream = self._kokoro.create_stream(
text, voice=self._voice_id, lang=self._lang_code, speed=1.0
@@ -172,10 +173,13 @@ class KokoroTTSService(TTSService):
)
yield TTSAudioRawFrame(
audio=audio_data, sample_rate=self.sample_rate, num_channels=1
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

20
src/pipecat/services/lmnt/tts.py
View File

@@ -113,8 +113,8 @@ class LmntTTSService(InterruptibleTTSService):
"language": self.language_to_service_language(language),
"format": "raw", # Use raw format for direct PCM data
}
self._started = False
self._receive_task = None
self._context_id: Optional[str] = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -170,8 +170,6 @@ class LmntTTSService(InterruptibleTTSService):
direction: The direction to push the frame.
"""
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
self._started = False
async def _connect(self):
"""Connect to LMNT WebSocket and start receive task."""
@@ -236,7 +234,7 @@ class LmntTTSService(InterruptibleTTSService):
except Exception as e:
await self.push_error(error_msg=f"Error disconnecting from LMNT: {e}", exception=e)
finally:
self._started = False
self._context_id = None
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -262,6 +260,7 @@ class LmntTTSService(InterruptibleTTSService):
audio=message,
sample_rate=self.sample_rate,
num_channels=1,
context_id=self._context_id,
)
await self.push_frame(frame)
else:
@@ -276,11 +275,12 @@ class LmntTTSService(InterruptibleTTSService):
logger.error(f"Invalid JSON message: {message}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate TTS audio from text using LMNT's streaming API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -292,10 +292,10 @@ class LmntTTSService(InterruptibleTTSService):
await self._connect()
try:
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._started = True
await self.start_ttfb_metrics()
# Store context_id for use in _receive_messages
self._context_id = context_id
yield TTSStartedFrame(context_id=context_id)
# Send text to LMNT
await self._get_websocket().send(json.dumps({"text": text}))
@@ -304,7 +304,7 @@ class LmntTTSService(InterruptibleTTSService):
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return

8
src/pipecat/services/minimax/tts.py
View File

@@ -284,11 +284,12 @@ class MiniMaxHttpTTSService(TTSService):
logger.debug(f"MiniMax TTS initialized with sample_rate: {self.sample_rate}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate TTS audio from text using MiniMax's streaming API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -318,7 +319,7 @@ class MiniMaxHttpTTSService(TTSService):
return
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
# Process the streaming response
buffer = bytearray()
@@ -377,6 +378,7 @@ class MiniMaxHttpTTSService(TTSService):
audio=audio_chunk,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
except ValueError as e:
logger.error(
@@ -394,4 +396,4 @@ class MiniMaxHttpTTSService(TTSService):
yield ErrorFrame(error=f"Unknown error occurred: {e}", exception=e)
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

40
src/pipecat/services/neuphonic/tts.py
View File

@@ -135,13 +135,11 @@ class NeuphonicTTSService(InterruptibleTTSService):
}
self.set_voice(voice_id)
# Indicates if we have sent TTSStartedFrame. It will reset to False when
# there's an interruption or TTSStoppedFrame.
self._started = False
self._cumulative_time = 0
self._receive_task = None
self._keepalive_task = None
self._context_id: Optional[str] = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -213,8 +211,6 @@ class NeuphonicTTSService(InterruptibleTTSService):
direction: The direction to push the frame.
"""
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
self._started = False
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames with special handling for speech control.
@@ -230,7 +226,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
# processing more frames until we receive a BotStoppedSpeakingFrame.
if isinstance(frame, TTSSpeakFrame):
await self.pause_processing_frames()
elif isinstance(frame, LLMFullResponseEndFrame) and self._started:
elif isinstance(frame, LLMFullResponseEndFrame):
await self.pause_processing_frames()
elif isinstance(frame, BotStoppedSpeakingFrame):
await self.resume_processing_frames()
@@ -304,7 +300,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._started = False
self._context_id = None
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -317,7 +313,9 @@ class NeuphonicTTSService(InterruptibleTTSService):
await self.stop_ttfb_metrics()
audio = base64.b64decode(msg["data"]["audio"])
frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
frame = TTSAudioRawFrame(
audio, self.sample_rate, 1, context_id=self._context_id
)
await self.push_frame(frame)
async def _keepalive_task_handler(self):
@@ -342,11 +340,12 @@ class NeuphonicTTSService(InterruptibleTTSService):
await self._websocket.send(json.dumps(msg))
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Neuphonic's streaming API.
Args:
text: The text to synthesize into speech.
context_id: Unique identifier for this TTS context.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -358,17 +357,17 @@ class NeuphonicTTSService(InterruptibleTTSService):
await self._connect()
try:
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._started = True
self._cumulative_time = 0
await self.start_ttfb_metrics()
# Store context_id for use in _receive_messages
self._context_id = context_id
yield TTSStartedFrame(context_id=context_id)
self._cumulative_time = 0
await self._send_text(text)
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return
@@ -502,11 +501,12 @@ class NeuphonicHttpTTSService(TTSService):
return None
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Neuphonic streaming API.
Args:
text: The text to convert to speech.
context_id: Unique identifier for this TTS context.
Yields:
Frame: Audio frames containing the synthesized speech and status information.
@@ -542,7 +542,7 @@ class NeuphonicHttpTTSService(TTSService):
return
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
# Process SSE stream line by line
async for line in response.content:
@@ -564,7 +564,9 @@ class NeuphonicHttpTTSService(TTSService):
audio_bytes = base64.b64decode(audio_b64)
await self.stop_ttfb_metrics()
yield TTSAudioRawFrame(audio_bytes, self.sample_rate, 1)
yield TTSAudioRawFrame(
audio_bytes, self.sample_rate, 1, context_id=context_id
)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
@@ -578,4 +580,4 @@ class NeuphonicHttpTTSService(TTSService):
yield ErrorFrame(error=f"Unknown error occurred: {e}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

10
src/pipecat/services/nvidia/tts.py
View File

@@ -153,11 +153,12 @@ class NvidiaTTSService(TTSService):
logger.debug(f"Initialized NvidiaTTSService with model: {self.model_name}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using NVIDIA Riva TTS.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech data.
@@ -193,7 +194,7 @@ class NvidiaTTSService(TTSService):
assert self._config is not None, "Synthesis configuration not created"
await self.start_ttfb_metrics()
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
logger.debug(f"{self}: Generating TTS [{text}]")
@@ -205,12 +206,13 @@ class NvidiaTTSService(TTSService):
audio=resp.audio,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield frame
await self.start_tts_usage_metrics(text)
yield TTSStoppedFrame()
except asyncio.TimeoutError:
yield TTSStoppedFrame(context_id=context_id)
except asyncio.TimeoutError as e:
logger.error(f"{self} timeout waiting for audio response")
yield ErrorFrame(error=f"{self} error: {e}")
except Exception as e:

9
src/pipecat/services/openai/tts.py
View File

@@ -168,11 +168,12 @@ class OpenAITTSService(TTSService):
)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using OpenAI's TTS API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech data.
@@ -212,12 +213,12 @@ class OpenAITTSService(TTSService):
CHUNK_SIZE = self.chunk_size
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
async for chunk in r.iter_bytes(CHUNK_SIZE):
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1, context_id=context_id)
yield frame
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
except BadRequestError as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")

19
src/pipecat/services/piper/tts.py
View File

@@ -86,11 +86,12 @@ class PiperTTSService(TTSService):
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Piper.
Args:
text: The text to convert to speech.
context_id: Unique identifier for this TTS context.
Yields:
Frame: Audio frames containing the synthesized speech and status frames.
@@ -116,11 +117,12 @@ class PiperTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
async for frame in self._stream_audio_frames_from_iterator(
async_iterator(self._voice.synthesize(text)),
in_sample_rate=self._voice.config.sample_rate,
context_id=context_id,
):
await self.stop_ttfb_metrics()
yield frame
@@ -130,7 +132,7 @@ class PiperTTSService(TTSService):
finally:
logger.debug(f"{self}: Finished TTS [{text}]")
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
# This assumes a running TTS service running:
@@ -184,11 +186,12 @@ class PiperHttpTTSService(TTSService):
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Piper's HTTP API.
Args:
text: The text to convert to speech.
context_id: Unique identifier for this TTS context.
Yields:
Frame: Audio frames containing the synthesized speech and status frames.
@@ -215,12 +218,14 @@ class PiperHttpTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
CHUNK_SIZE = self.chunk_size
async for frame in self._stream_audio_frames_from_iterator(
response.content.iter_chunked(CHUNK_SIZE), strip_wav_header=True
response.content.iter_chunked(CHUNK_SIZE),
strip_wav_header=True,
context_id=context_id,
):
await self.stop_ttfb_metrics()
yield frame
@@ -228,4 +233,4 @@ class PiperHttpTTSService(TTSService):
yield ErrorFrame(error=f"Unknown error occurred: {e}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

43
src/pipecat/services/playht/tts.py
View File

@@ -13,7 +13,6 @@ supporting both WebSocket streaming and HTTP-based synthesis.
import io
import json
import struct
import uuid
import warnings
from typing import AsyncGenerator, Optional
@@ -169,7 +168,7 @@ class PlayHTTTSService(InterruptibleTTSService):
self._user_id = user_id
self._websocket_url = None
self._receive_task = None
self._request_id = None
self._context_id = None
self._settings = {
"language": self.language_to_service_language(params.language)
@@ -285,7 +284,7 @@ class PlayHTTTSService(InterruptibleTTSService):
except Exception as e:
await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e)
finally:
self._request_id = None
self._context_id = None
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -328,7 +327,7 @@ class PlayHTTTSService(InterruptibleTTSService):
"""Handle interruption by stopping metrics and clearing request ID."""
await super()._handle_interruption(frame, direction)
await self.stop_all_metrics()
self._request_id = None
self._context_id = None
async def _receive_messages(self):
"""Receive messages from PlayHT websocket."""
@@ -338,7 +337,7 @@ class PlayHTTTSService(InterruptibleTTSService):
if message.startswith(b"RIFF"):
continue
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(message, self.sample_rate, 1)
frame = TTSAudioRawFrame(message, self.sample_rate, 1, context_id=self._context_id)
await self.push_frame(frame)
else:
logger.debug(f"Received text message: {message}")
@@ -349,20 +348,21 @@ class PlayHTTTSService(InterruptibleTTSService):
logger.debug(f"Started processing request: {msg.get('request_id')}")
elif msg.get("type") == "end":
# Handle end of stream
if "request_id" in msg and msg["request_id"] == self._request_id:
await self.push_frame(TTSStoppedFrame())
self._request_id = None
if "request_id" in msg and msg["request_id"] == self._context_id:
await self.push_frame(TTSStoppedFrame(context_id=self._context_id))
self._context_id = None
elif "error" in msg:
await self.push_error(error_msg=f"Error: {msg['error']}")
except json.JSONDecodeError:
logger.error(f"Invalid JSON message: {message}")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate TTS audio from text using PlayHT's WebSocket API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -374,10 +374,10 @@ class PlayHTTTSService(InterruptibleTTSService):
if not self._websocket or self._websocket.state is State.CLOSED:
await self._connect()
if not self._request_id:
if not self._context_id:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._request_id = str(uuid.uuid4())
yield TTSStartedFrame(context_id=context_id)
self._context_id = context_id
tts_command = {
"text": text,
@@ -388,7 +388,7 @@ class PlayHTTTSService(InterruptibleTTSService):
"language": self._settings["language"],
"speed": self._settings["speed"],
"seed": self._settings["seed"],
"request_id": self._request_id,
"request_id": self._context_id,
}
try:
@@ -396,7 +396,7 @@ class PlayHTTTSService(InterruptibleTTSService):
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return
@@ -540,11 +540,12 @@ class PlayHTHttpTTSService(TTSService):
return language_to_playht_language(language)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate TTS audio from text using PlayHT's HTTP API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -579,7 +580,7 @@ class PlayHTHttpTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
async with aiohttp.ClientSession() as session:
async with session.post(
@@ -616,16 +617,20 @@ class PlayHTHttpTTSService(TTSService):
audio_data = buffer[fh.tell() :]
if len(audio_data) > 0:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(audio_data, self.sample_rate, 1)
frame = TTSAudioRawFrame(
audio_data, self.sample_rate, 1, context_id=context_id
)
yield frame
in_header = False
elif len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
frame = TTSAudioRawFrame(
chunk, self.sample_rate, 1, context_id=context_id
)
yield frame
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)

95
src/pipecat/services/resembleai/tts.py
View File

@@ -84,9 +84,11 @@ class ResembleAITTSService(AudioContextWordTTSService):
self._websocket = None
self._request_id_counter = 0
self._current_request_id = None
self._receive_task = None
# Map request_id to context_id for tracking TTS requests
self._request_id_to_context: dict[int, str] = {}
# Per-request audio buffers to handle concurrent TTS requests
# ResembleAI may send odd-length data even for PCM_16, so buffering helps us
# create properly aligned frames while maintaining smooth audio output
@@ -122,7 +124,7 @@ class ResembleAITTSService(AudioContextWordTTSService):
"voice_uuid": self._voice_id,
"data": text,
"binary_response": False, # Use JSON frames to get timestamps
"request_id": self._request_id_counter,
"request_id": self._request_id_counter, # ResembleAI only accepts number
"output_format": self._settings["output_format"],
"sample_rate": self._settings["sample_rate"],
"precision": self._settings["precision"],
@@ -202,10 +204,10 @@ class ResembleAITTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._current_request_id = None
self._websocket = None
self._audio_buffers.clear()
self._playback_started.clear()
self._request_id_to_context.clear()
await self._call_event_handler("on_disconnected")
def _get_websocket(self):
@@ -230,18 +232,12 @@ class ResembleAITTSService(AudioContextWordTTSService):
"""
await super()._handle_interruption(frame, direction)
await self.stop_all_metrics()
# Note: Resemble AI doesn't have an explicit cancel mechanism,
# but we can stop processing by resetting our current request_id
self._current_request_id = None
async def flush_audio(self):
"""Flush any pending audio and finalize the current context."""
if not self._current_request_id:
return
logger.trace(f"{self}: flushing audio")
# For Resemble AI, we just wait for the audio_end message
# which is handled in _process_messages
self._current_request_id = None
async def _process_messages(self):
"""Process incoming WebSocket messages from Resemble AI."""
@@ -259,10 +255,10 @@ class ResembleAITTSService(AudioContextWordTTSService):
request_id = msg.get("request_id")
# Convert request_id to string for audio context tracking
request_id_str = str(request_id)
context_id = self._request_id_to_context.get(request_id, str(request_id))
# Check if this message belongs to a valid audio context
if not self.audio_context_available(request_id_str):
if not self.audio_context_available(context_id):
continue
if msg_type == "audio":
@@ -279,20 +275,20 @@ class ResembleAITTSService(AudioContextWordTTSService):
continue
# Get or create buffer for this request
if request_id_str not in self._audio_buffers:
self._audio_buffers[request_id_str] = bytearray()
self._playback_started[request_id_str] = False
buffer = self._audio_buffers[request_id_str]
if context_id not in self._audio_buffers:
self._audio_buffers[context_id] = bytearray()
self._playback_started[context_id] = False
buffer = self._audio_buffers[context_id]
# Add to buffer
buffer.extend(audio_bytes)
# Wait for jitter buffer to fill before starting playback
# This absorbs network latency gaps (ResembleAI sends in bursts)
if not self._playback_started.get(request_id_str, False):
if not self._playback_started.get(context_id, False):
if len(buffer) < self._jitter_buffer_bytes:
continue
self._playback_started[request_id_str] = True
self._playback_started[context_id] = True
# Send complete (even-byte) chunks for PCM_16 alignment
while len(buffer) >= self._buffer_threshold_bytes:
@@ -301,8 +297,8 @@ class ResembleAITTSService(AudioContextWordTTSService):
chunk_size -= 1
chunk_to_send = bytes(buffer[:chunk_size])
self._audio_buffers[request_id_str] = buffer[chunk_size:]
buffer = self._audio_buffers[request_id_str]
self._audio_buffers[context_id] = buffer[chunk_size:]
buffer = self._audio_buffers[context_id]
if len(chunk_to_send) == 0:
continue
@@ -311,8 +307,9 @@ class ResembleAITTSService(AudioContextWordTTSService):
audio=chunk_to_send,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
await self.append_to_audio_context(request_id_str, frame)
await self.append_to_audio_context(context_id, frame)
# Process timestamps if available
timestamps = msg.get("audio_timestamps", {})
@@ -328,13 +325,13 @@ class ResembleAITTSService(AudioContextWordTTSService):
word_times.append((char, start_time))
if word_times:
await self.add_word_timestamps(word_times)
await self.add_word_timestamps(word_times, context_id)
elif msg_type == "audio_end":
await self.stop_ttfb_metrics()
# Flush remaining buffer, ensuring even length for PCM_16
buffer = self._audio_buffers.get(request_id_str, bytearray())
buffer = self._audio_buffers.get(context_id, bytearray())
if buffer:
remaining = bytes(buffer)
# PCM_16 requires even number of bytes
@@ -345,20 +342,21 @@ class ResembleAITTSService(AudioContextWordTTSService):
audio=remaining,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
await self.append_to_audio_context(request_id_str, frame)
await self.append_to_audio_context(context_id, frame)
# Clean up buffer and playback tracking for this request
if request_id_str in self._audio_buffers:
del self._audio_buffers[request_id_str]
if request_id_str in self._playback_started:
del self._playback_started[request_id_str]
if context_id in self._audio_buffers:
del self._audio_buffers[context_id]
if context_id in self._playback_started:
del self._playback_started[context_id]
# Clean up request_id mapping
if request_id in self._request_id_to_context:
del self._request_id_to_context[request_id]
await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)])
await self.remove_audio_context(request_id_str)
# Clear current request if this was it
if self._current_request_id == request_id:
self._current_request_id = None
await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)], context_id)
await self.remove_audio_context(context_id)
elif msg_type == "error":
error_name = msg.get("error_name", "Unknown")
@@ -369,19 +367,15 @@ class ResembleAITTSService(AudioContextWordTTSService):
)
# Clean up buffer and playback tracking for this request
if request_id_str in self._audio_buffers:
del self._audio_buffers[request_id_str]
if request_id_str in self._playback_started:
del self._playback_started[request_id_str]
if context_id in self._audio_buffers:
del self._audio_buffers[context_id]
if context_id in self._playback_started:
del self._playback_started[context_id]
await self.push_frame(TTSStoppedFrame())
await self.push_frame(TTSStoppedFrame(context_id=context_id))
await self.stop_all_metrics()
await self.push_error(ErrorFrame(error=f"{self} error: {error_name} - {error_msg}"))
# Clear current request if this was it
if self._current_request_id == request_id:
self._current_request_id = None
# Check if this is an unrecoverable error (connection-level failure)
if status_code in [401, 403]:
# Close and reconnect for auth errors
@@ -402,11 +396,12 @@ class ResembleAITTSService(AudioContextWordTTSService):
await self._connect_websocket()
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Resemble AI's streaming API.
Args:
text: The text to synthesize into speech.
context_id: Unique identifier for this TTS context.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -417,15 +412,13 @@ class ResembleAITTSService(AudioContextWordTTSService):
if not self._websocket or self._websocket.state is State.CLOSED:
await self._connect()
if not self._current_request_id:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
# Track the current request_id we're processing
self._current_request_id = self._request_id_counter
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
# Create audio context using request_id (converted to string)
request_id_str = str(self._request_id_counter)
await self.create_audio_context(request_id_str)
# Map request_id to context_id for tracking
self._request_id_to_context[self._request_id_counter] = context_id
await self.create_audio_context(context_id)
msg = self._build_msg(text=text)
@@ -434,7 +427,7 @@ class ResembleAITTSService(AudioContextWordTTSService):
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return

44
src/pipecat/services/rime/tts.py
View File

@@ -12,7 +12,6 @@ using Rime's API for streaming and batch audio synthesis.
import base64
import json
import uuid
from typing import Any, AsyncGenerator, Mapping, Optional
import aiohttp
@@ -387,6 +386,7 @@ class RimeTTSService(AudioContextWordTTSService):
if not msg or not self.audio_context_available(msg["contextId"]):
continue
context_id = msg["contextId"]
if msg["type"] == "chunk":
# Process audio chunk
await self.stop_ttfb_metrics()
@@ -395,8 +395,9 @@ class RimeTTSService(AudioContextWordTTSService):
audio=base64.b64decode(msg["data"]),
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
await self.append_to_audio_context(msg["contextId"], frame)
await self.append_to_audio_context(context_id, frame)
elif msg["type"] == "timestamps":
# Process word timing information
@@ -409,7 +410,7 @@ class RimeTTSService(AudioContextWordTTSService):
# Calculate word timing pairs
word_pairs = self._calculate_word_times(words, starts, ends)
if word_pairs:
await self.add_word_timestamps(word_pairs)
await self.add_word_timestamps(word_pairs, context_id=context_id)
self._cumulative_time = ends[-1] + self._cumulative_time
logger.debug(f"Updated cumulative time to: {self._cumulative_time}")
@@ -432,11 +433,12 @@ class RimeTTSService(AudioContextWordTTSService):
await self.add_word_timestamps([("Reset", 0)])
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Rime's streaming API.
Args:
text: The text to convert to speech.
context_id: Unique identifier for this TTS context.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -449,9 +451,9 @@ class RimeTTSService(AudioContextWordTTSService):
try:
if not self._context_id:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
self._cumulative_time = 0
self._context_id = str(uuid.uuid4())
self._context_id = context_id
await self.create_audio_context(self._context_id)
msg = self._build_msg(text=text)
@@ -459,7 +461,7 @@ class RimeTTSService(AudioContextWordTTSService):
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return
@@ -558,11 +560,12 @@ class RimeHttpTTSService(TTSService):
return language_to_rime_language(language)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Rime's HTTP API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -601,13 +604,14 @@ class RimeHttpTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
CHUNK_SIZE = self.chunk_size
async for frame in self._stream_audio_frames_from_iterator(
response.content.iter_chunked(CHUNK_SIZE),
strip_wav_header=need_to_strip_wav_header,
context_id=context_id,
):
await self.stop_ttfb_metrics()
yield frame
@@ -616,7 +620,7 @@ class RimeHttpTTSService(TTSService):
yield ErrorFrame(error=f"Unknown error occurred: {e}")
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
class RimeNonJsonTTSService(InterruptibleTTSService):
@@ -716,8 +720,8 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
if params.extra:
self._settings.update(params.extra)
self._started = False
self._receive_task = None
self._context_id: Optional[str] = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -767,8 +771,6 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
direction: The direction to push the frame.
"""
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
self._started = False
async def _connect(self):
"""Establish WebSocket connection and start receive task."""
@@ -817,7 +819,7 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._started = False
self._context_id = None
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -847,17 +849,19 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
audio=message,
sample_rate=self.sample_rate,
num_channels=1,
context_id=self._context_id,
)
await self.push_frame(frame)
except Exception as e:
await self.push_error(error_msg=f"Error: {e}", exception=e)
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Rime's streaming API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -867,17 +871,17 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
if not self._websocket or self._websocket.state is State.CLOSED:
await self._connect()
try:
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._started = True
await self.start_ttfb_metrics()
# Store context_id for use in _receive_messages
self._context_id = context_id
yield TTSStartedFrame(context_id=context_id)
# Send bare text (not JSON)
await self._get_websocket().send(text)
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return

34
src/pipecat/services/sarvam/tts.py
View File

@@ -462,11 +462,12 @@ class SarvamHttpTTSService(TTSService):
self._settings["sample_rate"] = self.sample_rate
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Sarvam AI's API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -503,7 +504,7 @@ class SarvamHttpTTSService(TTSService):
url = f"{self._base_url}/text-to-speech"
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
async with self._session.post(url, json=payload, headers=headers) as response:
if response.status != 200:
@@ -533,6 +534,7 @@ class SarvamHttpTTSService(TTSService):
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
yield frame
@@ -541,7 +543,7 @@ class SarvamHttpTTSService(TTSService):
yield ErrorFrame(error=f"Error generating TTS: {e}", exception=e)
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
class SarvamTTSService(InterruptibleTTSService):
@@ -789,9 +791,9 @@ class SarvamTTSService(InterruptibleTTSService):
elif params.temperature != 0.6:
logger.warning(f"temperature parameter is ignored for {model}")
self._started = False
self._receive_task = None
self._keepalive_task = None
self._context_id: Optional[str] = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -859,8 +861,6 @@ class SarvamTTSService(InterruptibleTTSService):
direction: The direction to push the frame.
"""
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
self._started = False
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process a frame and flush audio if it's the end of a full response."""
@@ -956,7 +956,7 @@ class SarvamTTSService(InterruptibleTTSService):
except Exception as e:
await self.push_error(error_msg=f"Error closing websocket: {e}", exception=e)
finally:
self._started = False
self._context_id = None
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -974,7 +974,9 @@ class SarvamTTSService(InterruptibleTTSService):
# Check for interruption before processing audio
await self.stop_ttfb_metrics()
audio = base64.b64decode(msg["data"]["audio"])
frame = TTSAudioRawFrame(audio, self.sample_rate, 1)
frame = TTSAudioRawFrame(
audio, self.sample_rate, 1, context_id=self._context_id
)
await self.push_frame(frame)
elif msg.get("type") == "error":
error_msg = msg["data"]["message"]
@@ -984,7 +986,6 @@ class SarvamTTSService(InterruptibleTTSService):
if "too long" in error_msg.lower() or "timeout" in error_msg.lower():
logger.warning("Connection timeout detected, service may need restart")
self._started = False
await self.push_frame(ErrorFrame(error=f"TTS Error: {error_msg}"))
async def _keepalive_task_handler(self):
@@ -1009,16 +1010,17 @@ class SarvamTTSService(InterruptibleTTSService):
logger.warning("WebSocket not ready, cannot send text")
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech audio frames from input text using Sarvam TTS.
Sends text over WebSocket for synthesis and yields corresponding audio or status frames.
Args:
text: The text input to synthesize.
context_id: The context ID for tracking audio frames.
Yields:
Frame objects including TTSStartedFrame, TTSAudioRawFrame(s), or TTSStoppedFrame.
Frame objects including TTSStartedFrame, TTSAudioRawFrame(s, context_id=context_id), or TTSStoppedFrame.
"""
logger.debug(f"Generating TTS: [{text}]")
@@ -1027,15 +1029,15 @@ class SarvamTTSService(InterruptibleTTSService):
await self._connect()
try:
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
self._started = True
await self.start_ttfb_metrics()
# Store context_id for use in _receive_messages
self._context_id = context_id
yield TTSStartedFrame(context_id=context_id)
await self._send_text(text)
await self.start_tts_usage_metrics(text)
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
await self._disconnect()
await self._connect()
return

8
src/pipecat/services/speechmatics/tts.py
View File

@@ -106,11 +106,12 @@ class SpeechmaticsTTSService(TTSService):
return True
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Speechmatics' HTTP API.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -187,7 +188,7 @@ class SpeechmaticsTTSService(TTSService):
await self.start_tts_usage_metrics(text)
# Emit the TTS started frame
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
# Process the response in streaming chunks
first_chunk = True
@@ -216,6 +217,7 @@ class SpeechmaticsTTSService(TTSService):
audio=audio_data,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
# Successfully processed the response, break out of retry loop
@@ -225,7 +227,7 @@ class SpeechmaticsTTSService(TTSService):
yield ErrorFrame(error=f"Error generating TTS: {e}")
finally:
# Emit the TTS stopped frame
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)
def _get_endpoint_url(base_url: str, voice: str, sample_rate: int) -> str:

15
src/pipecat/services/xtts/tts.py
View File

@@ -148,11 +148,12 @@ class XTTSService(TTSService):
self._studio_speakers = await r.json()
@traced_tts
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using XTTS streaming server.
Args:
text: The text to synthesize into speech.
context_id: The context ID for tracking audio frames.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -186,7 +187,7 @@ class XTTSService(TTSService):
await self.start_tts_usage_metrics(text)
yield TTSStartedFrame()
yield TTSStartedFrame(context_id=context_id)
CHUNK_SIZE = self.chunk_size
@@ -211,7 +212,9 @@ class XTTSService(TTSService):
bytes(process_data), 24000, self.sample_rate
)
# Create the frame with the resampled audio
frame = TTSAudioRawFrame(resampled_audio, self.sample_rate, 1)
frame = TTSAudioRawFrame(
resampled_audio, self.sample_rate, 1, context_id=context_id
)
yield frame
# Process any remaining data in the buffer.
@@ -219,7 +222,9 @@ class XTTSService(TTSService):
resampled_audio = await self._resampler.resample(
bytes(buffer), 24000, self.sample_rate
)
frame = TTSAudioRawFrame(resampled_audio, self.sample_rate, 1)
frame = TTSAudioRawFrame(
resampled_audio, self.sample_rate, 1, context_id=context_id
)
yield frame
yield TTSStoppedFrame()
yield TTSStoppedFrame(context_id=context_id)