sync with main

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### 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` is processed by the `RTVIObserver` and will
+  be delivered to the client's `onServerMessage` callback or `ServerMessage`
+  event.
+
 - Added `GoogleLLMOpenAIBetaService` for Google LLM integration with an
   OpenAI-compatible interface. Added foundational example
   `14o-function-calling-gemini-openai-format.py`.

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(),
             ]
         )

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
 #

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)

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

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()

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()
 

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
 

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

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,

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"
 

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()

src/pipecat/services/google/google.py

@@ -1003,8 +1003,8 @@ class GoogleLLMService(LLMService):
 
         try:
             logger.debug(
-                # f"Generating chat: {self._system_instruction} | {context.get_messages_for_logging()}"
+                # f"{self}: Generating chat [{self._system_instruction}] | [{context.get_messages_for_logging()}]"
-                f"Generating chat: {context.get_messages_for_logging()}"
+                f"{self}: Generating chat [{context.get_messages_for_logging()}]"
             )
 
             messages = context.messages
@@ -1416,7 +1416,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()

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:

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()
 

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()

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",

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()

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")