From fa010c864414904ded30ff6f89d091b5cf152c7a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 26 Feb 2025 15:36:25 -0800 Subject: [PATCH 1/4] services: minor LLM and TTS logging improvements --- src/pipecat/services/anthropic.py | 2 +- src/pipecat/services/aws.py | 2 +- src/pipecat/services/azure.py | 4 ++-- src/pipecat/services/cartesia.py | 4 ++-- src/pipecat/services/deepgram.py | 2 +- src/pipecat/services/elevenlabs.py | 4 ++-- src/pipecat/services/fish.py | 2 +- src/pipecat/services/google/google.py | 6 +++--- src/pipecat/services/lmnt.py | 2 +- src/pipecat/services/openai.py | 4 ++-- src/pipecat/services/playht.py | 4 ++-- src/pipecat/services/rime.py | 4 ++-- src/pipecat/services/riva.py | 2 +- src/pipecat/services/xtts.py | 2 +- 14 files changed, 22 insertions(+), 22 deletions(-) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 1316802de..f10ee7ccb 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -152,7 +152,7 @@ class AnthropicLLMService(LLMService): await self.start_processing_metrics() logger.debug( - f"Generating chat: {context.system} | {context.get_messages_for_logging()}" + f"{self}: Generating chat [{context.system}] | [{context.get_messages_for_logging()}]" ) messages = context.messages diff --git a/src/pipecat/services/aws.py b/src/pipecat/services/aws.py index eac3f3e11..b83df2f77 100644 --- a/src/pipecat/services/aws.py +++ b/src/pipecat/services/aws.py @@ -197,7 +197,7 @@ class PollyTTSService(TTSService): return audio_data return None - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: await self.start_ttfb_metrics() diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index 11c34214c..83e5bbe6d 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -578,7 +578,7 @@ class AzureTTSService(AzureBaseTTSService): self._audio_queue.put_nowait(None) async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: if self._speech_synthesizer is None: @@ -645,7 +645,7 @@ class AzureHttpTTSService(AzureBaseTTSService): ) async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") await self.start_ttfb_metrics() diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 04a35087b..9bbde11bb 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -272,7 +272,7 @@ class CartesiaTTSService(AudioContextWordTTSService): logger.error(f"{self} error, unknown message type: {msg}") async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: if not self._websocket: @@ -358,7 +358,7 @@ class CartesiaHttpTTSService(TTSService): await self._client.close() async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: voice_controls = None diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 22879ab0e..3750164cf 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -70,7 +70,7 @@ class DeepgramTTSService(TTSService): return True async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") options = SpeakOptions( model=self._voice_id, diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 5e70735f8..4e6c84307 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -395,7 +395,7 @@ class ElevenLabsTTSService(InterruptibleWordTTSService): await self._websocket.send(json.dumps(msg)) async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: if not self._websocket: @@ -521,7 +521,7 @@ class ElevenLabsHttpTTSService(TTSService): Yields: Frames containing audio data and status information """ - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream" diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index e61579bfb..0b2729958 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -178,7 +178,7 @@ class FishAudioTTSService(InterruptibleTTSService): logger.error(f"Error processing message: {e}") async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating Fish TTS: [{text}]") + logger.debug(f"{self}: Generating Fish TTS: [{text}]") try: if not self._websocket or self._websocket.closed: await self._connect() diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index c0941ee33..7a48b0e4c 100644 --- a/src/pipecat/services/google/google.py +++ b/src/pipecat/services/google/google.py @@ -998,8 +998,8 @@ class GoogleLLMService(LLMService): try: logger.debug( - # f"Generating chat: {self._system_instruction} | {context.get_messages_for_logging()}" - f"Generating chat: {context.get_messages_for_logging()}" + # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]" + f"{self}: Generating chat [{context.get_messages_for_logging()}]" ) messages = context.messages @@ -1297,7 +1297,7 @@ class GoogleTTSService(TTSService): return ssml async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: await self.start_ttfb_metrics() diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index 8d4ca4803..31ddaf123 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -196,7 +196,7 @@ class LmntTTSService(InterruptibleTTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate TTS audio from text.""" - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: if not self._websocket: diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index e49bd3a90..71da3f5e2 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -178,7 +178,7 @@ class BaseOpenAILLMService(LLMService): async def _stream_chat_completions( self, context: OpenAILLMContext ) -> AsyncStream[ChatCompletionChunk]: - logger.debug(f"Generating chat: {context.get_messages_for_logging()}") + logger.debug(f"{self}: Generating chat [{context.get_messages_for_logging()}]") messages: List[ChatCompletionMessageParam] = context.get_messages() @@ -508,7 +508,7 @@ class OpenAITTSService(TTSService): ) async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: await self.start_ttfb_metrics() diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index e2c3b715d..80d313765 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -269,7 +269,7 @@ class PlayHTTTSService(InterruptibleTTSService): logger.error(f"Invalid JSON message: {message}") async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: # Reconnect if the websocket is closed @@ -392,7 +392,7 @@ class PlayHTHttpTTSService(TTSService): return language_to_playht_language(language) async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: options = self._create_options() diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 007ba958e..60083f55d 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -309,7 +309,7 @@ class RimeTTSService(AudioContextWordTTSService): Yields: Frames containing audio data and timing information. """ - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: if not self._websocket: await self._connect() @@ -376,7 +376,7 @@ class RimeHttpTTSService(TTSService): return True async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") headers = { "Accept": "audio/pcm", diff --git a/src/pipecat/services/riva.py b/src/pipecat/services/riva.py index 8d5ce79c4..de065aef4 100644 --- a/src/pipecat/services/riva.py +++ b/src/pipecat/services/riva.py @@ -101,7 +101,7 @@ class FastPitchTTSService(TTSService): await self.start_ttfb_metrics() yield TTSStartedFrame() - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") try: queue = asyncio.Queue() diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index e275b50b0..d2702d520 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -118,7 +118,7 @@ class XTTSService(TTSService): self._studio_speakers = await r.json() async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + logger.debug(f"{self}: Generating TTS [{text}]") if not self._studio_speakers: logger.error(f"{self} no studio speakers available") From 26c68ccd7c3ddcf9ffcea2b576c1a4f273c03426 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 26 Feb 2025 18:59:06 -0500 Subject: [PATCH 2/4] Add a new generic server to client message and frame type --- CHANGELOG.md | 10 ++++++++++ src/pipecat/frames/frames.py | 16 ++++++++++++++-- src/pipecat/processors/frameworks/rtvi.py | 18 ++++++++++++++++++ 3 files changed, 42 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff7b62e9b..9b6de3430 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,16 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- Added a new frame, `ServerMessageFrame`, and RTVI message `RTVIServerMessage` + which provides a generic mechanism for sending custom messages from server to + client. The `ServerMessageFrame` can be processed by either a `RTVIProcessor` + or `RTVIObserver` and will be delivered to the client's `onServerMessage` + callback. + ## [0.0.58] - 2025-02-26 ### Added diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b43aaaeae..b771b765f 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -568,7 +568,8 @@ class UserStoppedSpeakingFrame(SystemFrame): @dataclass class EmulateUserStartedSpeakingFrame(SystemFrame): """Emitted by internal processors upstream to emulate VAD behavior when a - user starts speaking.""" + user starts speaking. + """ pass @@ -576,7 +577,8 @@ class EmulateUserStartedSpeakingFrame(SystemFrame): @dataclass class EmulateUserStoppedSpeakingFrame(SystemFrame): """Emitted by internal processors upstream to emulate VAD behavior when a - user stops speaking.""" + user stops speaking. + """ pass @@ -704,6 +706,16 @@ class VisionImageRawFrame(InputImageRawFrame): return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})" +@dataclass +class ServerMessageFrame(SystemFrame): + """A frame for sending server messages to the client.""" + + data: Any + + def __str__(self): + return f"{self.name}(data: {self.data})" + + # # Control frames # diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index af55fbd42..4edef4203 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -38,6 +38,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMTextFrame, MetricsFrame, + ServerMessageFrame, StartFrame, SystemFrame, TranscriptionFrame, @@ -375,6 +376,12 @@ class RTVIMetricsMessage(BaseModel): data: Mapping[str, Any] +class RTVIServerMessage(BaseModel): + label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL + type: Literal["server-message"] = "server-message" + data: Any + + class RTVIFrameProcessor(FrameProcessor): def __init__(self, direction: FrameDirection = FrameDirection.DOWNSTREAM, **kwargs): super().__init__(**kwargs) @@ -710,6 +717,9 @@ class RTVIObserver(BaseObserver): mark_as_seen = False elif isinstance(frame, MetricsFrame): await self._handle_metrics(frame) + elif isinstance(frame, ServerMessageFrame): + message = RTVIServerMessage(data=frame.data) + await self.push_transport_message_urgent(message) if mark_as_seen: self._frames_seen.add(frame.id) @@ -895,6 +905,11 @@ class RTVIProcessor(FrameProcessor): async def handle_message(self, message: RTVIMessage): await self._message_queue.put(message) + async def _handle_server_message(self, frame: ServerMessageFrame): + """Handle server message frame by converting it to a transport message.""" + message = RTVIServerMessage(data=frame.data) + await self._push_transport_message(message) + async def handle_function_call( self, function_name: str, @@ -934,6 +949,9 @@ class RTVIProcessor(FrameProcessor): await self.push_frame(frame, direction) elif isinstance(frame, TransportMessageUrgentFrame): await self._handle_transport_message(frame) + elif isinstance(frame, ServerMessageFrame): + await self._handle_server_message(frame) + await self.push_frame(frame, direction) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) From 53d403af4bce22a4e8e77c40a7eac5dfd6319c3e Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 27 Feb 2025 12:50:43 -0500 Subject: [PATCH 3/4] Remove the RTVIServerMessage logic from the RTVIProcessor --- CHANGELOG.md | 6 +++--- src/pipecat/processors/frameworks/rtvi.py | 8 -------- 2 files changed, 3 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b6de3430..cdf100d41 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,9 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added a new frame, `ServerMessageFrame`, and RTVI message `RTVIServerMessage` which provides a generic mechanism for sending custom messages from server to - client. The `ServerMessageFrame` can be processed by either a `RTVIProcessor` - or `RTVIObserver` and will be delivered to the client's `onServerMessage` - callback. + client. The `ServerMessageFrame` is processed by the `RTVIObserver` and will + be delivered to the client's `onServerMessage` callback or `ServerMessage` + event. ## [0.0.58] - 2025-02-26 diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 4edef4203..9835a9648 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -905,11 +905,6 @@ class RTVIProcessor(FrameProcessor): async def handle_message(self, message: RTVIMessage): await self._message_queue.put(message) - async def _handle_server_message(self, frame: ServerMessageFrame): - """Handle server message frame by converting it to a transport message.""" - message = RTVIServerMessage(data=frame.data) - await self._push_transport_message(message) - async def handle_function_call( self, function_name: str, @@ -949,9 +944,6 @@ class RTVIProcessor(FrameProcessor): await self.push_frame(frame, direction) elif isinstance(frame, TransportMessageUrgentFrame): await self._handle_transport_message(frame) - elif isinstance(frame, ServerMessageFrame): - await self._handle_server_message(frame) - await self.push_frame(frame, direction) # All other system frames elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) From 803ea9d8bcfbfa6f18e3590328383cf2551979b1 Mon Sep 17 00:00:00 2001 From: Adrian Cowham Date: Thu, 27 Feb 2025 12:31:02 -0800 Subject: [PATCH 4/4] update the canonical client so that the audio recording is optional as long as there is a transcript --- examples/canonical-metrics/bot.py | 2 +- src/pipecat/services/canonical.py | 30 +++++++++++++++++++++++++++--- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index aa376c290..646636dc1 100644 --- a/examples/canonical-metrics/bot.py +++ b/examples/canonical-metrics/bot.py @@ -113,8 +113,8 @@ async def main(): llm, tts, transport.output(), - audio_buffer_processor, # captures audio into a buffer canonical, # uploads audio buffer to Canonical AI for metrics + audio_buffer_processor, # captures audio into a buffer context_aggregator.assistant(), ] ) diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py index 1c268ab22..7b62273d1 100644 --- a/src/pipecat/services/canonical.py +++ b/src/pipecat/services/canonical.py @@ -62,17 +62,21 @@ class CanonicalMetricsService(AIService): self, *, aiohttp_session: aiohttp.ClientSession, - audio_buffer_processor: AudioBufferProcessor, call_id: str, assistant: str, api_key: str, api_url: str = "https://voiceapp.canonical.chat/api/v1", assistant_speaks_first: bool = True, output_dir: str = "recordings", + audio_buffer_processor: Optional[AudioBufferProcessor] = None, context: Optional[OpenAILLMContext] = None, **kwargs, ): super().__init__(**kwargs) + # Validate that at least one of audio_buffer_processor or context is provided + if audio_buffer_processor is None and context is None: + raise ValueError("At least one of audio_buffer_processor or context must be specified") + self._aiohttp_session = aiohttp_session self._audio_buffer_processor = audio_buffer_processor self._api_key = api_key @@ -85,16 +89,36 @@ class CanonicalMetricsService(AIService): async def stop(self, frame: EndFrame): await super().stop(frame) - await self._process_audio() + await self._process_completion() async def cancel(self, frame: CancelFrame): await super().cancel(frame) - await self._process_audio() + await self._process_completion() async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) await self.push_frame(frame, direction) + async def _process_completion(self): + if self._audio_buffer_processor is not None: + await self._process_audio() + elif self._context is not None: + await self._process_transcript() + + async def _process_transcript(self): + params = { + "callId": self._call_id, + "assistant": {"id": self._assistant, "speaksFirst": self._assistant_speaks_first}, + "transcript": self._context.messages, + } + response = await self._aiohttp_session.post( + f"{self._api_url}/call", + headers=self._request_headers(), + json=params, + ) + if not response.ok: + logger.error(f"Failed to process transcript: {await response.text()}") + async def _process_audio(self): audio_buffer_processor = self._audio_buffer_processor