From a28a5e954a8adc34a0060efa997358bd107b5454 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 6 Oct 2024 16:47:10 -0700 Subject: [PATCH] add TransportMessageSystemFrame --- CHANGELOG.md | 4 ++++ src/pipecat/frames/frames.py | 9 +++++++- src/pipecat/processors/frameworks/rtvi.py | 27 ++++++++++++---------- src/pipecat/transports/base_output.py | 7 +++--- src/pipecat/transports/services/daily.py | 12 +++++++--- src/pipecat/transports/services/livekit.py | 17 ++++++++++++-- 6 files changed, 55 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 62637b317..9d232ae79 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `TransportMessageFrame` doesn't have an `urgent` field anymore, instead + there's now a `TransportMessageUrgentFrame` which is a `SystemFrame` and + therefore skip all internal queuing. + - For TTS services, convert inputted languages to match each service's language format diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e2ef78df5..174815651 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -269,7 +269,6 @@ class TTSSpeakFrame(DataFrame): @dataclass class TransportMessageFrame(DataFrame): message: Any - urgent: bool = False def __str__(self): return f"{self.name}(message: {self.message})" @@ -405,6 +404,14 @@ class BotInterruptionFrame(SystemFrame): pass +@dataclass +class TransportMessageUrgentFrame(SystemFrame): + message: Any + + def __str__(self): + return f"{self.name}(message: {self.message})" + + @dataclass class MetricsFrame(SystemFrame): """Emitted by processor that can compute metrics like latencies.""" diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 8a7763342..86dbf3d2c 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -31,6 +31,7 @@ from pipecat.frames.frames import ( TextFrame, TranscriptionFrame, TransportMessageFrame, + TransportMessageUrgentFrame, UserStartedSpeakingFrame, FunctionCallResultFrame, UserStoppedSpeakingFrame, @@ -349,9 +350,11 @@ class RTVIFrameProcessor(FrameProcessor): self._direction = direction async def _push_transport_message(self, model: BaseModel, exclude_none: bool = True): - frame = TransportMessageFrame( - message=model.model_dump(exclude_none=exclude_none), urgent=True - ) + frame = TransportMessageFrame(message=model.model_dump(exclude_none=exclude_none)) + await self.push_frame(frame, self._direction) + + async def _push_transport_message_urgent(self, model: BaseModel, exclude_none: bool = True): + frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none)) await self.push_frame(frame, self._direction) @@ -377,7 +380,7 @@ class RTVISpeakingProcessor(RTVIFrameProcessor): message = RTVIUserStoppedSpeakingMessage() if message: - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) async def _handle_bot_speaking(self, frame: Frame): message = None @@ -387,7 +390,7 @@ class RTVISpeakingProcessor(RTVIFrameProcessor): message = RTVIBotStoppedSpeakingMessage() if message: - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) class RTVIUserTranscriptionProcessor(RTVIFrameProcessor): @@ -418,7 +421,7 @@ class RTVIUserTranscriptionProcessor(RTVIFrameProcessor): ) if message: - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) class RTVIUserLLMTextProcessor(RTVIFrameProcessor): @@ -439,7 +442,7 @@ class RTVIUserLLMTextProcessor(RTVIFrameProcessor): message = messages[-1] if message["role"] == "user": message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=message["content"])) - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) class RTVIBotLLMProcessor(RTVIFrameProcessor): @@ -452,9 +455,9 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor): await self.push_frame(frame, direction) if isinstance(frame, LLMFullResponseStartFrame): - await self._push_transport_message(RTVIBotLLMStartedMessage()) + await self._push_transport_message_urgent(RTVIBotLLMStartedMessage()) elif isinstance(frame, LLMFullResponseEndFrame): - await self._push_transport_message(RTVIBotLLMStoppedMessage()) + await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage()) class RTVIBotTTSProcessor(RTVIFrameProcessor): @@ -467,9 +470,9 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor): await self.push_frame(frame, direction) if isinstance(frame, TTSStartedFrame): - await self._push_transport_message(RTVIBotTTSStartedMessage()) + await self._push_transport_message_urgent(RTVIBotTTSStartedMessage()) elif isinstance(frame, TTSStoppedFrame): - await self._push_transport_message(RTVIBotTTSStoppedMessage()) + await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage()) class RTVIBotLLMTextProcessor(RTVIFrameProcessor): @@ -486,7 +489,7 @@ class RTVIBotLLMTextProcessor(RTVIFrameProcessor): async def _handle_text(self, frame: TextFrame): message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text)) - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) class RTVIBotTTSTextProcessor(RTVIFrameProcessor): diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 6ee458f37..7ec503d24 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -33,6 +33,7 @@ from pipecat.frames.frames import ( TTSStoppedFrame, TextFrame, TransportMessageFrame, + TransportMessageUrgentFrame, ) from pipecat.transports.base_transport import TransportParams @@ -148,7 +149,7 @@ class BaseOutputTransport(FrameProcessor): await self._audio_out_task self._audio_out_task = None - async def send_message(self, frame: TransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): pass async def send_metrics(self, frame: MetricsFrame): @@ -186,6 +187,8 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, MetricsFrame): await self.push_frame(frame, direction) await self.send_metrics(frame) + elif isinstance(frame, TransportMessageUrgentFrame): + await self.send_message(frame) elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) # Control frames. @@ -198,8 +201,6 @@ class BaseOutputTransport(FrameProcessor): await self._handle_audio(frame) elif isinstance(frame, (OutputImageRawFrame, SpriteFrame)): await self._handle_image(frame) - elif isinstance(frame, TransportMessageFrame) and frame.urgent: - await self.send_message(frame) # TODO(aleix): Images and audio should support presentation timestamps. elif frame.pts: await self._sink_clock_queue.put((frame.pts, frame.id, frame)) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 50c2ae085..628fd61e2 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -35,6 +35,7 @@ from pipecat.frames.frames import ( StartFrame, TranscriptionFrame, TransportMessageFrame, + TransportMessageUrgentFrame, UserImageRawFrame, UserImageRequestFrame, ) @@ -70,6 +71,11 @@ class DailyTransportMessageFrame(TransportMessageFrame): participant_id: str | None = None +@dataclass +class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame): + participant_id: str | None = None + + class WebRTCVADAnalyzer(VADAnalyzer): def __init__(self, *, sample_rate=16000, num_channels=1, params: VADParams = VADParams()): super().__init__(sample_rate=sample_rate, num_channels=num_channels, params=params) @@ -234,12 +240,12 @@ class DailyTransportClient(EventHandler): def set_callbacks(self, callbacks: DailyCallbacks): self._callbacks = callbacks - async def send_message(self, frame: TransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): if not self._client: return participant_id = None - if isinstance(frame, DailyTransportMessageFrame): + if isinstance(frame, (DailyTransportMessageFrame, DailyTransportMessageUrgentFrame)): participant_id = frame.participant_id future = self._loop.create_future() @@ -736,7 +742,7 @@ class DailyOutputTransport(BaseOutputTransport): await super().cleanup() await self._client.cleanup() - async def send_message(self, frame: TransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): await self._client.send_message(frame) async def send_metrics(self, frame: MetricsFrame): diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 6e5e48d0b..d97b9596e 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -22,6 +22,7 @@ from pipecat.frames.frames import ( MetricsFrame, StartFrame, TransportMessageFrame, + TransportMessageUrgentFrame, ) from pipecat.metrics.metrics import ( LLMUsageMetricsData, @@ -51,6 +52,11 @@ class LiveKitTransportMessageFrame(TransportMessageFrame): participant_id: str | None = None +@dataclass +class LiveKitTransportMessageUrgentFrame(TransportMessageUrgentFrame): + participant_id: str | None = None + + class LiveKitParams(TransportParams): audio_out_sample_rate: int = 48000 audio_out_channels: int = 1 @@ -420,8 +426,8 @@ class LiveKitOutputTransport(BaseOutputTransport): await super().cancel(frame) await self._client.disconnect() - async def send_message(self, frame: TransportMessageFrame): - if isinstance(frame, LiveKitTransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + if isinstance(frame, (LiveKitTransportMessageFrame, LiveKitTransportMessageUrgentFrame)): await self._client.send_data(frame.message.encode(), frame.participant_id) else: await self._client.send_data(frame.message.encode()) @@ -596,6 +602,13 @@ class LiveKitTransport(BaseTransport): frame = LiveKitTransportMessageFrame(message=message, participant_id=participant_id) await self._output.send_message(frame) + async def send_message_urgent(self, message: str, participant_id: str | None = None): + if self._output: + frame = LiveKitTransportMessageUrgentFrame( + message=message, participant_id=participant_id + ) + await self._output.send_message(frame) + async def cleanup(self): if self._input: await self._input.cleanup()