diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2df70ebc7..4b558d0f1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,14 +11,248 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Added `LiveKitRESTHelper` utility class for managing LiveKit rooms via REST API.
+- Added `DeepgramSageMakerSTTService` which connects to a SageMaker hosted
+ Deepgram STT model. Added `07c-interruptible-deepgram-sagemaker.py`
+ foundational example.
+
+- Added `SageMakerBidiClient` to connect to SageMaker hosted BiDi compatible
+ services.
+
+- Added support for `include_timestamps` and `enable_logging` in
+ `ElevenLabsRealtimeSTTService`. When `include_timestamps` is enabled,
+ timestamp data is included in the `TranscriptionFrame`'s `result`
+ parameter.
+
- Added optional speaking rate control to `InworldTTSService`.
+- Introduced a new `AggregatedTextFrame` type to support passing text along with
+ an `aggregated_by` field to describe the type of text
+ included. `TTSTextFrame`s now inherit from `AggregatedTextFrame`. With this
+ inheritance, an observer can watch for `AggregatedTextFrame`s to accumlate the
+ perceived output and determine whether or not the text was spoken based on if
+ that frame is also a `TTSTextFrame`.
+
+ With this frame, the llm token stream can be transformed into custom
+ composable chunks, allowing for aggregation outside the TTS service. This
+ makes it possible to listen for or handle those aggregations and sets the
+ stage for doing things like composing a best effort of the perceived llm
+ output in a more digestable form and to do so whether or not it is processed
+ by a TTS or if even a TTS exists.
+
+- Introduced `LLMTextProcessor`: A new processor meant to allow customization
+ for how LLMTextFrames should be aggregated and considered. It's purpose is to
+ turn `LLMTextFrame`s into `AggregatedTextFrame`s. By default, a TTSService
+ will still aggregate `LLMTextFrame`s by sentence for the service to
+ consume. However, if you wish to override how the llm text is aggregated, you
+ should no longer override the TTS's internal text_aggregator, but instead,
+ insert this processor between your LLM and TTS in the pipeline.
+
+- New `bot-output` RTVI message to represent what the bot actually "says".
+
+ - The `RTVIObserver` now emits `bot-output` messages based off the new
+ `AggregatedTextFrame`s (`bot-tts-text` and `bot-llm-text` are still
+ supported and generated, but `bot-transcript` is now deprecated in lieu of
+ this new, more thorough, message).
+
+ - The new `RTVIBotOutputMessage` includes the fields:
+
+ - `spoken`: A boolean indicating whether the text was spoken by TTS
+
+ - `aggregated_by`: A string representing how the text was aggregated
+ ("sentence", "word", "my custom aggregation")
+
+ - Introduced new fields to `RTVIObserver` to support the new `bot-output`
+ messaging:
+
+ - `bot_output_enabled`: Defaults to True. Set to false to disable bot-output
+ messages.
+
+ - `skip_aggregator_types`: Defaults to `None`. Set to a list of strings that
+ match aggregation types that should not be included in bot-output
+ messages. (Ex. `credit_card`)
+
+ - Introduced new methods, `add_text_transformer()` and
+ `remove_text_transformer()`, to `RTVIObserver` to support providing (and
+ subsequently removing) callbacks for various types of aggregations (or all
+ aggregations with `*`) that can modify the text before being sent as a
+ `bot-output` or `tts-text` message. (Think obscuring the credit card or
+ inserting extra detail the client might want that the context doesn't need.)
+
+- In `MiniMaxHttpTTSService`:
+
+ - Added support for speech-2.6-hd and speech-2.6-turbo models
+
+ - Added languages: Afrikaans, Bulgarian, Catalan, Danish, Persian, Filipino,
+ Hebrew, Croatian, Hungarian, Malay, Norwegian, Nynorsk, Slovak, Slovenian,
+ Swedish, and Tamil
+
+ - Added new emotions: calm and fluent
+
### Changed
- Updated `daily-python` to 0.22.0.
+- `BaseTextAggregator` changes:
+
+ Modified the BaseTextAggregator type so that when text gets aggregated,
+ metadata can be associated with it. Currently, that just means a `type`, so
+ that the aggregation can be classified or described. Changes made to support
+ this:
+
+ - ⚠️ IMPORTANT: Aggregators are now expected to strip leading/trailing white
+ space characters before returning their aggregation from `aggregation()` or
+ `.text`. This way all aggregators have a consistent contract allowing
+ downstream use to know how to stitch aggregations back together.
+
+ - Introduced a new `Aggregation` dataclass to represent both the aggregated
+ `text` and a string identifying the `type` of aggregation (ex. "sentence",
+ "word", "my custom aggregation")
+
+ - ⚠️ Breaking change: `BaseTextAggregator.text` now returns an `Aggregation`
+ (instead of `str`).
+
+ Before:
+
+ ```python
+ aggregated_text = myAggregator.text
+ ```
+
+ Now:
+
+ ```python
+ aggregated_text = myAggregator.text.text
+ ```
+
+ - ⚠️ Breaking change: `BaseTextAggregator.aggregate()` now returns
+ `Optional[Aggregation]` (instead of `Optional[str]`).
+
+ Before:
+
+ ```python
+ aggregation = myAggregator.aggregate(text)
+ print(f"successfully aggregated text: {aggregation}")
+ ```
+
+ Now:
+
+ ```python
+ aggregation = myAggregator.aggregate(text)
+ if aggregation:
+ print(f"successfully aggregated text: {aggregation.text}")
+ ```
+
+ - `SimpleTextAggregator`, `SkipTagsAggregator`, `PatternPairAggregator`
+ updated to produce/consume `Aggregation` objects.
+
+ - All uses of the above Aggregators have been updated accordingly.
+
+- Augmented the `PatternPairAggregator` so that matched patterns can be treated
+ as their own aggregation, taking advantage of the new. To that end:
+
+ - Introduced a new, preferred version of `add_pattern` to support a new option
+ for treating a match as a separate aggregation returned from
+ `aggregate()`. This replaces the now deprecated `add_pattern_pair` method
+ and you provide a `MatchAction` in lieu of the `remove_match` field.
+
+ - `MatchAction` enum: `REMOVE`, `KEEP`, `AGGREGATE`, allowing customization
+ for how a match should be handled.
+
+ - `REMOVE`: The text along with its delimiters will be removed from the
+ streaming text. Sentence aggregation will continue on as if this text
+ did not exist.
+
+ - `KEEP`: The delimiters will be removed, but the content between them
+ will be kept. Sentence aggregation will continue on with the internal
+ text included.
+
+ - `AGGREGATE`: The delimiters will be removed and the content between will
+ be treated as a separate aggregation. Any text before the start of the
+ pattern will be returned early, whether or not a complete sentence was
+ found. Then the pattern will be returned. Then the aggregation will
+ continue on sentence matching after the closing delimiter is found. The
+ content between the delimiters is not aggregated by sentence. It is
+ aggregated as one single block of text.
+
+ - `PatternMatch` now extends `Aggregation` and provides richer info to
+ handlers.
+
+ - ⚠️ Breaking change: The `PatternMatch` type returned to handlers registered
+ via `on_pattern_match` has been updated to subclass from the new
+ `Aggregation` type, which means that `content` has been replaced with
+ `text` and `pattern_id` has been replaced with `type`:
+
+ ```python
+ async dev on_match_tag(match: PatternMatch):
+ pattern = match.type # instead of match.pattern_id
+ text = match.text # instead of match.content
+ ```
+
+- `TextFrame` now includes the field `append_to_context` to support setting
+ whether or not the encompassing text should be added to the LLM context (by
+ the LLM assistant aggregator). It defaults to `True`.
+
+- `TTSService` base class updates:
+
+ - `TTSService`s now accept a new `skip_aggregator_types` to avoid speaking
+ certain aggregation types (now determined/returned by the aggregator)
+
+ - Introduced the ability to do a just-in-time transform of text before it gets
+ sent to the TTS service via callbacks you can set up via a new init field,
+ `text_transforms` or a new method `add_text_transformer()`. This makes it
+ possible to do things like introduce TTS-specific tags for spelling or
+ emotion or change the pronunciation of something on the
+ fly. `remove_text_transformer` has also been added to support removing a
+ registered transform callback.
+
+ - TTS services push `AggregatedTextFrame` in addition to `TTSTextFrame`s when
+ either an aggregation occurs that should not be spoken or when the TTS
+ service supports word-by-word timestamping. In the latter case, the
+ `TTSService` preliminarily generates an `AggregatedTextFrame`, aggregated by
+ sentence to generate the full sentence content as early as possible.
+
+- Updated `CartesiaTTSService`:
+
+ - Modified use of custom default text_aggregator to avoid deprecation warnings
+ and push users towards use of transformers or the `LLMTextProcessor`
+
+ - Added convenience methods for taking advantage of Cartesia's SSML tags:
+ spell, emotion, pauses, volume, and speed.
+
+- Updated `RimeTTSService`:
+
+ - Modified use of custom default text_aggregator to avoid deprecation warnings
+ and push users towards use of transformers or the `LLMTextProcessor`
+
+ - Added convenience methods for taking advantage of Rime's customization
+ options: spell, pauses, pronunciations, and inline speed control.
+
+### Deprecated
+
+- The TTS constructor field, `text_aggregator` is deprecated in favor of the new
+ `LLMTextProcessor`. TTSServices still have an internal aggregator for support
+ of default behavior, but if you want to override the aggregation behavior, you
+ should use the new processor.
+
+- The RTVI `bot-transcription` event is deprecated in favor of the new
+ `bot-output` message which is the canonical representation of bot output
+ (spoken or not). The code still emits a transcription message for backwards
+ compatibility while transition occurs.
+
+- Deprecated `add_pattern_pair` in the `PatternPairAggregator` which takes a
+ `pattern_id` and `remove_match` field in favor of the new `add_pattern` method
+ which takes a `type` and an `action`
+
+- `english_normalization` input parameter for `MiniMaxHttpTTSService` is
+ deprecated, use `test_normalization` instead.
+
### Fixed
+- Fixed an issue in `ElevenLabsRealtimeSTTService` where dynamic language
+ updates were not working.
+
+- Fixed an issue in `ElevenLabsRealtimeSTTService` where setting the sample
+ rate would result in transcripts failing.
+
- Fixed `InworldTTSService` audio config payload to use camelCase keys expected
by the Inworld API.
@@ -80,20 +314,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Updated language mappings for the Google and Gemini TTS services to match
official documentation.
-- In `MiniMaxHttpTTSService`:
--- Added support for speech-2.6-hd and speech-2.6-turbo models
--- Added languages: Afrikaans, Bulgarian, Catalan, Danish, Persian, Filipino, Hebrew,
-Croatian, Hungarian, Malay, Norwegian, Nynorsk, Slovak, Slovenian, Swedish, and Tamil
--- Added new emotions: calm and fluent
-
### Deprecated
- The `api_key` parameter in `GeminiTTSService` is deprecated. Use
`credentials` or `credentials_path` instead for Google Cloud authentication.
-- `english_normalization` input parameter for `MiniMaxHttpTTSService` is deprecated,
-use `test_normalization` instead.
-
### Fixed
- Fixed a `SimliVideoService` connection issue.
diff --git a/env.example b/env.example
index 2865772ea..33c699259 100644
--- a/env.example
+++ b/env.example
@@ -44,6 +44,7 @@ DAILY_SAMPLE_ROOM_URL=https://...
# Deepgram
DEEPGRAM_API_KEY=...
+SAGEMAKER_ENDPOINT_NAME=...
# DeepSeek
DEEPSEEK_API_KEY=...
diff --git a/examples/foundational/07c-interruptible-deepgram-sagemaker.py b/examples/foundational/07c-interruptible-deepgram-sagemaker.py
new file mode 100644
index 000000000..db230a8ba
--- /dev/null
+++ b/examples/foundational/07c-interruptible-deepgram-sagemaker.py
@@ -0,0 +1,137 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.aws.llm import AWSBedrockLLMService
+from pipecat.services.deepgram.stt_sagemaker import DeepgramSageMakerSTTService
+from pipecat.services.deepgram.tts import DeepgramTTSService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+ "daily": lambda: DailyParams(
+ audio_in_enabled=True,
+ audio_out_enabled=True,
+ vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+ turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+ ),
+ "twilio": lambda: FastAPIWebsocketParams(
+ audio_in_enabled=True,
+ audio_out_enabled=True,
+ vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+ turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+ ),
+ "webrtc": lambda: TransportParams(
+ audio_in_enabled=True,
+ audio_out_enabled=True,
+ vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+ turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+ ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+ logger.info(f"Starting bot")
+
+ # Initialize Deepgram SageMaker STT Service
+ # This requires:
+ # - AWS credentials configured (via environment variables or AWS CLI)
+ # - A deployed SageMaker endpoint with Deepgram model
+ stt = DeepgramSageMakerSTTService(
+ endpoint_name=os.getenv("SAGEMAKER_ENDPOINT_NAME"),
+ region=os.getenv("AWS_REGION"),
+ )
+
+ tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-2-andromeda-en")
+
+ llm = AWSBedrockLLMService(
+ aws_region=os.getenv("AWS_REGION"),
+ model="us.amazon.nova-pro-v1:0",
+ params=AWSBedrockLLMService.InputParams(temperature=0.8),
+ )
+
+ messages = [
+ {
+ "role": "system",
+ "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
+ },
+ ]
+
+ context = LLMContext(messages)
+ context_aggregator = LLMContextAggregatorPair(context)
+
+ pipeline = Pipeline(
+ [
+ transport.input(), # Transport user input
+ stt, # STT
+ context_aggregator.user(), # User responses
+ llm, # LLM
+ tts, # TTS
+ transport.output(), # Transport bot output
+ context_aggregator.assistant(), # Assistant spoken responses
+ ]
+ )
+
+ task = PipelineTask(
+ pipeline,
+ params=PipelineParams(
+ enable_metrics=True,
+ enable_usage_metrics=True,
+ ),
+ idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+ )
+
+ @transport.event_handler("on_client_connected")
+ async def on_client_connected(transport, client):
+ logger.info(f"Client connected")
+ # Kick off the conversation.
+ messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+ await task.queue_frames([LLMRunFrame()])
+
+ @transport.event_handler("on_client_disconnected")
+ async def on_client_disconnected(transport, client):
+ logger.info(f"Client disconnected")
+ await task.cancel()
+
+ runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+ await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+ """Main bot entry point compatible with Pipecat Cloud."""
+ transport = await create_transport(runner_args, transport_params)
+ await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+ from pipecat.runner.run import main
+
+ main()
diff --git a/examples/foundational/35-pattern-pair-voice-switching.py b/examples/foundational/35-pattern-pair-voice-switching.py
index 7ed9eb268..3a102acfd 100644
--- a/examples/foundational/35-pattern-pair-voice-switching.py
+++ b/examples/foundational/35-pattern-pair-voice-switching.py
@@ -62,7 +62,11 @@ from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
-from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPairAggregator
+from pipecat.utils.text.pattern_pair_aggregator import (
+ MatchAction,
+ PatternMatch,
+ PatternPairAggregator,
+)
load_dotenv(override=True)
@@ -106,16 +110,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
pattern_aggregator = PatternPairAggregator()
# Add pattern for voice switching
- pattern_aggregator.add_pattern_pair(
- pattern_id="voice_tag",
+ pattern_aggregator.add_pattern(
+ type="voice",
start_pattern="",
end_pattern="",
- remove_match=True,
+ action=MatchAction.REMOVE, # Remove tags from final text
)
# Register handler for voice switching
async def on_voice_tag(match: PatternMatch):
- voice_name = match.content.strip().lower()
+ voice_name = match.text.strip().lower()
if voice_name in VOICE_IDS:
# First flush any existing audio to finish the current context
await tts.flush_audio()
@@ -125,7 +129,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
else:
logger.warning(f"Unknown voice: {voice_name}")
- pattern_aggregator.on_pattern_match("voice_tag", on_voice_tag)
+ pattern_aggregator.on_pattern_match("voice", on_voice_tag)
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
diff --git a/pyproject.toml b/pyproject.toml
index ff1452f11..eca73ef80 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -50,13 +50,13 @@ anthropic = [ "anthropic~=0.49.0" ]
assemblyai = [ "pipecat-ai[websockets-base]" ]
asyncai = [ "pipecat-ai[websockets-base]" ]
aws = [ "aioboto3~=15.0.0", "pipecat-ai[websockets-base]" ]
-aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.1.1; python_version>='3.12'" ]
+aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.2.0; python_version>='3.12'" ]
azure = [ "azure-cognitiveservices-speech~=1.42.0"]
cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ]
cerebras = []
-deepseek = []
daily = [ "daily-python~=0.22.0" ]
deepgram = [ "deepgram-sdk~=4.7.0" ]
+deepseek = []
elevenlabs = [ "pipecat-ai[websockets-base]" ]
fal = [ "fal-client~=0.5.9" ]
fireworks = []
@@ -69,19 +69,21 @@ gstreamer = [ "pygobject~=3.50.0" ]
heygen = [ "livekit>=1.0.13", "pipecat-ai[websockets-base]" ]
hume = [ "hume>=0.11.2" ]
inworld = []
-krisp = [ "pipecat-ai-krisp~=0.4.0" ]
koala = [ "pvkoala~=2.0.3" ]
+krisp = [ "pipecat-ai-krisp~=0.4.0" ]
langchain = [ "langchain~=0.3.20", "langchain-community~=0.3.20", "langchain-openai~=0.3.9" ]
livekit = [ "livekit~=1.0.13", "livekit-api~=1.0.5", "tenacity>=8.2.3,<10.0.0", "pyjwt>=2.10.1" ]
lmnt = [ "pipecat-ai[websockets-base]" ]
local = [ "pyaudio~=0.2.14" ]
+local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ]
+local-smart-turn-v3 = [ "transformers", "onnxruntime>=1.20.1,<2" ]
mcp = [ "mcp[cli]>=1.11.0,<2" ]
mem0 = [ "mem0ai~=0.1.94" ]
mistral = []
mlx-whisper = [ "mlx-whisper~=0.4.2" ]
moondream = [ "accelerate~=1.10.0", "einops~=0.8.0", "pyvips[binary]~=3.0.0", "timm~=1.0.13", "transformers>=4.48.0" ]
-nim = []
neuphonic = [ "pipecat-ai[websockets-base]" ]
+nim = []
noisereduce = [ "noisereduce~=3.0.3" ]
openai = [ "pipecat-ai[websockets-base]" ]
openpipe = [ "openpipe>=4.50.0,<6" ]
@@ -89,15 +91,14 @@ openrouter = []
perplexity = []
playht = [ "pipecat-ai[websockets-base]" ]
qwen = []
+remote-smart-turn = []
rime = [ "pipecat-ai[websockets-base]" ]
riva = [ "nvidia-riva-client~=2.21.1" ]
runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<0.122.0", "pipecat-ai-small-webrtc-prebuilt>=1.0.0"]
+sagemaker = ["aws_sdk_sagemaker_runtime_http2; python_version>='3.12'"]
sambanova = []
sarvam = [ "sarvamai==0.1.21", "pipecat-ai[websockets-base]" ]
sentry = [ "sentry-sdk>=2.28.0,<3" ]
-local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ]
-local-smart-turn-v3 = [ "transformers", "onnxruntime>=1.20.1,<2" ]
-remote-smart-turn = []
silero = [ "onnxruntime>=1.20.1,<2" ]
simli = [ "simli-ai~=1.0.3"]
soniox = [ "pipecat-ai[websockets-base]" ]
diff --git a/src/pipecat/extensions/ivr/ivr_navigator.py b/src/pipecat/extensions/ivr/ivr_navigator.py
index 05748d94f..1ddb41ed8 100644
--- a/src/pipecat/extensions/ivr/ivr_navigator.py
+++ b/src/pipecat/extensions/ivr/ivr_navigator.py
@@ -31,7 +31,11 @@ from pipecat.pipeline.pipeline import Pipeline
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.llm_service import LLMService
-from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPairAggregator
+from pipecat.utils.text.pattern_pair_aggregator import (
+ MatchAction,
+ PatternMatch,
+ PatternPairAggregator,
+)
class IVRStatus(Enum):
@@ -114,15 +118,15 @@ class IVRProcessor(FrameProcessor):
def _setup_xml_patterns(self):
"""Set up XML pattern detection and handlers."""
# Register DTMF pattern
- self._aggregator.add_pattern_pair("dtmf", "", "", remove_match=True)
+ self._aggregator.add_pattern("dtmf", "", "", action=MatchAction.REMOVE)
self._aggregator.on_pattern_match("dtmf", self._handle_dtmf_action)
# Register mode pattern
- self._aggregator.add_pattern_pair("mode", "", "", remove_match=True)
+ self._aggregator.add_pattern("mode", "", "", action=MatchAction.REMOVE)
self._aggregator.on_pattern_match("mode", self._handle_mode_action)
# Register IVR pattern
- self._aggregator.add_pattern_pair("ivr", "", "", remove_match=True)
+ self._aggregator.add_pattern("ivr", "", "", action=MatchAction.REMOVE)
self._aggregator.on_pattern_match("ivr", self._handle_ivr_action)
async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -148,7 +152,7 @@ class IVRProcessor(FrameProcessor):
result = await self._aggregator.aggregate(frame.text)
if result:
# Push aggregated text that doesn't contain XML patterns
- await self.push_frame(LLMTextFrame(result), direction)
+ await self.push_frame(LLMTextFrame(result.text), direction)
else:
await self.push_frame(frame, direction)
@@ -159,7 +163,7 @@ class IVRProcessor(FrameProcessor):
Args:
match: The pattern match containing DTMF content.
"""
- value = match.content
+ value = match.text
logger.debug(f"DTMF detected: {value}")
try:
@@ -180,7 +184,7 @@ class IVRProcessor(FrameProcessor):
Args:
match: The pattern match containing IVR status content.
"""
- status = match.content
+ status = match.text
logger.trace(f"IVR status detected: {status}")
# Convert string to enum, with validation
@@ -211,7 +215,7 @@ class IVRProcessor(FrameProcessor):
Args:
match: The pattern match containing mode content.
"""
- mode = match.content
+ mode = match.text
logger.debug(f"Mode detected: {mode}")
if mode == "conversation":
await self._handle_conversation()
diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py
index ddb4e5a14..e437d48a1 100644
--- a/src/pipecat/frames/frames.py
+++ b/src/pipecat/frames/frames.py
@@ -12,6 +12,7 @@ and LLM processing.
"""
from dataclasses import dataclass, field
+from enum import Enum
from typing import (
TYPE_CHECKING,
Any,
@@ -337,11 +338,14 @@ class TextFrame(DataFrame):
# mandatory fields of theirs to have defaults to preserve
# non-default-before-default argument order)
includes_inter_frame_spaces: bool = field(init=False)
+ # Whether this text frame should be appended to the LLM context.
+ append_to_context: bool = field(init=False)
def __post_init__(self):
super().__post_init__()
self.skip_tts = False
self.includes_inter_frame_spaces = False
+ self.append_to_context = True
def __str__(self):
pts = format_pts(self.pts)
@@ -358,8 +362,32 @@ class LLMTextFrame(TextFrame):
self.includes_inter_frame_spaces = True
+class AggregationType(str, Enum):
+ """Built-in aggregation strings."""
+
+ SENTENCE = "sentence"
+ WORD = "word"
+
+ def __str__(self):
+ return self.value
+
+
@dataclass
-class TTSTextFrame(TextFrame):
+class AggregatedTextFrame(TextFrame):
+ """Text frame representing an aggregation of TextFrames.
+
+ This frame contains multiple TextFrames aggregated together for processing
+ or output along with a field to indicate how they are aggregated.
+
+ Parameters:
+ aggregated_by: Method used to aggregate the text frames.
+ """
+
+ aggregated_by: AggregationType | str
+
+
+@dataclass
+class TTSTextFrame(AggregatedTextFrame):
"""Text frame generated by Text-to-Speech services."""
pass
diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py
index afc091d5a..ec13b643f 100644
--- a/src/pipecat/processors/aggregators/llm_response.py
+++ b/src/pipecat/processors/aggregators/llm_response.py
@@ -1001,7 +1001,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
await self.push_aggregation()
async def _handle_text(self, frame: TextFrame):
- if not self._started:
+ if not self._started or not frame.append_to_context:
return
if self._params.expect_stripped_words:
diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py
index 87974bed2..69fc649ce 100644
--- a/src/pipecat/processors/aggregators/llm_response_universal.py
+++ b/src/pipecat/processors/aggregators/llm_response_universal.py
@@ -811,7 +811,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self.push_aggregation()
async def _handle_text(self, frame: TextFrame):
- if not self._started:
+ if not self._started or not frame.append_to_context:
return
# Make sure we really have text (spaces count, too!)
diff --git a/src/pipecat/processors/aggregators/llm_text_processor.py b/src/pipecat/processors/aggregators/llm_text_processor.py
new file mode 100644
index 000000000..44a8dc24e
--- /dev/null
+++ b/src/pipecat/processors/aggregators/llm_text_processor.py
@@ -0,0 +1,106 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""LLM text processor module for processing and aggregating raw LLM output text.
+
+This processor will convert LLMTextFrames into AggregatedTextFrames based on the
+configured text aggregator. Using the customizable aggregator, it provides
+functionality to handle or manipulate LLM text frames before they are sent to other
+components such as TTS services or context aggregators. It can be used to pre-aggregate
+and categorize, modify, or filter direct output tokens from the LLM.
+"""
+
+from typing import Optional
+
+from pipecat.frames.frames import (
+ AggregatedTextFrame,
+ EndFrame,
+ Frame,
+ InterruptionFrame,
+ LLMFullResponseEndFrame,
+ LLMTextFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
+
+
+class LLMTextProcessor(FrameProcessor):
+ """A processor for handling or manipulating LLM text frames before they are processed further.
+
+ This processor will convert LLMTextFrames into AggregatedTextFrames based on the configured
+ text aggregator. Using the customizable aggregator, it provides functionality to handle or
+ manipulate LLM text frames before they are sent to other components such as TTS services or
+ context aggregators. It can be used to pre-aggregate and categorize, modify, or filter direct
+ output tokens from the LLM.
+ """
+
+ def __init__(self, *, text_aggregator: Optional[BaseTextAggregator] = None, **kwargs):
+ """Initialize the LLM text processor.
+
+ Args:
+ text_aggregator: An optional text aggregator to use for processing LLM text frames. By
+ default, a SimpleTextAggregator aggregating by sentence will be used.
+ **kwargs: Additional arguments passed to parent class.
+
+ TODO: Allow transformations per aggregation type or all (and deprecate the TTS filters).
+ """
+ super().__init__(**kwargs)
+ self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator()
+
+ async def process_frame(self, frame: Frame, direction: FrameDirection):
+ """Process an LLMTextFrames using the aggregator to generate AggregatedTextFrames.
+
+ Args:
+ frame: The frame to process.
+ direction: The direction of frame flow in the pipeline.
+ """
+ await super().process_frame(frame, direction)
+
+ if isinstance(frame, InterruptionFrame):
+ await self._handle_interruption(frame)
+ await self.push_frame(frame, direction)
+ elif isinstance(frame, LLMTextFrame):
+ await self._handle_llm_text(frame)
+ elif isinstance(frame, LLMFullResponseEndFrame):
+ await self._handle_llm_end(frame.skip_tts)
+ await self.push_frame(frame, direction)
+ elif isinstance(frame, EndFrame):
+ await self._handle_llm_end()
+ await self.push_frame(frame, direction)
+ else:
+ await self.push_frame(frame, direction)
+
+ async def _handle_interruption(self, _):
+ """Handle interruptions by resetting the text aggregator."""
+ await self._text_aggregator.handle_interruption()
+
+ async def reset(self):
+ """Reset the internal state of the text processor and its aggregator."""
+ await self._text_aggregator.reset()
+
+ async def _handle_llm_text(self, in_frame: LLMTextFrame):
+ aggregation = await self._text_aggregator.aggregate(in_frame.text)
+ if aggregation:
+ out_frame = AggregatedTextFrame(
+ text=aggregation.text,
+ aggregated_by=aggregation.type,
+ )
+ out_frame.skip_tts = in_frame.skip_tts
+ await self.push_frame(out_frame)
+
+ async def _handle_llm_end(self, skip_tts: bool = False):
+ # Flush any remaining aggregated text at the end of the LLM response
+ aggregation = self._text_aggregator.text
+ await self._text_aggregator.reset()
+ text = aggregation.text.strip()
+ if text:
+ out_frame = AggregatedTextFrame(
+ text=text,
+ aggregated_by=aggregation.type,
+ )
+ out_frame.skip_tts = skip_tts
+ await self.push_frame(out_frame)
diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py
index f04cbd395..7bd84fb42 100644
--- a/src/pipecat/processors/frameworks/rtvi.py
+++ b/src/pipecat/processors/frameworks/rtvi.py
@@ -24,6 +24,7 @@ from typing import (
Literal,
Mapping,
Optional,
+ Tuple,
Union,
)
@@ -32,6 +33,8 @@ from pydantic import BaseModel, Field, PrivateAttr, ValidationError
from pipecat.audio.utils import calculate_audio_volume
from pipecat.frames.frames import (
+ AggregatedTextFrame,
+ AggregationType,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -704,6 +707,29 @@ class RTVITextMessageData(BaseModel):
text: str
+class RTVIBotOutputMessageData(RTVITextMessageData):
+ """Data for bot output RTVI messages.
+
+ Extends RTVITextMessageData to include metadata about the output.
+ """
+
+ spoken: bool = False # Indicates if the text has been spoken by TTS
+ aggregated_by: AggregationType | str
+ # Indicates what form the text is in (e.g., by word, sentence, etc.)
+
+
+class RTVIBotOutputMessage(BaseModel):
+ """Message containing bot output text.
+
+ An event meant to holistically represent what the bot is outputting,
+ along with metadata about the output and if it has been spoken.
+ """
+
+ label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
+ type: Literal["bot-output"] = "bot-output"
+ data: RTVIBotOutputMessageData
+
+
class RTVIBotTranscriptionMessage(BaseModel):
"""Message containing bot transcription text.
@@ -896,6 +922,7 @@ class RTVIObserverParams:
Parameter `errors_enabled` is deprecated. Error messages are always enabled.
Parameters:
+ bot_output_enabled: Indicates if bot output messages should be sent.
bot_llm_enabled: Indicates if the bot's LLM messages should be sent.
bot_tts_enabled: Indicates if the bot's TTS messages should be sent.
bot_speaking_enabled: Indicates if the bot's started/stopped speaking messages should be sent.
@@ -907,9 +934,17 @@ class RTVIObserverParams:
metrics_enabled: Indicates if metrics messages should be sent.
system_logs_enabled: Indicates if system logs should be sent.
errors_enabled: [Deprecated] Indicates if errors messages should be sent.
+ skip_aggregator_types: List of aggregation types to skip sending as tts/output messages.
+ Note: if using this to avoid sending secure information, be sure to also disable
+ bot_llm_enabled to avoid leaking through LLM messages.
+ bot_output_transforms: A list of callables to transform text before just before sending it
+ to TTS. Each callable takes the aggregated text and its type, and returns the
+ transformed text. To register, provide a list of tuples of
+ (aggregation_type | '*', transform_function).
audio_level_period_secs: How often audio levels should be sent if enabled.
"""
+ bot_output_enabled: bool = True
bot_llm_enabled: bool = True
bot_tts_enabled: bool = True
bot_speaking_enabled: bool = True
@@ -921,6 +956,15 @@ class RTVIObserverParams:
metrics_enabled: bool = True
system_logs_enabled: bool = False
errors_enabled: Optional[bool] = None
+ skip_aggregator_types: Optional[List[AggregationType | str]] = None
+ bot_output_transforms: Optional[
+ List[
+ Tuple[
+ AggregationType | str,
+ Callable[[str, AggregationType | str], Awaitable[str]],
+ ]
+ ]
+ ] = None
audio_level_period_secs: float = 0.15
@@ -973,8 +1017,45 @@ class RTVIObserver(BaseObserver):
DeprecationWarning,
)
+ self._aggregation_transforms: List[
+ Tuple[AggregationType | str, Callable[[str, AggregationType | str], Awaitable[str]]]
+ ] = self._params.bot_output_transforms or []
+
+ def add_bot_output_transformer(
+ self,
+ transform_function: Callable[[str, AggregationType | str], Awaitable[str]],
+ aggregation_type: AggregationType | str = "*",
+ ):
+ """Transform text for a specific aggregation type before sending as Bot Output or TTS.
+
+ Args:
+ transform_function: The function to apply for transformation. This function should take
+ the text and aggregation type as input and return the transformed text.
+ Ex.: async def my_transform(text: str, aggregation_type: str) -> str:
+ aggregation_type: The type of aggregation to transform. This value defaults to "*" to
+ handle all text before sending to the client.
+ """
+ self._aggregation_transforms.append((aggregation_type, transform_function))
+
+ def remove_bot_output_transformer(
+ self,
+ transform_function: Callable[[str, AggregationType | str], Awaitable[str]],
+ aggregation_type: AggregationType | str = "*",
+ ):
+ """Remove a text transformer for a specific aggregation type.
+
+ Args:
+ transform_function: The function to remove.
+ aggregation_type: The type of aggregation to remove the transformer for.
+ """
+ self._aggregation_transforms = [
+ (agg_type, func)
+ for agg_type, func in self._aggregation_transforms
+ if not (agg_type == aggregation_type and func == transform_function)
+ ]
+
async def _logger_sink(self, message):
- """Logger sink so we cna send system logs to RTVI clients."""
+ """Logger sink so we can send system logs to RTVI clients."""
message = RTVISystemLogMessage(data=RTVITextMessageData(text=message))
await self.send_rtvi_message(message)
@@ -1048,12 +1129,15 @@ class RTVIObserver(BaseObserver):
await self.send_rtvi_message(RTVIBotTTSStartedMessage())
elif isinstance(frame, TTSStoppedFrame) and self._params.bot_tts_enabled:
await self.send_rtvi_message(RTVIBotTTSStoppedMessage())
- elif isinstance(frame, TTSTextFrame) and self._params.bot_tts_enabled:
- if isinstance(src, BaseOutputTransport):
- message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text))
- await self.send_rtvi_message(message)
- else:
+ elif isinstance(frame, AggregatedTextFrame) and (
+ self._params.bot_output_enabled or self._params.bot_tts_enabled
+ ):
+ if isinstance(frame, TTSTextFrame) and not isinstance(src, BaseOutputTransport):
+ # This check is to make sure we handle the frame when it has gone
+ # through the transport and has correct timing.
mark_as_seen = False
+ else:
+ await self._handle_aggregated_llm_text(frame)
elif isinstance(frame, MetricsFrame) and self._params.metrics_enabled:
await self._handle_metrics(frame)
elif isinstance(frame, RTVIServerMessageFrame):
@@ -1084,15 +1168,6 @@ class RTVIObserver(BaseObserver):
if mark_as_seen:
self._frames_seen.add(frame.id)
- async def _push_bot_transcription(self):
- """Push accumulated bot transcription as a message."""
- if len(self._bot_transcription) > 0:
- message = RTVIBotTranscriptionMessage(
- data=RTVITextMessageData(text=self._bot_transcription)
- )
- await self.send_rtvi_message(message)
- self._bot_transcription = ""
-
async def _handle_interruptions(self, frame: Frame):
"""Handle user speaking interruption frames."""
message = None
@@ -1115,14 +1190,45 @@ class RTVIObserver(BaseObserver):
if message:
await self.send_rtvi_message(message)
+ async def _handle_aggregated_llm_text(self, frame: AggregatedTextFrame):
+ """Handle aggregated LLM text output frames."""
+ # Skip certain aggregator types if configured to do so.
+ if (
+ self._params.skip_aggregator_types
+ and frame.aggregated_by in self._params.skip_aggregator_types
+ ):
+ return
+
+ text = frame.text
+ type = frame.aggregated_by
+ for aggregation_type, transform in self._aggregation_transforms:
+ if aggregation_type == type or aggregation_type == "*":
+ text = await transform(text, type)
+
+ isTTS = isinstance(frame, TTSTextFrame)
+ if self._params.bot_output_enabled:
+ message = RTVIBotOutputMessage(
+ data=RTVIBotOutputMessageData(text=text, spoken=isTTS, aggregated_by=type)
+ )
+ await self.send_rtvi_message(message)
+
+ if isTTS and self._params.bot_tts_enabled:
+ tts_message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=text))
+ await self.send_rtvi_message(tts_message)
+
async def _handle_llm_text_frame(self, frame: LLMTextFrame):
"""Handle LLM text output frames."""
message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text))
await self.send_rtvi_message(message)
+ # TODO (mrkb): Remove all this logic when we fully deprecate bot-transcription messages.
self._bot_transcription += frame.text
- if match_endofsentence(self._bot_transcription):
- await self._push_bot_transcription()
+
+ if match_endofsentence(self._bot_transcription) and len(self._bot_transcription) > 0:
+ await self.send_rtvi_message(
+ RTVIBotTranscriptionMessage(data=RTVITextMessageData(text=self._bot_transcription))
+ )
+ self._bot_transcription = ""
async def _handle_user_transcriptions(self, frame: Frame):
"""Handle user transcription frames."""
@@ -1248,7 +1354,7 @@ class RTVIProcessor(FrameProcessor):
# Default to 0.3.0 which is the last version before actually having a
# "client-version".
self._client_version = [0, 3, 0]
- self._skip_tts: bool = False # Keep in sync with llm_service.py
+ self._llm_skip_tts: bool = False # Keep in sync with llm_service.py's configuration.
self._registered_actions: Dict[str, RTVIAction] = {}
self._registered_services: Dict[str, RTVIService] = {}
@@ -1441,7 +1547,7 @@ class RTVIProcessor(FrameProcessor):
elif isinstance(frame, RTVIActionFrame):
await self._action_queue.put(frame)
elif isinstance(frame, LLMConfigureOutputFrame):
- self._skip_tts = frame.skip_tts
+ self._llm_skip_tts = frame.skip_tts
await self.push_frame(frame, direction)
# Other frames
else:
@@ -1697,9 +1803,9 @@ class RTVIProcessor(FrameProcessor):
opts = data.options if data.options is not None else RTVISendTextOptions()
if opts.run_immediately:
await self.interrupt_bot()
- cur_skip_tts = self._skip_tts
+ cur_llm_skip_tts = self._llm_skip_tts
should_skip_tts = not opts.audio_response
- toggle_skip_tts = cur_skip_tts != should_skip_tts
+ toggle_skip_tts = cur_llm_skip_tts != should_skip_tts
if toggle_skip_tts:
output_frame = LLMConfigureOutputFrame(skip_tts=should_skip_tts)
await self.push_frame(output_frame)
@@ -1709,7 +1815,7 @@ class RTVIProcessor(FrameProcessor):
)
await self.push_frame(text_frame)
if toggle_skip_tts:
- output_frame = LLMConfigureOutputFrame(skip_tts=cur_skip_tts)
+ output_frame = LLMConfigureOutputFrame(skip_tts=cur_llm_skip_tts)
await self.push_frame(output_frame)
async def _handle_update_context(self, data: RTVIAppendToContextData):
diff --git a/src/pipecat/services/aws/__init__.py b/src/pipecat/services/aws/__init__.py
index 3cdd4cc5a..6f6903f75 100644
--- a/src/pipecat/services/aws/__init__.py
+++ b/src/pipecat/services/aws/__init__.py
@@ -10,6 +10,7 @@ from pipecat.services import DeprecatedModuleProxy
from .llm import *
from .nova_sonic import *
+from .sagemaker import *
from .stt import *
from .tts import *
diff --git a/src/pipecat/services/aws/nova_sonic/llm.py b/src/pipecat/services/aws/nova_sonic/llm.py
index 2572b03cb..95240f748 100644
--- a/src/pipecat/services/aws/nova_sonic/llm.py
+++ b/src/pipecat/services/aws/nova_sonic/llm.py
@@ -27,6 +27,7 @@ from pydantic import BaseModel, Field
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter, Role
from pipecat.frames.frames import (
+ AggregationType,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
@@ -1027,7 +1028,7 @@ class AWSNovaSonicLLMService(LLMService):
logger.debug(f"Assistant response text added: {text}")
# Report the text of the assistant response.
- frame = TTSTextFrame(text)
+ frame = TTSTextFrame(text, aggregated_by=AggregationType.SENTENCE)
frame.includes_inter_frame_spaces = True
await self.push_frame(frame)
@@ -1062,7 +1063,9 @@ class AWSNovaSonicLLMService(LLMService):
# TTSTextFrame would be ignored otherwise (the interruption frame
# would have cleared the assistant aggregator state).
await self.push_frame(LLMFullResponseStartFrame())
- frame = TTSTextFrame(self._assistant_text_buffer)
+ frame = TTSTextFrame(
+ self._assistant_text_buffer, aggregated_by=AggregationType.SENTENCE
+ )
frame.includes_inter_frame_spaces = True
await self.push_frame(frame)
self._may_need_repush_assistant_text = False
diff --git a/src/pipecat/services/aws/sagemaker/__init__.py b/src/pipecat/services/aws/sagemaker/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/pipecat/services/aws/sagemaker/bidi_client.py b/src/pipecat/services/aws/sagemaker/bidi_client.py
new file mode 100644
index 000000000..5e02af03d
--- /dev/null
+++ b/src/pipecat/services/aws/sagemaker/bidi_client.py
@@ -0,0 +1,283 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""AWS SageMaker bidirectional streaming client.
+
+This module provides a client for streaming bidirectional communication with
+SageMaker endpoints using the HTTP/2 protocol. Supports sending audio, text,
+and JSON data to SageMaker model endpoints and receiving streaming responses.
+"""
+
+import os
+from typing import Optional
+
+from loguru import logger
+
+try:
+ from aws_sdk_sagemaker_runtime_http2.client import SageMakerRuntimeHTTP2Client
+ from aws_sdk_sagemaker_runtime_http2.config import Config, HTTPAuthSchemeResolver
+ from aws_sdk_sagemaker_runtime_http2.models import (
+ InvokeEndpointWithBidirectionalStreamInput,
+ RequestPayloadPart,
+ RequestStreamEventPayloadPart,
+ ResponseStreamEvent,
+ )
+ from smithy_aws_core.auth.sigv4 import SigV4AuthScheme
+ from smithy_aws_core.identity import EnvironmentCredentialsResolver
+ from smithy_core.aio.eventstream import DuplexEventStream
+except ModuleNotFoundError as e:
+ logger.error(f"Exception: {e}")
+ logger.error(
+ "In order to use SageMaker BiDi client, you need to `pip install pipecat-ai[sagemaker]`."
+ )
+ raise Exception(f"Missing module: {e}")
+
+
+class SageMakerBidiClient:
+ """Client for bidirectional streaming with AWS SageMaker endpoints.
+
+ Handles low-level HTTP/2 bidirectional streaming protocol for communicating
+ with SageMaker model endpoints. Provides methods for sending various data
+ types (audio, text, JSON) and receiving streaming responses.
+
+ This client uses AWS SigV4 authentication and supports credential resolution
+ from environment variables, AWS CLI configuration, and instance metadata.
+
+ Example::
+
+ client = SageMakerBidiClient(
+ endpoint_name="my-deepgram-endpoint",
+ region="us-east-2",
+ model_invocation_path="v1/listen",
+ model_query_string="model=nova-3&language=en"
+ )
+ await client.start_session()
+ await client.send_audio_chunk(audio_bytes)
+ response = await client.receive_response()
+ await client.close_session()
+ """
+
+ def __init__(
+ self,
+ endpoint_name: str,
+ region: str,
+ model_invocation_path: str = "",
+ model_query_string: str = "",
+ ):
+ """Initialize the SageMaker BiDi client.
+
+ Args:
+ endpoint_name: Name of the SageMaker endpoint to connect to.
+ region: AWS region where the endpoint is deployed.
+ model_invocation_path: API path for the model invocation (e.g., "v1/listen").
+ model_query_string: Query string parameters for the model (e.g., "model=nova-3").
+ """
+ self.endpoint_name = endpoint_name
+ self.region = region
+ self.model_invocation_path = model_invocation_path
+ self.model_query_string = model_query_string
+ self.bidi_endpoint = f"https://runtime.sagemaker.{region}.amazonaws.com:8443"
+ self._client: Optional[SageMakerRuntimeHTTP2Client] = None
+ self._stream: Optional[
+ DuplexEventStream[RequestStreamEventPayloadPart, ResponseStreamEvent, any]
+ ] = None
+ self._output_stream = None
+ self._is_active = False
+
+ def _initialize_client(self):
+ """Initialize the SageMaker Runtime HTTP2 client with AWS credentials.
+
+ Creates and configures the SageMaker Runtime HTTP2 client with SigV4
+ authentication. Attempts to resolve AWS credentials from environment
+ variables, AWS CLI configuration, or instance metadata.
+ """
+ logger.debug(f"Initializing SageMaker BiDi client for region: {self.region}")
+ logger.debug(f"Using endpoint URI: {self.bidi_endpoint}")
+
+ # Check for AWS credentials
+ has_env_creds = bool(os.getenv("AWS_ACCESS_KEY_ID") and os.getenv("AWS_SECRET_ACCESS_KEY"))
+
+ if not has_env_creds:
+ logger.warning(
+ "AWS credentials not found in environment variables. "
+ "Attempting to use EnvironmentCredentialsResolver which will check "
+ "AWS CLI configuration and instance metadata."
+ )
+
+ config = Config(
+ endpoint_uri=self.bidi_endpoint,
+ region=self.region,
+ aws_credentials_identity_resolver=EnvironmentCredentialsResolver(),
+ auth_scheme_resolver=HTTPAuthSchemeResolver(),
+ auth_schemes={"aws.auth#sigv4": SigV4AuthScheme(service="sagemaker")},
+ )
+ self._client = SageMakerRuntimeHTTP2Client(config=config)
+
+ async def start_session(self):
+ """Start a bidirectional streaming session with the SageMaker endpoint.
+
+ Initializes the client if needed, creates the bidirectional stream, and
+ establishes the connection to the SageMaker endpoint. Must be called
+ before sending or receiving data.
+
+ Returns:
+ The output stream for receiving responses.
+
+ Raises:
+ RuntimeError: If client initialization or connection fails.
+ """
+ if not self._client:
+ self._initialize_client()
+
+ logger.debug(f"Starting BiDi session with endpoint: {self.endpoint_name}")
+ logger.debug(f"Model invocation path: {self.model_invocation_path}")
+ logger.debug(f"Model query string: {self.model_query_string}")
+
+ # Create the bidirectional stream
+ stream_input = InvokeEndpointWithBidirectionalStreamInput(
+ endpoint_name=self.endpoint_name,
+ model_invocation_path=self.model_invocation_path,
+ model_query_string=self.model_query_string,
+ )
+
+ try:
+ self._stream = await self._client.invoke_endpoint_with_bidirectional_stream(
+ stream_input
+ )
+ self._is_active = True
+
+ # Get output stream
+ output = await self._stream.await_output()
+ self._output_stream = output[1]
+
+ logger.debug("BiDi session started successfully")
+ return self._output_stream
+
+ except Exception as e:
+ logger.error(f"Failed to start BiDi session: {e}")
+ self._is_active = False
+ raise RuntimeError(f"Failed to start SageMaker BiDi session: {e}")
+
+ async def send_data(self, data_bytes: bytes, data_type: Optional[str] = None):
+ """Send a chunk of data to the stream.
+
+ Generic method for sending any type of data to the SageMaker endpoint.
+ Use the convenience methods (send_audio_chunk, send_text, send_json)
+ for common data types.
+
+ Args:
+ data_bytes: Raw bytes to send.
+ data_type: Optional data type header. Common values are "BINARY" for
+ audio/binary data and "UTF8" for text/JSON data.
+
+ Raises:
+ RuntimeError: If session is not active or send fails.
+ """
+ if not self._is_active or not self._stream:
+ raise RuntimeError("BiDi session not active")
+
+ try:
+ payload = RequestPayloadPart(bytes_=data_bytes, data_type=data_type)
+ event = RequestStreamEventPayloadPart(value=payload)
+ await self._stream.input_stream.send(event)
+ except Exception as e:
+ logger.error(f"Failed to send data: {e}")
+ raise
+
+ async def send_audio_chunk(self, audio_bytes: bytes):
+ """Send a chunk of audio data to the stream.
+
+ Convenience method for sending audio data. Automatically sets the data
+ type to "BINARY".
+
+ Args:
+ audio_bytes: Raw audio bytes to send (e.g., PCM audio data).
+
+ Raises:
+ RuntimeError: If session is not active or send fails.
+ """
+ await self.send_data(audio_bytes, data_type="BINARY")
+
+ async def send_text(self, text: str):
+ """Send text data to the stream.
+
+ Convenience method for sending text data. Automatically encodes the text
+ as UTF-8 and sets the data type to "UTF8".
+
+ Args:
+ text: Text string to send.
+
+ Raises:
+ RuntimeError: If session is not active or send fails.
+ """
+ await self.send_data(text.encode("utf-8"), data_type="UTF8")
+
+ async def send_json(self, data: dict):
+ """Send JSON data to the stream.
+
+ Convenience method for sending JSON-encoded messages. Useful for control
+ messages like KeepAlive or CloseStream. Automatically serializes the
+ dictionary to JSON, encodes as UTF-8, and sets the data type to "UTF8".
+
+ Args:
+ data: Dictionary to send as JSON (e.g., {"type": "KeepAlive"}).
+
+ Raises:
+ RuntimeError: If session is not active or send fails.
+ """
+ import json
+
+ await self.send_data(json.dumps(data).encode("utf-8"), data_type="UTF8")
+
+ async def receive_response(self) -> Optional[ResponseStreamEvent]:
+ """Receive a response from the stream.
+
+ Blocks until a response is available from the SageMaker endpoint. Returns
+ None when the stream is closed.
+
+ Returns:
+ The response event containing payload data, or None if stream is closed.
+
+ Raises:
+ RuntimeError: If session is not active.
+ """
+ if not self._is_active or not self._output_stream:
+ raise RuntimeError("BiDi session not active")
+
+ try:
+ result = await self._output_stream.receive()
+ return result
+ except Exception as e:
+ logger.error(f"Failed to receive response: {e}")
+ raise
+
+ async def close_session(self):
+ """Close the bidirectional streaming session.
+
+ Gracefully closes the input stream and marks the session as inactive.
+ Safe to call multiple times.
+ """
+ if not self._is_active:
+ return
+
+ logger.debug("Closing BiDi session...")
+ self._is_active = False
+
+ try:
+ if self._stream:
+ await self._stream.input_stream.close()
+ logger.debug("BiDi session closed successfully")
+ except Exception as e:
+ logger.warning(f"Error closing BiDi session: {e}")
+
+ @property
+ def is_active(self) -> bool:
+ """Check if the session is currently active.
+
+ Returns:
+ True if session is active, False otherwise.
+ """
+ return self._is_active
diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py
index f8881200c..d42802cf2 100644
--- a/src/pipecat/services/cartesia/tts.py
+++ b/src/pipecat/services/cartesia/tts.py
@@ -10,7 +10,8 @@ import base64
import json
import uuid
import warnings
-from typing import AsyncGenerator, List, Literal, Optional, Union
+from enum import Enum
+from typing import AsyncGenerator, List, Literal, Optional
from loguru import logger
from pydantic import BaseModel, Field
@@ -125,6 +126,72 @@ def language_to_cartesia_language(language: Language) -> Optional[str]:
return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
+class CartesiaEmotion(str, Enum):
+ """Predefined Emotions supported by Cartesia."""
+
+ # Primary emotions supported by Cartesia
+ NEUTRAL = "neutral"
+ ANGRY = "angry"
+ EXCITED = "excited"
+ CONTENT = "content"
+ SAD = "sad"
+ SCARED = "scared"
+ # Additional emotions supported by Cartesia
+ HAPPY = "happy"
+ ENTHUSIASTIC = "enthusiastic"
+ ELATED = "elated"
+ EUPHORIC = "euphoric"
+ TRIUMPHANT = "triumphant"
+ AMAZED = "amazed"
+ SURPRISED = "surprised"
+ FLIRTATIOUS = "flirtatious"
+ JOKING_COMEDIC = "joking/comedic"
+ CURIOUS = "curious"
+ PEACEFUL = "peaceful"
+ SERENE = "serene"
+ CALM = "calm"
+ GRATEFUL = "grateful"
+ AFFECTIONATE = "affectionate"
+ TRUST = "trust"
+ SYMPATHETIC = "sympathetic"
+ ANTICIPATION = "anticipation"
+ MYSTERIOUS = "mysterious"
+ MAD = "mad"
+ OUTRAGED = "outraged"
+ FRUSTRATED = "frustrated"
+ AGITATED = "agitated"
+ THREATENED = "threatened"
+ DISGUSTED = "disgusted"
+ CONTEMPT = "contempt"
+ ENVIOUS = "envious"
+ SARCASTIC = "sarcastic"
+ IRONIC = "ironic"
+ DEJECTED = "dejected"
+ MELANCHOLIC = "melancholic"
+ DISAPPOINTED = "disappointed"
+ HURT = "hurt"
+ GUILTY = "guilty"
+ BORED = "bored"
+ TIRED = "tired"
+ REJECTED = "rejected"
+ NOSTALGIC = "nostalgic"
+ WISTFUL = "wistful"
+ APOLOGETIC = "apologetic"
+ HESITANT = "hesitant"
+ INSECURE = "insecure"
+ CONFUSED = "confused"
+ RESIGNED = "resigned"
+ ANXIOUS = "anxious"
+ PANICKED = "panicked"
+ ALARMED = "alarmed"
+ PROUD = "proud"
+ CONFIDENT = "confident"
+ DISTANT = "distant"
+ SKEPTICAL = "skeptical"
+ CONTEMPLATIVE = "contemplative"
+ DETERMINED = "determined"
+
+
class CartesiaTTSService(AudioContextWordTTSService):
"""Cartesia TTS service with WebSocket streaming and word timestamps.
@@ -182,6 +249,10 @@ class CartesiaTTSService(AudioContextWordTTSService):
container: Audio container format.
params: Additional input parameters for voice customization.
text_aggregator: Custom text aggregator for processing input text.
+
+ .. deprecated:: 0.0.95
+ Use an LLMTextProcessor before the TTSService for custom text aggregation.
+
aggregate_sentences: Whether to aggregate sentences within the TTSService.
**kwargs: Additional arguments passed to the parent service.
"""
@@ -200,10 +271,18 @@ class CartesiaTTSService(AudioContextWordTTSService):
push_text_frames=False,
pause_frame_processing=True,
sample_rate=sample_rate,
- text_aggregator=text_aggregator or SkipTagsAggregator([("", "")]),
+ text_aggregator=text_aggregator,
**kwargs,
)
+ if not text_aggregator:
+ # Always skip tags added for spelled-out text
+ # Note: This is primarily to support backwards compatibility.
+ # The preferred way of taking advantage of Cartesia SSML Tags is
+ # to use an LLMTextProcessor and/or a text_transformer to identify
+ # and insert these tags for the purpose of the TTS service alone.
+ self._text_aggregator = SkipTagsAggregator([("", "")])
+
params = params or CartesiaTTSService.InputParams()
self._api_key = api_key
@@ -257,6 +336,27 @@ class CartesiaTTSService(AudioContextWordTTSService):
"""
return language_to_cartesia_language(language)
+ # A set of Cartesia-specific helpers for text transformations
+ def SPELL(text: str) -> str:
+ """Wrap text in Cartesia spell tag."""
+ return f"{text}"
+
+ def EMOTION_TAG(emotion: CartesiaEmotion) -> str:
+ """Convenience method to create an emotion tag."""
+ return f''
+
+ def PAUSE_TAG(seconds: float) -> str:
+ """Convenience method to create a pause tag."""
+ return f''
+
+ def VOLUME_TAG(volume: float) -> str:
+ """Convenience method to create a volume tag."""
+ return f''
+
+ def SPEED_TAG(speed: float) -> str:
+ """Convenience method to create a speed tag."""
+ return f''
+
def _is_cjk_language(self, language: str) -> bool:
"""Check if the given language is CJK (Chinese, Japanese, Korean).
diff --git a/src/pipecat/services/deepgram/__init__.py b/src/pipecat/services/deepgram/__init__.py
index c23ebbec9..227ac5c64 100644
--- a/src/pipecat/services/deepgram/__init__.py
+++ b/src/pipecat/services/deepgram/__init__.py
@@ -10,6 +10,7 @@ from pipecat.services import DeprecatedModuleProxy
from .flux import *
from .stt import *
+from .stt_sagemaker import *
from .tts import *
sys.modules[__name__] = DeprecatedModuleProxy(globals(), "deepgram", "deepgram.[stt,tts]")
diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py
new file mode 100644
index 000000000..6d28feefa
--- /dev/null
+++ b/src/pipecat/services/deepgram/stt_sagemaker.py
@@ -0,0 +1,447 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Deepgram speech-to-text service for AWS SageMaker.
+
+This module provides a Pipecat STT service that connects to Deepgram models
+deployed on AWS SageMaker endpoints. Uses HTTP/2 bidirectional streaming for
+low-latency real-time transcription with support for interim results, multiple
+languages, and various Deepgram features.
+"""
+
+import asyncio
+import json
+from typing import AsyncGenerator, Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+ CancelFrame,
+ EndFrame,
+ ErrorFrame,
+ Frame,
+ InterimTranscriptionFrame,
+ StartFrame,
+ TranscriptionFrame,
+ UserStartedSpeakingFrame,
+ UserStoppedSpeakingFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient
+from pipecat.services.stt_service import STTService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.time import time_now_iso8601
+from pipecat.utils.tracing.service_decorators import traced_stt
+
+try:
+ from deepgram import LiveOptions
+except ModuleNotFoundError as e:
+ logger.error(f"Exception: {e}")
+ logger.error(
+ "In order to use DeepgramSageMakerSTTService, you need to `pip install pipecat-ai[deepgram,sagemaker]`."
+ )
+ raise Exception(f"Missing module: {e}")
+
+
+class DeepgramSageMakerSTTService(STTService):
+ """Deepgram speech-to-text service for AWS SageMaker.
+
+ Provides real-time speech recognition using Deepgram models deployed on
+ AWS SageMaker endpoints. Uses HTTP/2 bidirectional streaming for low-latency
+ transcription with support for interim results, speaker diarization, and
+ multiple languages.
+
+ Requirements:
+
+ - AWS credentials configured (via environment variables, AWS CLI, or instance metadata)
+ - A deployed SageMaker endpoint with Deepgram model: https://developers.deepgram.com/docs/deploy-amazon-sagemaker
+ - Deepgram SDK for LiveOptions configuration
+
+ Example::
+
+ stt = DeepgramSageMakerSTTService(
+ endpoint_name="my-deepgram-endpoint",
+ region="us-east-2",
+ live_options=LiveOptions(
+ model="nova-3",
+ language="en",
+ interim_results=True,
+ punctuate=True,
+ ),
+ )
+ """
+
+ def __init__(
+ self,
+ *,
+ endpoint_name: str,
+ region: str,
+ sample_rate: Optional[int] = None,
+ live_options: Optional[LiveOptions] = None,
+ **kwargs,
+ ):
+ """Initialize the Deepgram SageMaker STT service.
+
+ Args:
+ endpoint_name: Name of the SageMaker endpoint with Deepgram model
+ deployed (e.g., "my-deepgram-nova-3-endpoint").
+ region: AWS region where the endpoint is deployed (e.g., "us-east-2").
+ sample_rate: Audio sample rate in Hz. If None, uses value from
+ live_options or defaults to the value from StartFrame.
+ live_options: Deepgram LiveOptions for detailed configuration. If None,
+ uses sensible defaults (nova-3 model, English, interim results enabled).
+ **kwargs: Additional arguments passed to the parent STTService.
+ """
+ sample_rate = sample_rate or (live_options.sample_rate if live_options else None)
+ super().__init__(sample_rate=sample_rate, **kwargs)
+
+ self._endpoint_name = endpoint_name
+ self._region = region
+
+ # Create default options similar to DeepgramSTTService
+ default_options = LiveOptions(
+ encoding="linear16",
+ language=Language.EN,
+ model="nova-3",
+ channels=1,
+ interim_results=True,
+ punctuate=True,
+ )
+
+ # Merge with provided options
+ merged_options = default_options.to_dict()
+ if live_options:
+ default_model = default_options.model
+ merged_options.update(live_options.to_dict())
+ # Handle the "None" string bug from deepgram-sdk
+ if "model" in merged_options and merged_options["model"] == "None":
+ merged_options["model"] = default_model
+
+ # Convert Language enum to string if needed
+ if "language" in merged_options and isinstance(merged_options["language"], Language):
+ merged_options["language"] = merged_options["language"].value
+
+ self.set_model_name(merged_options["model"])
+ self._settings = merged_options
+
+ self._client: Optional[SageMakerBidiClient] = None
+ self._response_task: Optional[asyncio.Task] = None
+ self._keepalive_task: Optional[asyncio.Task] = None
+
+ def can_generate_metrics(self) -> bool:
+ """Check if this service can generate processing metrics.
+
+ Returns:
+ True, as Deepgram SageMaker service supports metrics generation.
+ """
+ return True
+
+ async def set_model(self, model: str):
+ """Set the Deepgram model and reconnect.
+
+ Disconnects from the current session, updates the model setting, and
+ establishes a new connection with the updated model.
+
+ Args:
+ model: The Deepgram model name to use (e.g., "nova-3").
+ """
+ await super().set_model(model)
+ logger.info(f"Switching STT model to: [{model}]")
+ self._settings["model"] = model
+ await self._disconnect()
+ await self._connect()
+
+ async def set_language(self, language: Language):
+ """Set the recognition language and reconnect.
+
+ Disconnects from the current session, updates the language setting, and
+ establishes a new connection with the updated language.
+
+ Args:
+ language: The language to use for speech recognition (e.g., Language.EN,
+ Language.ES).
+ """
+ logger.info(f"Switching STT language to: [{language}]")
+ self._settings["language"] = language
+ await self._disconnect()
+ await self._connect()
+
+ async def start(self, frame: StartFrame):
+ """Start the Deepgram SageMaker STT service.
+
+ Args:
+ frame: The start frame containing initialization parameters.
+ """
+ await super().start(frame)
+ self._settings["sample_rate"] = self.sample_rate
+ await self._connect()
+
+ async def stop(self, frame: EndFrame):
+ """Stop the Deepgram SageMaker STT service.
+
+ Args:
+ frame: The end frame.
+ """
+ await super().stop(frame)
+ await self._disconnect()
+
+ async def cancel(self, frame: CancelFrame):
+ """Cancel the Deepgram SageMaker STT service.
+
+ Args:
+ frame: The cancel frame.
+ """
+ await super().cancel(frame)
+ await self._disconnect()
+
+ async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+ """Send audio data to Deepgram for transcription.
+
+ Args:
+ audio: Raw audio bytes to transcribe.
+
+ Yields:
+ Frame: None (transcription results come via BiDi stream callbacks).
+ """
+ if self._client and self._client.is_active:
+ try:
+ await self._client.send_audio_chunk(audio)
+ except Exception as e:
+ logger.error(f"Error sending audio to SageMaker: {e}")
+ await self.push_error(ErrorFrame(error=f"SageMaker STT error: {e}"))
+ yield None
+
+ async def _connect(self):
+ """Connect to the SageMaker endpoint and start the BiDi session.
+
+ Builds the Deepgram query string from settings, creates the BiDi client,
+ starts the streaming session, and launches background tasks for processing
+ responses and sending KeepAlive messages.
+ """
+ logger.debug("Connecting to Deepgram on SageMaker...")
+
+ # Update sample rate in settings
+ self._settings["sample_rate"] = self.sample_rate
+
+ # Build query string from settings, converting booleans to strings
+ query_params = {}
+ for key, value in self._settings.items():
+ if value is not None:
+ # Convert boolean values to lowercase strings for Deepgram API
+ if isinstance(value, bool):
+ query_params[key] = str(value).lower()
+ else:
+ query_params[key] = str(value)
+
+ query_string = "&".join(f"{k}={v}" for k, v in query_params.items())
+
+ # Create BiDi client
+ self._client = SageMakerBidiClient(
+ endpoint_name=self._endpoint_name,
+ region=self._region,
+ model_invocation_path="v1/listen",
+ model_query_string=query_string,
+ )
+
+ try:
+ # Start the session
+ await self._client.start_session()
+
+ # Start processing responses in the background
+ self._response_task = self.create_task(self._process_responses())
+
+ # Start keepalive task to maintain connection
+ self._keepalive_task = self.create_task(self._send_keepalive())
+
+ logger.debug("Connected to Deepgram on SageMaker")
+ await self._call_event_handler("on_connected")
+
+ except Exception as e:
+ logger.error(f"Failed to connect to SageMaker: {e}")
+ await self.push_error(ErrorFrame(error=f"SageMaker connection error: {e}"))
+ await self._call_event_handler("on_connection_error", str(e))
+
+ async def _disconnect(self):
+ """Disconnect from the SageMaker endpoint.
+
+ Sends a CloseStream message to Deepgram, cancels background tasks
+ (KeepAlive and response processing), and closes the BiDi session.
+ Safe to call multiple times.
+ """
+ if self._client and self._client.is_active:
+ logger.debug("Disconnecting from Deepgram on SageMaker...")
+
+ # Send CloseStream message to Deepgram
+ try:
+ await self._client.send_json({"type": "CloseStream"})
+ except Exception as e:
+ logger.warning(f"Failed to send CloseStream message: {e}")
+
+ # Cancel keepalive task
+ if self._keepalive_task and not self._keepalive_task.done():
+ await self.cancel_task(self._keepalive_task)
+
+ # Cancel response processing task
+ if self._response_task and not self._response_task.done():
+ await self.cancel_task(self._response_task)
+
+ # Close the BiDi session
+ await self._client.close_session()
+
+ logger.debug("Disconnected from Deepgram on SageMaker")
+ await self._call_event_handler("on_disconnected")
+
+ async def _send_keepalive(self):
+ """Send periodic KeepAlive messages to maintain the connection.
+
+ Sends a KeepAlive JSON message to Deepgram every 5 seconds while the
+ connection is active. This prevents the connection from timing out during
+ periods of silence.
+ """
+ while self._client and self._client.is_active:
+ await asyncio.sleep(5)
+ if self._client and self._client.is_active:
+ try:
+ await self._client.send_json({"type": "KeepAlive"})
+ except Exception as e:
+ logger.warning(f"Failed to send KeepAlive: {e}")
+
+ async def _process_responses(self):
+ """Process streaming responses from Deepgram on SageMaker.
+
+ Continuously receives responses from the BiDi stream, decodes the payload,
+ parses JSON responses from Deepgram, and processes transcription results.
+ Runs as a background task until the connection is closed or cancelled.
+ """
+ try:
+ while self._client and self._client.is_active:
+ result = await self._client.receive_response()
+
+ if result is None:
+ break
+
+ # Check if this is a PayloadPart with bytes
+ if hasattr(result, "value") and hasattr(result.value, "bytes_"):
+ if result.value.bytes_:
+ response_data = result.value.bytes_.decode("utf-8")
+
+ try:
+ # Parse JSON response from Deepgram
+ parsed = json.loads(response_data)
+
+ # Extract and process transcript if available
+ if "channel" in parsed:
+ await self._handle_transcript_response(parsed)
+
+ except json.JSONDecodeError:
+ logger.warning(f"Non-JSON response: {response_data}")
+
+ except asyncio.CancelledError:
+ logger.debug("Response processor cancelled")
+ except Exception as e:
+ logger.error(f"Error processing responses: {e}", exc_info=True)
+ await self.push_error(ErrorFrame(error=f"SageMaker response error: {e}"))
+ finally:
+ logger.debug("Response processor stopped")
+
+ async def _handle_transcript_response(self, parsed: dict):
+ """Handle a transcript response from Deepgram.
+
+ Extracts the transcript text, determines if it's final or interim, extracts
+ language information, and pushes the appropriate frame (TranscriptionFrame
+ or InterimTranscriptionFrame) downstream.
+
+ Args:
+ parsed: The parsed JSON response from Deepgram containing channel,
+ alternatives, transcript, and metadata.
+ """
+ alternatives = parsed.get("channel", {}).get("alternatives", [])
+ if not alternatives or not alternatives[0].get("transcript"):
+ return
+
+ transcript = alternatives[0]["transcript"]
+ if not transcript.strip():
+ return
+
+ # Stop TTFB metrics on first transcript
+ await self.stop_ttfb_metrics()
+
+ is_final = parsed.get("is_final", False)
+ speech_final = parsed.get("speech_final", False)
+
+ # Extract language if available
+ language = None
+ if alternatives[0].get("languages"):
+ language = alternatives[0]["languages"][0]
+ language = Language(language)
+
+ if is_final and speech_final:
+ # Final transcription
+ await self.push_frame(
+ TranscriptionFrame(
+ transcript,
+ self._user_id,
+ time_now_iso8601(),
+ language,
+ result=parsed,
+ )
+ )
+ await self._handle_transcription(transcript, is_final, language)
+ await self.stop_processing_metrics()
+ else:
+ # Interim transcription
+ await self.push_frame(
+ InterimTranscriptionFrame(
+ transcript,
+ self._user_id,
+ time_now_iso8601(),
+ language,
+ result=parsed,
+ )
+ )
+
+ @traced_stt
+ async def _handle_transcription(
+ self, transcript: str, is_final: bool, language: Optional[Language] = None
+ ):
+ """Handle a transcription result with tracing.
+
+ This method is decorated with @traced_stt for observability and tracing
+ integration. The actual transcription processing is handled by the parent
+ class and observers.
+
+ Args:
+ transcript: The transcribed text.
+ is_final: Whether this is a final transcription result.
+ language: The detected language of the transcription, if available.
+ """
+ pass
+
+ async def start_metrics(self):
+ """Start TTFB and processing metrics collection."""
+ await self.start_ttfb_metrics()
+ await self.start_processing_metrics()
+
+ async def process_frame(self, frame: Frame, direction: FrameDirection):
+ """Process frames with Deepgram SageMaker-specific handling.
+
+ Args:
+ frame: The frame to process.
+ direction: The direction of frame processing.
+ """
+ await super().process_frame(frame, direction)
+
+ # Start metrics when user starts speaking (if VAD is not provided by Deepgram)
+ if isinstance(frame, UserStartedSpeakingFrame):
+ await self.start_metrics()
+ elif isinstance(frame, UserStoppedSpeakingFrame):
+ # Send finalize message to Deepgram when user stops speaking
+ # This tells Deepgram to flush any remaining audio and return final results
+ if self._client and self._client.is_active:
+ try:
+ await self._client.send_json({"type": "Finalize"})
+ except Exception as e:
+ logger.warning(f"Error sending Finalize message: {e}")
diff --git a/src/pipecat/services/elevenlabs/stt.py b/src/pipecat/services/elevenlabs/stt.py
index 8cbb40d63..95a802edd 100644
--- a/src/pipecat/services/elevenlabs/stt.py
+++ b/src/pipecat/services/elevenlabs/stt.py
@@ -416,6 +416,8 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
Only used when commit_strategy is VAD. None uses ElevenLabs default.
min_silence_duration_ms: Minimum silence duration for VAD (50-2000ms).
Only used when commit_strategy is VAD. None uses ElevenLabs default.
+ include_timestamps: Whether to include word-level timestamps in transcripts.
+ enable_logging: Whether to enable logging on ElevenLabs' side.
"""
language_code: Optional[str] = None
@@ -424,6 +426,8 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
vad_threshold: Optional[float] = None
min_speech_duration_ms: Optional[int] = None
min_silence_duration_ms: Optional[int] = None
+ include_timestamps: bool = False
+ enable_logging: bool = False
def __init__(
self,
@@ -459,6 +463,8 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
self._audio_format = "" # initialized in start()
self._receive_task = None
+ self._settings = {"language": params.language_code}
+
def can_generate_metrics(self) -> bool:
"""Check if the service can generate processing metrics.
@@ -477,7 +483,13 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
Changing language requires reconnecting to the WebSocket.
"""
logger.info(f"Switching STT language to: [{language}]")
- self._params.language_code = language.value if isinstance(language, Language) else language
+ new_language = (
+ language_to_elevenlabs_language(language)
+ if isinstance(language, Language)
+ else language
+ )
+ self._params.language_code = new_language
+ self._settings["language"] = new_language
# Reconnect with new settings
await self._disconnect()
await self._connect()
@@ -620,10 +632,16 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
if self._params.language_code:
params.append(f"language_code={self._params.language_code}")
- params.append(f"encoding={self._audio_format}")
- params.append(f"sample_rate={self.sample_rate}")
+ params.append(f"audio_format={self._audio_format}")
params.append(f"commit_strategy={self._params.commit_strategy.value}")
+ # Add optional parameters
+ if self._params.include_timestamps:
+ params.append(f"include_timestamps={str(self._params.include_timestamps).lower()}")
+
+ if self._params.enable_logging:
+ params.append(f"enable_logging={str(self._params.enable_logging).lower()}")
+
# Add VAD parameters if using VAD commit strategy and values are specified
if self._params.commit_strategy == CommitStrategy.VAD:
if self._params.vad_silence_threshold_secs is not None:
@@ -712,15 +730,20 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
elif message_type == "committed_transcript_with_timestamps":
await self._on_committed_transcript_with_timestamps(data)
- elif message_type == "input_error":
- error_msg = data.get("error", "Unknown input error")
- logger.error(f"ElevenLabs input error: {error_msg}")
- await self.push_error(ErrorFrame(f"Input error: {error_msg}"))
+ elif message_type == "error":
+ error_msg = data.get("error", "Unknown error")
+ logger.error(f"ElevenLabs error: {error_msg}")
+ await self.push_error(ErrorFrame(f"Error: {error_msg}"))
- elif message_type in ["auth_error", "quota_exceeded", "transcriber_error", "error"]:
- error_msg = data.get("error", data.get("message", "Unknown error"))
- logger.error(f"ElevenLabs error ({message_type}): {error_msg}")
- await self.push_error(ErrorFrame(f"{message_type}: {error_msg}"))
+ elif message_type == "auth_error":
+ error_msg = data.get("error", "Authentication error")
+ logger.error(f"ElevenLabs auth error: {error_msg}")
+ await self.push_error(ErrorFrame(f"Auth error: {error_msg}"))
+
+ elif message_type == "quota_exceeded_error":
+ error_msg = data.get("error", "Quota exceeded")
+ logger.error(f"ElevenLabs quota exceeded: {error_msg}")
+ await self.push_error(ErrorFrame(f"Quota exceeded: {error_msg}"))
else:
logger.debug(f"Unknown message type: {message_type}")
@@ -765,6 +788,11 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
Args:
data: Committed transcript data.
"""
+ # If timestamps are enabled, skip this message and wait for the
+ # committed_transcript_with_timestamps message which contains all the data
+ if self._params.include_timestamps:
+ return
+
text = data.get("text", "").strip()
if not text:
return
@@ -792,6 +820,18 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
async def _on_committed_transcript_with_timestamps(self, data: dict):
"""Handle committed transcript with word-level timestamps.
+ This message is sent when include_timestamps=true. The result data includes:
+ - text: The transcribed text
+ - language_code: Detected language (if available)
+ - words: Array of word objects with timing information:
+ - text: The word text
+ - start: Start time in seconds
+ - end: End time in seconds
+ - type: "word" or "spacing"
+ - speaker_id: Speaker identifier (if available)
+ - logprob: Log probability score (if available)
+ - characters: Array of character strings (if available)
+
Args:
data: Committed transcript data with timestamps.
"""
@@ -799,9 +839,24 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
if not text:
return
- logger.debug(f"Committed transcript with timestamps: [{text}]")
- logger.trace(f"Timestamps: {data.get('words', [])}")
+ await self.stop_ttfb_metrics()
+ await self.stop_processing_metrics()
- # This is sent after the committed_transcript, so we don't need to
- # push another TranscriptionFrame, but we could use the timestamps
- # for additional processing if needed in the future
+ # Get language if provided
+ language = data.get("language_code")
+
+ logger.debug(f"Committed transcript with timestamps: [{text}]")
+
+ await self._handle_transcription(text, True, language)
+
+ # This message is sent after committed_transcript when include_timestamps=true.
+ # It contains the full transcript data including text and word-level timestamps.
+ await self.push_frame(
+ TranscriptionFrame(
+ text,
+ self._user_id,
+ time_now_iso8601(),
+ language,
+ result=data,
+ )
+ )
diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py
index 7e0b0f494..2c6e8e463 100644
--- a/src/pipecat/services/google/gemini_live/llm.py
+++ b/src/pipecat/services/google/gemini_live/llm.py
@@ -27,6 +27,7 @@ from pydantic import BaseModel, Field
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.frames.frames import (
+ AggregationType,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -1644,7 +1645,7 @@ class GeminiLiveLLMService(LLMService):
await self.push_frame(TTSStartedFrame())
await self.push_frame(LLMFullResponseStartFrame())
- frame = TTSTextFrame(text=text)
+ frame = TTSTextFrame(text=text, aggregated_by=AggregationType.SENTENCE)
# Gemini Live text already includes any necessary inter-chunk spaces
frame.includes_inter_frame_spaces = True
diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py
index 8eaa3d6fa..755e64040 100644
--- a/src/pipecat/services/openai/realtime/llm.py
+++ b/src/pipecat/services/openai/realtime/llm.py
@@ -19,6 +19,7 @@ from pipecat.adapters.services.open_ai_realtime_adapter import (
OpenAIRealtimeLLMAdapter,
)
from pipecat.frames.frames import (
+ AggregationType,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
@@ -684,7 +685,7 @@ class OpenAIRealtimeLLMService(LLMService):
# We receive audio transcript deltas (as opposed to text deltas) when
# the output modality is "audio" (the default)
if evt.delta:
- frame = TTSTextFrame(evt.delta)
+ frame = TTSTextFrame(evt.delta, aggregated_by=AggregationType.SENTENCE)
# OpenAI Realtime text already includes any necessary inter-chunk spaces
frame.includes_inter_frame_spaces = True
await self.push_frame(frame)
diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py
index af0600882..d0cb39bf6 100644
--- a/src/pipecat/services/openai_realtime_beta/openai.py
+++ b/src/pipecat/services/openai_realtime_beta/openai.py
@@ -17,6 +17,7 @@ from loguru import logger
from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
from pipecat.frames.frames import (
+ AggregationType,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
@@ -652,7 +653,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
async def _handle_evt_audio_transcript_delta(self, evt):
if evt.delta:
await self.push_frame(LLMTextFrame(evt.delta))
- await self.push_frame(TTSTextFrame(evt.delta))
+ await self.push_frame(TTSTextFrame(evt.delta, aggregated_by=AggregationType.SENTENCE))
async def _handle_evt_speech_started(self, evt):
await self._truncate_current_audio_response()
diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py
index 7b62f20fa..c9f461350 100644
--- a/src/pipecat/services/rime/tts.py
+++ b/src/pipecat/services/rime/tts.py
@@ -113,6 +113,10 @@ class RimeTTSService(AudioContextWordTTSService):
sample_rate: Audio sample rate in Hz.
params: Additional configuration parameters.
text_aggregator: Custom text aggregator for processing input text.
+
+ .. deprecated:: 0.0.95
+ Use an LLMTextProcessor before the TTSService for custom text aggregation.
+
aggregate_sentences: Whether to aggregate sentences within the TTSService.
**kwargs: Additional arguments passed to parent class.
"""
@@ -123,10 +127,17 @@ class RimeTTSService(AudioContextWordTTSService):
push_stop_frames=True,
pause_frame_processing=True,
sample_rate=sample_rate,
- text_aggregator=text_aggregator or SkipTagsAggregator([("spell(", ")")]),
**kwargs,
)
+ if not text_aggregator:
+ # Always skip tags added for spelled-out text
+ # Note: This is primarily to support backwards compatibility.
+ # The preferred way of taking advantage of Rime spelling is
+ # to use an LLMTextProcessor and/or a text_transformer to identify
+ # and insert these tags for the purpose of the TTS service alone.
+ self._text_aggregator = SkipTagsAggregator([("spell(", ")")])
+
params = params or RimeTTSService.InputParams()
# Store service configuration
@@ -152,6 +163,7 @@ class RimeTTSService(AudioContextWordTTSService):
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
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -181,6 +193,31 @@ class RimeTTSService(AudioContextWordTTSService):
self._model = model
await super().set_model(model)
+ # A set of Rime-specific helpers for text transformations
+ def SPELL(text: str) -> str:
+ """Wrap text in Rime spell function."""
+ return f"spell({text})"
+
+ def PAUSE_TAG(seconds: float) -> str:
+ """Convenience method to create a pause tag."""
+ return f"<{seconds * 1000}>"
+
+ def PRONOUNCE(self, text: str, word: str, phoneme: str) -> str:
+ """Convenience method to support Rime's custom pronunciations feature.
+
+ https://docs.rime.ai/api-reference/custom-pronunciation
+ """
+ self._extra_msg_fields["phonemizeBetweenBrackets"] = True
+ return text.replace(word, f"{phoneme}")
+
+ def INLINE_SPEED(self, text: str, speed: float) -> str:
+ """Convenience method to support inline speeds."""
+ if not self._extra_msg_fields:
+ self._extra_msg_fields = {}
+ speed_vals = self._extra_msg_fields.get("inlineSpeedAlpha", "").split(",")
+ self._extra_msg_fields["inlineSpeedAlpha"] = ",".join(speed_vals + [str(speed)])
+ return f"[{text}]"
+
async def _update_settings(self, settings: Mapping[str, Any]):
"""Update service settings and reconnect if voice changed."""
prev_voice = self._voice_id
@@ -193,7 +230,11 @@ class RimeTTSService(AudioContextWordTTSService):
def _build_msg(self, text: str = "") -> dict:
"""Build JSON message for Rime API."""
- return {"text": text, "contextId": self._context_id}
+ msg = {"text": text, "contextId": self._context_id}
+ if self._extra_msg_fields:
+ msg |= self._extra_msg_fields
+ self._extra_msg_fields = {}
+ return msg
def _build_clear_msg(self) -> dict:
"""Build clear operation message."""
diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py
index f0d602a40..8d7c4e6bb 100644
--- a/src/pipecat/services/tts_service.py
+++ b/src/pipecat/services/tts_service.py
@@ -12,6 +12,8 @@ from typing import (
Any,
AsyncGenerator,
AsyncIterator,
+ Awaitable,
+ Callable,
Dict,
List,
Mapping,
@@ -23,6 +25,8 @@ from typing import (
from loguru import logger
from pipecat.frames.frames import (
+ AggregatedTextFrame,
+ AggregationType,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -101,6 +105,16 @@ class TTSService(AIService):
sample_rate: Optional[int] = None,
# Text aggregator to aggregate incoming tokens and decide when to push to the TTS.
text_aggregator: Optional[BaseTextAggregator] = None,
+ # Types of text aggregations that should not be spoken.
+ skip_aggregator_types: Optional[List[str]] = [],
+ # A list of callables to transform text before just before sending it to TTS.
+ # Each callable takes the aggregated text and its type, and returns the transformed text.
+ # To register, provide a list of tuples of (aggregation_type | '*', transform_function).
+ text_transforms: Optional[
+ List[
+ Tuple[AggregationType | str, Callable[[str, str | AggregationType], Awaitable[str]]]
+ ]
+ ] = None,
# Text filter executed after text has been aggregated.
text_filters: Optional[Sequence[BaseTextFilter]] = None,
text_filter: Optional[BaseTextFilter] = None,
@@ -120,6 +134,16 @@ class TTSService(AIService):
pause_frame_processing: Whether to pause frame processing during audio generation.
sample_rate: Output sample rate for generated audio.
text_aggregator: Custom text aggregator for processing incoming text.
+
+ .. deprecated:: 0.0.95
+ Use an LLMTextProcessor before the TTSService for custom text aggregation.
+
+ skip_aggregator_types: List of aggregation types that should not be spoken.
+ text_transforms: A list of callables to transform text before just before sending it
+ to TTS. Each callable takes the aggregated text and its type, and returns the
+ transformed text. To register, provide a list of tuples of
+ (aggregation_type | '*', transform_function).
+
text_filters: Sequence of text filters to apply after aggregation.
text_filter: Single text filter (deprecated, use text_filters).
@@ -142,7 +166,21 @@ class TTSService(AIService):
self._voice_id: str = ""
self._settings: Dict[str, Any] = {}
self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator()
- self._aggregated_text_includes_inter_frame_spaces: bool = False
+ if text_aggregator:
+ import warnings
+
+ with warnings.catch_warnings():
+ warnings.simplefilter("always")
+ warnings.warn(
+ "Parameter 'text_aggregator' is deprecated. Use an LLMTextProcessor before the TTSService for custom text aggregation.",
+ DeprecationWarning,
+ )
+
+ self._skip_aggregator_types: List[str] = skip_aggregator_types or []
+ self._text_transforms: List[
+ Tuple[AggregationType | str, Callable[[str, AggregationType | str], Awaitable[str]]]
+ ] = text_transforms or []
+ # TODO: Deprecate _text_filters when added to LLMTextProcessor
self._text_filters: Sequence[BaseTextFilter] = text_filters or []
self._transport_destination: Optional[str] = transport_destination
self._tracing_enabled: bool = False
@@ -282,6 +320,39 @@ class TTSService(AIService):
await self.cancel_task(self._stop_frame_task)
self._stop_frame_task = None
+ def add_text_transformer(
+ self,
+ transform_function: Callable[[str, AggregationType | str], Awaitable[str]],
+ aggregation_type: AggregationType | str = "*",
+ ):
+ """Transform text for a specific aggregation type.
+
+ Args:
+ transform_function: The function to apply for transformation. This function should take
+ the text and aggregation type as input and return the transformed text.
+ Ex.: async def my_transform(text: str, aggregation_type: str) -> str:
+ aggregation_type: The type of aggregation to transform. This value defaults to "*" indicating
+ the function should handle all text before sending to TTS.
+ """
+ self._text_transforms.append((aggregation_type, transform_function))
+
+ def remove_text_transformer(
+ self,
+ transform_function: Callable[[str, AggregationType | str], Awaitable[str]],
+ aggregation_type: AggregationType | str = "*",
+ ):
+ """Remove a text transformer for a specific aggregation type.
+
+ Args:
+ transform_function: The function to remove.
+ aggregation_type: The type of aggregation to remove the transformer for.
+ """
+ self._text_transforms = [
+ (agg_type, func)
+ for agg_type, func in self._text_transforms
+ if not (agg_type == aggregation_type and func == transform_function)
+ ]
+
async def _update_settings(self, settings: Mapping[str, Any]):
for key, value in settings.items():
if key in self._settings:
@@ -337,6 +408,8 @@ class TTSService(AIService):
and frame.skip_tts
):
await self.push_frame(frame, direction)
+ elif isinstance(frame, AggregatedTextFrame):
+ await self._push_tts_frames(frame)
elif (
isinstance(frame, TextFrame)
and not isinstance(frame, InterimTranscriptionFrame)
@@ -352,17 +425,16 @@ class TTSService(AIService):
# pause to avoid audio overlapping.
await self._maybe_pause_frame_processing()
- sentence = self._text_aggregator.text
- includes_inter_frame_spaces = self._aggregated_text_includes_inter_frame_spaces
+ pending_aggregation = self._text_aggregator.text
# Reset aggregator state
await self._text_aggregator.reset()
self._processing_text = False
- self._aggregated_text_includes_inter_frame_spaces = False
- await self._push_tts_frames(
- sentence, includes_inter_frame_spaces=includes_inter_frame_spaces
- )
+ if pending_aggregation.text:
+ await self._push_tts_frames(
+ AggregatedTextFrame(pending_aggregation.text, pending_aggregation.type)
+ )
if isinstance(frame, LLMFullResponseEndFrame):
if self._push_text_frames:
await self.push_frame(frame, direction)
@@ -372,7 +444,7 @@ 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(frame.text, includes_inter_frame_spaces=False)
+ await self._push_tts_frames(AggregatedTextFrame(frame.text, AggregationType.SENTENCE))
# 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()
@@ -462,21 +534,35 @@ class TTSService(AIService):
async def _process_text_frame(self, frame: TextFrame):
text: Optional[str] = None
+ includes_inter_frame_spaces: bool = False
if not self._aggregate_sentences:
text = frame.text
+ includes_inter_frame_spaces = frame.includes_inter_frame_spaces
+ aggregated_by = "token"
else:
- text = await self._text_aggregator.aggregate(frame.text)
- # Assumption: whether inter-frame spaces are included shouldn't
- # change during aggregation, so we can just use the latest frame's
- # value
- self._aggregated_text_includes_inter_frame_spaces = frame.includes_inter_frame_spaces
+ aggregate = await self._text_aggregator.aggregate(frame.text)
+ if aggregate:
+ text = aggregate.text
+ aggregated_by = aggregate.type
if text:
+ logger.trace(f"Pushing TTS frames for text: {text}, {aggregated_by}")
await self._push_tts_frames(
- text, includes_inter_frame_spaces=frame.includes_inter_frame_spaces
+ AggregatedTextFrame(text, aggregated_by), includes_inter_frame_spaces
)
- async def _push_tts_frames(self, text: str, includes_inter_frame_spaces: bool):
+ async def _push_tts_frames(
+ self, src_frame: AggregatedTextFrame, includes_inter_frame_spaces: Optional[bool] = False
+ ):
+ type = src_frame.aggregated_by
+ text = src_frame.text
+
+ # Skip sending to TTS if the aggregation type is in the skip list. Simply
+ # push the original frame downstream.
+ if type in self._skip_aggregator_types:
+ await self.push_frame(src_frame)
+ return
+
# Remove leading newlines only
text = text.lstrip("\n")
@@ -492,20 +578,44 @@ class TTSService(AIService):
await self.start_processing_metrics()
- # Process all filter.
+ # Process all filters.
for filter in self._text_filters:
await filter.reset_interruption()
text = await filter.filter(text)
- if text:
- await self.process_generator(self.run_tts(text))
+ if not text.strip():
+ await self.stop_processing_metrics()
+ return
+
+ # 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
+ await self.push_frame(src_frame)
+
+ # Note: Text transformations are meant to only affect the text sent to the TTS for
+ # TTS-specific purposes. This allows for explicit TTS modifications (e.g., inserting
+ # TTS supported tags for spelling or emotion or replacing an @ with "at"). For TTS
+ # services that support word-level timestamps, this CAN affect the resulting context
+ # since the TTSTextFrames are generated from the TTS output stream
+ transformed_text = text
+ for aggregation_type, transform in self._text_transforms:
+ if aggregation_type == type or aggregation_type == "*":
+ transformed_text = await transform(transformed_text, type)
+ await self.process_generator(self.run_tts(transformed_text))
await self.stop_processing_metrics()
if self._push_text_frames:
- # We send the original text after the audio. This way, if we are
- # interrupted, the text is not added to the assistant context.
- frame = TTSTextFrame(text)
+ # In TTS services that support word timestamps, the TTSTextFrames
+ # are pushed as words are spoken. However, in the case where the TTS service
+ # does not support word timestamps (i.e. _push_text_frames is True), we send
+ # the original (non-transformed) text after the TTS generation has completed.
+ # This way, if we are interrupted, the text is not added to the assistant
+ # context and the context that IS added does not include TTS-specific tags
+ # or transformations.
+ frame = TTSTextFrame(text, aggregated_by=type)
frame.includes_inter_frame_spaces = includes_inter_frame_spaces
await self.push_frame(frame)
@@ -635,7 +745,7 @@ class WordTTSService(TTSService):
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)
+ frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD)
frame.pts = self._initial_word_timestamp + timestamp
if frame:
last_pts = frame.pts
diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py
index 6ccce4b31..94b8cb1a4 100644
--- a/src/pipecat/tests/utils.py
+++ b/src/pipecat/tests/utils.py
@@ -203,8 +203,16 @@ async def run_test(
if not isinstance(frame, EndFrame) or not send_end_frame:
received_down_frames.append(frame)
- print("received DOWN frames =", received_down_frames)
- print("expected DOWN frames =", expected_down_frames)
+ down_frames_printed = "["
+ for frame in received_down_frames:
+ down_frames_printed += f"{frame.__class__.__name__}, "
+ down_frames_printed += "]"
+ expected_frames_printed = "["
+ for frame in expected_down_frames:
+ expected_frames_printed += f"{frame.__name__}, "
+ expected_frames_printed += "]"
+ print("received DOWN frames =", down_frames_printed)
+ print("expected DOWN frames =", expected_frames_printed)
assert len(received_down_frames) == len(expected_down_frames)
diff --git a/src/pipecat/utils/text/base_text_aggregator.py b/src/pipecat/utils/text/base_text_aggregator.py
index 27e50fff5..5c39ce769 100644
--- a/src/pipecat/utils/text/base_text_aggregator.py
+++ b/src/pipecat/utils/text/base_text_aggregator.py
@@ -12,9 +12,47 @@ aggregated text should be sent for speech synthesis.
"""
from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
from typing import Optional
+class AggregationType(str, Enum):
+ """Built-in aggregation strings."""
+
+ SENTENCE = "sentence"
+ WORD = "word"
+
+ def __str__(self):
+ return self.value
+
+
+@dataclass
+class Aggregation:
+ """Data class representing aggregated text and its type.
+
+ An Aggregation object is created whenever a stream of text is aggregated by
+ a text aggregator. It contains the aggregated text and a type indicating
+ the nature of the aggregation.
+
+ Parameters:
+ text: The aggregated text content.
+ type: The type of aggregation the text represents (e.g., 'sentence', 'word', 'token',
+ 'my_custom_aggregation').
+ """
+
+ text: str
+ type: str
+
+ def __str__(self) -> str:
+ """Return a string representation of the aggregation.
+
+ Returns:
+ A descriptive string showing the type and text of the aggregation.
+ """
+ return f"Aggregation by {self.type}: {self.text}"
+
+
class BaseTextAggregator(ABC):
"""Base class for text aggregators in the Pipecat framework.
@@ -30,7 +68,7 @@ class BaseTextAggregator(ABC):
@property
@abstractmethod
- def text(self) -> str:
+ def text(self) -> Aggregation:
"""Get the currently aggregated text.
Subclasses must implement this property to return the text that has
@@ -42,25 +80,33 @@ class BaseTextAggregator(ABC):
pass
@abstractmethod
- async def aggregate(self, text: str) -> Optional[str]:
+ async def aggregate(self, text: str) -> Optional[Aggregation]:
"""Aggregate the specified text with the currently accumulated text.
This method should be implemented to define how the new text contributes
- to the aggregation process. It returns the updated aggregated text if
- it's ready to be processed, or None otherwise.
+ to the aggregation process. It returns the aggregated text and a string
+ describing how it was aggregated if it's ready to be processed,
+ or None otherwise.
Subclasses should implement their specific logic for:
- How to combine new text with existing accumulated text
- When to consider the aggregated text ready for processing
- What criteria determine text completion (e.g., sentence boundaries)
+ - When a completion occurs, the method should return an Aggregation object
+ containing the aggregated text and its type. The text should be stripped
+ of leading/trailing whitespace so that consumers can rely on a consistent
+ format.
Args:
text: The text to be aggregated.
Returns:
- The updated aggregated text if ready for processing, or None if more
- text is needed before the aggregated content is ready.
+ An Aggregation object if ready for processing, or None if more
+ text is needed before the aggregated content is ready. If an Aggregation
+ object is returned, it should consist of the updated aggregated text,
+ stripped of leading/trailing whitespace, and a string indicating the
+ type of aggregation (e.g., 'sentence', 'word', 'token', 'my_custom_aggregation').
"""
pass
diff --git a/src/pipecat/utils/text/pattern_pair_aggregator.py b/src/pipecat/utils/text/pattern_pair_aggregator.py
index ac074f2de..c140e3243 100644
--- a/src/pipecat/utils/text/pattern_pair_aggregator.py
+++ b/src/pipecat/utils/text/pattern_pair_aggregator.py
@@ -8,19 +8,41 @@
This module provides an aggregator that identifies and processes content between
pattern pairs (like XML tags or custom delimiters) in streaming text, with
-support for custom handlers and configurable pattern removal.
+support for custom handlers and configurable actions for when a pattern is found.
"""
import re
-from typing import Awaitable, Callable, Optional, Tuple
+from enum import Enum
+from typing import Awaitable, Callable, List, Optional, Tuple
from loguru import logger
from pipecat.utils.string import match_endofsentence
-from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator
-class PatternMatch:
+class MatchAction(Enum):
+ """Actions to take when a pattern pair is matched.
+
+ Parameters:
+ REMOVE: The text along with its delimiters will be removed from the streaming text.
+ Sentence aggregation will continue on as if this text did not exist.
+ KEEP: The delimiters will be removed, but the content between them will be kept.
+ Sentence aggregation will continue on with the internal text included.
+ AGGREGATE: The delimiters will be removed and the content between will be treated
+ as a separate aggregation. Any text before the start of the pattern will be
+ returned early, whether or not a complete sentence was found. Then the pattern
+ will be returned. Then the aggregation will continue on sentence matching after
+ the closing delimiter is found. The content between the delimiters is not
+ aggregated by sentence. It is aggregated as one single block of text.
+ """
+
+ REMOVE = "remove"
+ KEEP = "keep"
+ AGGREGATE = "aggregate"
+
+
+class PatternMatch(Aggregation):
"""Represents a matched pattern pair with its content.
A PatternMatch object is created when a complete pattern pair is found
@@ -29,25 +51,25 @@ class PatternMatch:
content between the patterns.
"""
- def __init__(self, pattern_id: str, full_match: str, content: str):
+ def __init__(self, content: str, type: str, full_match: str):
"""Initialize a pattern match.
Args:
- pattern_id: The identifier of the matched pattern pair.
+ type: The type of the matched pattern pair. It should be representative
+ of the content type (e.g., 'sentence', 'code', 'speaker', 'custom').
full_match: The complete text including start and end patterns.
content: The text content between the start and end patterns.
"""
- self.pattern_id = pattern_id
+ super().__init__(text=content, type=type)
self.full_match = full_match
- self.content = content
def __str__(self) -> str:
"""Return a string representation of the pattern match.
Returns:
- A descriptive string showing the pattern ID and content.
+ A descriptive string showing the pattern type and content.
"""
- return f"PatternMatch(id={self.pattern_id}, content={self.content})"
+ return f"PatternMatch(type={self.type}, text={self.text}, full_match={self.full_match})"
class PatternPairAggregator(BaseTextAggregator):
@@ -55,16 +77,21 @@ class PatternPairAggregator(BaseTextAggregator):
This aggregator buffers text until it can identify complete pattern pairs
(defined by start and end patterns), processes the content between these
- patterns using registered handlers, and returns text at sentence boundaries.
- It's particularly useful for processing structured content in streaming text,
- such as XML tags, markdown formatting, or custom delimiters.
+ patterns using registered handlers. By default, its aggregation method
+ returns text at sentence boundaries, and remove the content found between
+ any matched patterns. However, matched patterns can also be configured to
+ returned as a separate aggregation object containing the content between
+ their start and end patterns or left in, so that only the delimiters are
+ removed and a callback can be triggered.
+
+ This aggregator is particularly useful for processing structured content in
+ streaming text, such as XML tags, markdown formatting, or custom delimiters.
The aggregator ensures that patterns spanning multiple text chunks are
- correctly identified and handles cases where patterns contain sentence
- boundaries.
+ correctly identified.
"""
- def __init__(self):
+ def __init__(self, **kwargs):
"""Initialize the pattern pair aggregator.
Creates an empty aggregator with no patterns or handlers registered.
@@ -75,16 +102,27 @@ class PatternPairAggregator(BaseTextAggregator):
self._handlers = {}
@property
- def text(self) -> str:
- """Get the currently buffered text.
+ def text(self) -> Aggregation:
+ """Get the currently aggregated text.
Returns:
- The current text buffer content that hasn't been processed yet.
+ The text that has been accumulated in the buffer.
"""
- return self._text
+ pattern_start = self._match_start_of_pattern(self._text)
+ stripped_text = self._text.strip()
+ type = (
+ pattern_start[1].get("type", AggregationType.SENTENCE)
+ if pattern_start
+ else AggregationType.SENTENCE
+ )
+ return Aggregation(text=stripped_text, type=type)
- def add_pattern_pair(
- self, pattern_id: str, start_pattern: str, end_pattern: str, remove_match: bool = True
+ def add_pattern(
+ self,
+ type: str,
+ start_pattern: str,
+ end_pattern: str,
+ action: MatchAction = MatchAction.REMOVE,
) -> "PatternPairAggregator":
"""Add a pattern pair to detect in the text.
@@ -93,41 +131,94 @@ class PatternPairAggregator(BaseTextAggregator):
the end pattern, and treat the content between them as a match.
Args:
- pattern_id: Unique identifier for this pattern pair.
+ type: Identifier for this pattern pair. Should be unique and ideally descriptive.
+ (e.g., 'code', 'speaker', 'custom'). type can not be 'sentence' or 'word' as
+ those are reserved for the default behavior.
start_pattern: Pattern that marks the beginning of content.
end_pattern: Pattern that marks the end of content.
- remove_match: Whether to remove the matched content from the text.
+ action: What to do when a complete pattern is matched:
+ - MatchAction.REMOVE: Remove the matched pattern from the text.
+ - MatchAction.KEEP: Keep the matched pattern in the text and treat it as
+ normal text. This allows you to register handlers for
+ the pattern without affecting the aggregation logic.
+ - MatchAction.AGGREGATE: Return the matched pattern as a separate
+ aggregation object.
Returns:
Self for method chaining.
"""
- self._patterns[pattern_id] = {
+ if type in [AggregationType.SENTENCE, AggregationType.WORD]:
+ raise ValueError(
+ f"The aggregation type '{type}' is reserved for default behavior and can not be used for custom patterns."
+ )
+ self._patterns[type] = {
"start": start_pattern,
"end": end_pattern,
- "remove_match": remove_match,
+ "type": type,
+ "action": action,
}
return self
+ def add_pattern_pair(
+ self, pattern_id: str, start_pattern: str, end_pattern: str, remove_match: bool = True
+ ):
+ """Add a pattern pair to detect in the text.
+
+ .. deprecated:: 0.0.95
+ This function is deprecated and will be removed in a future version.
+ Use `add_pattern` with a type and MatchAction instead.
+
+ This method calls `add_pattern` setting type with the provided pattern_id and action
+ to either MatchAction.REMOVE or MatchAction.KEEP based on `remove_match`.
+
+ Args:
+ pattern_id: Identifier for this pattern pair. Should be unique and ideally descriptive.
+ (e.g., 'code', 'speaker', 'custom'). pattern_id can not be 'sentence' or 'word'
+ as those arereserved for the default behavior.
+ start_pattern: Pattern that marks the beginning of content.
+ end_pattern: Pattern that marks the end of content.
+ remove_match: If True, the matched pattern will be removed from the text. (Same as MatchAction.REMOVE)
+ If False, it will be kept and treated as normal text. (Same as MatchAction.KEEP)
+ """
+ import warnings
+
+ with warnings.catch_warnings():
+ warnings.simplefilter("once")
+ warnings.warn(
+ "add_pattern_pair with a pattern_id or remove_match is deprecated and will be"
+ " removed in a future version. Use add_pattern with a type and MatchAction instead",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+
+ action = MatchAction.REMOVE if remove_match else MatchAction.KEEP
+ return self.add_pattern(
+ type=pattern_id,
+ start_pattern=start_pattern,
+ end_pattern=end_pattern,
+ action=action,
+ )
+
def on_pattern_match(
- self, pattern_id: str, handler: Callable[[PatternMatch], Awaitable[None]]
+ self, type: str, handler: Callable[[PatternMatch], Awaitable[None]]
) -> "PatternPairAggregator":
"""Register a handler for when a pattern pair is matched.
The handler will be called whenever a complete match for the
- specified pattern ID is found in the text.
+ specified type is found in the text.
Args:
- pattern_id: ID of the pattern pair to match.
+ type: The type of the pattern pair to trigger the handler.
handler: Async function to call when pattern is matched.
The function should accept a PatternMatch object.
Returns:
Self for method chaining.
"""
- self._handlers[pattern_id] = handler
+ self._handlers[type] = handler
return self
- async def _process_complete_patterns(self, text: str) -> Tuple[str, bool]:
+ async def _process_complete_patterns(self, text: str) -> Tuple[List[PatternMatch], str]:
"""Process all complete pattern pairs in the text.
Searches for all complete pattern pairs in the text, calls the
@@ -137,19 +228,19 @@ class PatternPairAggregator(BaseTextAggregator):
text: The text to process for pattern matches.
Returns:
- Tuple of (processed_text, was_modified) where:
+ Tuple of (all_matches, processed_text) where:
- - processed_text is the text after processing patterns
- - was_modified indicates whether any changes were made
+ - all_matches is a list of all pattern matches found. Note: There really should only ever be 1.
+ - processed_text is the text after processing patterns. If no patterns are found, it will be the same as input text.
"""
+ all_matches = []
processed_text = text
- modified = False
- for pattern_id, pattern_info in self._patterns.items():
+ for type, pattern_info in self._patterns.items():
# Escape special regex characters in the patterns
start = re.escape(pattern_info["start"])
end = re.escape(pattern_info["end"])
- remove_match = pattern_info["remove_match"]
+ action = pattern_info["action"]
# Create regex to match from start pattern to end pattern
# The .*? is non-greedy to handle nested patterns
@@ -165,24 +256,25 @@ class PatternPairAggregator(BaseTextAggregator):
# Create pattern match object
pattern_match = PatternMatch(
- pattern_id=pattern_id, full_match=full_match, content=content
+ content=content.strip(), type=type, full_match=full_match
)
# Call the appropriate handler if registered
- if pattern_id in self._handlers:
+ if type in self._handlers:
try:
- await self._handlers[pattern_id](pattern_match)
+ await self._handlers[type](pattern_match)
except Exception as e:
- logger.error(f"Error in pattern handler for {pattern_id}: {e}")
+ logger.error(f"Error in pattern handler for {type}: {e}")
# Remove the pattern from the text if configured
- if remove_match:
+ if action == MatchAction.REMOVE:
processed_text = processed_text.replace(full_match, "", 1)
- modified = True
+ else:
+ all_matches.append(pattern_match)
- return processed_text, modified
+ return all_matches, processed_text
- def _has_incomplete_patterns(self, text: str) -> bool:
+ def _match_start_of_pattern(self, text: str) -> Optional[Tuple[int, dict]]:
"""Check if text contains incomplete pattern pairs.
Determines whether the text contains any start patterns without
@@ -192,9 +284,10 @@ class PatternPairAggregator(BaseTextAggregator):
text: The text to check for incomplete patterns.
Returns:
- True if there are incomplete patterns, False otherwise.
+ A tuple of (start_index, pattern_info) if an incomplete pattern is found,
+ or None if no patterns are found or all patterns are complete.
"""
- for pattern_id, pattern_info in self._patterns.items():
+ for type, pattern_info in self._patterns.items():
start = pattern_info["start"]
end = pattern_info["end"]
@@ -203,12 +296,16 @@ class PatternPairAggregator(BaseTextAggregator):
end_count = text.count(end)
# If there are more starts than ends, we have incomplete patterns
+ # Again, this is written generically but there only ever should
+ # be one pattern active at a time, so the counts should be 0 or 1.
+ # Which is why we base the return on the first found.
if start_count > end_count:
- return True
+ start_index = text.find(start)
+ return [start_index, pattern_info]
- return False
+ return None
- async def aggregate(self, text: str) -> Optional[str]:
+ async def aggregate(self, text: str) -> Optional[PatternMatch]:
"""Aggregate text and process pattern pairs.
This method adds the new text to the buffer, processes any complete pattern
@@ -227,16 +324,36 @@ class PatternPairAggregator(BaseTextAggregator):
self._text += text
# Process any complete patterns in the buffer
- processed_text, modified = await self._process_complete_patterns(self._text)
+ patterns, processed_text = await self._process_complete_patterns(self._text)
- # Only update the buffer if modifications were made
- if modified:
- self._text = processed_text
+ self._text = processed_text
+
+ if len(patterns) > 0:
+ if len(patterns) > 1:
+ logger.warning(
+ f"Multiple patterns matched: {[p.type for p in patterns]}. Only the first pattern will be returned."
+ )
+ # If the pattern found is set to be aggregated, return it
+ action = self._patterns[patterns[0].type].get("action", MatchAction.REMOVE)
+ if action == MatchAction.AGGREGATE:
+ self._text = ""
+ return patterns[0]
# Check if we have incomplete patterns
- if self._has_incomplete_patterns(self._text):
- # Still waiting for complete patterns
- return None
+ pattern_start = self._match_start_of_pattern(self._text)
+ if pattern_start is not None:
+ # If the start pattern is at the beginning or should not be separately aggregated, return None
+ if (
+ pattern_start[0] == 0
+ or pattern_start[1].get("action", MatchAction.REMOVE) != MatchAction.AGGREGATE
+ ):
+ return None
+ # Otherwise, strip the text up to the start pattern and return it
+ result = self._text[: pattern_start[0]]
+ self._text = self._text[pattern_start[0] :]
+ return PatternMatch(
+ content=result.strip(), type=AggregationType.SENTENCE, full_match=result
+ )
# Find sentence boundary if no incomplete patterns
eos_marker = match_endofsentence(self._text)
@@ -244,7 +361,9 @@ class PatternPairAggregator(BaseTextAggregator):
# Extract text up to the sentence boundary
result = self._text[:eos_marker]
self._text = self._text[eos_marker:]
- return result
+ return PatternMatch(
+ content=result.strip(), type=AggregationType.SENTENCE, full_match=result
+ )
# No complete sentence found yet
return None
diff --git a/src/pipecat/utils/text/simple_text_aggregator.py b/src/pipecat/utils/text/simple_text_aggregator.py
index f9eb7d83a..56eab7032 100644
--- a/src/pipecat/utils/text/simple_text_aggregator.py
+++ b/src/pipecat/utils/text/simple_text_aggregator.py
@@ -14,7 +14,7 @@ text processing scenarios.
from typing import Optional
from pipecat.utils.string import match_endofsentence
-from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator
class SimpleTextAggregator(BaseTextAggregator):
@@ -33,15 +33,15 @@ class SimpleTextAggregator(BaseTextAggregator):
self._text = ""
@property
- def text(self) -> str:
+ def text(self) -> Aggregation:
"""Get the currently aggregated text.
Returns:
The text that has been accumulated in the buffer.
"""
- return self._text
+ return Aggregation(text=self._text.strip(), type=AggregationType.SENTENCE)
- async def aggregate(self, text: str) -> Optional[str]:
+ async def aggregate(self, text: str) -> Optional[Aggregation]:
"""Aggregate text and return completed sentences.
Adds the new text to the buffer and checks for end-of-sentence markers.
@@ -64,7 +64,9 @@ class SimpleTextAggregator(BaseTextAggregator):
result = self._text[:eos_end_marker]
self._text = self._text[eos_end_marker:]
- return result
+ if result:
+ return Aggregation(text=result.strip(), type=AggregationType.SENTENCE)
+ return None
async def handle_interruption(self):
"""Handle interruptions by clearing the text buffer.
diff --git a/src/pipecat/utils/text/skip_tags_aggregator.py b/src/pipecat/utils/text/skip_tags_aggregator.py
index 6f6f8455c..3c8b95aab 100644
--- a/src/pipecat/utils/text/skip_tags_aggregator.py
+++ b/src/pipecat/utils/text/skip_tags_aggregator.py
@@ -14,7 +14,7 @@ as a unit regardless of internal punctuation.
from typing import Optional, Sequence
from pipecat.utils.string import StartEndTags, match_endofsentence, parse_start_end_tags
-from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+from pipecat.utils.text.base_text_aggregator import Aggregation, AggregationType, BaseTextAggregator
class SkipTagsAggregator(BaseTextAggregator):
@@ -43,15 +43,15 @@ class SkipTagsAggregator(BaseTextAggregator):
self._current_tag_index: int = 0
@property
- def text(self) -> str:
+ def text(self) -> Aggregation:
"""Get the currently buffered text.
Returns:
The current text buffer content that hasn't been processed yet.
"""
- return self._text
+ return Aggregation(text=self._text.strip(), type=AggregationType.SENTENCE)
- async def aggregate(self, text: str) -> Optional[str]:
+ async def aggregate(self, text: str) -> Optional[Aggregation]:
"""Aggregate text while respecting tag boundaries.
This method adds the new text to the buffer, processes any complete
@@ -63,8 +63,9 @@ class SkipTagsAggregator(BaseTextAggregator):
text: New text to add to the buffer.
Returns:
- Processed text up to a sentence boundary (when not within tags),
- or None if more text is needed to complete a sentence or close tags.
+ An Aggregation object containing text up to a sentence boundary and
+ marked as SENTENCE type or None if more text is needed to complete a
+ sentence or close tags.
"""
# Add new text to buffer
self._text += text
@@ -80,7 +81,7 @@ class SkipTagsAggregator(BaseTextAggregator):
# Extract text up to the sentence boundary
result = self._text[:eos_marker]
self._text = self._text[eos_marker:]
- return result
+ return Aggregation(text=result.strip(), type=AggregationType.SENTENCE)
# No complete sentence found yet
return None
diff --git a/tests/test_pattern_pair_aggregator.py b/tests/test_pattern_pair_aggregator.py
index 8426dcf39..20b44a03c 100644
--- a/tests/test_pattern_pair_aggregator.py
+++ b/tests/test_pattern_pair_aggregator.py
@@ -7,30 +7,42 @@
import unittest
from unittest.mock import AsyncMock
-from pipecat.utils.text.pattern_pair_aggregator import PatternMatch, PatternPairAggregator
+from pipecat.utils.text.pattern_pair_aggregator import (
+ MatchAction,
+ PatternMatch,
+ PatternPairAggregator,
+)
class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
def setUp(self):
self.aggregator = PatternPairAggregator()
self.test_handler = AsyncMock()
+ self.code_handler = AsyncMock()
# Add a test pattern
self.aggregator.add_pattern_pair(
pattern_id="test_pattern",
start_pattern="",
end_pattern="",
- remove_match=True,
+ )
+ self.aggregator.add_pattern(
+ type="code_pattern",
+ start_pattern="",
+ end_pattern="",
+ action=MatchAction.AGGREGATE,
)
# Register the mock handler
self.aggregator.on_pattern_match("test_pattern", self.test_handler)
+ self.aggregator.on_pattern_match("code_pattern", self.code_handler)
async def test_pattern_match_and_removal(self):
# First part doesn't complete the pattern
result = await self.aggregator.aggregate("Hello pattern")
self.assertIsNone(result)
- self.assertEqual(self.aggregator.text, "Hello pattern")
+ self.assertEqual(self.aggregator.text.text, "Hello pattern")
+ self.assertEqual(self.aggregator.text.type, "test_pattern")
# Second part completes the pattern and includes an exclamation point
result = await self.aggregator.aggregate(" content!")
@@ -39,20 +51,50 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
self.test_handler.assert_called_once()
call_args = self.test_handler.call_args[0][0]
self.assertIsInstance(call_args, PatternMatch)
- self.assertEqual(call_args.pattern_id, "test_pattern")
+ self.assertEqual(call_args.type, "test_pattern")
self.assertEqual(call_args.full_match, "pattern content")
- self.assertEqual(call_args.content, "pattern content")
+ self.assertEqual(call_args.text, "pattern content")
# The exclamation point should be treated as a sentence boundary,
# so the result should include just text up to and including "!"
- self.assertEqual(result, "Hello !")
+ self.assertEqual(result.text, "Hello !")
+ self.assertEqual(result.type, "sentence")
+
+ # Next sentence should be processed separately. Spaces around the sentence
+ # should be stripped in the returned Aggregation.
+ result = await self.aggregator.aggregate(" This is another sentence.")
+ self.assertEqual(result.text, "This is another sentence.")
+
+ # Buffer should be empty after returning a complete sentence
+ self.assertEqual(self.aggregator.text.text, "")
+
+ async def test_pattern_match_and_aggregate(self):
+ # First part doesn't complete the pattern
+ result = await self.aggregator.aggregate("Here is code pattern")
+ self.assertEqual(result.text, "Here is code")
+ self.assertEqual(self.aggregator.text.text, "pattern")
+ self.assertEqual(self.aggregator.text.type, "code_pattern")
+
+ # Second part completes the pattern and includes an exclamation point
+ result = await self.aggregator.aggregate(" content")
+
+ # Verify the handler was called with correct PatternMatch object
+ self.code_handler.assert_called_once()
+ call_args = self.code_handler.call_args[0][0]
+ self.assertIsInstance(call_args, PatternMatch)
+ self.assertEqual(call_args.type, "code_pattern")
+ self.assertEqual(call_args.full_match, "pattern content")
+ self.assertEqual(call_args.text, "pattern content")
+ self.assertEqual(result.text, "pattern content")
+ self.assertEqual(result.type, "code_pattern")
# Next sentence should be processed separately
result = await self.aggregator.aggregate(" This is another sentence.")
- self.assertEqual(result, " This is another sentence.")
+ self.assertEqual(result.text, "This is another sentence.")
+ self.assertEqual(result.type, "sentence")
# Buffer should be empty after returning a complete sentence
- self.assertEqual(self.aggregator.text, "")
+ self.assertEqual(self.aggregator.text.text, "")
async def test_incomplete_pattern(self):
# Add text with incomplete pattern
@@ -65,26 +107,30 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
self.test_handler.assert_not_called()
# Buffer should contain the incomplete text
- self.assertEqual(self.aggregator.text, "Hello pattern content")
+ self.assertEqual(self.aggregator.text.text, "Hello pattern content")
+ self.assertEqual(self.aggregator.text.type, "test_pattern")
# Reset and confirm buffer is cleared
await self.aggregator.reset()
- self.assertEqual(self.aggregator.text, "")
+ self.assertEqual(self.aggregator.text.text, "")
async def test_multiple_patterns(self):
# Set up multiple patterns and handlers
voice_handler = AsyncMock()
emphasis_handler = AsyncMock()
- self.aggregator.add_pattern_pair(
- pattern_id="voice", start_pattern="", end_pattern="", remove_match=True
+ self.aggregator.add_pattern(
+ type="voice",
+ start_pattern="",
+ end_pattern="",
+ action=MatchAction.REMOVE,
)
- self.aggregator.add_pattern_pair(
- pattern_id="emphasis",
+ self.aggregator.add_pattern(
+ type="emphasis",
start_pattern="",
end_pattern="",
- remove_match=False, # Keep emphasis tags
+ action=MatchAction.KEEP, # Keep emphasis tags
)
self.aggregator.on_pattern_match("voice", voice_handler)
@@ -97,19 +143,19 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
# Both handlers should be called with correct data
voice_handler.assert_called_once()
voice_match = voice_handler.call_args[0][0]
- self.assertEqual(voice_match.pattern_id, "voice")
- self.assertEqual(voice_match.content, "female")
+ self.assertEqual(voice_match.type, "voice")
+ self.assertEqual(voice_match.text, "female")
emphasis_handler.assert_called_once()
emphasis_match = emphasis_handler.call_args[0][0]
- self.assertEqual(emphasis_match.pattern_id, "emphasis")
- self.assertEqual(emphasis_match.content, "very")
+ self.assertEqual(emphasis_match.type, "emphasis")
+ self.assertEqual(emphasis_match.text, "very")
# Voice pattern should be removed, emphasis pattern should remain
- self.assertEqual(result, "Hello I am very excited to meet you!")
+ self.assertEqual(result.text, "Hello I am very excited to meet you!")
# Buffer should be empty
- self.assertEqual(self.aggregator.text, "")
+ self.assertEqual(self.aggregator.text.text, "")
async def test_handle_interruption(self):
# Start with incomplete pattern
@@ -120,7 +166,7 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
await self.aggregator.handle_interruption()
# Buffer should be cleared
- self.assertEqual(self.aggregator.text, "")
+ self.assertEqual(self.aggregator.text.text, "")
# Handler should not have been called
self.test_handler.assert_not_called()
@@ -138,10 +184,10 @@ class TestPatternPairAggregator(unittest.IsolatedAsyncioTestCase):
# Handler should be called with entire content
self.test_handler.assert_called_once()
call_args = self.test_handler.call_args[0][0]
- self.assertEqual(call_args.content, "This is sentence one. This is sentence two.")
+ self.assertEqual(call_args.text, "This is sentence one. This is sentence two.")
# Pattern should be removed, resulting in text with sentences merged
- self.assertEqual(result, "Hello Final sentence.")
+ self.assertEqual(result.text, "Hello Final sentence.")
# Buffer should be empty
- self.assertEqual(self.aggregator.text, "")
+ self.assertEqual(self.aggregator.text.text, "")
diff --git a/tests/test_piper_tts.py b/tests/test_piper_tts.py
index 75893f93f..a006f555c 100644
--- a/tests/test_piper_tts.py
+++ b/tests/test_piper_tts.py
@@ -13,6 +13,7 @@ import pytest
from aiohttp import web
from pipecat.frames.frames import (
+ AggregatedTextFrame,
ErrorFrame,
TTSAudioRawFrame,
TTSSpeakFrame,
@@ -74,6 +75,7 @@ async def test_run_piper_tts_success(aiohttp_client):
]
expected_returned_frames = [
+ AggregatedTextFrame,
TTSStartedFrame,
TTSAudioRawFrame,
TTSAudioRawFrame,
@@ -121,7 +123,7 @@ async def test_run_piper_tts_error(aiohttp_client):
TTSSpeakFrame(text="Error case."),
]
- expected_down_frames = [TTSStoppedFrame, TTSTextFrame]
+ expected_down_frames = [AggregatedTextFrame, TTSStoppedFrame, TTSTextFrame]
expected_up_frames = [ErrorFrame]
diff --git a/tests/test_simple_text_aggregator.py b/tests/test_simple_text_aggregator.py
index ff6dd1847..f8e2ee553 100644
--- a/tests/test_simple_text_aggregator.py
+++ b/tests/test_simple_text_aggregator.py
@@ -15,15 +15,21 @@ class TestSimpleTextAggregator(unittest.IsolatedAsyncioTestCase):
async def test_reset_aggregations(self):
assert await self.aggregator.aggregate("Hello ") == None
- assert self.aggregator.text == "Hello "
+ assert self.aggregator.text.text == "Hello"
await self.aggregator.reset()
- assert self.aggregator.text == ""
+ assert self.aggregator.text.text == ""
async def test_simple_sentence(self):
assert await self.aggregator.aggregate("Hello ") == None
- assert await self.aggregator.aggregate("Pipecat!") == "Hello Pipecat!"
- assert self.aggregator.text == ""
+ aggregate = await self.aggregator.aggregate("Pipecat!")
+ assert aggregate.text == "Hello Pipecat!"
+ assert aggregate.type == "sentence"
+ assert self.aggregator.text.text == ""
async def test_multiple_sentences(self):
- assert await self.aggregator.aggregate("Hello Pipecat! How are ") == "Hello Pipecat!"
- assert await self.aggregator.aggregate("you?") == " How are you?"
+ aggregate = await self.aggregator.aggregate("Hello Pipecat! How are ")
+ assert aggregate.text == "Hello Pipecat!"
+ # Aggregators should strip leading/trailing spaces when returning text
+ assert self.aggregator.text.text == "How are"
+ aggregate = await self.aggregator.aggregate("you?")
+ assert aggregate.text == "How are you?"
diff --git a/tests/test_skip_tags_aggregator.py b/tests/test_skip_tags_aggregator.py
index f6cbb7b93..702b991ce 100644
--- a/tests/test_skip_tags_aggregator.py
+++ b/tests/test_skip_tags_aggregator.py
@@ -18,16 +18,18 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
# No tags involved, aggregate at end of sentence.
result = await self.aggregator.aggregate("Hello Pipecat!")
- self.assertEqual(result, "Hello Pipecat!")
- self.assertEqual(self.aggregator.text, "")
+ self.assertEqual(result.text, "Hello Pipecat!")
+ self.assertEqual(result.type, "sentence")
+ self.assertEqual(self.aggregator.text.text, "")
async def test_basic_tags(self):
await self.aggregator.reset()
# Tags involved, avoid aggregation during tags.
result = await self.aggregator.aggregate("My email is foo@pipecat.ai.")
- self.assertEqual(result, "My email is foo@pipecat.ai.")
- self.assertEqual(self.aggregator.text, "")
+ self.assertEqual(result.text, "My email is foo@pipecat.ai.")
+ self.assertEqual(result.type, "sentence")
+ self.assertEqual(self.aggregator.text.text, "")
async def test_streaming_tags(self):
await self.aggregator.reset()
@@ -35,20 +37,22 @@ class TestSkipTagsAggregator(unittest.IsolatedAsyncioTestCase):
# Tags involved, stream small chunk of texts.
result = await self.aggregator.aggregate("My email is foo.")
self.assertIsNone(result)
- self.assertEqual(self.aggregator.text, "My email is foo.")
+ self.assertEqual(self.aggregator.text.text, "My email is foo.")
result = await self.aggregator.aggregate("bar@pipecat.")
self.assertIsNone(result)
- self.assertEqual(self.aggregator.text, "My email is foo.bar@pipecat.")
+ self.assertEqual(self.aggregator.text.text, "My email is foo.bar@pipecat.")
result = await self.aggregator.aggregate("aifoo.bar@pipecat.aifoo.bar@pipecat.ai.")
- self.assertEqual(result, "My email is foo.bar@pipecat.ai.")
- self.assertEqual(self.aggregator.text, "")
+ self.assertEqual(result.text, "My email is foo.bar@pipecat.ai.")
+ self.assertEqual(self.aggregator.text.text, "")
+ self.assertEqual(self.aggregator.text.type, "sentence")
diff --git a/tests/test_transcript_processor.py b/tests/test_transcript_processor.py
index 19366086c..d86e42101 100644
--- a/tests/test_transcript_processor.py
+++ b/tests/test_transcript_processor.py
@@ -11,6 +11,7 @@ from datetime import datetime, timezone
from typing import List, Tuple, cast
from pipecat.frames.frames import (
+ AggregationType,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -130,11 +131,11 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(), # Wait for StartedSpeaking to process
- TTSTextFrame(text="Hello"),
- TTSTextFrame(text="world!"),
- TTSTextFrame(text="How"),
- TTSTextFrame(text="are"),
- TTSTextFrame(text="you?"),
+ TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="How", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="are", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="you?", aggregated_by=AggregationType.WORD),
SleepFrame(), # Wait for text frames to queue
BotStoppedSpeakingFrame(),
]
@@ -195,9 +196,9 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(),
- TTSTextFrame(text=""), # Empty text
- TTSTextFrame(text=" "), # Just whitespace
- TTSTextFrame(text="\n"), # Just newline
+ TTSTextFrame(text="", aggregated_by=AggregationType.WORD), # Empty text
+ TTSTextFrame(text=" ", aggregated_by=AggregationType.WORD), # Just whitespace
+ TTSTextFrame(text="\n", aggregated_by=AggregationType.WORD), # Just newline
BotStoppedSpeakingFrame(),
# Pipeline ends here; run_test will automatically send EndFrame
]
@@ -235,14 +236,14 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(),
- TTSTextFrame(text="Hello"),
- TTSTextFrame(text="world!"),
+ TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD),
SleepFrame(),
InterruptionFrame(), # User interrupts here
SleepFrame(),
BotStartedSpeakingFrame(),
- TTSTextFrame(text="New"),
- TTSTextFrame(text="response"),
+ TTSTextFrame(text="New", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="response", aggregated_by=AggregationType.WORD),
SleepFrame(),
BotStoppedSpeakingFrame(),
]
@@ -299,8 +300,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(),
- TTSTextFrame(text="Hello"),
- TTSTextFrame(text="world"),
+ TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="world", aggregated_by=AggregationType.WORD),
# Pipeline ends here; run_test will automatically send EndFrame
]
@@ -338,8 +339,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(),
- TTSTextFrame(text="Hello"),
- TTSTextFrame(text="world"),
+ TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="world", aggregated_by=AggregationType.WORD),
SleepFrame(), # Ensure messages are processed
CancelFrame(),
]
@@ -401,8 +402,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(),
- TTSTextFrame(text="Assistant"),
- TTSTextFrame(text="message"),
+ TTSTextFrame(text="Assistant", aggregated_by=AggregationType.WORD),
+ TTSTextFrame(text="message", aggregated_by=AggregationType.WORD),
BotStoppedSpeakingFrame(),
]
@@ -439,7 +440,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
# Test the specific pattern shared
def make_tts_text_frame(text: str) -> TTSTextFrame:
- frame = TTSTextFrame(text=text)
+ frame = TTSTextFrame(text=text, aggregated_by=AggregationType.WORD)
frame.includes_inter_frame_spaces = True
return frame
diff --git a/uv.lock b/uv.lock
index 8f7585828..eb2fca39c 100644
--- a/uv.lock
+++ b/uv.lock
@@ -419,16 +419,30 @@ wheels = [
[[package]]
name = "aws-sdk-bedrock-runtime"
-version = "0.1.1"
+version = "0.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "smithy-aws-core", extra = ["eventstream", "json"], marker = "python_full_version >= '3.12'" },
{ name = "smithy-core", marker = "python_full_version >= '3.12'" },
{ name = "smithy-http", extra = ["awscrt"], marker = "python_full_version >= '3.12'" },
]
-sdist = { url = "https://files.pythonhosted.org/packages/1d/78/48574454b3cac869df67665e4a403ebfc3abfcfba2c2ff01ccfd67d55f8f/aws_sdk_bedrock_runtime-0.1.1.tar.gz", hash = "sha256:c896f99e675c3a1ab600633a07b785f3dc9fe8ab94f640b1f992b63da2dfc784", size = 82446, upload-time = "2025-10-21T20:25:25.845Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/db/94/f2451bb09c106e5690bbb88fc366637cdcec942b352ed9bb788804c877e0/aws_sdk_bedrock_runtime-0.2.0.tar.gz", hash = "sha256:8de52dd4492e74c73244d4b41a52304e1db368814a10e49dbbf8f4e8e412cd0e", size = 88156, upload-time = "2025-11-22T00:35:44.978Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/83/07/62c0b70223d178c138f29124ac2f7973a6ba803abc7735b6a01a85217f3d/aws_sdk_bedrock_runtime-0.1.1-py3-none-any.whl", hash = "sha256:c0336b377b2112cf88197d3d44302fbeb3efb1101989fa49ae55e78f49cfe345", size = 74954, upload-time = "2025-10-21T20:25:24.973Z" },
+ { url = "https://files.pythonhosted.org/packages/eb/6b/07fbddd31dd6e38c967fe088b5e91a7cc3a2bc0f645f18b4e5d45bc03f1f/aws_sdk_bedrock_runtime-0.2.0-py3-none-any.whl", hash = "sha256:19594de50a52d199d73efca153c0a2328bd781827715a6e012d50b11085236cc", size = 79875, upload-time = "2025-11-22T00:35:44.092Z" },
+]
+
+[[package]]
+name = "aws-sdk-sagemaker-runtime-http2"
+version = "0.1.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+ { name = "smithy-aws-core", extra = ["eventstream", "json"], marker = "python_full_version >= '3.12'" },
+ { name = "smithy-core", marker = "python_full_version >= '3.12'" },
+ { name = "smithy-http", extra = ["awscrt"], marker = "python_full_version >= '3.12'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/6e/ca/00f9c55887fc0f3fa345995dd871d40ff81473ab1591e56b4b4483d99d00/aws_sdk_sagemaker_runtime_http2-0.1.0.tar.gz", hash = "sha256:5077ec0c4440495b15004bbf04e27bc0bc137f1f8950d32195c6b45d7788d837", size = 20863, upload-time = "2025-11-22T00:20:56.358Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/9c/24/2e2f727c51c20f4625cd19364d9421dbd7c893fe2b53a46eb0caaf6263a2/aws_sdk_sagemaker_runtime_http2-0.1.0-py3-none-any.whl", hash = "sha256:1aebb728ba6c6d14e58e29ecf89b51f7abbe8786d34144f8a7d59a419e80bd2f", size = 21911, upload-time = "2025-11-22T00:20:55.054Z" },
]
[[package]]
@@ -4442,9 +4456,6 @@ assemblyai = [
asyncai = [
{ name = "websockets" },
]
-auth = [
- { name = "pyjwt" },
-]
aws = [
{ name = "aioboto3" },
{ name = "websockets" },
@@ -4572,6 +4583,9 @@ runner = [
{ name = "python-dotenv" },
{ name = "uvicorn" },
]
+sagemaker = [
+ { name = "aws-sdk-sagemaker-runtime-http2", marker = "python_full_version >= '3.12'" },
+]
sarvam = [
{ name = "sarvamai" },
{ name = "websockets" },
@@ -4657,7 +4671,8 @@ requires-dist = [
{ name = "aiortc", marker = "extra == 'webrtc'", specifier = ">=1.13.0,<2" },
{ name = "anthropic", marker = "extra == 'anthropic'", specifier = "~=0.49.0" },
{ name = "audioop-lts", marker = "python_full_version >= '3.13'", specifier = "~=0.2.1" },
- { name = "aws-sdk-bedrock-runtime", marker = "python_full_version >= '3.12' and extra == 'aws-nova-sonic'", specifier = "~=0.1.1" },
+ { name = "aws-sdk-bedrock-runtime", marker = "python_full_version >= '3.12' and extra == 'aws-nova-sonic'", specifier = "~=0.2.0" },
+ { name = "aws-sdk-sagemaker-runtime-http2", marker = "python_full_version >= '3.12' and extra == 'sagemaker'" },
{ name = "azure-cognitiveservices-speech", marker = "extra == 'azure'", specifier = "~=1.42.0" },
{ name = "cartesia", marker = "extra == 'cartesia'", specifier = "~=2.0.3" },
{ name = "coremltools", marker = "extra == 'local-smart-turn'", specifier = ">=8.0" },
@@ -4724,7 +4739,6 @@ requires-dist = [
{ name = "pyaudio", marker = "extra == 'local'", specifier = "~=0.2.14" },
{ name = "pydantic", specifier = ">=2.10.6,<3" },
{ name = "pygobject", marker = "extra == 'gstreamer'", specifier = "~=3.50.0" },
- { name = "pyjwt", marker = "extra == 'auth'", specifier = ">=2.10.1" },
{ name = "pyloudnorm", specifier = "~=0.1.1" },
{ name = "python-dotenv", marker = "extra == 'runner'", specifier = ">=1.0.0,<2.0.0" },
{ name = "pyvips", extras = ["binary"], marker = "extra == 'moondream'", specifier = "~=3.0.0" },
@@ -4749,7 +4763,7 @@ requires-dist = [
{ name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" },
{ name = "websockets", marker = "extra == 'websockets-base'", specifier = ">=13.1,<16.0" },
]
-provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "local-smart-turn-v3", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper", "auth"]
+provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "local-smart-turn-v3", "remote-smart-turn", "sagemaker", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"]
[package.metadata.requires-dev]
dev = [
@@ -6526,16 +6540,16 @@ wheels = [
[[package]]
name = "smithy-aws-core"
-version = "0.1.1"
+version = "0.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aws-sdk-signers", marker = "python_full_version >= '3.12'" },
{ name = "smithy-core", marker = "python_full_version >= '3.12'" },
{ name = "smithy-http", marker = "python_full_version >= '3.12'" },
]
-sdist = { url = "https://files.pythonhosted.org/packages/56/d3/f847e0fd36b95aa36ce3a4c9ce1a08e16b2aa9a56b71714045c9c924e282/smithy_aws_core-0.1.1.tar.gz", hash = "sha256:78dfd7040fc2bc72b6af293096642fc9a7bfd2db28ddbdf7c4110535eab9d662", size = 11196, upload-time = "2025-10-21T20:21:18.648Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/c1/c8/5970c869527972b23a1733de3993d54283c84a2340e84acdd48a11aa0ff4/smithy_aws_core-0.2.0.tar.gz", hash = "sha256:dfa1ecd311d6f0a16f48c86d793085e2a0a33a46de897d129dd1f142a43bf7f6", size = 11344, upload-time = "2025-11-21T18:33:01.928Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/d0/04/87cb06f0f6d664b5cffdef6d4042dd52c11c138436084d30ffdaa3543031/smithy_aws_core-0.1.1-py3-none-any.whl", hash = "sha256:0d1634f276c2999dc2a04fafef63b9d28309de50d939d1d49df952773a7063c4", size = 18963, upload-time = "2025-10-21T20:21:17.692Z" },
+ { url = "https://files.pythonhosted.org/packages/88/25/739c0005a6cb4effbc2d37fe23590660b508fe314200f4acf94410a8f315/smithy_aws_core-0.2.0-py3-none-any.whl", hash = "sha256:d112082ef77758e1977f8694cf690ac35c76570c12a6590fccd5da085a3ce507", size = 18966, upload-time = "2025-11-21T18:33:00.812Z" },
]
[package.optional-dependencies]
@@ -6548,35 +6562,35 @@ json = [
[[package]]
name = "smithy-aws-event-stream"
-version = "0.1.0"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
- { name = "smithy-core", marker = "python_full_version >= '3.12'" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/49/26/8ff24194efed60b2df18f610ea05fa2a4c6546858b80a0a51335a4943b9b/smithy_aws_event_stream-0.1.0.tar.gz", hash = "sha256:6634691a3bf5d4801a2c29f0761db2dc4771f3ae43cdee50c10d4b4bb2f86475", size = 12216, upload-time = "2025-09-29T19:37:14.659Z" }
-wheels = [
- { url = "https://files.pythonhosted.org/packages/90/c4/2b63d31af58fc359577e5515bf730348a235f2f2fa10e17af8640495c81c/smithy_aws_event_stream-0.1.0-py3-none-any.whl", hash = "sha256:17a7300a85cb90df4c6c23f895ca6343361fa419203c3cf80019edd7d3b5f036", size = 15581, upload-time = "2025-09-29T19:37:13.589Z" },
-]
-
-[[package]]
-name = "smithy-core"
-version = "0.1.0"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/b9/8d/16028d03456071d21de7591f1e1e6a1cc81b2389e53ef8663dbf59caf9cd/smithy_core-0.1.0.tar.gz", hash = "sha256:b159b8905264e1e4c613eab9f74cec0b2f5b8119c42fbadddb4da0a8ed8050e9", size = 48415, upload-time = "2025-09-29T19:37:16.873Z" }
-wheels = [
- { url = "https://files.pythonhosted.org/packages/ca/5b/563cb2beadcfa40597b0c3ff3f2d42e21f065b14782c4ba9cb41a44b745f/smithy_core-0.1.0-py3-none-any.whl", hash = "sha256:cb44e9355fb89e89f2c6ba6a1d59c5db4f2f7282c72d31d9307b6202d66cd0fa", size = 62895, upload-time = "2025-09-29T19:37:15.917Z" },
-]
-
-[[package]]
-name = "smithy-http"
version = "0.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "smithy-core", marker = "python_full_version >= '3.12'" },
]
-sdist = { url = "https://files.pythonhosted.org/packages/3c/1c/44e99a7dfb8c39bf0c3d998accdf4573a7a3488863b90f21af260cec2d45/smithy_http-0.2.0.tar.gz", hash = "sha256:2382562fa9af326be455f14b18615f16ffe9db756e51b2a4ca0d23e1b881cff8", size = 26729, upload-time = "2025-10-21T20:21:06.146Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/31/90/78283c21484f8cf9862982e53bc2769b784910735fb5fb2400a17bfb5fdd/smithy_aws_event_stream-0.2.0.tar.gz", hash = "sha256:99700a11346e7ab1435ff2e53e6f6d60a1e857f2b2ee1941d40b54270adf3323", size = 12278, upload-time = "2025-11-21T18:33:03.79Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/d4/e2/d475fad81ac74ec0e145cb6d72afe5ecde4e2358bd632c2fd5d3f4bc87dc/smithy_http-0.2.0-py3-none-any.whl", hash = "sha256:49ee2402d7737798d70f99f491fbfb2a5767283ae562e21b6f86e3fd14f3e3e0", size = 37328, upload-time = "2025-10-21T20:21:05.362Z" },
+ { url = "https://files.pythonhosted.org/packages/ca/f5/08b997eee81b55150496ce565f0e03c72d0c80e5b218170bdeae7c46a5a4/smithy_aws_event_stream-0.2.0-py3-none-any.whl", hash = "sha256:679a0c7d944e67d3a55d287541b3ca1e61f9d6a62e13401367dcc034e75aa55d", size = 15567, upload-time = "2025-11-21T18:33:02.711Z" },
+]
+
+[[package]]
+name = "smithy-core"
+version = "0.2.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/c7/f6/140f0be9331dd7cd8fa012b3ca4735df39a1a81d03eea89728f997249116/smithy_core-0.2.0.tar.gz", hash = "sha256:05c3e3309df5dcb9cf53e241bd57a96510e4575186443ea157db9dbb59b6c85e", size = 50334, upload-time = "2025-11-21T18:33:05.697Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/16/e3/d0defa2acf50b91625fe15e3ddb0c8e41ff64363a1f4cd9b8f19ae2ec0c6/smithy_core-0.2.0-py3-none-any.whl", hash = "sha256:db4620da3497abb60f79ac1d8a738d3eac46d7e820bfb50c777c36e932915239", size = 64777, upload-time = "2025-11-21T18:33:04.591Z" },
+]
+
+[[package]]
+name = "smithy-http"
+version = "0.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+ { name = "smithy-core", marker = "python_full_version >= '3.12'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/1c/c7/4d8be56e897f99f3b6ffcdf52ba00a468febc939fca85b90f1c122450830/smithy_http-0.3.0.tar.gz", hash = "sha256:55dcc3af315eee6863d2f3f58ada1d9cb4bcc3a57faac10a1b21d4a93722f520", size = 28674, upload-time = "2025-11-21T18:33:07.387Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/2d/e5/59ae79ecdc9a935ad10512c581b3054ebb1afd90498ecc8afaf141dbc22b/smithy_http-0.3.0-py3-none-any.whl", hash = "sha256:972924304febd77c7134a7cffab83ce3b48423ff966dcc1f257e2c0d58fa9b18", size = 40520, upload-time = "2025-11-21T18:33:06.312Z" },
]
[package.optional-dependencies]
@@ -6586,15 +6600,15 @@ awscrt = [
[[package]]
name = "smithy-json"
-version = "0.1.0"
+version = "0.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "ijson", marker = "python_full_version >= '3.12'" },
{ name = "smithy-core", marker = "python_full_version >= '3.12'" },
]
-sdist = { url = "https://files.pythonhosted.org/packages/e2/5b/0ecb10007475e1b8faca3bbff1be2fc6edb3ea12ffc5e939e2249be95325/smithy_json-0.1.0.tar.gz", hash = "sha256:84fb48e445b87d850c240d837702c16b259ea53bad76b655ac1bbd8094d48912", size = 7086, upload-time = "2025-09-29T19:37:20.432Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/89/cf/e319a2a299b27bc0addf46ee3d4b9c25ec0817e3a0507b2b7a33eddc19f1/smithy_json-0.2.0.tar.gz", hash = "sha256:0946066fdda15d6a579dfdd4b61a547ab915eb057bd176fc2bc17d01dc789499", size = 7157, upload-time = "2025-11-21T18:33:08.968Z" }
wheels = [
- { url = "https://files.pythonhosted.org/packages/62/95/e11c04e56aae12b62e38c49000004a1dc598a64dc207018c08448efde322/smithy_json-0.1.0-py3-none-any.whl", hash = "sha256:80ff64734dccdabf1ba6a2908555b97e60f62c07c3a27df48e421ee058413cb9", size = 9914, upload-time = "2025-09-29T19:37:19.459Z" },
+ { url = "https://files.pythonhosted.org/packages/2e/b1/33012ac5b2e5940a00b6e1ccc313330e6f8692152a151f72a398cd6be0e0/smithy_json-0.2.0-py3-none-any.whl", hash = "sha256:5018a4e61731afa3094a02d737d4f956dbf270c271410c089045a17d86fc3b3b", size = 9911, upload-time = "2025-11-21T18:33:08.267Z" },
]
[[package]]