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

Rename AggregatedLLMTextFrame to AggregatedTextFrame and made built-in types an enum

Browse Source
This commit is contained in:
mattie ruth backman
2025-11-07 10:32:54 -05:00
parent 124f147a37
commit 8ab0c92681
11 changed files with 75 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
CHANGELOG.md
View File

@@ -16,27 +16,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
services that subclass `TTSService` can indicate whether the text in the
`TTSTextFrame`s they push already contain any necessary inter-frame spaces.
- New bot-output RTVI message to represent what the bot actually "says".
- RTVIBotOutputMessage / RTVIBotOutputMessageData — includes:
- spoken: bool — whether the text was spoken by TTS
- aggregated_by: Optional[str|\"word\"|\"sentence\"] — how the text was aggregated
- RTVIObserver now emits bot-output messages (bot-tts-text and bot-llm-text are still
supported and generated. bot-transcript is now deprecated in lieu of this new, more
thorough, message).
- Introduced new `AggregatedLLMTextFrame` type to support representing effective llm
- Introduced new `AggregatedTextFrame` type to support representing effective llm
types an enum)
output whether or not it is processed by the TTS. This new frame type includes the
field `aggregated_by` to represent the conceptual format by which the given text
is aggregated. `TTSTextFrame`s now inherit from `AggregatedLLMTextFrame`.
is aggregated. `TTSTextFrame`s now inherit from `AggregatedTextFrame`.
- New `bot-output` RTVI message to represent what the bot actually "says".
- The `RTVIObserver` now emits `bot-output` messages based off the new `AggregatedLLMTextFrame`s
- 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",
"custom")
"my custom aggregation")
- Updated the base aggregator type:
- Introduced a new `Aggregation` dataclass to represent both the aggregated `text` and
@@ -86,10 +79,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `PatternMatch` now extends `Aggregation` and provides richer info to handlers.
- Added support for aggregating `LLMTextFrame`s from within the assistant `LLMAssistantAggregator`
when `skip_tts` is set to `True`, generating `AggregatedLLMTextFrame`s, therefore supporting
`bot-output` even when TTS is turned off. You can customize the aggregator used using the new
`llm_text_aggregator` field in the `LLMAssistantAggregatorParams`. NOTE: This feature is only
supported when using the new universal context.
when `skip_tts` is set to `True`, generating `AggregatedTextFrame`s, therefore supporting
the new `bot-output` event when TTS is turned off. You can customize the aggregator used using
the new `llm_text_aggregator` field in the `LLMAssistantAggregatorParams`. NOTE: This feature is
only supported when using the new `LLMContext`.
### Changed
@@ -116,11 +109,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- TTS flow respects aggregation metadata
- `TTSService` accepts a new `skip_aggregator_types` to avoid speaking certain aggregation types
(asnow determined/returned by the aggregator)
- TTS services push `AggregatedLLMTextFrame` in addition to `TTSTextFrame`s when either an
(now determined/returned by the aggregator)
- 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
`AggregatedLLMTextFrame`, aggregated by sentence to generate the full sentence content as early
`AggregatedTextFrame`, aggregated by sentence to generate the full sentence content as early
as possible.
### Deprecated

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

@@ -12,6 +12,7 @@ and LLM processing.
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import (
TYPE_CHECKING,
Any,
@@ -358,22 +359,29 @@ class LLMTextFrame(TextFrame):
pass
@dataclass
class AggregatedLLMTextFrame(TextFrame):
"""Text frame representing an aggregation of LLMTextFrames.
class AggregationType(Enum):
"""Built-in aggregation strings."""
This frame contains multiple LLMTextFrames aggregated together for
processing or output along with a field to indicate how they are aggregated.
SENTENCE = "sentence"
WORD = "word"
@dataclass
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: Literal["sentence", "word"] | str
aggregated_by: AggregationType | str
@dataclass
class TTSTextFrame(AggregatedLLMTextFrame):
class TTSTextFrame(AggregatedTextFrame):
"""Text frame generated by Text-to-Speech services."""
pass

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

@@ -24,7 +24,7 @@ from pipecat.audio.interruptions.base_interruption_strategy import BaseInterrupt
from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import (
AggregatedLLMTextFrame,
AggregatedTextFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -627,6 +627,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self.push_frame(frame, direction)
elif isinstance(frame, LLMFullResponseStartFrame):
await self._handle_llm_start(frame)
# as a subclass of TextFrame, LLMTextFrame must be checked first
elif isinstance(frame, LLMTextFrame):
await self._handle_llm_text(frame)
elif isinstance(frame, LLMFullResponseEndFrame):
@@ -854,7 +855,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
if not aggregate:
return
llm_frame = AggregatedLLMTextFrame(text=aggregate.text, aggregated_by=aggregate.type)
llm_frame = AggregatedTextFrame(text=aggregate.text, aggregated_by=aggregate.type)
await self.push_frame(llm_frame)
if should_reset_aggregator:
await self._llm_text_aggregator.reset()

9
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -32,7 +32,8 @@ from pydantic import BaseModel, Field, PrivateAttr, ValidationError
from pipecat.audio.utils import calculate_audio_volume
from pipecat.frames.frames import (
AggregatedLLMTextFrame,
AggregatedTextFrame,
AggregationType,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -712,7 +713,7 @@ class RTVIBotOutputMessageData(RTVITextMessageData):
"""
spoken: bool = True # Indicates if the text has been spoken by TTS
aggregated_by: Optional[Literal["word", "sentence"] | str] = None
aggregated_by: Optional[AggregationType | str] = None
# Indicates what form the text is in (e.g., by word, sentence, etc.)
@@ -1074,7 +1075,7 @@ 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, AggregatedLLMTextFrame) and (
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):
@@ -1135,7 +1136,7 @@ class RTVIObserver(BaseObserver):
if message:
await self.send_rtvi_message(message)
async def _handle_aggregated_llm_text(self, frame: AggregatedLLMTextFrame):
async def _handle_aggregated_llm_text(self, frame: AggregatedTextFrame):
"""Handle aggregated LLM text output frames."""
isTTS = isinstance(frame, TTSTextFrame)
if self._params.bot_output_enabled:

7
src/pipecat/services/aws/nova_sonic/llm.py
View File

@@ -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, aggregated_by="sentence")
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, aggregated_by="sentence")
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

3
src/pipecat/services/google/gemini_live/llm.py
View File

@@ -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,
@@ -1646,7 +1647,7 @@ class GeminiLiveLLMService(LLMService):
await self.push_frame(TTSStartedFrame())
await self.push_frame(LLMFullResponseStartFrame())
frame = TTSTextFrame(text=text, aggregated_by="sentence")
frame = TTSTextFrame(text=text, aggregated_by=AggregationType.SENTENCE)
# Gemini Live text already includes any necessary inter-chunk spaces
frame.includes_inter_frame_spaces = True

3
src/pipecat/services/openai/realtime/llm.py
View File

@@ -19,6 +19,7 @@ from pipecat.adapters.services.open_ai_realtime_adapter import (
OpenAIRealtimeLLMAdapter,
)
from pipecat.frames.frames import (
AggregationType,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
@@ -686,7 +687,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, aggregated_by="sentence")
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)

3
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -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, aggregated_by="sentence"))
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()

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

@@ -23,7 +23,8 @@ from typing import (
from loguru import logger
from pipecat.frames.frames import (
AggregatedLLMTextFrame,
AggregatedTextFrame,
AggregationType,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
@@ -388,7 +389,7 @@ class TTSService(AIService):
elif isinstance(frame, TTSSpeakFrame):
# Store if we were processing text or not so we can set it back.
processing_text = self._processing_text
await self._push_tts_frames(frame.text, aggregated_by="sentence")
await self._push_tts_frames(frame.text, aggregated_by=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()
@@ -494,8 +495,8 @@ class TTSService(AIService):
async def _push_tts_frames(self, text: str, aggregated_by: str):
if aggregated_by in self._skip_aggregator_types:
# If this type of aggregation should be skipped, we just push the text as
# a basic AggregatedLLMTextFrame without sending it to TTS to speak.
await self.push_frame(AggregatedLLMTextFrame(text, aggregated_by=aggregated_by))
# a basic AggregatedTextFrame without sending it to TTS to speak.
await self.push_frame(AggregatedTextFrame(text, aggregated_by=aggregated_by))
return
# Remove leading newlines only
@@ -526,11 +527,11 @@ class TTSService(AIService):
# is set to False and these are sent word by word as part of the
# _words_task_handler in the WordTTSService subclass. However, to
# support use cases where an observer may want the full text before
# the audio is generated, we send an AggregatedLLMTextFrame here, but
# the audio is generated, we send an AggregatedTextFrame here, but
# we set append_to_context to False so it does not cause duplication
# in the context. This is primarily used by the RTVIObserver to
# generate a complete bot-output.
frame = AggregatedLLMTextFrame(text, aggregated_by=aggregated_by)
frame = AggregatedTextFrame(text, aggregated_by=aggregated_by)
frame.append_to_context = False
await self.push_frame(frame)
await self.process_generator(self.run_tts(text))
@@ -669,7 +670,7 @@ class WordTTSService(TTSService):
frame = TTSStoppedFrame()
frame.pts = last_pts
else:
frame = TTSTextFrame(word, aggregated_by="word")
frame = TTSTextFrame(word, aggregated_by=AggregationType.WORD)
frame.pts = self._initial_word_timestamp + timestamp
if frame:
last_pts = frame.pts

2
tests/test_piper_tts.py
View File

@@ -13,7 +13,7 @@ import pytest
from aiohttp import web
from pipecat.frames.frames import (
AggregatedLLMTextFrame,
AggregatedTextFrame,
ErrorFrame,
TTSAudioRawFrame,
TTSSpeakFrame,

39
tests/test_transcript_processor.py
View File

@@ -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", aggregated_by="word"),
TTSTextFrame(text="world!", aggregated_by="word"),
TTSTextFrame(text="How", aggregated_by="word"),
TTSTextFrame(text="are", aggregated_by="word"),
TTSTextFrame(text="you?", aggregated_by="word"),
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="", aggregated_by="word"), # Empty text
TTSTextFrame(text=" ", aggregated_by="word"), # Just whitespace
TTSTextFrame(text="\n", aggregated_by="word"), # 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", aggregated_by="word"),
TTSTextFrame(text="world!", aggregated_by="word"),
TTSTextFrame(text="Hello", aggregated_by=AggregationType.WORD),
TTSTextFrame(text="world!", aggregated_by=AggregationType.WORD),
SleepFrame(),
InterruptionFrame(), # User interrupts here
SleepFrame(),
BotStartedSpeakingFrame(),
TTSTextFrame(text="New", aggregated_by="word"),
TTSTextFrame(text="response", aggregated_by="word"),
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", aggregated_by="word"),
TTSTextFrame(text="world", aggregated_by="word"),
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", aggregated_by="word"),
TTSTextFrame(text="world", aggregated_by="word"),
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", aggregated_by="word"),
TTSTextFrame(text="message", aggregated_by="word"),
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, aggregated_by="word")
frame = TTSTextFrame(text=text, aggregated_by=AggregationType.WORD)
frame.includes_inter_frame_spaces = True
return frame