diff --git a/src/pipecat/serializers/base_serializer.py b/src/pipecat/serializers/base_serializer.py index 9cc38cdc6..edf005a81 100644 --- a/src/pipecat/serializers/base_serializer.py +++ b/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. diff --git a/src/pipecat/serializers/exotel.py b/src/pipecat/serializers/exotel.py index 61d6eeada..74e3f28b0 100644 --- a/src/pipecat/serializers/exotel.py +++ b/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 diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index d5d37d12b..24b68eb81 100644 --- a/src/pipecat/serializers/genesys.py +++ b/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) - self._params = params or GenesysAudioHookSerializer.InputParams() + super().__init__(params or GenesysAudioHookSerializer.InputParams(), **kwargs) 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: diff --git a/src/pipecat/serializers/plivo.py b/src/pipecat/serializers/plivo.py index 2a57d3698..3341e5ffa 100644 --- a/src/pipecat/serializers/plivo.py +++ b/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 diff --git a/src/pipecat/serializers/protobuf.py b/src/pipecat/serializers/protobuf.py index 912f20b42..fd21af2bf 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/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): - """Initialize the Protobuf frame serializer.""" - pass + def __init__(self, params: Optional[FrameSerializer.InputParams] = None): + """Initialize the Protobuf frame serializer. + + 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), ) diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 2e60399fd..72fec4f28 100644 --- a/src/pipecat/serializers/twilio.py +++ b/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 diff --git a/src/pipecat/serializers/vonage.py b/src/pipecat/serializers/vonage.py index 3213e8bba..01fed08f8 100644 --- a/src/pipecat/serializers/vonage.py +++ b/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)