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

Refactored audio context management in TTS services to improve encapsulation and reduce code duplication

Browse Source
This commit is contained in:
filipi87
2026-02-20 14:57:44 -03:00
parent c9615c8db6
commit 125c423356
8 changed files with 167 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
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._context_id = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -255,7 +253,7 @@ class AsyncAITTSService(AudioContextTTSService):
if self._websocket:
logger.debug("Disconnecting from Async")
# Close all contexts and the socket
if self._context_id:
if self.has_active_audio_context():
await self._websocket.send(json.dumps({"terminate": True}))
await self._websocket.close()
logger.debug("Disconnected from Async")
@@ -263,7 +261,7 @@ class AsyncAITTSService(AudioContextTTSService):
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._websocket = None
self._context_id = None
await self.remove_active_audio_context()
await self._call_event_handler("on_disconnected")
def _get_websocket(self):
@@ -271,26 +269,13 @@ class AsyncAITTSService(AudioContextTTSService):
return self._websocket
raise Exception("Websocket not connected")
def create_context_id(self) -> str:
"""Generate a unique context ID for a TTS request in case we don't have one already in progress.
Returns:
A unique string identifier for the TTS context.
"""
# If a context ID does not exist, create a new one.
# If an ID exists, continue using the current ID.
# When interruptions happen, user speech results in
# an interruption, which resets the context ID.
if not self._context_id:
return str(uuid.uuid4())
return self._context_id
async def flush_audio(self):
"""Flush any pending audio."""
if not self._context_id or not self._websocket:
context_id = self.get_active_audio_context_id()
if not context_id or not self._websocket:
return
logger.trace(f"{self}: flushing audio")
msg = self._build_msg(text=" ", context_id=self._context_id, force=True)
msg = self._build_msg(text=" ", context_id=context_id, force=True)
await self._websocket.send(msg)
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
@@ -318,11 +303,11 @@ class AsyncAITTSService(AudioContextTTSService):
# Check if this message belongs to the current context.
if not self.audio_context_available(received_ctx_id):
if self._context_id == received_ctx_id:
if self.get_active_audio_context_id() == received_ctx_id:
logger.debug(
f"Received a delayed message, recreating the context: {self._context_id}"
f"Received a delayed message, recreating the context: {received_ctx_id}"
)
await self.create_audio_context(self._context_id)
await self.create_audio_context(received_ctx_id)
else:
# This can happen if a message is received _after_ we have closed a context
# due to user interruption but _before_ the `isFinal` message for the context
@@ -343,10 +328,11 @@ class AsyncAITTSService(AudioContextTTSService):
await asyncio.sleep(KEEPALIVE_SLEEP)
try:
if self._websocket and self._websocket.state is State.OPEN:
if self._context_id:
context_id = self.get_active_audio_context_id()
if context_id:
keepalive_message = {
"transcript": " ",
"context_id": self._context_id,
"context_id": context_id,
}
logger.trace("Sending keepalive message")
else:
@@ -362,19 +348,16 @@ class AsyncAITTSService(AudioContextTTSService):
async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
"""Handle interruption by closing the current context."""
context_id = self.get_active_audio_context_id()
await super()._handle_interruption(frame, direction)
# Close the current context when interrupted without closing the websocket
if self._context_id and self._websocket:
if context_id and self._websocket:
try:
await self._websocket.send(
json.dumps(
{"context_id": self._context_id, "close_context": True, "transcript": ""}
)
json.dumps({"context_id": context_id, "close_context": True, "transcript": ""})
)
except Exception as e:
logger.error(f"Error closing context on interruption: {e}")
self._context_id = None
@traced_tts
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
@@ -394,16 +377,13 @@ class AsyncAITTSService(AudioContextTTSService):
await self._connect()
try:
if not self._context_id:
if not self.has_active_audio_context():
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
if not self.audio_context_available(context_id):
await self.create_audio_context(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)
msg = self._build_msg(text=text, force=True, context_id=context_id)
await self._get_websocket().send(msg)
await self.start_tts_usage_metrics(text)

38
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
@@ -307,7 +306,6 @@ class CartesiaTTSService(AudioContextWordTTSService):
self.set_model_name(model)
self.set_voice(voice_id)
self._context_id = None
self._receive_task = None
def can_generate_metrics(self) -> bool:
@@ -430,7 +428,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
msg = {
"transcript": text,
"continue": continue_transcript,
"context_id": self._context_id,
"context_id": self.get_active_audio_context_id(),
"model_id": self.model_name,
"voice": voice_config,
"output_format": self._settings["output_format"],
@@ -523,7 +521,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._context_id = None
await self.remove_active_audio_context()
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -533,35 +531,22 @@ class CartesiaTTSService(AudioContextWordTTSService):
raise Exception("Websocket not connected")
async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
context_id = self.get_active_audio_context_id()
await super()._handle_interruption(frame, direction)
await self.stop_all_metrics()
if self._context_id:
cancel_msg = json.dumps({"context_id": self._context_id, "cancel": True})
if context_id:
cancel_msg = json.dumps({"context_id": context_id, "cancel": True})
await self._get_websocket().send(cancel_msg)
self._context_id = None
def create_context_id(self) -> str:
"""Generate a unique context ID for a TTS request in case we don't have one already in progress.
Returns:
A unique string identifier for the TTS context.
"""
# If a context ID does not exist, create a new one.
# If an ID exists, continue using the current ID.
# When interruptions happen, user speech results in
# an interruption, which resets the context ID.
if not self._context_id:
return str(uuid.uuid4())
return self._context_id
async def flush_audio(self):
"""Flush any pending audio and finalize the current context."""
if not self._context_id or not self._websocket:
context_id = self.get_active_audio_context_id()
if not context_id or not self._websocket:
return
logger.trace(f"{self}: flushing audio")
msg = self._build_msg(text="", continue_transcript=False)
await self._websocket.send(msg)
self._context_id = None
self.reset_active_audio_context()
async def _process_messages(self):
async for message in self._get_websocket():
@@ -593,7 +578,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
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
self.reset_active_audio_context()
else:
await self.push_error(error_msg=f"Error, unknown message type: {msg}")
@@ -622,11 +607,10 @@ class CartesiaTTSService(AudioContextWordTTSService):
if not self._websocket or self._websocket.state is State.CLOSED:
await self._connect()
if not self._context_id:
if not self.has_active_audio_context():
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
self._context_id = context_id
await self.create_audio_context(self._context_id)
await self.create_audio_context(context_id)
msg = self._build_msg(text=text)

73
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
@@ -343,7 +342,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
self._partial_word_start_time = 0.0
# Context management for v1 multi API
self._context_id = None
self._receive_task = None
self._keepalive_task = None
@@ -411,18 +409,19 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
)
await self._disconnect()
await self._connect()
elif voice_settings_changed and self._context_id:
elif voice_settings_changed and self.has_active_audio_context():
# Voice settings can be updated by closing current context
# so new one gets created with updated voice settings
logger.debug(f"Voice settings changed, closing current context to apply changes")
context_id = self.get_active_audio_context_id()
try:
if self._websocket:
await self._websocket.send(
json.dumps({"context_id": self._context_id, "close_context": True})
json.dumps({"context_id": context_id, "close_context": True})
)
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
self._context_id = None
self.reset_active_audio_context()
async def start(self, frame: StartFrame):
"""Start the ElevenLabs TTS service.
@@ -454,10 +453,11 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
async def flush_audio(self):
"""Flush any pending audio and finalize the current context."""
if not self._context_id or not self._websocket:
context_id = self.get_active_audio_context_id()
if not context_id or not self._websocket:
return
logger.trace(f"{self}: flushing audio")
msg = {"context_id": self._context_id, "flush": True}
msg = {"context_id": context_id, "flush": True}
await self._websocket.send(json.dumps(msg))
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
@@ -470,7 +470,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await super().push_frame(frame, direction)
if isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("Reset", 0)], self._context_id)
await self.add_word_timestamps([("Reset", 0)], self.get_active_audio_context_id())
async def _connect(self):
await super()._connect()
@@ -545,14 +545,14 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
if self._websocket:
logger.debug("Disconnecting from ElevenLabs")
# Close all contexts and the socket
if self._context_id:
if self.has_active_audio_context():
await self._websocket.send(json.dumps({"close_socket": True}))
await self._websocket.close()
logger.debug("Disconnected from ElevenLabs")
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._context_id = None
await self.remove_active_audio_context()
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -563,11 +563,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
"""Handle interruption by closing the current context."""
# Close the current context when interrupted without closing the websocket
context_id = self.get_active_audio_context_id()
await super()._handle_interruption(frame, direction)
# Close the current context when interrupted without closing the websocket
if self._context_id and self._websocket:
logger.trace(f"Closing context {self._context_id} due to interruption")
if context_id and self._websocket:
logger.trace(f"Closing context {context_id} due to interruption")
try:
# ElevenLabs requires that Pipecat manages the contexts and closes them
# when they're not longer in use. Since an InterruptionFrame is pushed
@@ -576,11 +577,10 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
# Note: We do not need to call remove_audio_context here, as the context is
# automatically reset when super ()._handle_interruption is called.
await self._websocket.send(
json.dumps({"context_id": self._context_id, "close_context": True})
json.dumps({"context_id": context_id, "close_context": True})
)
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
self._context_id = None
self._partial_word = ""
self._partial_word_start_time = 0.0
@@ -600,11 +600,11 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
# Check if this message belongs to the current context.
if not self.audio_context_available(received_ctx_id):
if self._context_id == received_ctx_id:
if self.get_active_audio_context_id() == received_ctx_id:
logger.debug(
f"Received a delayed message, recreating the context: {self._context_id}"
f"Received a delayed message, recreating the context: {received_ctx_id}"
)
await self.create_audio_context(self._context_id)
await self.create_audio_context(received_ctx_id)
else:
# This can happen if a message is received _after_ we have closed a context
# due to user interruption but _before_ the `isFinal` message for the context
@@ -657,13 +657,14 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await asyncio.sleep(KEEPALIVE_SLEEP)
try:
if self._websocket and self._websocket.state is State.OPEN:
if self._context_id:
context_id = self.get_active_audio_context_id()
if context_id:
# Send keepalive with context ID to keep the connection alive
keepalive_message = {
"text": "",
"context_id": self._context_id,
"context_id": context_id,
}
logger.trace(f"Sending keepalive for context {self._context_id}")
logger.trace(f"Sending keepalive for context {context_id}")
else:
# It's possible to have a user interruption which clears the context
# without generating a new TTS response. In this case, we'll just send
@@ -677,24 +678,11 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
async def _send_text(self, text: str):
"""Send text to the WebSocket for synthesis."""
if self._websocket and self._context_id:
msg = {"text": text, "context_id": self._context_id}
context_id = self.get_active_audio_context_id()
if self._websocket and context_id:
msg = {"text": text, "context_id": context_id}
await self._websocket.send(json.dumps(msg))
def create_context_id(self) -> str:
"""Generate a unique context ID for a TTS request in case we don't have one already in progress.
Returns:
A unique string identifier for the TTS context.
"""
# If a context ID does not exist, create a new one.
# If an ID exists, continue using the current ID.
# When interruptions happens, user speech results in
# an interruption, which resets the context ID.
if not self._context_id:
return str(uuid.uuid4())
return self._context_id
@traced_tts
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using ElevenLabs' streaming WebSocket API.
@@ -713,19 +701,18 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await self._connect()
try:
if not self._context_id:
if not self.has_active_audio_context():
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
self._context_id = context_id
self._cumulative_time = 0
self._partial_word = ""
self._partial_word_start_time = 0.0
if not self.audio_context_available(self._context_id):
await self.create_audio_context(self._context_id)
if not self.audio_context_available(context_id):
await self.create_audio_context(context_id)
# Initialize context with voice settings and pronunciation dictionaries
msg = {"text": " ", "context_id": self._context_id}
msg = {"text": " ", "context_id": context_id}
if self._voice_settings:
msg["voice_settings"] = self._voice_settings
if self._pronunciation_dictionary_locators:
@@ -734,7 +721,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
for locator in self._pronunciation_dictionary_locators
]
await self._websocket.send(json.dumps(msg))
logger.trace(f"Created new context {self._context_id}")
logger.trace(f"Created new context {context_id}")
await self._send_text(text)
await self.start_tts_usage_metrics(text)

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

@@ -6,7 +6,6 @@
import base64
import json
import uuid
from typing import Any, AsyncGenerator, Mapping, Optional
from loguru import logger
@@ -97,7 +96,6 @@ class GradiumTTSService(AudioContextWordTTSService):
# State tracking
self._receive_task = None
self._context_id: Optional[str] = None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -129,8 +127,9 @@ class GradiumTTSService(AudioContextWordTTSService):
def _build_msg(self, text: str = "") -> dict:
"""Build JSON message for Gradium API."""
msg = {"text": text, "type": "text"}
if self._context_id:
msg["client_req_id"] = self._context_id
context_id = self.get_active_audio_context_id()
if context_id:
msg["client_req_id"] = context_id
return msg
async def start(self, frame: StartFrame):
@@ -229,6 +228,7 @@ class GradiumTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
await self.remove_active_audio_context()
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -240,12 +240,13 @@ class GradiumTTSService(AudioContextWordTTSService):
async def flush_audio(self):
"""Flush any pending audio synthesis."""
if not self._context_id or not self._websocket:
context_id = self.get_active_audio_context_id()
if not context_id or not self._websocket:
return
try:
msg = {"type": "end_of_stream", "client_req_id": self._context_id}
msg = {"type": "end_of_stream", "client_req_id": context_id}
await self._websocket.send(json.dumps(msg))
self._context_id = None
self.reset_active_audio_context()
except ConnectionClosedOK:
logger.debug(f"{self}: connection closed normally during flush")
except Exception as e:
@@ -265,7 +266,6 @@ class GradiumTTSService(AudioContextWordTTSService):
"""
await super()._handle_interruption(frame, direction)
await self.stop_all_metrics()
self._context_id = None
async def _receive_messages(self):
"""Process incoming websocket messages, demultiplexing by client_req_id."""
@@ -306,20 +306,6 @@ class GradiumTTSService(AudioContextWordTTSService):
await self.stop_all_metrics()
await self.push_error(error_msg=f"Error: {msg.get('message', msg)}")
def create_context_id(self) -> str:
"""Generate a unique context ID for a TTS request in case we don't have one already in progress.
Returns:
A unique string identifier for the TTS context.
"""
# If a context ID does not exist, create a new one.
# If an ID exists, continue using the current ID.
# When interruptions happens, user speech results in
# an interruption, which resets the context ID.
if not self._context_id:
return str(uuid.uuid4())
return self._context_id
@traced_tts
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Gradium's streaming API.
@@ -338,11 +324,10 @@ class GradiumTTSService(AudioContextWordTTSService):
await self._connect()
try:
if not self._context_id:
if not self.has_active_audio_context():
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
self._context_id = context_id
await self.create_audio_context(self._context_id)
await self.create_audio_context(context_id)
msg = self._build_msg(text=text)
await self._get_websocket().send(json.dumps(msg))

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

@@ -517,7 +517,6 @@ class InworldTTSService(AudioContextWordTTSService):
self._receive_task = None
self._keepalive_task = None
self._context_id = None
# Track cumulative time across generations for monotonic timestamps within a turn.
# When auto_mode is enabled, the server controls generations and timestamps reset
@@ -573,9 +572,10 @@ class InworldTTSService(AudioContextWordTTSService):
keeping the context open for subsequent text. The context is only
closed on interruption, disconnect, or end of session.
"""
if self._context_id and self._websocket:
logger.trace(f"Flushing audio for context {self._context_id}")
await self._send_flush(self._context_id)
context_id = self.get_active_audio_context_id()
if context_id and self._websocket:
logger.trace(f"Flushing audio for context {context_id}")
await self._send_flush(context_id)
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
"""Push a frame and handle state changes.
@@ -640,7 +640,7 @@ class InworldTTSService(AudioContextWordTTSService):
frame: The interruption frame.
direction: The direction of the interruption.
"""
old_context_id = self._context_id
old_context_id = self.get_active_audio_context_id()
logger.trace(f"{self}: Handling interruption, old context: {old_context_id}")
await super()._handle_interruption(frame, direction)
@@ -652,7 +652,6 @@ class InworldTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
self._context_id = None
self._cumulative_time = 0.0
self._generation_end_time = 0.0
logger.trace(f"{self}: Interruption handled, context reset to None")
@@ -736,9 +735,10 @@ class InworldTTSService(AudioContextWordTTSService):
if self._websocket:
logger.debug("Disconnecting from Inworld WebSocket TTS")
if self._context_id:
context_id = self.get_active_audio_context_id()
if context_id:
try:
await self._send_close_context(self._context_id)
await self._send_close_context(context_id)
except Exception:
pass
await self._websocket.close()
@@ -746,7 +746,7 @@ class InworldTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
finally:
self._context_id = None
await self.remove_active_audio_context()
self._websocket = None
self._cumulative_time = 0.0
self._generation_end_time = 0.0
@@ -772,7 +772,7 @@ class InworldTTSService(AudioContextWordTTSService):
]
logger.debug(
f"{self}: Received message types={msg_types}, ctx_id={ctx_id}, "
f"current_ctx={self._context_id}, available={self.audio_context_available(ctx_id) if ctx_id else 'N/A'}"
f"current_ctx={self.get_active_audio_context_id()}, available={self.audio_context_available(ctx_id) if ctx_id else 'N/A'}"
)
# Check for errors
@@ -784,7 +784,9 @@ class InworldTTSService(AudioContextWordTTSService):
# Handle "Context not found" error (code 5)
# This can happen when a keepalive message is sent but no context is available.
if error_code == 5 and "not found" in error_msg.lower():
logger.debug(f"{self}: Context {ctx_id or self._context_id} not found.")
logger.debug(
f"{self}: Context {ctx_id or self.get_active_audio_context_id()} not found."
)
continue
# For other errors, push error frame
@@ -799,11 +801,9 @@ class InworldTTSService(AudioContextWordTTSService):
# If the context isn't available but matches our current context ID,
# recreate it (handles race conditions during interruption recovery).
if ctx_id and not self.audio_context_available(ctx_id):
if self._context_id == ctx_id:
logger.trace(
f"{self}: Recreating audio context for current context: {self._context_id}"
)
await self.create_audio_context(self._context_id)
if self.get_active_audio_context_id() == ctx_id:
logger.trace(f"{self}: Recreating audio context for current context: {ctx_id}")
await self.create_audio_context(ctx_id)
else:
# This is a message from an old/closed context - skip it
logger.trace(f"{self}: Skipping message from unavailable context: {ctx_id}")
@@ -849,8 +849,8 @@ class InworldTTSService(AudioContextWordTTSService):
logger.trace(f"{self}: Context closed on server: {ctx_id}")
await self.stop_ttfb_metrics()
# Only reset if this is our current context
if ctx_id == self._context_id:
self._context_id = None
if ctx_id == self.get_active_audio_context_id():
self.reset_active_audio_context()
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)], ctx_id)
@@ -862,12 +862,13 @@ class InworldTTSService(AudioContextWordTTSService):
await asyncio.sleep(KEEPALIVE_SLEEP)
try:
if self._websocket and self._websocket.state is State.OPEN:
if self._context_id:
context_id = self.get_active_audio_context_id()
if context_id:
keepalive_message = {
"send_text": {"text": ""},
"contextId": self._context_id,
"contextId": context_id,
}
logger.trace(f"Sending keepalive for context {self._context_id}")
logger.trace(f"Sending keepalive for context {context_id}")
else:
keepalive_message = {"send_text": {"text": ""}}
logger.trace("Sending keepalive without context")
@@ -938,20 +939,6 @@ class InworldTTSService(AudioContextWordTTSService):
msg = {"close_context": {}, "contextId": context_id}
await self.send_with_retry(json.dumps(msg), self._report_error)
def create_context_id(self) -> str:
"""Generate a unique context ID for a TTS request in case we don't have one already in progress.
Returns:
A unique string identifier for the TTS context.
"""
# If a context ID does not exist, create a new one.
# If an ID exists, continue using the current ID.
# When interruptions happen, user speech results in
# an interruption, which resets the context ID.
if not self._context_id:
return str(uuid.uuid4())
return self._context_id
@traced_tts
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.
@@ -970,19 +957,13 @@ class InworldTTSService(AudioContextWordTTSService):
await self._connect()
try:
if not self._context_id:
if not self.has_active_audio_context():
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=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.create_audio_context(context_id)
await self._send_context(context_id)
await self._send_text(self._context_id, text)
await self._send_text(context_id, text)
await self.start_tts_usage_metrics(text)
except Exception as e:

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

@@ -25,8 +25,6 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import AudioContextWordTTSService
from pipecat.transcriptions.language import Language
from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.tracing.service_decorators import traced_tts
try:
@@ -70,6 +68,7 @@ class ResembleAITTSService(AudioContextWordTTSService):
"""
super().__init__(
sample_rate=sample_rate,
reuse_context_id_within_turn=False,
**kwargs,
)

36
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
@@ -167,7 +166,6 @@ class RimeTTSService(AudioContextWordTTSService):
self._settings = self._build_settings()
# State tracking
self._context_id = None # Tracks current turn
self._receive_task = None
self._cumulative_time = 0 # Accumulates time across messages
self._extra_msg_fields = {} # Extra fields for next message
@@ -339,7 +337,7 @@ class RimeTTSService(AudioContextWordTTSService):
def _build_msg(self, text: str = "") -> dict:
"""Build JSON message for Rime API."""
msg = {"text": text, "contextId": self._context_id}
msg = {"text": text, "contextId": self.get_active_audio_context_id()}
if self._extra_msg_fields:
msg |= self._extra_msg_fields
self._extra_msg_fields = {}
@@ -427,7 +425,7 @@ class RimeTTSService(AudioContextWordTTSService):
except Exception as e:
await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e)
finally:
self._context_id = None
await self.remove_active_audio_context()
self._websocket = None
await self._call_event_handler("on_disconnected")
@@ -439,11 +437,11 @@ class RimeTTSService(AudioContextWordTTSService):
async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
"""Handle interruption by clearing current context."""
context_id = self.get_active_audio_context_id()
await super()._handle_interruption(frame, direction)
await self.stop_all_metrics()
if self._context_id:
if context_id:
await self._get_websocket().send(json.dumps(self._build_clear_msg()))
self._context_id = None
def _calculate_word_times(self, words: list, starts: list, ends: list) -> list:
"""Calculate word timing pairs with proper spacing and punctuation.
@@ -474,28 +472,15 @@ class RimeTTSService(AudioContextWordTTSService):
return word_pairs
def create_context_id(self) -> str:
"""Generate a unique context ID for a TTS request in case we don't have one already in progress.
Returns:
A unique string identifier for the TTS context.
"""
# If a context ID does not exist, create a new one.
# If an ID exists, continue using the current ID.
# When interruptions happen, user speech results in
# an interruption, which resets the context ID.
if not self._context_id:
return str(uuid.uuid4())
return self._context_id
async def flush_audio(self):
"""Flush any pending audio synthesis."""
if not self._context_id or not self._websocket:
context_id = self.get_active_audio_context_id()
if not context_id or not self._websocket:
return
logger.trace(f"{self}: flushing audio")
await self._get_websocket().send(json.dumps({"operation": "flush"}))
self._context_id = None
self.reset_active_audio_context()
async def _receive_messages(self):
"""Process incoming websocket messages."""
@@ -537,7 +522,7 @@ class RimeTTSService(AudioContextWordTTSService):
await self.push_frame(TTSStoppedFrame())
await self.stop_all_metrics()
await self.push_error(error_msg=f"Error: {msg['message']}")
self._context_id = None
self.reset_active_audio_context()
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
"""Push frame and handle end-of-turn conditions.
@@ -568,12 +553,11 @@ class RimeTTSService(AudioContextWordTTSService):
await self._connect()
try:
if not self._context_id:
if not self.has_active_audio_context():
await self.start_ttfb_metrics()
yield TTSStartedFrame(context_id=context_id)
self._cumulative_time = 0
self._context_id = context_id
await self.create_audio_context(self._context_id)
await self.create_audio_context(context_id)
msg = self._build_msg(text=text)
await self._get_websocket().send(json.dumps(msg))

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

@@ -1042,14 +1042,23 @@ class AudioContextTTSService(WebsocketTTSService):
audio from context ID "A" will be played first.
"""
def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
def __init__(
self,
*,
reuse_context_id_within_turn: bool = True,
reconnect_on_error: bool = True,
**kwargs,
):
"""Initialize the Audio Context TTS service.
Args:
reuse_context_id_within_turn: Whether the service should reuse context IDs within the same turn.
reconnect_on_error: Whether to automatically reconnect on websocket errors.
**kwargs: Additional arguments passed to the parent WebsocketTTSService.
"""
super().__init__(reconnect_on_error=reconnect_on_error, **kwargs)
self._reuse_context_id_within_turn = reuse_context_id_within_turn
self._context_id = None
self._contexts: Dict[str, asyncio.Queue] = {}
self._audio_context_task = None
@@ -1059,6 +1068,10 @@ class AudioContextTTSService(WebsocketTTSService):
Args:
context_id: Unique identifier for the audio context.
"""
# Set the context ID if not already set
if not self._context_id:
self._context_id = context_id
await self._contexts_queue.put(context_id)
self._contexts[context_id] = asyncio.Queue()
logger.trace(f"{self} created audio context {context_id}")
@@ -1091,6 +1104,32 @@ class AudioContextTTSService(WebsocketTTSService):
else:
logger.warning(f"{self} unable to remove context {context_id}")
def has_active_audio_context(self) -> bool:
"""Check if there is an active audio context.
Returns:
True if an active audio context exists, False otherwise.
"""
return self._context_id is not None and self.audio_context_available(self._context_id)
def get_active_audio_context_id(self) -> Optional[str]:
"""Get the active audio context ID.
Returns:
The active context ID, or None if no context is active.
"""
return self._context_id
async def remove_active_audio_context(self):
"""Remove the active audio context."""
if self._context_id:
await self.remove_audio_context(self._context_id)
self.reset_active_audio_context()
def reset_active_audio_context(self):
"""Reset the active audio context."""
self._context_id = None
def audio_context_available(self, context_id: str) -> bool:
"""Check whether the given audio context is registered.
@@ -1102,6 +1141,20 @@ class AudioContextTTSService(WebsocketTTSService):
"""
return context_id in self._contexts
def create_context_id(self) -> str:
"""Generate or reuse a context ID based on concurrent TTS support.
If _reuse_context_id_within_turn is False and a context already exists,
the existing context ID is returned. Otherwise, a new unique context
ID is generated.
Returns:
A context ID string for the TTS request.
"""
if self._reuse_context_id_within_turn and self._context_id:
return self._context_id
return super().create_context_id()
async def start(self, frame: StartFrame):
"""Start the audio context TTS service.
@@ -1137,6 +1190,7 @@ class AudioContextTTSService(WebsocketTTSService):
async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
await super()._handle_interruption(frame, direction)
await self._stop_audio_context_task()
self.reset_active_audio_context()
self._create_audio_context_task()
def _create_audio_context_task(self):
@@ -1155,6 +1209,7 @@ class AudioContextTTSService(WebsocketTTSService):
running = True
while running:
context_id = await self._contexts_queue.get()
self._context_id = context_id
if context_id:
# Process the audio context until the context doesn't have more
@@ -1163,11 +1218,15 @@ class AudioContextTTSService(WebsocketTTSService):
# We just finished processing the context, so we can safely remove it.
del self._contexts[context_id]
self.reset_active_audio_context()
# Append some silence between sentences.
silence = b"\x00" * self.sample_rate
frame = TTSAudioRawFrame(
audio=silence, sample_rate=self.sample_rate, num_channels=1
audio=silence,
sample_rate=self.sample_rate,
num_channels=1,
context_id=context_id,
)
await self.push_frame(frame)
else: