Merge pull request #3650 from pipecat-ai/filipi/twilio_issues

Ignoring RTVI messages inside the Serializers by default.

src/pipecat/serializers/base_serializer.py

@@ -7,8 +7,17 @@
 """Frame serialization interfaces for Pipecat."""
 
 from abc import ABC, abstractmethod
+from typing import Optional
 
-from pipecat.frames.frames import Frame, StartFrame
+from pydantic import BaseModel
+
+from pipecat.frames.frames import (
+    Frame,
+    OutputTransportMessageFrame,
+    OutputTransportMessageUrgentFrame,
+    StartFrame,
+)
+from pipecat.processors.frameworks.rtvi import RTVI_MESSAGE_LABEL
 from pipecat.utils.base_object import BaseObject
 
 
@@ -20,6 +29,46 @@ class FrameSerializer(BaseObject):
     serialize/deserialize methods.
     """
 
+    class InputParams(BaseModel):
+        """Base configuration parameters for FrameSerializer.
+
+        Parameters:
+            ignore_rtvi_messages: Whether to ignore RTVI protocol messages during serialization.
+                Defaults to True to prevent RTVI messages from being sent to external transports.
+        """
+
+        ignore_rtvi_messages: bool = True
+
+    def __init__(self, params: Optional[InputParams] = None, **kwargs):
+        """Initialize the FrameSerializer.
+
+        Args:
+            params: Configuration parameters.
+            **kwargs: Additional arguments passed to BaseObject (e.g., name).
+        """
+        super().__init__(**kwargs)
+        self._params = params or FrameSerializer.InputParams()
+
+    def should_ignore_frame(self, frame: Frame) -> bool:
+        """Check if a frame should be ignored during serialization.
+
+        This method filters out RTVI protocol messages when ignore_rtvi_messages is enabled.
+        Subclasses can override this to add additional filtering logic.
+
+        Args:
+            frame: The frame to check.
+
+        Returns:
+            True if the frame should be ignored, False otherwise.
+        """
+        if (
+            self._params.ignore_rtvi_messages
+            and isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame))
+            and frame.message.get("label") == RTVI_MESSAGE_LABEL
+        ):
+            return True
+        return False
+
     async def setup(self, frame: StartFrame):
         """Initialize the serializer with startup configuration.
 

src/pipecat/serializers/exotel.py

@@ -39,12 +39,13 @@ class ExotelFrameSerializer(FrameSerializer):
         https://support.exotel.com/support/solutions/articles/3000108630-working-with-the-stream-and-voicebot-applet
     """
 
-    class InputParams(BaseModel):
+    class InputParams(FrameSerializer.InputParams):
         """Configuration parameters for ExotelFrameSerializer.
 
         Parameters:
             exotel_sample_rate: Sample rate used by Exotel, defaults to 8000 Hz.
             sample_rate: Optional override for pipeline input sample rate.
+            ignore_rtvi_messages: Inherited from base FrameSerializer, defaults to True.
         """
 
         exotel_sample_rate: int = 8000
@@ -60,9 +61,10 @@ class ExotelFrameSerializer(FrameSerializer):
             call_sid: The associated Exotel Call SID (optional, not used in this implementation).
             params: Configuration parameters.
         """
+        super().__init__(params or ExotelFrameSerializer.InputParams())
+
         self._stream_sid = stream_sid
         self._call_sid = call_sid
-        self._params = params or ExotelFrameSerializer.InputParams()
 
         self._exotel_sample_rate = self._params.exotel_sample_rate
         self._sample_rate = 0  # Pipeline input rate
@@ -113,6 +115,8 @@ class ExotelFrameSerializer(FrameSerializer):
 
             return json.dumps(answer)
         elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
+            if self.should_ignore_frame(frame):
+                return None
             return json.dumps(frame.message)
 
         return None

src/pipecat/serializers/genesys.py

@@ -131,7 +131,7 @@ class GenesysAudioHookSerializer(FrameSerializer):
 
     PROTOCOL_VERSION = "2"
 
-    class InputParams(BaseModel):
+    class InputParams(FrameSerializer.InputParams):
         """Configuration parameters for GenesysAudioHookSerializer.
 
         Attributes:
@@ -144,6 +144,7 @@ class GenesysAudioHookSerializer(FrameSerializer):
             supported_languages: List of language codes the bot supports (e.g., ["en-US", "es-ES"]).
             selected_language: Default language code to use.
             start_paused: Whether to start the session in paused state.
+            ignore_rtvi_messages: Inherited from base FrameSerializer, defaults to True.
         """
 
         genesys_sample_rate: int = 8000
@@ -167,8 +168,7 @@ class GenesysAudioHookSerializer(FrameSerializer):
             params: Configuration parameters.
             **kwargs: Additional arguments passed to BaseObject (e.g., name).
         """
-        super().__init__(**kwargs)
+        super().__init__(params or GenesysAudioHookSerializer.InputParams(), **kwargs)
-        self._params = params or GenesysAudioHookSerializer.InputParams()
 
         self._genesys_sample_rate = self._params.genesys_sample_rate
         self._sample_rate = 0  # Pipeline input rate, set in setup()
@@ -604,6 +604,9 @@ class GenesysAudioHookSerializer(FrameSerializer):
             return json.dumps(self.create_barge_in_event())
 
         elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
+            # Filter out RTVI messages using base class method
+            if self.should_ignore_frame(frame):
+                return None
             # Only pass through AudioHook protocol messages (those with "version" field)
             # Filter out RTVI and other non-AudioHook messages
             if isinstance(frame.message, dict) and "version" in frame.message:

src/pipecat/serializers/plivo.py

@@ -42,13 +42,14 @@ class PlivoFrameSerializer(FrameSerializer):
     credentials to be provided.
     """
 
-    class InputParams(BaseModel):
+    class InputParams(FrameSerializer.InputParams):
         """Configuration parameters for PlivoFrameSerializer.
 
         Parameters:
             plivo_sample_rate: Sample rate used by Plivo, defaults to 8000 Hz.
             sample_rate: Optional override for pipeline input sample rate.
             auto_hang_up: Whether to automatically terminate call on EndFrame.
+            ignore_rtvi_messages: Inherited from base FrameSerializer, defaults to True.
         """
 
         plivo_sample_rate: int = 8000
@@ -72,11 +73,12 @@ class PlivoFrameSerializer(FrameSerializer):
             auth_token: Plivo auth token (required for auto hang-up).
             params: Configuration parameters.
         """
+        super().__init__(params or PlivoFrameSerializer.InputParams())
+
         self._stream_id = stream_id
         self._call_id = call_id
         self._auth_id = auth_id
         self._auth_token = auth_token
-        self._params = params or PlivoFrameSerializer.InputParams()
 
         self._plivo_sample_rate = self._params.plivo_sample_rate
         self._sample_rate = 0  # Pipeline input rate
@@ -140,6 +142,8 @@ class PlivoFrameSerializer(FrameSerializer):
 
             return json.dumps(answer)
         elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
+            if self.should_ignore_frame(frame):
+                return None
             return json.dumps(frame.message)
 
         # Return None for unhandled frames

src/pipecat/serializers/protobuf.py

@@ -8,6 +8,7 @@
 
 import dataclasses
 import json
+from typing import Optional
 
 from loguru import logger
 
@@ -60,9 +61,13 @@ class ProtobufFrameSerializer(FrameSerializer):
     }
     DESERIALIZABLE_FIELDS = {v: k for k, v in DESERIALIZABLE_TYPES.items()}
 
-    def __init__(self):
+    def __init__(self, params: Optional[FrameSerializer.InputParams] = None):
-        """Initialize the Protobuf frame serializer."""
+        """Initialize the Protobuf frame serializer.
-        pass
+
+        Args:
+            params: Configuration parameters.
+        """
+        super().__init__(params)
 
     async def serialize(self, frame: Frame) -> str | bytes | None:
         """Serialize a frame to Protocol Buffer binary format.
@@ -75,6 +80,8 @@ class ProtobufFrameSerializer(FrameSerializer):
         """
         # Wrapping this messages as a JSONFrame to send
         if isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
+            if self.should_ignore_frame(frame):
+                return None
             frame = MessageFrame(
                 data=json.dumps(frame.message),
             )

src/pipecat/serializers/twilio.py

@@ -42,13 +42,14 @@ class TwilioFrameSerializer(FrameSerializer):
     credentials to be provided.
     """
 
-    class InputParams(BaseModel):
+    class InputParams(FrameSerializer.InputParams):
         """Configuration parameters for TwilioFrameSerializer.
 
         Parameters:
             twilio_sample_rate: Sample rate used by Twilio, defaults to 8000 Hz.
             sample_rate: Optional override for pipeline input sample rate.
             auto_hang_up: Whether to automatically terminate call on EndFrame.
+            ignore_rtvi_messages: Inherited from base FrameSerializer, defaults to True.
         """
 
         twilio_sample_rate: int = 8000
@@ -76,7 +77,7 @@ class TwilioFrameSerializer(FrameSerializer):
             edge: Twilio edge location (e.g., "sydney", "dublin"). Must be specified with region.
             params: Configuration parameters.
         """
-        self._params = params or TwilioFrameSerializer.InputParams()
+        super().__init__(params or TwilioFrameSerializer.InputParams())
 
         # Validate hangup-related parameters if auto_hang_up is enabled
         if self._params.auto_hang_up:
@@ -167,6 +168,8 @@ class TwilioFrameSerializer(FrameSerializer):
 
             return json.dumps(answer)
         elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
+            if self.should_ignore_frame(frame):
+                return None
             return json.dumps(frame.message)
 
         # Return None for unhandled frames

src/pipecat/serializers/vonage.py

@@ -37,13 +37,14 @@ class VonageFrameSerializer(FrameSerializer):
         Ref docs: https://developer.vonage.com/en/video/guides/audio-connector
     """
 
-    class InputParams(BaseModel):
+    class InputParams(FrameSerializer.InputParams):
         """Configuration parameters for VonageFrameSerializer.
 
         Parameters:
             vonage_sample_rate: Sample rate used by Vonage, defaults to 16000 Hz.
                 Common values: 8000, 16000, 24000 Hz.
             sample_rate: Optional override for pipeline input sample rate.
+            ignore_rtvi_messages: Inherited from base FrameSerializer, defaults to True.
         """
 
         vonage_sample_rate: int = 16000
@@ -55,7 +56,7 @@ class VonageFrameSerializer(FrameSerializer):
         Args:
             params: Configuration parameters.
         """
-        self._params = params or VonageFrameSerializer.InputParams()
+        super().__init__(params or VonageFrameSerializer.InputParams())
 
         self._vonage_sample_rate = self._params.vonage_sample_rate
         self._sample_rate = 0  # Pipeline input rate
@@ -100,6 +101,8 @@ class VonageFrameSerializer(FrameSerializer):
             # Vonage expects raw binary PCM data (not base64 encoded)
             return serialized_data
         elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
+            if self.should_ignore_frame(frame):
+                return None
             # Allow sending custom JSON commands (e.g., notify)
             return json.dumps(frame.message)