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

Merge pull request #3584 from pipecat-ai/filipi/speak_frame

Browse Source 
TTS services improvements.
This commit is contained in:
Filipi da Silva Fuchter
2026-02-10 12:11:47 -05:00
committed by GitHub
parent 2dd3e2f1e7 e00b98343e
commit ad179b0852
37 changed files with 555 additions and 401 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
changelog/3584.added.2.md Normal file
View File

@@ -0,0 +1,3 @@
- Added `append_to_context` parameter to `TTSSpeakFrame` for conditional LLM context addition.
- Allows fine-grained control over whether text should be added to conversation context
- Defaults to `True` to maintain backward compatibility

4
changelog/3584.added.md Normal file
View File

@@ -0,0 +1,4 @@
- Added TTS context tracking system with `context_id` field to trace audio generation through the pipeline.
- `TTSAudioRawFrame`, `TTSStartedFrame`, `TTSStoppedFrame` now include `context_id`
- `AggregatedTextFrame` and `TTSTextFrame` now include `context_id`
- Enables tracking which TTS request generated specific audio chunks

3
changelog/3584.changed.2.md Normal file
View File

@@ -0,0 +1,3 @@
- Simplified context aggregators to use `frame.append_to_context` flag instead of tracking internal state.
- Cleaner logic in `LLMResponseAggregator` and `LLMResponseUniversalAggregator`
- More consistent behavior across aggregator implementations

4
changelog/3584.changed.3.md Normal file
View File

@@ -0,0 +1,4 @@
- ⚠️ `TTSService.run_tts()` now requires a `context_id` parameter for context tracking.
- Custom TTS service implementations must update their `run_tts()` signature
- Before: `async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:`
- After: `async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:`

3
changelog/3584.changed.md Normal file
View File

@@ -0,0 +1,3 @@
- Updated all 30+ TTS service implementations to support context tracking with `context_id`.
- Services now generate and propagate context IDs through TTS frames
- Enables end-to-end tracing of TTS requests through the pipeline

6
examples/foundational/14d-function-calling-openai-video.py
View File

@@ -99,7 +99,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
@llm.event_handler("on_function_calls_started")
async def on_function_calls_started(service, function_calls):
await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False))
fetch_image_function = FunctionSchema(
name="fetch_user_image",
@@ -174,6 +174,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Client disconnected")
await task.cancel()
@tts.event_handler("on_tts_request")
async def on_tts_request(tts, context_id: str, text: str):
logger.debug(f"On TTS request: {context_id}: {text}")
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)

35
src/pipecat/frames/frames.py
View File

@@ -279,9 +279,12 @@ class TTSAudioRawFrame(OutputAudioRawFrame):
"""Audio data frame generated by Text-to-Speech services.
A chunk of output audio generated by a TTS service, ready for playback.
Parameters:
context_id: Unique identifier for the TTS context that generated this audio.
"""
pass
context_id: Optional[str] = None
@dataclass
@@ -343,6 +346,11 @@ class TextFrame(DataFrame):
Parameters:
text: The text content.
skip_tts: Whether this text should be skipped by the TTS service.
includes_inter_frame_spaces: Whether any necessary inter-frame (leading/trailing) spaces are already
included in the text.
append_to_context: Whether this text should be appended to the LLM context.
Defaults to True.
"""
text: str
@@ -397,9 +405,11 @@ class AggregatedTextFrame(TextFrame):
Parameters:
aggregated_by: Method used to aggregate the text frames.
context_id: Unique identifier for the TTS context that generated this text.
"""
aggregated_by: AggregationType | str
context_id: Optional[str] = None
@dataclass
@@ -411,9 +421,13 @@ class VisionTextFrame(LLMTextFrame):
@dataclass
class TTSTextFrame(AggregatedTextFrame):
"""Text frame generated by Text-to-Speech services."""
"""Text frame generated by Text-to-Speech services.
pass
Parameters:
context_id: Unique identifier for the TTS context that generated this text.
"""
context_id: Optional[str] = None
@dataclass
@@ -923,9 +937,11 @@ class TTSSpeakFrame(DataFrame):
Parameters:
text: The text to be spoken.
append_to_context: Whether to append the text to the context.
"""
text: str
append_to_context: Optional[bool] = None
@dataclass
@@ -2023,16 +2039,23 @@ class TTSStartedFrame(ControlFrame):
TTSStoppedFrame. These frames can be used for aggregating audio frames in a
transport to optimize the size of frames sent to the session, without
needing to control this in the TTS service.
Parameters:
context_id: Unique identifier for this TTS context.
"""
pass
context_id: Optional[str] = None
@dataclass
class TTSStoppedFrame(ControlFrame):
"""Frame indicating the end of a TTS response."""
"""Frame indicating the end of a TTS response.
pass
Parameters:
context_id: Unique identifier for this TTS context.
"""
context_id: Optional[str] = None
@dataclass

2
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -1059,7 +1059,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
await self.push_aggregation()
async def _handle_text(self, frame: TextFrame):
if not self._started or not frame.append_to_context:
if not frame.append_to_context:
return
if self._params.expect_stripped_words:

21
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -52,6 +52,7 @@ from pipecat.frames.frames import (
StartFrame,
TextFrame,
TranscriptionFrame,
TranslationFrame,
UserImageRawFrame,
UserMuteStartedFrame,
UserMuteStoppedFrame,
@@ -795,7 +796,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
DeprecationWarning,
)
self._started = 0
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._function_calls_image_results: Dict[str, UserImageRawFrame] = {}
self._context_updated_tasks: Set[asyncio.Task] = set()
@@ -917,12 +917,10 @@ class LLMAssistantAggregator(LLMContextAggregator):
async def _handle_interruptions(self, frame: InterruptionFrame):
await self._trigger_assistant_turn_stopped()
self._started = 0
await self.reset()
async def _handle_end_or_cancel(self, frame: Frame):
await self._trigger_assistant_turn_stopped()
self._started = 0
async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
@@ -1067,15 +1065,17 @@ class LLMAssistantAggregator(LLMContextAggregator):
)
async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
self._started += 1
await self._trigger_assistant_turn_started()
async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
self._started -= 1
await self._trigger_assistant_turn_stopped()
async def _handle_text(self, frame: TextFrame):
if not self._started or not frame.append_to_context:
# Skip TextFrame types not intended to build the assistant context
if isinstance(frame, (TranscriptionFrame, TranslationFrame, InterimTranscriptionFrame)):
return
if not frame.append_to_context:
return
# Make sure we really have text (spaces count, too!)
@@ -1089,18 +1089,12 @@ class LLMAssistantAggregator(LLMContextAggregator):
)
async def _handle_thought_start(self, frame: LLMThoughtStartFrame):
if not self._started:
return
await self._reset_thought_aggregation()
self._thought_append_to_context = frame.append_to_context
self._thought_llm = frame.llm
self._thought_start_time = time_now_iso8601()
async def _handle_thought_text(self, frame: LLMThoughtTextFrame):
if not self._started:
return
# Make sure we really have text (spaces count, too!)
if len(frame.text) == 0:
return
@@ -1112,9 +1106,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
)
async def _handle_thought_end(self, frame: LLMThoughtEndFrame):
if not self._started:
return
thought = concatenate_aggregated_text(self._thought_aggregation)
if self._thought_append_to_context:

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:

109
src/pipecat/services/tts_service.py
View File

@@ -7,7 +7,9 @@
"""Base classes for Text-to-speech services."""
import asyncio
import uuid
from abc import abstractmethod
from dataclasses import dataclass
from typing import (
Any,
AsyncGenerator,
@@ -58,6 +60,17 @@ from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
from pipecat.utils.time import seconds_to_nanoseconds
@dataclass
class TTSContext:
"""Context information for a TTS request.
Attributes:
append_to_context: Whether this TTS output should be appended to the conversation context.
"""
append_to_context: bool = True
class TTSService(AIService):
"""Base class for text-to-speech services.
@@ -66,9 +79,10 @@ class TTSService(AIService):
sentence aggregation, silence insertion, and frame processing control.
Event handlers:
on_connected: Called when connected to the STT service.
on_connected: Called when disconnected from the STT service.
on_connection_error: Called when a connection to the STT service error occurs.
on_connected: Called when connected to the TTS service.
on_disconnected: Called when disconnected from the TTS service.
on_connection_error: Called when a connection to the TTS service error occurs.
on_tts_request: Called before a TTS request is made, with the context ID and text.
Example::
@@ -81,8 +95,12 @@ class TTSService(AIService):
logger.debug(f"TTS disconnected")
@tts.event_handler("on_connection_error")
async def on_connection_error(stt: TTSService, error: str):
async def on_connection_error(tts: TTSService, error: str):
logger.error(f"TTS connection error: {error}")
@tts.event_handler("on_tts_request")
async def on_tts_request(tts: TTSService, context_id: str, text: str):
logger.debug(f"TTS request: {context_id} - {text}")
"""
def __init__(
@@ -209,10 +227,12 @@ class TTSService(AIService):
self._stop_frame_queue: asyncio.Queue = asyncio.Queue()
self._processing_text: bool = False
self._tts_contexts: Dict[str, TTSContext] = {}
self._register_event_handler("on_connected")
self._register_event_handler("on_disconnected")
self._register_event_handler("on_connection_error")
self._register_event_handler("on_tts_request")
@property
def sample_rate(self) -> int:
@@ -256,15 +276,26 @@ class TTSService(AIService):
"""
self._voice_id = voice
def create_context_id(self) -> str:
"""Generate a unique context ID for a TTS request.
This method can be overridden by subclasses to provide custom context ID generation.
Returns:
A unique string identifier for the TTS context.
"""
return str(uuid.uuid4())
# Converts the text to audio.
@abstractmethod
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.
This method must be implemented by subclasses to provide actual TTS functionality.
Args:
text: The text to synthesize into speech.
context_id: Unique identifier for this TTS context.
Yields:
Frame: Audio frames containing the synthesized speech.
@@ -463,7 +494,10 @@ class TTSService(AIService):
# Store if we were processing text or not so we can set it back.
processing_text = self._processing_text
# Assumption: text in TTSSpeakFrame does not include inter-frame spaces
await self._push_tts_frames(AggregatedTextFrame(frame.text, AggregationType.SENTENCE))
await self._push_tts_frames(
AggregatedTextFrame(frame.text, AggregationType.SENTENCE),
append_tts_text_to_context=frame.append_to_context,
)
# We pause processing incoming frames because we are sending data to
# the TTS. We pause to avoid audio overlapping.
await self._maybe_pause_frame_processing()
@@ -484,6 +518,12 @@ class TTSService(AIService):
frame: The frame to push.
direction: The direction to push the frame.
"""
# Clean up context when we see TTSStoppedFrame
if isinstance(frame, TTSStoppedFrame) and frame.context_id:
if frame.context_id in self._tts_contexts:
logger.debug(f"{self} cleaning up TTS context {frame.context_id}")
del self._tts_contexts[frame.context_id]
if self._push_silence_after_stop and isinstance(frame, TTSStoppedFrame):
silence_num_bytes = int(self._silence_time_s * self.sample_rate * 2) # 16-bit
silence_frame = TTSAudioRawFrame(
@@ -513,6 +553,7 @@ class TTSService(AIService):
*,
strip_wav_header: bool = False,
in_sample_rate: Optional[int] = None,
context_id: Optional[str] = None,
) -> AsyncGenerator[Frame, None]:
"""Stream audio frames from an async byte iterator with optional resampling.
@@ -526,6 +567,7 @@ class TTSService(AIService):
strip_wav_header: Strip WAV header and parse source sample rate from it.
in_sample_rate: Source sample rate for raw PCM data. Overrides
WAV-detected rate if both are provided.
context_id: Unique identifier for this TTS context.
"""
buffer = bytearray()
@@ -555,7 +597,10 @@ class TTSService(AIService):
buffer = buffer[aligned_length:] # keep any leftover byte
if len(aligned_chunk) > 0:
yield TTSAudioRawFrame(aligned_chunk, self.sample_rate, 1)
frame = TTSAudioRawFrame(
bytes(aligned_chunk), self.sample_rate, 1, context_id=context_id
)
yield frame
if len(buffer) > 0:
# Make sure we don't need an extra padding byte.
@@ -601,7 +646,10 @@ class TTSService(AIService):
)
async def _push_tts_frames(
self, src_frame: AggregatedTextFrame, includes_inter_frame_spaces: Optional[bool] = False
self,
src_frame: AggregatedTextFrame,
includes_inter_frame_spaces: Optional[bool] = False,
append_tts_text_to_context: Optional[bool] = True,
):
type = src_frame.aggregated_by
text = src_frame.text
@@ -636,11 +684,15 @@ class TTSService(AIService):
await self.stop_processing_metrics()
return
# Create context ID and store metadata
context_id = self.create_context_id()
# To support use cases that may want to know the text before it's spoken, we
# push the AggregatedTextFrame version before transforming and sending to TTS.
# However, we do not want to add this text to the assistant context until it
# is spoken, so we set append_to_context to False.
src_frame.append_to_context = False
src_frame.context_id = context_id
await self.push_frame(src_frame)
# Note: Text transformations are meant to only affect the text sent to the TTS for
@@ -653,9 +705,19 @@ class TTSService(AIService):
if aggregation_type == type or aggregation_type == "*":
transformed_text = await transform(transformed_text, type)
self._tts_contexts[context_id] = TTSContext(
append_to_context=append_tts_text_to_context
if append_tts_text_to_context is not None
else True
)
# Apply any final text preparation (e.g., trailing space)
prepared_text = self._prepare_text_for_tts(transformed_text)
await self.process_generator(self.run_tts(prepared_text))
# Trigger event before starting TTS
await self._call_event_handler("on_tts_request", context_id, prepared_text)
await self.process_generator(self.run_tts(prepared_text, context_id))
await self.stop_processing_metrics()
@@ -669,6 +731,10 @@ class TTSService(AIService):
# or transformations.
frame = TTSTextFrame(text, aggregated_by=type)
frame.includes_inter_frame_spaces = includes_inter_frame_spaces
frame.context_id = context_id
# Only override append_to_context if explicitly set
if append_tts_text_to_context is not None:
frame.append_to_context = append_tts_text_to_context
await self.push_frame(frame)
async def _stop_frame_handler(self):
@@ -721,18 +787,24 @@ class WordTTSService(TTSService):
"""Reset word timestamp tracking."""
self._initial_word_timestamp = -1
async def add_word_timestamps(self, word_times: List[Tuple[str, float]]):
async def add_word_timestamps(
self, word_times: List[Tuple[str, float]], context_id: Optional[str] = None
):
"""Add word timestamps to the processing queue.
Args:
word_times: List of (word, timestamp) tuples where timestamp is in seconds.
context_id: Unique identifier for the TTS context.
"""
# Transform to include context_id in each tuple
word_times_with_context = [(word, timestamp, context_id) for word, timestamp in word_times]
if self._initial_word_timestamp == -1:
# Cache word timestamps and don't add them until we have started
# (i.e. we have some audio).
self._initial_word_times.extend(word_times)
self._initial_word_times.extend(word_times_with_context)
else:
await self._add_word_timestamps(word_times)
await self._add_word_timestamps(word_times_with_context)
async def start(self, frame: StartFrame):
"""Start the word TTS service.
@@ -790,15 +862,15 @@ class WordTTSService(TTSService):
await self.cancel_task(self._words_task)
self._words_task = None
async def _add_word_timestamps(self, word_times: List[Tuple[str, float]]):
for word, timestamp in word_times:
await self._words_queue.put((word, seconds_to_nanoseconds(timestamp)))
async def _add_word_timestamps(self, word_times_with_context: List[Tuple[str, float, str]]):
for word, timestamp, context_id in word_times_with_context:
await self._words_queue.put((word, seconds_to_nanoseconds(timestamp), context_id))
async def _words_task_handler(self):
last_pts = 0
while True:
frame = None
(word, timestamp) = await self._words_queue.get()
(word, timestamp, context_id) = await self._words_queue.get()
if word == "Reset" and timestamp == 0:
await self.reset_word_timestamps()
if self._llm_response_started:
@@ -808,11 +880,16 @@ class WordTTSService(TTSService):
elif word == "TTSStoppedFrame" and timestamp == 0:
frame = TTSStoppedFrame()
frame.pts = last_pts
frame.context_id = context_id
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)
frame.pts = self._initial_word_timestamp + timestamp
frame.context_id = context_id
# Look up append_to_context from context metadata
if context_id in self._tts_contexts:
frame.append_to_context = self._tts_contexts[context_id].append_to_context
if frame:
last_pts = frame.pts
await self.push_frame(frame)

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)