diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py deleted file mode 100644 index eb1e79f3e..000000000 --- a/src/pipecat/processors/frameworks/rtvi.py +++ /dev/null @@ -1,2169 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""RTVI (Real-Time Voice Interface) protocol implementation for Pipecat. - -This module provides the RTVI protocol implementation for real-time voice interactions -between clients and AI agents. It includes message handling, action processing, -and frame observation for the RTVI protocol. -""" - -import asyncio -import base64 -import time -from dataclasses import dataclass, field -from enum import Enum -from typing import ( - Any, - Awaitable, - Callable, - Dict, - List, - Literal, - Mapping, - Optional, - Set, - Tuple, - Union, -) - -from loguru import logger -from pydantic import BaseModel, Field, PrivateAttr, ValidationError - -from pipecat import version as pipecat_version -from pipecat.audio.utils import calculate_audio_volume -from pipecat.frames.frames import ( - AggregatedTextFrame, - AggregationType, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - CancelFrame, - DataFrame, - EndFrame, - EndTaskFrame, - ErrorFrame, - Frame, - FunctionCallCancelFrame, - FunctionCallInProgressFrame, - FunctionCallResultFrame, - FunctionCallsStartedFrame, - InputAudioRawFrame, - InputTransportMessageFrame, - InterimTranscriptionFrame, - LLMConfigureOutputFrame, - LLMContextFrame, - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMMessagesAppendFrame, - LLMTextFrame, - MetricsFrame, - OutputTransportMessageUrgentFrame, - StartFrame, - SystemFrame, - TranscriptionFrame, - TTSAudioRawFrame, - TTSStartedFrame, - TTSStoppedFrame, - TTSTextFrame, - UserMuteStartedFrame, - UserMuteStoppedFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, -) -from pipecat.metrics.metrics import ( - LLMUsageMetricsData, - ProcessingMetricsData, - TTFBMetricsData, - TTSUsageMetricsData, -) -from pipecat.observers.base_observer import BaseObserver, FramePushed -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.llm_service import ( - FunctionCallParams, # TODO(aleix): we shouldn't import `services` from `processors` -) -from pipecat.transports.base_input import BaseInputTransport -from pipecat.transports.base_output import BaseOutputTransport -from pipecat.transports.base_transport import BaseTransport -from pipecat.utils.string import match_endofsentence - -RTVI_PROTOCOL_VERSION = "1.2.0" - -RTVI_MESSAGE_LABEL = "rtvi-ai" -RTVIMessageLiteral = Literal["rtvi-ai"] - -ActionResult = Union[bool, int, float, str, list, dict] - - -class RTVIServiceOption(BaseModel): - """Configuration option for an RTVI service. - - Defines a configurable option that can be set for an RTVI service, - including its name, type, and handler function. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - name: str - type: Literal["bool", "number", "string", "array", "object"] - handler: Callable[["RTVIProcessor", str, "RTVIServiceOptionConfig"], Awaitable[None]] = Field( - exclude=True - ) - - -class RTVIService(BaseModel): - """An RTVI service definition. - - Represents a service that can be configured and used within the RTVI protocol, - containing a name and list of configurable options. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - name: str - options: List[RTVIServiceOption] - _options_dict: Dict[str, RTVIServiceOption] = PrivateAttr(default={}) - - def model_post_init(self, __context: Any) -> None: - """Initialize the options dictionary after model creation.""" - self._options_dict = {} - for option in self.options: - self._options_dict[option.name] = option - return super().model_post_init(__context) - - -class RTVIActionArgumentData(BaseModel): - """Data for an RTVI action argument. - - Contains the name and value of an argument passed to an RTVI action. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - name: str - value: Any - - -class RTVIActionArgument(BaseModel): - """Definition of an RTVI action argument. - - Specifies the name and expected type of an argument for an RTVI action. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - name: str - type: Literal["bool", "number", "string", "array", "object"] - - -class RTVIAction(BaseModel): - """An RTVI action definition. - - Represents an action that can be executed within the RTVI protocol, - including its service, name, arguments, and handler function. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - service: str - action: str - arguments: List[RTVIActionArgument] = Field(default_factory=list) - result: Literal["bool", "number", "string", "array", "object"] - handler: Callable[["RTVIProcessor", str, Dict[str, Any]], Awaitable[ActionResult]] = Field( - exclude=True - ) - _arguments_dict: Dict[str, RTVIActionArgument] = PrivateAttr(default={}) - - def model_post_init(self, __context: Any) -> None: - """Initialize the arguments dictionary after model creation.""" - self._arguments_dict = {} - for arg in self.arguments: - self._arguments_dict[arg.name] = arg - return super().model_post_init(__context) - - -class RTVIServiceOptionConfig(BaseModel): - """Configuration value for an RTVI service option. - - Contains the name and value to set for a specific service option. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - name: str - value: Any - - -class RTVIServiceConfig(BaseModel): - """Configuration for an RTVI service. - - Contains the service name and list of option configurations to apply. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - service: str - options: List[RTVIServiceOptionConfig] - - -class RTVIConfig(BaseModel): - """Complete RTVI configuration. - - Contains the full configuration for all RTVI services. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - config: List[RTVIServiceConfig] - - -# -# Client -> Pipecat messages. -# - - -# deprecated -class RTVIUpdateConfig(BaseModel): - """Request to update RTVI configuration. - - Contains new configuration settings and whether to interrupt the bot. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - config: List[RTVIServiceConfig] - interrupt: bool = False - - -class RTVIActionRunArgument(BaseModel): - """Argument for running an RTVI action. - - Contains the name and value of an argument to pass to an action. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - name: str - value: Any - - -class RTVIActionRun(BaseModel): - """Request to run an RTVI action. - - Contains the service, action name, and optional arguments. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - service: str - action: str - arguments: Optional[List[RTVIActionRunArgument]] = None - - -@dataclass -class RTVIActionFrame(DataFrame): - """Frame containing an RTVI action to execute. - - Parameters: - rtvi_action_run: The action to execute. - message_id: Optional message ID for response correlation. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - rtvi_action_run: RTVIActionRun - message_id: Optional[str] = None - - -class RTVIRawClientMessageData(BaseModel): - """Data structure expected from client messages sent to the RTVI server.""" - - t: str - d: Optional[Any] = None - - -class RTVIClientMessage(BaseModel): - """Cleansed data structure for client messages for handling.""" - - msg_id: str - type: str - data: Optional[Any] = None - - -@dataclass -class RTVIClientMessageFrame(SystemFrame): - """A frame for sending messages from the client to the RTVI server. - - This frame is meant for custom messaging from the client to the server - and expects a server-response message. - """ - - msg_id: str - type: str - data: Optional[Any] = None - - -@dataclass -class RTVIServerResponseFrame(SystemFrame): - """A frame for responding to a client RTVI message. - - This frame should be sent in response to an RTVIClientMessageFrame - and include the original RTVIClientMessageFrame to ensure the response - is properly attributed to the original request. To respond with an error, - set the `error` field to a string describing the error. This will result - in the client receiving a `response-error` message instead of a - `server-response` message. - """ - - client_msg: RTVIClientMessageFrame - data: Optional[Any] = None - error: Optional[str] = None - - -class RTVIRawServerResponseData(BaseModel): - """Data structure for server responses to client messages.""" - - t: str - d: Optional[Any] = None - - -class RTVIServerResponse(BaseModel): - """The RTVI-formatted message response from the server to the client. - - This message is used to respond to custom messages sent by the client. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["server-response"] = "server-response" - id: str - data: RTVIRawServerResponseData - - -class RTVIMessage(BaseModel): - """Base RTVI message structure. - - Represents the standard format for RTVI protocol messages. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: str - id: str - data: Optional[Dict[str, Any]] = None - - -# -# Pipecat -> Client responses and messages. -# - - -class RTVIErrorResponseData(BaseModel): - """Data for an RTVI error response. - - Contains the error message to send back to the client. - """ - - error: str - - -class RTVIErrorResponse(BaseModel): - """RTVI error response message. - - RTVI Formatted error response message for relaying failed client requests. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["error-response"] = "error-response" - id: str - data: RTVIErrorResponseData - - -class RTVIErrorData(BaseModel): - """Data for an RTVI error event. - - Contains error information including whether it's fatal. - """ - - error: str - fatal: bool # Indicates the pipeline has stopped due to this error - - -class RTVIError(BaseModel): - """RTVI error event message. - - RTVI Formatted error message for relaying errors in the pipeline. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["error"] = "error" - data: RTVIErrorData - - -class RTVIDescribeConfigData(BaseModel): - """Data for describing available RTVI configuration. - - Contains the list of available services and their options. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - config: List[RTVIService] - - -class RTVIDescribeConfig(BaseModel): - """Message describing available RTVI configuration. - - Sent in response to a describe-config request. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["config-available"] = "config-available" - id: str - data: RTVIDescribeConfigData - - -class RTVIDescribeActionsData(BaseModel): - """Data for describing available RTVI actions. - - Contains the list of available actions that can be executed. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - actions: List[RTVIAction] - - -class RTVIDescribeActions(BaseModel): - """Message describing available RTVI actions. - - Sent in response to a describe-actions request. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["actions-available"] = "actions-available" - id: str - data: RTVIDescribeActionsData - - -class RTVIConfigResponse(BaseModel): - """Response containing current RTVI configuration. - - Sent in response to a get-config request. - - .. deprecated:: 0.0.75 - Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["config"] = "config" - id: str - data: RTVIConfig - - -class RTVIActionResponseData(BaseModel): - """Data for an RTVI action response. - - Contains the result of executing an action. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - result: ActionResult - - -class RTVIActionResponse(BaseModel): - """Response to an RTVI action execution. - - Sent after successfully executing an action. - - .. deprecated:: 0.0.75 - Actions have been removed as part of the RTVI protocol 1.0.0. - Use custom client and server messages instead. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["action-response"] = "action-response" - id: str - data: RTVIActionResponseData - - -class AboutClientData(BaseModel): - """Data about the RTVI client. - - Contains information about the client, including which RTVI library it - is using, what platform it is on and any additional details, if available. - """ - - library: str - library_version: Optional[str] = None - platform: Optional[str] = None - platform_version: Optional[str] = None - platform_details: Optional[Any] = None - - -class RTVIClientReadyData(BaseModel): - """Data format of client ready messages. - - Contains the RTVIprotocol version and client information. - """ - - version: str - about: AboutClientData - - -class RTVIBotReadyData(BaseModel): - """Data for bot ready notification. - - Contains protocol version and initial configuration. - """ - - version: str - # The config field is deprecated and will not be included if - # the client's rtvi version is 1.0.0 or higher. - config: Optional[List[RTVIServiceConfig]] = None - about: Optional[Mapping[str, Any]] = None - - -class RTVIBotReady(BaseModel): - """Message indicating bot is ready for interaction. - - Sent after bot initialization is complete. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-ready"] = "bot-ready" - id: str - data: RTVIBotReadyData - - -class RTVILLMFunctionCallMessageData(BaseModel): - """Data for LLM function call notification. - - Contains function call details including name, ID, and arguments. - - .. deprecated:: 0.0.102 - Use ``RTVILLMFunctionCallInProgressMessageData`` instead. - """ - - function_name: str - tool_call_id: str - args: Mapping[str, Any] - - -class RTVILLMFunctionCallMessage(BaseModel): - """Message notifying of an LLM function call. - - Sent when the LLM makes a function call. - - .. deprecated:: 0.0.102 - Use ``RTVILLMFunctionCallInProgressMessage`` with the - ``llm-function-call-in-progress`` event type instead. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["llm-function-call"] = "llm-function-call" - data: RTVILLMFunctionCallMessageData - - -class RTVISendTextOptions(BaseModel): - """Options for sending text input to the LLM. - - Contains options for how the pipeline should process the text input. - """ - - run_immediately: bool = True - audio_response: bool = True - - -class RTVISendTextData(BaseModel): - """Data format for sending text input to the LLM. - - Contains the text content to send and any options for how the pipeline should process it. - - """ - - content: str - options: Optional[RTVISendTextOptions] = None - - -class RTVIAppendToContextData(BaseModel): - """Data format for appending messages to the context. - - Contains the role, content, and whether to run the message immediately. - - .. deprecated:: 0.0.85 - The RTVI message, append-to-context, has been deprecated. Use send-text - or custom client and server messages instead. - """ - - role: Literal["user", "assistant"] | str - content: Any - run_immediately: bool = False - - -class RTVIAppendToContext(BaseModel): - """RTVI Message format to append content to the LLM context.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["append-to-context"] = "append-to-context" - data: RTVIAppendToContextData - - -class RTVILLMFunctionCallStartMessageData(BaseModel): - """Data for LLM function call start notification. - - Contains the function name being called. Fields may be omitted based on - the configured function_call_report_level for security. - """ - - function_name: Optional[str] = None - - -class RTVILLMFunctionCallStartMessage(BaseModel): - """Message notifying that an LLM function call has started. - - Sent when the LLM begins a function call. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["llm-function-call-started"] = "llm-function-call-started" - data: RTVILLMFunctionCallStartMessageData - - -class RTVILLMFunctionCallResultData(BaseModel): - """Data for LLM function call result. - - Contains function call details and result. - """ - - function_name: str - tool_call_id: str - arguments: dict - result: dict | str - - -class RTVILLMFunctionCallInProgressMessageData(BaseModel): - """Data for LLM function call in-progress notification. - - Contains function call details including name, ID, and arguments. - Fields may be omitted based on the configured function_call_report_level for security. - """ - - tool_call_id: str - function_name: Optional[str] = None - arguments: Optional[Mapping[str, Any]] = None - - -class RTVILLMFunctionCallInProgressMessage(BaseModel): - """Message notifying that an LLM function call is in progress. - - Sent when the LLM function call execution begins. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["llm-function-call-in-progress"] = "llm-function-call-in-progress" - data: RTVILLMFunctionCallInProgressMessageData - - -class RTVILLMFunctionCallStoppedMessageData(BaseModel): - """Data for LLM function call stopped notification. - - Contains details about the function call that stopped, including - whether it was cancelled or completed with a result. - Fields may be omitted based on the configured function_call_report_level for security. - """ - - tool_call_id: str - cancelled: bool - function_name: Optional[str] = None - result: Optional[Any] = None - - -class RTVILLMFunctionCallStoppedMessage(BaseModel): - """Message notifying that an LLM function call has stopped. - - Sent when a function call completes (with result) or is cancelled. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["llm-function-call-stopped"] = "llm-function-call-stopped" - data: RTVILLMFunctionCallStoppedMessageData - - -class RTVIBotLLMStartedMessage(BaseModel): - """Message indicating bot LLM processing has started.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-llm-started"] = "bot-llm-started" - - -class RTVIBotLLMStoppedMessage(BaseModel): - """Message indicating bot LLM processing has stopped.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-llm-stopped"] = "bot-llm-stopped" - - -class RTVIBotTTSStartedMessage(BaseModel): - """Message indicating bot TTS processing has started.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-tts-started"] = "bot-tts-started" - - -class RTVIBotTTSStoppedMessage(BaseModel): - """Message indicating bot TTS processing has stopped.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-tts-stopped"] = "bot-tts-stopped" - - -class RTVITextMessageData(BaseModel): - """Data for text-based RTVI messages. - - Contains text content. - """ - - text: str - - -class RTVIBotOutputMessageData(RTVITextMessageData): - """Data for bot output RTVI messages. - - Extends RTVITextMessageData to include metadata about the output. - """ - - spoken: bool = False # Indicates if the text has been spoken by TTS - aggregated_by: AggregationType | str - # Indicates what form the text is in (e.g., by word, sentence, etc.) - - -class RTVIBotOutputMessage(BaseModel): - """Message containing bot output text. - - An event meant to holistically represent what the bot is outputting, - along with metadata about the output and if it has been spoken. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-output"] = "bot-output" - data: RTVIBotOutputMessageData - - -class RTVIBotTranscriptionMessage(BaseModel): - """Message containing bot transcription text. - - Sent when the bot's speech is transcribed. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-transcription"] = "bot-transcription" - data: RTVITextMessageData - - -class RTVIBotLLMTextMessage(BaseModel): - """Message containing bot LLM text output. - - Sent when the bot's LLM generates text. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-llm-text"] = "bot-llm-text" - data: RTVITextMessageData - - -class RTVIBotTTSTextMessage(BaseModel): - """Message containing bot TTS text output. - - Sent when text is being processed by TTS. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-tts-text"] = "bot-tts-text" - data: RTVITextMessageData - - -class RTVIAudioMessageData(BaseModel): - """Data for audio-based RTVI messages. - - Contains audio data and metadata. - """ - - audio: str - sample_rate: int - num_channels: int - - -class RTVIBotTTSAudioMessage(BaseModel): - """Message containing bot TTS audio output. - - Sent when the bot's TTS generates audio. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-tts-audio"] = "bot-tts-audio" - data: RTVIAudioMessageData - - -class RTVIUserTranscriptionMessageData(BaseModel): - """Data for user transcription messages. - - Contains transcription text and metadata. - """ - - text: str - user_id: str - timestamp: str - final: bool - - -class RTVIUserTranscriptionMessage(BaseModel): - """Message containing user transcription. - - Sent when user speech is transcribed. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["user-transcription"] = "user-transcription" - data: RTVIUserTranscriptionMessageData - - -class RTVIUserLLMTextMessage(BaseModel): - """Message containing user text input for LLM. - - Sent when user text is processed by the LLM. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["user-llm-text"] = "user-llm-text" - data: RTVITextMessageData - - -class RTVIUserStartedSpeakingMessage(BaseModel): - """Message indicating user has started speaking.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["user-started-speaking"] = "user-started-speaking" - - -class RTVIUserStoppedSpeakingMessage(BaseModel): - """Message indicating user has stopped speaking.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["user-stopped-speaking"] = "user-stopped-speaking" - - -class RTVIUserMuteStartedMessage(BaseModel): - """Message indicating user has been muted.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["user-mute-started"] = "user-mute-started" - - -class RTVIUserMuteStoppedMessage(BaseModel): - """Message indicating user has been unmuted.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["user-mute-stopped"] = "user-mute-stopped" - - -class RTVIBotStartedSpeakingMessage(BaseModel): - """Message indicating bot has started speaking.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-started-speaking"] = "bot-started-speaking" - - -class RTVIBotStoppedSpeakingMessage(BaseModel): - """Message indicating bot has stopped speaking.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-stopped-speaking"] = "bot-stopped-speaking" - - -class RTVIMetricsMessage(BaseModel): - """Message containing performance metrics. - - Sent to provide performance and usage metrics. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["metrics"] = "metrics" - data: Mapping[str, Any] - - -class RTVIServerMessage(BaseModel): - """Generic server message. - - Used for custom server-to-client messages. - """ - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["server-message"] = "server-message" - data: Any - - -class RTVIAudioLevelMessageData(BaseModel): - """Data format for sending audio levels.""" - - value: float - - -class RTVIUserAudioLevelMessage(BaseModel): - """Message indicating user audio level.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["user-audio-level"] = "user-audio-level" - data: RTVIAudioLevelMessageData - - -class RTVIBotAudioLevelMessage(BaseModel): - """Message indicating bot audio level.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["bot-audio-level"] = "bot-audio-level" - data: RTVIAudioLevelMessageData - - -class RTVISystemLogMessage(BaseModel): - """Message including a system log.""" - - label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL - type: Literal["system-log"] = "system-log" - data: RTVITextMessageData - - -@dataclass -class RTVIServerMessageFrame(SystemFrame): - """A frame for sending server messages to the client. - - Parameters: - data: The message data to send to the client. - """ - - data: Any - - def __str__(self): - """String representation of the RTVI server message frame.""" - return f"{self.name}(data: {self.data})" - - -class RTVIFunctionCallReportLevel(str, Enum): - """Level of detail to include in function call RTVI events. - - Controls what information is exposed in function call events for security. - - Values: - DISABLED: No events emitted for this function call. - NONE: Events only with tool_call_id, no function name or metadata (most secure). - NAME: Events with function name, no arguments or results. - FULL: Events with function name, arguments, and results. - """ - - DISABLED = "disabled" - NONE = "none" - NAME = "name" - FULL = "full" - - -@dataclass -class RTVIObserverParams: - """Parameters for configuring RTVI Observer behavior. - - .. deprecated:: 0.0.87 - Parameter `errors_enabled` is deprecated. Error messages are always enabled. - - Parameters: - bot_output_enabled: Indicates if bot output messages should be sent. - bot_llm_enabled: Indicates if the bot's LLM messages should be sent. - bot_tts_enabled: Indicates if the bot's TTS messages should be sent. - bot_speaking_enabled: Indicates if the bot's started/stopped speaking messages should be sent. - bot_audio_level_enabled: Indicates if bot's audio level messages should be sent. - user_llm_enabled: Indicates if the user's LLM input messages should be sent. - user_speaking_enabled: Indicates if the user's started/stopped speaking messages should be sent. - user_transcription_enabled: Indicates if user's transcription messages should be sent. - user_audio_level_enabled: Indicates if user's audio level messages should be sent. - metrics_enabled: Indicates if metrics messages should be sent. - system_logs_enabled: Indicates if system logs should be sent. - errors_enabled: [Deprecated] Indicates if errors messages should be sent. - ignored_sources: List of frame processors whose frames should be silently ignored - by this observer. Useful for suppressing RTVI messages from secondary pipeline - branches (e.g. a silent evaluation LLM) that should not be visible to clients. - Sources can also be added and removed dynamically via ``add_ignored_source()`` - and ``remove_ignored_source()``. - skip_aggregator_types: List of aggregation types to skip sending as tts/output messages. - Note: if using this to avoid sending secure information, be sure to also disable - bot_llm_enabled to avoid leaking through LLM messages. - bot_output_transforms: A list of callables to transform text before just before sending it - to TTS. Each callable takes the aggregated text and its type, and returns the - transformed text. To register, provide a list of tuples of - (aggregation_type | '*', transform_function). - audio_level_period_secs: How often audio levels should be sent if enabled. - function_call_report_level: Controls what information is exposed in function call - events for security. A dict mapping function names to levels, where ``"*"`` - sets the default level for unlisted functions:: - - function_call_report_level={ - "*": RTVIFunctionCallReportLevel.NONE, # Default: events with no metadata - "get_weather": RTVIFunctionCallReportLevel.FULL, # Expose everything - } - - Levels: - - DISABLED: No events emitted for this function. - - NONE: Events with tool_call_id only (most secure when events needed). - - NAME: Adds function name to events. - - FULL: Adds function name, arguments, and results. - - Defaults to ``{"*": RTVIFunctionCallReportLevel.NONE}``. - """ - - bot_output_enabled: bool = True - bot_llm_enabled: bool = True - bot_tts_enabled: bool = True - bot_speaking_enabled: bool = True - bot_audio_level_enabled: bool = False - user_llm_enabled: bool = True - user_speaking_enabled: bool = True - user_mute_enabled: bool = True - user_transcription_enabled: bool = True - user_audio_level_enabled: bool = False - metrics_enabled: bool = True - system_logs_enabled: bool = False - errors_enabled: Optional[bool] = None - ignored_sources: List[FrameProcessor] = field(default_factory=list) - skip_aggregator_types: Optional[List[AggregationType | str]] = None - bot_output_transforms: Optional[ - List[ - Tuple[ - AggregationType | str, - Callable[[str, AggregationType | str], Awaitable[str]], - ] - ] - ] = None - audio_level_period_secs: float = 0.15 - function_call_report_level: Dict[str, RTVIFunctionCallReportLevel] = field( - default_factory=lambda: {"*": RTVIFunctionCallReportLevel.NONE} - ) - - -class RTVIObserver(BaseObserver): - """Pipeline frame observer for RTVI server message handling. - - This observer monitors pipeline frames and converts them into appropriate RTVI messages - for client communication. It handles various frame types including speech events, - transcriptions, LLM responses, and TTS events. - - Note: - This observer only handles outgoing messages. Incoming RTVI client messages - are handled by the RTVIProcessor. - """ - - def __init__( - self, - rtvi: Optional["RTVIProcessor"] = None, - *, - params: Optional[RTVIObserverParams] = None, - **kwargs, - ): - """Initialize the RTVI observer. - - Args: - rtvi: The RTVI processor to push frames to. - params: Settings to enable/disable specific messages. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - self._rtvi = rtvi - self._params = params or RTVIObserverParams() - - self._ignored_sources: Set[FrameProcessor] = set(self._params.ignored_sources) - self._frames_seen = set() - - self._bot_transcription = "" - self._last_user_audio_level = 0 - self._last_bot_audio_level = 0 - - # Track bot speaking state for queuing aggregated text frames - self._bot_is_speaking = False - self._queued_aggregated_text_frames: List[AggregatedTextFrame] = [] - - if self._params.system_logs_enabled: - self._system_logger_id = logger.add(self._logger_sink) - - if self._params.errors_enabled is not None: - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Parameter `errors_enabled` is deprecated. Error messages are always enabled.", - DeprecationWarning, - ) - - self._aggregation_transforms: List[ - Tuple[AggregationType | str, Callable[[str, AggregationType | str], Awaitable[str]]] - ] = self._params.bot_output_transforms or [] - - def add_bot_output_transformer( - self, - transform_function: Callable[[str, AggregationType | str], Awaitable[str]], - aggregation_type: AggregationType | str = "*", - ): - """Transform text for a specific aggregation type before sending as Bot Output or TTS. - - Args: - transform_function: The function to apply for transformation. This function should take - the text and aggregation type as input and return the transformed text. - Ex.: async def my_transform(text: str, aggregation_type: str) -> str: - aggregation_type: The type of aggregation to transform. This value defaults to "*" to - handle all text before sending to the client. - """ - self._aggregation_transforms.append((aggregation_type, transform_function)) - - def remove_bot_output_transformer( - self, - transform_function: Callable[[str, AggregationType | str], Awaitable[str]], - aggregation_type: AggregationType | str = "*", - ): - """Remove a text transformer for a specific aggregation type. - - Args: - transform_function: The function to remove. - aggregation_type: The type of aggregation to remove the transformer for. - """ - self._aggregation_transforms = [ - (agg_type, func) - for agg_type, func in self._aggregation_transforms - if not (agg_type == aggregation_type and func == transform_function) - ] - - def add_ignored_source(self, source: FrameProcessor): - """Ignore all frames pushed by the given processor. - - Any frame whose source matches ``source`` will be silently skipped, - preventing RTVI messages from being emitted for activity in that - processor. Useful for suppressing events from secondary pipeline - branches (e.g. a silent evaluation LLM) that should not be visible - to clients. - - Args: - source: The frame processor to ignore. - """ - self._ignored_sources.add(source) - - def remove_ignored_source(self, source: FrameProcessor): - """Stop ignoring frames pushed by the given processor. - - Reverses a previous call to ``add_ignored_source()``. If ``source`` - was not previously ignored this is a no-op. - - Args: - source: The frame processor to stop ignoring. - """ - self._ignored_sources.discard(source) - - def _get_function_call_report_level(self, function_name: str) -> RTVIFunctionCallReportLevel: - """Get the report level for a specific function call. - - Args: - function_name: The name of the function to get the report level for. - - Returns: - The report level for the function. Looks up the function name first, - then falls back to "*" key, then NONE. - """ - levels = self._params.function_call_report_level - if function_name in levels: - return levels[function_name] - return levels.get("*", RTVIFunctionCallReportLevel.NONE) - - async def _logger_sink(self, message): - """Logger sink so we can send system logs to RTVI clients.""" - message = RTVISystemLogMessage(data=RTVITextMessageData(text=message)) - await self.send_rtvi_message(message) - - async def cleanup(self): - """Cleanup RTVI observer resources.""" - await super().cleanup() - if self._params.system_logs_enabled: - logger.remove(self._system_logger_id) - - async def send_rtvi_message(self, model: BaseModel, exclude_none: bool = True): - """Send an RTVI message. - - By default, we push a transport frame. But this function can be - overriden by subclass to send RTVI messages in different ways. - - Args: - model: The message to send. - exclude_none: Whether to exclude None values from the model dump. - - """ - if self._rtvi: - await self._rtvi.push_transport_message(model, exclude_none) - - async def on_push_frame(self, data: FramePushed): - """Process a frame being pushed through the pipeline. - - Args: - data: Frame push event data containing source, frame, direction, and timestamp. - """ - src = data.source - frame = data.frame - direction = data.direction - - # Frames from explicitly ignored sources are always skipped. - if self._ignored_sources and src in self._ignored_sources: - return - - # For broadcast frames (pushed in both directions), only process - # the downstream copy to avoid sending duplicate RTVI messages. - if frame.broadcast_sibling_id is not None and direction != FrameDirection.DOWNSTREAM: - return - - # If we have already seen this frame, let's skip it. - if frame.id in self._frames_seen: - return - - # This tells whether the frame is already processed. If false, we will try - # again the next time we see the frame. - mark_as_seen = True - - if ( - isinstance(frame, (UserStartedSpeakingFrame, UserStoppedSpeakingFrame)) - and self._params.user_speaking_enabled - ): - await self._handle_interruptions(frame) - elif ( - isinstance(frame, (UserMuteStartedFrame, UserMuteStoppedFrame)) - and self._params.user_mute_enabled - ): - await self._handle_user_mute(frame) - elif ( - isinstance(frame, (BotStartedSpeakingFrame, BotStoppedSpeakingFrame)) - and self._params.bot_speaking_enabled - ): - await self._handle_bot_speaking(frame) - elif ( - isinstance(frame, (TranscriptionFrame, InterimTranscriptionFrame)) - and self._params.user_transcription_enabled - ): - await self._handle_user_transcriptions(frame) - elif ( - isinstance(frame, (OpenAILLMContextFrame, LLMContextFrame)) - and self._params.user_llm_enabled - ): - await self._handle_context(frame) - elif isinstance(frame, LLMFullResponseStartFrame) and self._params.bot_llm_enabled: - await self.send_rtvi_message(RTVIBotLLMStartedMessage()) - elif isinstance(frame, LLMFullResponseEndFrame) and self._params.bot_llm_enabled: - await self.send_rtvi_message(RTVIBotLLMStoppedMessage()) - elif isinstance(frame, LLMTextFrame) and self._params.bot_llm_enabled: - await self._handle_llm_text_frame(frame) - elif isinstance(frame, TTSStartedFrame) and self._params.bot_tts_enabled: - await self.send_rtvi_message(RTVIBotTTSStartedMessage()) - elif isinstance(frame, TTSStoppedFrame) and self._params.bot_tts_enabled: - await self.send_rtvi_message(RTVIBotTTSStoppedMessage()) - elif isinstance(frame, AggregatedTextFrame) and ( - self._params.bot_output_enabled or self._params.bot_tts_enabled - ): - if isinstance(frame, TTSTextFrame) and not isinstance(src, BaseOutputTransport): - # This check is to make sure we handle the frame when it has gone - # through the transport and has correct timing. - mark_as_seen = False - else: - await self._handle_aggregated_llm_text(frame) - elif isinstance(frame, MetricsFrame) and self._params.metrics_enabled: - await self._handle_metrics(frame) - elif isinstance(frame, FunctionCallsStartedFrame): - for function_call in frame.function_calls: - report_level = self._get_function_call_report_level(function_call.function_name) - if report_level == RTVIFunctionCallReportLevel.DISABLED: - continue - data = RTVILLMFunctionCallStartMessageData() - if report_level in ( - RTVIFunctionCallReportLevel.NAME, - RTVIFunctionCallReportLevel.FULL, - ): - data.function_name = function_call.function_name - message = RTVILLMFunctionCallStartMessage(data=data) - await self.send_rtvi_message(message) - elif isinstance(frame, FunctionCallInProgressFrame): - report_level = self._get_function_call_report_level(frame.function_name) - if report_level != RTVIFunctionCallReportLevel.DISABLED: - data = RTVILLMFunctionCallInProgressMessageData(tool_call_id=frame.tool_call_id) - if report_level in ( - RTVIFunctionCallReportLevel.NAME, - RTVIFunctionCallReportLevel.FULL, - ): - data.function_name = frame.function_name - if report_level == RTVIFunctionCallReportLevel.FULL: - data.arguments = frame.arguments - message = RTVILLMFunctionCallInProgressMessage(data=data) - await self.send_rtvi_message(message) - elif isinstance(frame, FunctionCallCancelFrame): - report_level = self._get_function_call_report_level(frame.function_name) - if report_level != RTVIFunctionCallReportLevel.DISABLED: - data = RTVILLMFunctionCallStoppedMessageData( - tool_call_id=frame.tool_call_id, - cancelled=True, - ) - if report_level in ( - RTVIFunctionCallReportLevel.NAME, - RTVIFunctionCallReportLevel.FULL, - ): - data.function_name = frame.function_name - message = RTVILLMFunctionCallStoppedMessage(data=data) - await self.send_rtvi_message(message) - elif isinstance(frame, FunctionCallResultFrame): - report_level = self._get_function_call_report_level(frame.function_name) - if report_level != RTVIFunctionCallReportLevel.DISABLED: - data = RTVILLMFunctionCallStoppedMessageData( - tool_call_id=frame.tool_call_id, - cancelled=False, - ) - if report_level in ( - RTVIFunctionCallReportLevel.NAME, - RTVIFunctionCallReportLevel.FULL, - ): - data.function_name = frame.function_name - if report_level == RTVIFunctionCallReportLevel.FULL: - data.result = frame.result if frame.result else None - message = RTVILLMFunctionCallStoppedMessage(data=data) - await self.send_rtvi_message(message) - elif isinstance(frame, RTVIServerMessageFrame): - message = RTVIServerMessage(data=frame.data) - await self.send_rtvi_message(message) - elif isinstance(frame, RTVIServerResponseFrame): - if frame.error is not None: - await self._send_error_response(frame) - else: - await self._send_server_response(frame) - elif isinstance(frame, InputAudioRawFrame) and self._params.user_audio_level_enabled: - curr_time = time.time() - diff_time = curr_time - self._last_user_audio_level - if diff_time > self._params.audio_level_period_secs: - level = calculate_audio_volume(frame.audio, frame.sample_rate) - message = RTVIUserAudioLevelMessage(data=RTVIAudioLevelMessageData(value=level)) - await self.send_rtvi_message(message) - self._last_user_audio_level = curr_time - elif isinstance(frame, TTSAudioRawFrame) and self._params.bot_audio_level_enabled: - curr_time = time.time() - diff_time = curr_time - self._last_bot_audio_level - if diff_time > self._params.audio_level_period_secs: - level = calculate_audio_volume(frame.audio, frame.sample_rate) - message = RTVIBotAudioLevelMessage(data=RTVIAudioLevelMessageData(value=level)) - await self.send_rtvi_message(message) - self._last_bot_audio_level = curr_time - - if mark_as_seen: - self._frames_seen.add(frame.id) - - async def _handle_interruptions(self, frame: Frame): - """Handle user speaking interruption frames.""" - message = None - if isinstance(frame, UserStartedSpeakingFrame): - message = RTVIUserStartedSpeakingMessage() - elif isinstance(frame, UserStoppedSpeakingFrame): - message = RTVIUserStoppedSpeakingMessage() - - if message: - await self.send_rtvi_message(message) - - async def _handle_user_mute(self, frame: Frame): - """Handle user mute/unmute frames.""" - message = None - if isinstance(frame, UserMuteStartedFrame): - message = RTVIUserMuteStartedMessage() - elif isinstance(frame, UserMuteStoppedFrame): - message = RTVIUserMuteStoppedMessage() - - if message: - await self.send_rtvi_message(message) - - async def _handle_bot_speaking(self, frame: Frame): - """Handle bot speaking event frames.""" - if isinstance(frame, BotStartedSpeakingFrame): - message = RTVIBotStartedSpeakingMessage() - await self.send_rtvi_message(message) - # Flush any queued aggregated text frames - for queued_frame in self._queued_aggregated_text_frames: - await self._send_aggregated_llm_text(queued_frame) - self._queued_aggregated_text_frames.clear() - self._bot_is_speaking = True - elif isinstance(frame, BotStoppedSpeakingFrame): - message = RTVIBotStoppedSpeakingMessage() - await self.send_rtvi_message(message) - self._bot_is_speaking = False - - async def _handle_aggregated_llm_text(self, frame: AggregatedTextFrame): - """Handle aggregated LLM text output frames.""" - if self._bot_is_speaking: - # Bot has already started speaking, send directly - await self._send_aggregated_llm_text(frame) - else: - # Bot hasn't started speaking yet, queue the frame - self._queued_aggregated_text_frames.append(frame) - - async def _send_aggregated_llm_text(self, frame: AggregatedTextFrame): - """Send aggregated LLM text messages.""" - # Skip certain aggregator types if configured to do so. - if ( - self._params.skip_aggregator_types - and frame.aggregated_by in self._params.skip_aggregator_types - ): - return - - text = frame.text - type = frame.aggregated_by - for aggregation_type, transform in self._aggregation_transforms: - if aggregation_type == type or aggregation_type == "*": - text = await transform(text, type) - - isTTS = isinstance(frame, TTSTextFrame) - if self._params.bot_output_enabled: - message = RTVIBotOutputMessage( - data=RTVIBotOutputMessageData(text=text, spoken=isTTS, aggregated_by=type) - ) - await self.send_rtvi_message(message) - - if isTTS and self._params.bot_tts_enabled: - tts_message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=text)) - await self.send_rtvi_message(tts_message) - - async def _handle_llm_text_frame(self, frame: LLMTextFrame): - """Handle LLM text output frames.""" - message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text)) - await self.send_rtvi_message(message) - - # TODO (mrkb): Remove all this logic when we fully deprecate bot-transcription messages. - self._bot_transcription += frame.text - - if match_endofsentence(self._bot_transcription) and len(self._bot_transcription) > 0: - await self.send_rtvi_message( - RTVIBotTranscriptionMessage(data=RTVITextMessageData(text=self._bot_transcription)) - ) - self._bot_transcription = "" - - async def _handle_user_transcriptions(self, frame: Frame): - """Handle user transcription frames.""" - message = None - if isinstance(frame, TranscriptionFrame): - message = RTVIUserTranscriptionMessage( - data=RTVIUserTranscriptionMessageData( - text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp, final=True - ) - ) - elif isinstance(frame, InterimTranscriptionFrame): - message = RTVIUserTranscriptionMessage( - data=RTVIUserTranscriptionMessageData( - text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp, final=False - ) - ) - - if message: - await self.send_rtvi_message(message) - - async def _handle_context(self, frame: OpenAILLMContextFrame | LLMContextFrame): - """Process LLM context frames to extract user messages for the RTVI client.""" - try: - if isinstance(frame, OpenAILLMContextFrame): - messages = frame.context.messages - else: - messages = frame.context.get_messages() - if not messages: - return - - message = messages[-1] - - # Handle Google LLM format (protobuf objects with attributes) - # Note: not possible if frame is a universal LLMContextFrame - if hasattr(message, "role") and message.role == "user" and hasattr(message, "parts"): - text = "".join(part.text for part in message.parts if hasattr(part, "text")) - if text: - rtvi_message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=text)) - await self.send_rtvi_message(rtvi_message) - - # Handle OpenAI format (original implementation) - elif isinstance(message, dict): - if message["role"] == "user": - content = message["content"] - if isinstance(content, list): - text = " ".join(item["text"] for item in content if "text" in item) - else: - text = content - rtvi_message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=text)) - await self.send_rtvi_message(rtvi_message) - - except Exception as e: - logger.warning(f"Caught an error while trying to handle context: {e}") - - async def _handle_metrics(self, frame: MetricsFrame): - """Handle metrics frames and convert to RTVI metrics messages.""" - metrics = {} - for d in frame.data: - if isinstance(d, TTFBMetricsData): - if "ttfb" not in metrics: - metrics["ttfb"] = [] - metrics["ttfb"].append(d.model_dump(exclude_none=True)) - elif isinstance(d, ProcessingMetricsData): - if "processing" not in metrics: - metrics["processing"] = [] - metrics["processing"].append(d.model_dump(exclude_none=True)) - elif isinstance(d, LLMUsageMetricsData): - if "tokens" not in metrics: - metrics["tokens"] = [] - metrics["tokens"].append(d.value.model_dump(exclude_none=True)) - elif isinstance(d, TTSUsageMetricsData): - if "characters" not in metrics: - metrics["characters"] = [] - metrics["characters"].append(d.model_dump(exclude_none=True)) - - message = RTVIMetricsMessage(data=metrics) - await self.send_rtvi_message(message) - - async def _send_server_response(self, frame: RTVIServerResponseFrame): - """Send a response to the client for a specific request.""" - message = RTVIServerResponse( - id=str(frame.client_msg.msg_id), - data=RTVIRawServerResponseData(t=frame.client_msg.type, d=frame.data), - ) - await self.send_rtvi_message(message) - - async def _send_error_response(self, frame: RTVIServerResponseFrame): - """Send a response to the client for a specific request.""" - message = RTVIErrorResponse( - id=str(frame.client_msg.msg_id), data=RTVIErrorResponseData(error=frame.error) - ) - await self.send_rtvi_message(message) - - -class RTVIProcessor(FrameProcessor): - """Main processor for handling RTVI protocol messages and actions. - - This processor manages the RTVI protocol communication including client-server - handshaking, configuration management, action execution, and message routing. - It serves as the central hub for RTVI protocol operations. - """ - - def __init__( - self, - *, - config: Optional[RTVIConfig] = None, - transport: Optional[BaseTransport] = None, - **kwargs, - ): - """Initialize the RTVI processor. - - Args: - config: Initial RTVI configuration. - transport: Transport layer for communication. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - self._config = config or RTVIConfig(config=[]) - - self._bot_ready = False - self._client_ready = False - self._client_ready_id = "" - # Default to 0.3.0 which is the last version before actually having a - # "client-version". - self._client_version = [0, 3, 0] - self._llm_skip_tts: bool = False # Keep in sync with llm_service.py's configuration. - - self._registered_actions: Dict[str, RTVIAction] = {} - self._registered_services: Dict[str, RTVIService] = {} - - # A task to process incoming action frames. - self._action_task: Optional[asyncio.Task] = None - - # A task to process incoming transport messages. - self._message_task: Optional[asyncio.Task] = None - - self._register_event_handler("on_bot_started") - self._register_event_handler("on_client_ready") - self._register_event_handler("on_client_message") - - self._input_transport = None - self._transport = transport - if self._transport: - input_transport = self._transport.input() - if isinstance(input_transport, BaseInputTransport): - self._input_transport = input_transport - self._input_transport.enable_audio_in_stream_on_start(False) - - def register_action(self, action: RTVIAction): - """Register an action that can be executed via RTVI. - - Args: - action: The action to register. - """ - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "The actions API is deprecated, use server and client messages instead.", - DeprecationWarning, - ) - - id = self._action_id(action.service, action.action) - self._registered_actions[id] = action - - def register_service(self, service: RTVIService): - """Register a service that can be configured via RTVI. - - Args: - service: The service to register. - """ - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "The actions API is deprecated, use server and client messages instead.", - DeprecationWarning, - ) - - self._registered_services[service.name] = service - - def create_rtvi_observer(self, *, params: Optional[RTVIObserverParams] = None, **kwargs): - """Creates a new RTVI Observer. - - Args: - params: Settings to enable/disable specific messages. - **kwargs: Additional arguments passed to the observer. - - Returns: - A new RTVI observer. - """ - return RTVIObserver(self, params=params, **kwargs) - - async def set_client_ready(self): - """Mark the client as ready and trigger the ready event.""" - self._client_ready = True - await self._call_event_handler("on_client_ready") - - async def set_bot_ready(self, about: Mapping[str, Any] = None): - """Mark the bot as ready and send the bot-ready message. - - Args: - about: Optional information about the bot to include in the ready message. - If left as None, the Pipecat library and version will be used. - """ - self._bot_ready = True - # Only call the (deprecated) _update_config method if the we're using a - # config (which is deprecated). Otherwise we'd always print an - # unnecessary deprecation warning. - if self._config.config: - await self._update_config(self._config, False) - await self._send_bot_ready(about=about) - - async def interrupt_bot(self): - """Send a bot interruption frame upstream.""" - await self.broadcast_interruption() - - async def send_server_message(self, data: Any): - """Send a server message to the client.""" - message = RTVIServerMessage(data=data) - await self._send_server_message(message) - - async def send_server_response(self, client_msg: RTVIClientMessage, data: Any): - """Send a server response for a given client message.""" - message = RTVIServerResponse( - id=client_msg.msg_id, data=RTVIRawServerResponseData(t=client_msg.type, d=data) - ) - await self._send_server_message(message) - - async def send_error_response(self, client_msg: RTVIClientMessage, error: str): - """Send an error response for a given client message.""" - await self._send_error_response(id=client_msg.msg_id, error=error) - - async def send_error(self, error: str): - """Send an error message to the client. - - Args: - error: The error message to send. - """ - await self._send_error_frame(ErrorFrame(error=error)) - - async def push_transport_message(self, model: BaseModel, exclude_none: bool = True): - """Push a transport message frame.""" - frame = OutputTransportMessageUrgentFrame( - message=model.model_dump(exclude_none=exclude_none) - ) - await self.push_frame(frame) - - async def handle_message(self, message: RTVIMessage): - """Handle an incoming RTVI message. - - Args: - message: The RTVI message to handle. - """ - await self._message_queue.put(message) - - async def handle_function_call(self, params: FunctionCallParams): - """Handle a function call from the LLM. - - Args: - params: The function call parameters. - - .. deprecated:: 0.0.102 - This method is deprecated. Function call events are now automatically - sent by ``RTVIObserver`` using the ``llm-function-call-in-progress`` event. - Configure reporting level via ``RTVIObserverParams.function_call_report_level``. - """ - import warnings - - warnings.warn( - "handle_function_call is deprecated. Function call events are now " - "automatically sent by RTVIObserver using llm-function-call-in-progress.", - DeprecationWarning, - stacklevel=2, - ) - fn = RTVILLMFunctionCallMessageData( - function_name=params.function_name, - tool_call_id=params.tool_call_id, - args=params.arguments, - ) - message = RTVILLMFunctionCallMessage(data=fn) - await self.push_transport_message(message, exclude_none=False) - - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process incoming frames through the RTVI processor. - - Args: - frame: The frame to process. - direction: The direction of frame flow. - """ - await super().process_frame(frame, direction) - - # Specific system frames - if isinstance(frame, StartFrame): - # Push StartFrame before start(), because we want StartFrame to be - # processed by every processor before any other frame is processed. - await self.push_frame(frame, direction) - await self._start(frame) - elif isinstance(frame, CancelFrame): - await self._cancel(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, ErrorFrame): - await self._send_error_frame(frame) - await self.push_frame(frame, direction) - elif isinstance(frame, InputTransportMessageFrame): - await self._handle_transport_message(frame) - # All other system frames - elif isinstance(frame, SystemFrame): - await self.push_frame(frame, direction) - # Control frames - elif isinstance(frame, EndFrame): - # Push EndFrame before stop(), because stop() waits on the task to - # finish and the task finishes when EndFrame is processed. - await self.push_frame(frame, direction) - await self._stop(frame) - # Data frames - elif isinstance(frame, RTVIActionFrame): - await self._action_queue.put(frame) - elif isinstance(frame, LLMConfigureOutputFrame): - self._llm_skip_tts = frame.skip_tts - await self.push_frame(frame, direction) - # Other frames - else: - await self.push_frame(frame, direction) - - async def _start(self, frame: StartFrame): - """Start the RTVI processor tasks.""" - if not self._action_task: - self._action_queue = asyncio.Queue() - self._action_task = self.create_task(self._action_task_handler()) - if not self._message_task: - self._message_queue = asyncio.Queue() - self._message_task = self.create_task(self._message_task_handler()) - await self._call_event_handler("on_bot_started") - - async def _stop(self, frame: EndFrame): - """Stop the RTVI processor tasks.""" - await self._cancel_tasks() - - async def _cancel(self, frame: CancelFrame): - """Cancel the RTVI processor tasks.""" - await self._cancel_tasks() - - async def _cancel_tasks(self): - """Cancel all running tasks.""" - if self._action_task: - await self.cancel_task(self._action_task) - self._action_task = None - - if self._message_task: - await self.cancel_task(self._message_task) - self._message_task = None - - async def _action_task_handler(self): - """Handle incoming action frames.""" - while True: - frame = await self._action_queue.get() - await self._handle_action(frame.message_id, frame.rtvi_action_run) - self._action_queue.task_done() - - async def _message_task_handler(self): - """Handle incoming transport messages.""" - while True: - message = await self._message_queue.get() - await self._handle_message(message) - self._message_queue.task_done() - - async def _handle_transport_message(self, frame: InputTransportMessageFrame): - """Handle an incoming transport message frame.""" - try: - transport_message = frame.message - if transport_message.get("label") != RTVI_MESSAGE_LABEL: - logger.warning(f"Ignoring not RTVI message: {transport_message}") - return - message = RTVIMessage.model_validate(transport_message) - await self._message_queue.put(message) - except ValidationError as e: - await self.send_error(f"Invalid RTVI transport message: {e}") - logger.warning(f"Invalid RTVI transport message: {e}") - - async def _handle_message(self, message: RTVIMessage): - """Handle a parsed RTVI message.""" - try: - match message.type: - case "client-ready": - data = None - try: - data = RTVIClientReadyData.model_validate(message.data) - except ValidationError: - # Not all clients have been updated to RTVI 1.0.0. - # For now, that's okay, we just log their info as unknown. - data = None - pass - await self._handle_client_ready(message.id, data) - case "describe-actions": - await self._handle_describe_actions(message.id) - case "describe-config": - await self._handle_describe_config(message.id) - case "get-config": - await self._handle_get_config(message.id) - case "update-config": - update_config = RTVIUpdateConfig.model_validate(message.data) - await self._handle_update_config(message.id, update_config) - case "disconnect-bot": - await self.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM) - case "client-message": - data = RTVIRawClientMessageData.model_validate(message.data) - await self._handle_client_message(message.id, data) - case "action": - action = RTVIActionRun.model_validate(message.data) - action_frame = RTVIActionFrame(message_id=message.id, rtvi_action_run=action) - await self._action_queue.put(action_frame) - case "llm-function-call-result": - data = RTVILLMFunctionCallResultData.model_validate(message.data) - await self._handle_function_call_result(data) - case "send-text": - data = RTVISendTextData.model_validate(message.data) - await self._handle_send_text(data) - case "append-to-context": - logger.warning( - f"The append-to-context message is deprecated, use send-text instead." - ) - data = RTVIAppendToContextData.model_validate(message.data) - await self._handle_update_context(data) - case "raw-audio" | "raw-audio-batch": - await self._handle_audio_buffer(message.data) - - case _: - await self._send_error_response(message.id, f"Unsupported type {message.type}") - - except ValidationError as e: - await self._send_error_response(message.id, f"Invalid message: {e}") - logger.warning(f"Invalid message: {e}") - except Exception as e: - await self._send_error_response(message.id, f"Exception processing message: {e}") - logger.warning(f"Exception processing message: {e}") - - async def _handle_client_ready(self, request_id: str, data: RTVIClientReadyData | None): - """Handle the client-ready message from the client.""" - version = data.version if data else None - logger.debug(f"Received client-ready: version {version}") - if version: - try: - self._client_version = [int(v) for v in version.split(".")] - except ValueError: - logger.warning(f"Invalid client version format: {version}") - about = data.about if data else {"library": "unknown"} - logger.debug(f"Client Details: {about}") - if self._input_transport: - await self._input_transport.start_audio_in_streaming() - - self._client_ready_id = request_id - await self.set_client_ready() - - async def _handle_audio_buffer(self, data): - """Handle incoming audio buffer data.""" - if not self._input_transport: - return - - # Extract audio batch ensuring it's a list - audio_list = data.get("base64AudioBatch") or [data.get("base64Audio")] - - try: - for base64_audio in filter(None, audio_list): # Filter out None values - pcm_bytes = base64.b64decode(base64_audio) - frame = InputAudioRawFrame( - audio=pcm_bytes, - sample_rate=data["sampleRate"], - num_channels=data["numChannels"], - ) - await self._input_transport.push_audio_frame(frame) - - except (KeyError, TypeError, ValueError) as e: - # Handle missing keys, decoding errors, and invalid types - logger.error(f"Error processing audio buffer: {e}") - - async def _handle_describe_config(self, request_id: str): - """Handle a describe-config request.""" - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.", - DeprecationWarning, - ) - - services = list(self._registered_services.values()) - message = RTVIDescribeConfig(id=request_id, data=RTVIDescribeConfigData(config=services)) - await self.push_transport_message(message) - - async def _handle_describe_actions(self, request_id: str): - """Handle a describe-actions request.""" - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "The Actions API is deprecated, use custom server and client messages instead.", - DeprecationWarning, - ) - - actions = list(self._registered_actions.values()) - message = RTVIDescribeActions(id=request_id, data=RTVIDescribeActionsData(actions=actions)) - await self.push_transport_message(message) - - async def _handle_get_config(self, request_id: str): - """Handle a get-config request.""" - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.", - DeprecationWarning, - ) - - message = RTVIConfigResponse(id=request_id, data=self._config) - await self.push_transport_message(message) - - def _update_config_option(self, service: str, config: RTVIServiceOptionConfig): - """Update a specific configuration option.""" - for service_config in self._config.config: - if service_config.service == service: - for option_config in service_config.options: - if option_config.name == config.name: - option_config.value = config.value - return - # If we couldn't find a value for this config, we simply need to - # add it. - service_config.options.append(config) - - async def _update_service_config(self, config: RTVIServiceConfig): - """Update configuration for a specific service.""" - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.", - DeprecationWarning, - ) - - service = self._registered_services[config.service] - for option in config.options: - handler = service._options_dict[option.name].handler - await handler(self, service.name, option) - self._update_config_option(service.name, option) - - async def _update_config(self, data: RTVIConfig, interrupt: bool): - """Update the RTVI configuration.""" - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.", - DeprecationWarning, - ) - - if interrupt: - await self.interrupt_bot() - for service_config in data.config: - await self._update_service_config(service_config) - - async def _handle_update_config(self, request_id: str, data: RTVIUpdateConfig): - """Handle an update-config request.""" - await self._update_config(RTVIConfig(config=data.config), data.interrupt) - await self._handle_get_config(request_id) - - async def _handle_send_text(self, data: RTVISendTextData): - """Handle a send-text message from the client.""" - opts = data.options if data.options is not None else RTVISendTextOptions() - if opts.run_immediately: - await self.interrupt_bot() - cur_llm_skip_tts = self._llm_skip_tts - should_skip_tts = not opts.audio_response - toggle_skip_tts = cur_llm_skip_tts != should_skip_tts - if toggle_skip_tts: - output_frame = LLMConfigureOutputFrame(skip_tts=should_skip_tts) - await self.push_frame(output_frame) - text_frame = LLMMessagesAppendFrame( - messages=[{"role": "user", "content": data.content}], - run_llm=opts.run_immediately, - ) - await self.push_frame(text_frame) - if toggle_skip_tts: - output_frame = LLMConfigureOutputFrame(skip_tts=cur_llm_skip_tts) - await self.push_frame(output_frame) - - async def _handle_update_context(self, data: RTVIAppendToContextData): - if data.run_immediately: - await self.interrupt_bot() - frame = LLMMessagesAppendFrame( - messages=[{"role": data.role, "content": data.content}], - run_llm=data.run_immediately, - ) - await self.push_frame(frame) - - async def _handle_client_message(self, msg_id: str, data: RTVIRawClientMessageData): - """Handle a client message frame.""" - if not data: - await self._send_error_response(msg_id, "Malformed client message") - return - - # Create a RTVIClientMessageFrame to push the message - frame = RTVIClientMessageFrame(msg_id=msg_id, type=data.t, data=data.d) - await self.push_frame(frame) - await self._call_event_handler( - "on_client_message", - RTVIClientMessage( - msg_id=msg_id, - type=data.t, - data=data.d, - ), - ) - - async def _handle_function_call_result(self, data): - """Handle a function call result from the client.""" - frame = FunctionCallResultFrame( - function_name=data.function_name, - tool_call_id=data.tool_call_id, - arguments=data.arguments, - result=data.result, - ) - await self.push_frame(frame) - - async def _handle_action(self, request_id: Optional[str], data: RTVIActionRun): - """Handle an action execution request.""" - action_id = self._action_id(data.service, data.action) - if action_id not in self._registered_actions: - await self._send_error_response(request_id, f"Action {action_id} not registered") - return - action = self._registered_actions[action_id] - arguments = {} - if data.arguments: - for arg in data.arguments: - arguments[arg.name] = arg.value - result = await action.handler(self, action.service, arguments) - # Only send a response if request_id is present. Things that don't care about - # action responses (such as webhooks) don't set a request_id - if request_id: - message = RTVIActionResponse(id=request_id, data=RTVIActionResponseData(result=result)) - await self.push_transport_message(message) - - async def _send_bot_ready(self, about: Mapping[str, Any] = None): - """Send the bot-ready message to the client. - - Args: - about: Optional information about the bot to include in the ready message. - If left as None, the pipecat library and version will be used. - """ - config = None - if self._client_version and self._client_version[0] < 1: - config = self._config.config - if not about: - about = {"library": "pipecat-ai", "library_version": f"{pipecat_version()}"} - message = RTVIBotReady( - id=self._client_ready_id, - data=RTVIBotReadyData(version=RTVI_PROTOCOL_VERSION, about=about, config=config), - ) - await self.push_transport_message(message) - - async def _send_server_message(self, message: RTVIServerMessage | RTVIServerResponse): - """Send a message or response to the client.""" - await self.push_transport_message(message) - - async def _send_error_frame(self, frame: ErrorFrame): - """Send an error frame as an RTVI error message.""" - message = RTVIError(data=RTVIErrorData(error=frame.error, fatal=frame.fatal)) - await self.push_transport_message(message) - - async def _send_error_response(self, id: str, error: str): - """Send an error response message.""" - message = RTVIErrorResponse(id=id, data=RTVIErrorResponseData(error=error)) - await self.push_transport_message(message) - - def _action_id(self, service: str, action: str) -> str: - """Generate an action ID from service and action names.""" - return f"{service}:{action}" diff --git a/src/pipecat/processors/frameworks/rtvi/__init__.py b/src/pipecat/processors/frameworks/rtvi/__init__.py new file mode 100644 index 000000000..12f6022ee --- /dev/null +++ b/src/pipecat/processors/frameworks/rtvi/__init__.py @@ -0,0 +1,73 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""RTVI (Real-Time Voice Interface) protocol implementation for Pipecat.""" + +from pipecat.processors.frameworks.rtvi.frames import ( + RTVIActionFrame, + RTVIClientMessageFrame, + RTVIServerMessageFrame, + RTVIServerResponseFrame, +) +from pipecat.processors.frameworks.rtvi.models_v0 import ( + ActionResult, + RTVIAction, + RTVIActionArgument, + RTVIActionArgumentData, + RTVIActionResponse, + RTVIActionResponseData, + RTVIActionRun, + RTVIActionRunArgument, + RTVIBotReadyDataDeprecated, + RTVIConfig, + RTVIConfigResponse, + RTVIDescribeActions, + RTVIDescribeActionsData, + RTVIDescribeConfig, + RTVIDescribeConfigData, + RTVIService, + RTVIServiceConfig, + RTVIServiceOption, + RTVIServiceOptionConfig, + RTVIUpdateConfig, +) +from pipecat.processors.frameworks.rtvi.observer import ( + RTVIFunctionCallReportLevel, + RTVIObserver, + RTVIObserverParams, +) +from pipecat.processors.frameworks.rtvi.processor import RTVIProcessor + +__all__ = [ + "ActionResult", + "RTVIAction", + "RTVIActionArgument", + "RTVIActionArgumentData", + "RTVIActionFrame", + "RTVIActionResponse", + "RTVIActionResponseData", + "RTVIActionRun", + "RTVIActionRunArgument", + "RTVIBotReadyDataDeprecated", + "RTVIClientMessageFrame", + "RTVIConfig", + "RTVIConfigResponse", + "RTVIDescribeActions", + "RTVIDescribeActionsData", + "RTVIDescribeConfig", + "RTVIDescribeConfigData", + "RTVIFunctionCallReportLevel", + "RTVIObserver", + "RTVIObserverParams", + "RTVIProcessor", + "RTVIServerMessageFrame", + "RTVIServerResponseFrame", + "RTVIService", + "RTVIServiceConfig", + "RTVIServiceOption", + "RTVIServiceOptionConfig", + "RTVIUpdateConfig", +] diff --git a/src/pipecat/processors/frameworks/rtvi/frames.py b/src/pipecat/processors/frameworks/rtvi/frames.py new file mode 100644 index 000000000..37d25c764 --- /dev/null +++ b/src/pipecat/processors/frameworks/rtvi/frames.py @@ -0,0 +1,74 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""RTVI pipeline frame definitions.""" + +from dataclasses import dataclass +from typing import Any, Optional + +from pipecat.frames.frames import DataFrame, SystemFrame + + +@dataclass +class RTVIActionFrame(DataFrame): + """Frame containing an RTVI action to execute. + + Parameters: + rtvi_action_run: The action to execute. + message_id: Optional message ID for response correlation. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + rtvi_action_run: Any + message_id: Optional[str] = None + + +@dataclass +class RTVIServerMessageFrame(SystemFrame): + """A frame for sending server messages to the client. + + Parameters: + data: The message data to send to the client. + """ + + data: Any + + def __str__(self): + """String representation of the RTVI server message frame.""" + return f"{self.name}(data: {self.data})" + + +@dataclass +class RTVIClientMessageFrame(SystemFrame): + """A frame for sending messages from the client to the RTVI server. + + This frame is meant for custom messaging from the client to the server + and expects a server-response message. + """ + + msg_id: str + type: str + data: Optional[Any] = None + + +@dataclass +class RTVIServerResponseFrame(SystemFrame): + """A frame for responding to a client RTVI message. + + This frame should be sent in response to an RTVIClientMessageFrame + and include the original RTVIClientMessageFrame to ensure the response + is properly attributed to the original request. To respond with an error, + set the `error` field to a string describing the error. This will result + in the client receiving a `response-error` message instead of a + `server-response` message. + """ + + client_msg: RTVIClientMessageFrame + data: Optional[Any] = None + error: Optional[str] = None diff --git a/src/pipecat/processors/frameworks/rtvi/models_v0.py b/src/pipecat/processors/frameworks/rtvi/models_v0.py new file mode 100644 index 000000000..b3826902e --- /dev/null +++ b/src/pipecat/processors/frameworks/rtvi/models_v0.py @@ -0,0 +1,334 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""RTVI pre-1.0 protocol models (deprecated). + +All classes here are kept for backward compatibility only. Pipeline configuration +and the actions API were removed in RTVI protocol 1.0.0. Use custom client and +server messages instead. +""" + +from typing import ( + Any, + Awaitable, + Callable, + Dict, + List, + Literal, + Optional, + Union, +) + +from pydantic import BaseModel, Field, PrivateAttr + +import pipecat.processors.frameworks.rtvi.models_v1 as RTVI + +ActionResult = Union[bool, int, float, str, list, dict] + + +class RTVIServiceOption(BaseModel): + """Configuration option for an RTVI service. + + Defines a configurable option that can be set for an RTVI service, + including its name, type, and handler function. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + name: str + type: Literal["bool", "number", "string", "array", "object"] + handler: Callable[["RTVIProcessor", str, "RTVIServiceOptionConfig"], Awaitable[None]] = Field( + exclude=True + ) + + +class RTVIService(BaseModel): + """An RTVI service definition. + + Represents a service that can be configured and used within the RTVI protocol, + containing a name and list of configurable options. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + name: str + options: List[RTVIServiceOption] + _options_dict: Dict[str, RTVIServiceOption] = PrivateAttr(default={}) + + def model_post_init(self, __context: Any) -> None: + """Initialize the options dictionary after model creation.""" + self._options_dict = {} + for option in self.options: + self._options_dict[option.name] = option + return super().model_post_init(__context) + + +class RTVIActionArgumentData(BaseModel): + """Data for an RTVI action argument. + + Contains the name and value of an argument passed to an RTVI action. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + name: str + value: Any + + +class RTVIActionArgument(BaseModel): + """Definition of an RTVI action argument. + + Specifies the name and expected type of an argument for an RTVI action. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + name: str + type: Literal["bool", "number", "string", "array", "object"] + + +class RTVIAction(BaseModel): + """An RTVI action definition. + + Represents an action that can be executed within the RTVI protocol, + including its service, name, arguments, and handler function. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + service: str + action: str + arguments: List[RTVIActionArgument] = Field(default_factory=list) + result: Literal["bool", "number", "string", "array", "object"] + handler: Callable[["RTVIProcessor", str, Dict[str, Any]], Awaitable[ActionResult]] = Field( + exclude=True + ) + _arguments_dict: Dict[str, RTVIActionArgument] = PrivateAttr(default={}) + + def model_post_init(self, __context: Any) -> None: + """Initialize the arguments dictionary after model creation.""" + self._arguments_dict = {} + for arg in self.arguments: + self._arguments_dict[arg.name] = arg + return super().model_post_init(__context) + + +class RTVIServiceOptionConfig(BaseModel): + """Configuration value for an RTVI service option. + + Contains the name and value to set for a specific service option. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + name: str + value: Any + + +class RTVIServiceConfig(BaseModel): + """Configuration for an RTVI service. + + Contains the service name and list of option configurations to apply. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + service: str + options: List[RTVIServiceOptionConfig] + + +class RTVIConfig(BaseModel): + """Complete RTVI configuration. + + Contains the full configuration for all RTVI services. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + config: List[RTVIServiceConfig] + + +# +# Client -> Pipecat messages. +# + + +class RTVIUpdateConfig(BaseModel): + """Request to update RTVI configuration. + + Contains new configuration settings and whether to interrupt the bot. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + config: List[RTVIServiceConfig] + interrupt: bool = False + + +class RTVIActionRunArgument(BaseModel): + """Argument for running an RTVI action. + + Contains the name and value of an argument to pass to an action. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + name: str + value: Any + + +class RTVIActionRun(BaseModel): + """Request to run an RTVI action. + + Contains the service, action name, and optional arguments. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + service: str + action: str + arguments: Optional[List[RTVIActionRunArgument]] = None + + +# +# Pipecat -> Client responses and messages. +# + + +class RTVIBotReadyDataDeprecated(RTVI.BotReadyData): + """Data for bot ready notification. + + Contains protocol version and initial configuration. + """ + + # The config field is deprecated and will not be included if + # the client's rtvi version is 1.0.0 or higher. + config: Optional[List[RTVIServiceConfig]] = None + + +class RTVIDescribeConfigData(BaseModel): + """Data for describing available RTVI configuration. + + Contains the list of available services and their options. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + config: List[RTVIService] + + +class RTVIDescribeConfig(BaseModel): + """Message describing available RTVI configuration. + + Sent in response to a describe-config request. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + label: RTVI.MessageLiteral = RTVI.MESSAGE_LABEL + type: Literal["config-available"] = "config-available" + id: str + data: RTVIDescribeConfigData + + +class RTVIDescribeActionsData(BaseModel): + """Data for describing available RTVI actions. + + Contains the list of available actions that can be executed. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + actions: List[RTVIAction] + + +class RTVIDescribeActions(BaseModel): + """Message describing available RTVI actions. + + Sent in response to a describe-actions request. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + label: RTVI.MessageLiteral = RTVI.MESSAGE_LABEL + type: Literal["actions-available"] = "actions-available" + id: str + data: RTVIDescribeActionsData + + +class RTVIConfigResponse(BaseModel): + """Response containing current RTVI configuration. + + Sent in response to a get-config request. + + .. deprecated:: 0.0.75 + Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + label: RTVI.MessageLiteral = RTVI.MESSAGE_LABEL + type: Literal["config"] = "config" + id: str + data: RTVIConfig + + +class RTVIActionResponseData(BaseModel): + """Data for an RTVI action response. + + Contains the result of executing an action. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + result: ActionResult + + +class RTVIActionResponse(BaseModel): + """Response to an RTVI action execution. + + Sent after successfully executing an action. + + .. deprecated:: 0.0.75 + Actions have been removed as part of the RTVI protocol 1.0.0. + Use custom client and server messages instead. + """ + + label: RTVI.MessageLiteral = RTVI.MESSAGE_LABEL + type: Literal["action-response"] = "action-response" + id: str + data: RTVIActionResponseData diff --git a/src/pipecat/processors/frameworks/rtvi/models_v1.py b/src/pipecat/processors/frameworks/rtvi/models_v1.py new file mode 100644 index 000000000..f38986cb0 --- /dev/null +++ b/src/pipecat/processors/frameworks/rtvi/models_v1.py @@ -0,0 +1,581 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""RTVI protocol v1 message models. + +Contains all RTVI protocol v1 message definitions and data structures. +Import this module under the ``RTVI`` alias to use as a namespace:: + + import pipecat.processors.frameworks.rtvi.models_v1 as RTVI + + msg = RTVI.BotReady(id="1", data=RTVI.BotReadyData(version=RTVI.PROTOCOL_VERSION)) +""" + +from typing import ( + Any, + Dict, + Literal, + Mapping, + Optional, +) + +from pydantic import BaseModel + +from pipecat.frames.frames import ( + AggregationType, +) + +# -- Constants -- +PROTOCOL_VERSION = "1.2.0" + +MESSAGE_LABEL = "rtvi-ai" +MessageLiteral = Literal["rtvi-ai"] + +# -- Base Message Structure -- + + +class Message(BaseModel): + """Base RTVI message structure. + + Represents the standard format for RTVI protocol messages. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: str + id: str + data: Optional[Dict[str, Any]] = None + + +# -- Client -> Pipecat messages. + + +class RawClientMessageData(BaseModel): + """Data structure expected from client messages sent to the RTVI server.""" + + t: str + d: Optional[Any] = None + + +class ClientMessage(BaseModel): + """Cleansed data structure for client messages for handling.""" + + msg_id: str + type: str + data: Optional[Any] = None + + +class RawServerResponseData(BaseModel): + """Data structure for server responses to client messages.""" + + t: str + d: Optional[Any] = None + + +class ServerResponse(BaseModel): + """The RTVI-formatted message response from the server to the client. + + This message is used to respond to custom messages sent by the client. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["server-response"] = "server-response" + id: str + data: RawServerResponseData + + +class AboutClientData(BaseModel): + """Data about the RTVI client. + + Contains information about the client, including which RTVI library it + is using, what platform it is on and any additional details, if available. + """ + + library: str + library_version: Optional[str] = None + platform: Optional[str] = None + platform_version: Optional[str] = None + platform_details: Optional[Any] = None + + +class ClientReadyData(BaseModel): + """Data format of client ready messages. + + Contains the RTVI protocol version and client information. + """ + + version: str + about: AboutClientData + + +# -- Pipecat -> Client errors + + +class ErrorResponseData(BaseModel): + """Data for an RTVI error response. + + Contains the error message to send back to the client. + """ + + error: str + + +class ErrorResponse(BaseModel): + """RTVI error response message. + + RTVI formatted error response message for relaying failed client requests. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["error-response"] = "error-response" + id: str + data: ErrorResponseData + + +class ErrorData(BaseModel): + """Data for an RTVI error event. + + Contains error information including whether it's fatal. + """ + + error: str + fatal: bool # Indicates the pipeline has stopped due to this error + + +class Error(BaseModel): + """RTVI error event message. + + RTVI formatted error message for relaying errors in the pipeline. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["error"] = "error" + data: ErrorData + + +# -- Pipecat -> Client responses and messages. + + +class BotReadyData(BaseModel): + """Data for bot ready notification. + + Contains protocol version and initial configuration. + """ + + version: str + about: Optional[Mapping[str, Any]] = None + + +class BotReady(BaseModel): + """Message indicating bot is ready for interaction. + + Sent after bot initialization is complete. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-ready"] = "bot-ready" + id: str + data: BotReadyData + + +class LLMFunctionCallMessageData(BaseModel): + """Data for LLM function call notification. + + Contains function call details including name, ID, and arguments. + + .. deprecated:: 0.0.102 + Use ``LLMFunctionCallInProgressMessageData`` instead. + """ + + function_name: str + tool_call_id: str + args: Mapping[str, Any] + + +class LLMFunctionCallMessage(BaseModel): + """Message notifying of an LLM function call. + + Sent when the LLM makes a function call. + + .. deprecated:: 0.0.102 + Use ``LLMFunctionCallInProgressMessage`` with the + ``llm-function-call-in-progress`` event type instead. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["llm-function-call"] = "llm-function-call" + data: LLMFunctionCallMessageData + + +class SendTextOptions(BaseModel): + """Options for sending text input to the LLM. + + Contains options for how the pipeline should process the text input. + """ + + run_immediately: bool = True + audio_response: bool = True + + +class SendTextData(BaseModel): + """Data format for sending text input to the LLM. + + Contains the text content to send and any options for how the pipeline should process it. + """ + + content: str + options: Optional[SendTextOptions] = None + + +class AppendToContextData(BaseModel): + """Data format for appending messages to the context. + + Contains the role, content, and whether to run the message immediately. + + .. deprecated:: 0.0.85 + The RTVI message, append-to-context, has been deprecated. Use send-text + or custom client and server messages instead. + """ + + role: Literal["user", "assistant"] | str + content: Any + run_immediately: bool = False + + +class AppendToContext(BaseModel): + """RTVI message format to append content to the LLM context. + + .. deprecated:: 0.0.85 + The RTVI message, append-to-context, has been deprecated. Use send-text + or custom client and server messages instead. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["append-to-context"] = "append-to-context" + data: AppendToContextData + + +class LLMFunctionCallStartMessageData(BaseModel): + """Data for LLM function call start notification. + + Contains the function name being called. Fields may be omitted based on + the configured function_call_report_level for security. + """ + + function_name: Optional[str] = None + + +class LLMFunctionCallStartMessage(BaseModel): + """Message notifying that an LLM function call has started. + + Sent when the LLM begins a function call. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["llm-function-call-started"] = "llm-function-call-started" + data: LLMFunctionCallStartMessageData + + +class LLMFunctionCallResultData(BaseModel): + """Data for LLM function call result. + + Contains function call details and result. + """ + + function_name: str + tool_call_id: str + arguments: dict + result: dict | str + + +class LLMFunctionCallInProgressMessageData(BaseModel): + """Data for LLM function call in-progress notification. + + Contains function call details including name, ID, and arguments. + Fields may be omitted based on the configured function_call_report_level for security. + """ + + tool_call_id: str + function_name: Optional[str] = None + arguments: Optional[Mapping[str, Any]] = None + + +class LLMFunctionCallInProgressMessage(BaseModel): + """Message notifying that an LLM function call is in progress. + + Sent when the LLM function call execution begins. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["llm-function-call-in-progress"] = "llm-function-call-in-progress" + data: LLMFunctionCallInProgressMessageData + + +class LLMFunctionCallStoppedMessageData(BaseModel): + """Data for LLM function call stopped notification. + + Contains details about the function call that stopped, including + whether it was cancelled or completed with a result. + Fields may be omitted based on the configured function_call_report_level for security. + """ + + tool_call_id: str + cancelled: bool + function_name: Optional[str] = None + result: Optional[Any] = None + + +class LLMFunctionCallStoppedMessage(BaseModel): + """Message notifying that an LLM function call has stopped. + + Sent when a function call completes (with result) or is cancelled. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["llm-function-call-stopped"] = "llm-function-call-stopped" + data: LLMFunctionCallStoppedMessageData + + +class BotLLMStartedMessage(BaseModel): + """Message indicating bot LLM processing has started.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-llm-started"] = "bot-llm-started" + + +class BotLLMStoppedMessage(BaseModel): + """Message indicating bot LLM processing has stopped.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-llm-stopped"] = "bot-llm-stopped" + + +class BotTTSStartedMessage(BaseModel): + """Message indicating bot TTS processing has started.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-tts-started"] = "bot-tts-started" + + +class BotTTSStoppedMessage(BaseModel): + """Message indicating bot TTS processing has stopped.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-tts-stopped"] = "bot-tts-stopped" + + +class TextMessageData(BaseModel): + """Data for text-based RTVI messages. + + Contains text content. + """ + + text: str + + +class BotOutputMessageData(TextMessageData): + """Data for bot output RTVI messages. + + Extends TextMessageData to include metadata about the output. + """ + + spoken: bool = False # Indicates if the text has been spoken by TTS + aggregated_by: AggregationType | str + # Indicates what form the text is in (e.g., by word, sentence, etc.) + + +class BotOutputMessage(BaseModel): + """Message containing bot output text. + + An event meant to holistically represent what the bot is outputting, + along with metadata about the output and if it has been spoken. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-output"] = "bot-output" + data: BotOutputMessageData + + +class BotTranscriptionMessage(BaseModel): + """Message containing bot transcription text. + + Sent when the bot's speech is transcribed. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-transcription"] = "bot-transcription" + data: TextMessageData + + +class BotLLMTextMessage(BaseModel): + """Message containing bot LLM text output. + + Sent when the bot's LLM generates text. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-llm-text"] = "bot-llm-text" + data: TextMessageData + + +class BotTTSTextMessage(BaseModel): + """Message containing bot TTS text output. + + Sent when text is being processed by TTS. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-tts-text"] = "bot-tts-text" + data: TextMessageData + + +class AudioMessageData(BaseModel): + """Data for audio-based RTVI messages. + + Contains audio data and metadata. + """ + + audio: str + sample_rate: int + num_channels: int + + +class BotTTSAudioMessage(BaseModel): + """Message containing bot TTS audio output. + + Sent when the bot's TTS generates audio. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-tts-audio"] = "bot-tts-audio" + data: AudioMessageData + + +class UserTranscriptionMessageData(BaseModel): + """Data for user transcription messages. + + Contains transcription text and metadata. + """ + + text: str + user_id: str + timestamp: str + final: bool + + +class UserTranscriptionMessage(BaseModel): + """Message containing user transcription. + + Sent when user speech is transcribed. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["user-transcription"] = "user-transcription" + data: UserTranscriptionMessageData + + +class UserLLMTextMessage(BaseModel): + """Message containing user text input for LLM. + + Sent when user text is processed by the LLM. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["user-llm-text"] = "user-llm-text" + data: TextMessageData + + +class UserStartedSpeakingMessage(BaseModel): + """Message indicating user has started speaking.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["user-started-speaking"] = "user-started-speaking" + + +class UserStoppedSpeakingMessage(BaseModel): + """Message indicating user has stopped speaking.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["user-stopped-speaking"] = "user-stopped-speaking" + + +class UserMuteStartedMessage(BaseModel): + """Message indicating user has been muted.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["user-mute-started"] = "user-mute-started" + + +class UserMuteStoppedMessage(BaseModel): + """Message indicating user has been unmuted.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["user-mute-stopped"] = "user-mute-stopped" + + +class BotStartedSpeakingMessage(BaseModel): + """Message indicating bot has started speaking.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-started-speaking"] = "bot-started-speaking" + + +class BotStoppedSpeakingMessage(BaseModel): + """Message indicating bot has stopped speaking.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-stopped-speaking"] = "bot-stopped-speaking" + + +class MetricsMessage(BaseModel): + """Message containing performance metrics. + + Sent to provide performance and usage metrics. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["metrics"] = "metrics" + data: Mapping[str, Any] + + +class ServerMessage(BaseModel): + """Generic server message. + + Used for custom server-to-client messages. + """ + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["server-message"] = "server-message" + data: Any + + +class AudioLevelMessageData(BaseModel): + """Data format for sending audio levels.""" + + value: float + + +class UserAudioLevelMessage(BaseModel): + """Message indicating user audio level.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["user-audio-level"] = "user-audio-level" + data: AudioLevelMessageData + + +class BotAudioLevelMessage(BaseModel): + """Message indicating bot audio level.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["bot-audio-level"] = "bot-audio-level" + data: AudioLevelMessageData + + +class SystemLogMessage(BaseModel): + """Message including a system log.""" + + label: MessageLiteral = MESSAGE_LABEL + type: Literal["system-log"] = "system-log" + data: TextMessageData diff --git a/src/pipecat/processors/frameworks/rtvi/observer.py b/src/pipecat/processors/frameworks/rtvi/observer.py new file mode 100644 index 000000000..83837ad6e --- /dev/null +++ b/src/pipecat/processors/frameworks/rtvi/observer.py @@ -0,0 +1,664 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""RTVI observer for converting pipeline frames to outgoing RTVI messages.""" + +import time +from dataclasses import dataclass, field +from enum import Enum +from typing import ( + TYPE_CHECKING, + Awaitable, + Callable, + Dict, + List, + Optional, + Set, + Tuple, +) + +from loguru import logger +from pydantic import BaseModel + +import pipecat.processors.frameworks.rtvi.models_v1 as RTVI +from pipecat.audio.utils import calculate_audio_volume +from pipecat.frames.frames import ( + AggregatedTextFrame, + AggregationType, + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + Frame, + FunctionCallCancelFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, + FunctionCallsStartedFrame, + InputAudioRawFrame, + InterimTranscriptionFrame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, + MetricsFrame, + TranscriptionFrame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, + TTSTextFrame, + UserMuteStartedFrame, + UserMuteStoppedFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.metrics.metrics import ( + LLMUsageMetricsData, + ProcessingMetricsData, + TTFBMetricsData, + TTSUsageMetricsData, +) +from pipecat.observers.base_observer import BaseObserver, FramePushed +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.frameworks.rtvi.frames import ( + RTVIServerMessageFrame, + RTVIServerResponseFrame, +) +from pipecat.transports.base_output import BaseOutputTransport +from pipecat.utils.string import match_endofsentence + +if TYPE_CHECKING: + from pipecat.processors.frameworks.rtvi.processor import RTVIProcessor + + +class RTVIFunctionCallReportLevel(str, Enum): + """Level of detail to include in function call RTVI events. + + Controls what information is exposed in function call events for security. + + Values: + DISABLED: No events emitted for this function call. + NONE: Events only with tool_call_id, no function name or metadata (most secure). + NAME: Events with function name, no arguments or results. + FULL: Events with function name, arguments, and results. + """ + + DISABLED = "disabled" + NONE = "none" + NAME = "name" + FULL = "full" + + +@dataclass +class RTVIObserverParams: + """Parameters for configuring RTVI Observer behavior. + + .. deprecated:: 0.0.87 + Parameter `errors_enabled` is deprecated. Error messages are always enabled. + + Parameters: + bot_output_enabled: Indicates if bot output messages should be sent. + bot_llm_enabled: Indicates if the bot's LLM messages should be sent. + bot_tts_enabled: Indicates if the bot's TTS messages should be sent. + bot_speaking_enabled: Indicates if the bot's started/stopped speaking messages should be sent. + bot_audio_level_enabled: Indicates if bot's audio level messages should be sent. + user_llm_enabled: Indicates if the user's LLM input messages should be sent. + user_speaking_enabled: Indicates if the user's started/stopped speaking messages should be sent. + user_transcription_enabled: Indicates if user's transcription messages should be sent. + user_audio_level_enabled: Indicates if user's audio level messages should be sent. + metrics_enabled: Indicates if metrics messages should be sent. + system_logs_enabled: Indicates if system logs should be sent. + errors_enabled: [Deprecated] Indicates if errors messages should be sent. + ignored_sources: List of frame processors whose frames should be silently ignored + by this observer. Useful for suppressing RTVI messages from secondary pipeline + branches (e.g. a silent evaluation LLM) that should not be visible to clients. + Sources can also be added and removed dynamically via ``add_ignored_source()`` + and ``remove_ignored_source()``. + skip_aggregator_types: List of aggregation types to skip sending as tts/output messages. + Note: if using this to avoid sending secure information, be sure to also disable + bot_llm_enabled to avoid leaking through LLM messages. + bot_output_transforms: A list of callables to transform text before just before sending it + to TTS. Each callable takes the aggregated text and its type, and returns the + transformed text. To register, provide a list of tuples of + (aggregation_type | '*', transform_function). + audio_level_period_secs: How often audio levels should be sent if enabled. + function_call_report_level: Controls what information is exposed in function call + events for security. A dict mapping function names to levels, where ``"*"`` + sets the default level for unlisted functions:: + + function_call_report_level={ + "*": RTVIFunctionCallReportLevel.NONE, # Default: events with no metadata + "get_weather": RTVIFunctionCallReportLevel.FULL, # Expose everything + } + + Levels: + - DISABLED: No events emitted for this function. + - NONE: Events with tool_call_id only (most secure when events needed). + - NAME: Adds function name to events. + - FULL: Adds function name, arguments, and results. + + Defaults to ``{"*": RTVIFunctionCallReportLevel.NONE}``. + """ + + bot_output_enabled: bool = True + bot_llm_enabled: bool = True + bot_tts_enabled: bool = True + bot_speaking_enabled: bool = True + bot_audio_level_enabled: bool = False + user_llm_enabled: bool = True + user_speaking_enabled: bool = True + user_mute_enabled: bool = True + user_transcription_enabled: bool = True + user_audio_level_enabled: bool = False + metrics_enabled: bool = True + system_logs_enabled: bool = False + errors_enabled: Optional[bool] = None + ignored_sources: List[FrameProcessor] = field(default_factory=list) + skip_aggregator_types: Optional[List[AggregationType | str]] = None + bot_output_transforms: Optional[ + List[ + Tuple[ + AggregationType | str, + Callable[[str, AggregationType | str], Awaitable[str]], + ] + ] + ] = None + audio_level_period_secs: float = 0.15 + function_call_report_level: Dict[str, RTVIFunctionCallReportLevel] = field( + default_factory=lambda: {"*": RTVIFunctionCallReportLevel.NONE} + ) + + +class RTVIObserver(BaseObserver): + """Pipeline frame observer for RTVI server message handling. + + This observer monitors pipeline frames and converts them into appropriate RTVI messages + for client communication. It handles various frame types including speech events, + transcriptions, LLM responses, and TTS events. + + Note: + This observer only handles outgoing messages. Incoming RTVI client messages + are handled by the RTVIProcessor. + """ + + def __init__( + self, + rtvi: Optional["RTVIProcessor"] = None, + *, + params: Optional[RTVIObserverParams] = None, + **kwargs, + ): + """Initialize the RTVI observer. + + Args: + rtvi: The RTVI processor to push frames to. + params: Settings to enable/disable specific messages. + **kwargs: Additional arguments passed to parent class. + """ + super().__init__(**kwargs) + self._rtvi = rtvi + self._params = params or RTVIObserverParams() + + self._ignored_sources: Set[FrameProcessor] = set(self._params.ignored_sources) + self._frames_seen = set() + + self._bot_transcription = "" + self._last_user_audio_level = 0 + self._last_bot_audio_level = 0 + + # Track bot speaking state for queuing aggregated text frames + self._bot_is_speaking = False + self._queued_aggregated_text_frames: List[AggregatedTextFrame] = [] + + if self._params.system_logs_enabled: + self._system_logger_id = logger.add(self._logger_sink) + + if self._params.errors_enabled is not None: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Parameter `errors_enabled` is deprecated. Error messages are always enabled.", + DeprecationWarning, + ) + + self._aggregation_transforms: List[ + Tuple[AggregationType | str, Callable[[str, AggregationType | str], Awaitable[str]]] + ] = self._params.bot_output_transforms or [] + + def add_bot_output_transformer( + self, + transform_function: Callable[[str, AggregationType | str], Awaitable[str]], + aggregation_type: AggregationType | str = "*", + ): + """Transform text for a specific aggregation type before sending as Bot Output or TTS. + + Args: + transform_function: The function to apply for transformation. This function should take + the text and aggregation type as input and return the transformed text. + Ex.: async def my_transform(text: str, aggregation_type: str) -> str: + aggregation_type: The type of aggregation to transform. This value defaults to "*" to + handle all text before sending to the client. + """ + self._aggregation_transforms.append((aggregation_type, transform_function)) + + def remove_bot_output_transformer( + self, + transform_function: Callable[[str, AggregationType | str], Awaitable[str]], + aggregation_type: AggregationType | str = "*", + ): + """Remove a text transformer for a specific aggregation type. + + Args: + transform_function: The function to remove. + aggregation_type: The type of aggregation to remove the transformer for. + """ + self._aggregation_transforms = [ + (agg_type, func) + for agg_type, func in self._aggregation_transforms + if not (agg_type == aggregation_type and func == transform_function) + ] + + def add_ignored_source(self, source: FrameProcessor): + """Ignore all frames pushed by the given processor. + + Any frame whose source matches ``source`` will be silently skipped, + preventing RTVI messages from being emitted for activity in that + processor. Useful for suppressing events from secondary pipeline + branches (e.g. a silent evaluation LLM) that should not be visible + to clients. + + Args: + source: The frame processor to ignore. + """ + self._ignored_sources.add(source) + + def remove_ignored_source(self, source: FrameProcessor): + """Stop ignoring frames pushed by the given processor. + + Reverses a previous call to ``add_ignored_source()``. If ``source`` + was not previously ignored this is a no-op. + + Args: + source: The frame processor to stop ignoring. + """ + self._ignored_sources.discard(source) + + def _get_function_call_report_level(self, function_name: str) -> RTVIFunctionCallReportLevel: + """Get the report level for a specific function call. + + Args: + function_name: The name of the function to get the report level for. + + Returns: + The report level for the function. Looks up the function name first, + then falls back to "*" key, then NONE. + """ + levels = self._params.function_call_report_level + if function_name in levels: + return levels[function_name] + return levels.get("*", RTVIFunctionCallReportLevel.NONE) + + async def _logger_sink(self, message): + """Logger sink so we can send system logs to RTVI clients.""" + message = RTVI.SystemLogMessage(data=RTVI.TextMessageData(text=message)) + await self.send_rtvi_message(message) + + async def cleanup(self): + """Cleanup RTVI observer resources.""" + await super().cleanup() + if self._params.system_logs_enabled: + logger.remove(self._system_logger_id) + + async def send_rtvi_message(self, model: BaseModel, exclude_none: bool = True): + """Send an RTVI message. + + By default, we push a transport frame. But this function can be + overriden by subclass to send RTVI messages in different ways. + + Args: + model: The message to send. + exclude_none: Whether to exclude None values from the model dump. + + """ + if self._rtvi: + await self._rtvi.push_transport_message(model, exclude_none) + + async def on_push_frame(self, data: FramePushed): + """Process a frame being pushed through the pipeline. + + Args: + data: Frame push event data containing source, frame, direction, and timestamp. + """ + src = data.source + frame = data.frame + direction = data.direction + + # Frames from explicitly ignored sources are always skipped. + if self._ignored_sources and src in self._ignored_sources: + return + + # For broadcast frames (pushed in both directions), only process + # the downstream copy to avoid sending duplicate RTVI messages. + if frame.broadcast_sibling_id is not None and direction != FrameDirection.DOWNSTREAM: + return + + # If we have already seen this frame, let's skip it. + if frame.id in self._frames_seen: + return + + # This tells whether the frame is already processed. If false, we will try + # again the next time we see the frame. + mark_as_seen = True + + if ( + isinstance(frame, (UserStartedSpeakingFrame, UserStoppedSpeakingFrame)) + and self._params.user_speaking_enabled + ): + await self._handle_interruptions(frame) + elif ( + isinstance(frame, (UserMuteStartedFrame, UserMuteStoppedFrame)) + and self._params.user_mute_enabled + ): + await self._handle_user_mute(frame) + elif ( + isinstance(frame, (BotStartedSpeakingFrame, BotStoppedSpeakingFrame)) + and self._params.bot_speaking_enabled + ): + await self._handle_bot_speaking(frame) + elif ( + isinstance(frame, (TranscriptionFrame, InterimTranscriptionFrame)) + and self._params.user_transcription_enabled + ): + await self._handle_user_transcriptions(frame) + elif ( + isinstance(frame, (OpenAILLMContextFrame, LLMContextFrame)) + and self._params.user_llm_enabled + ): + await self._handle_context(frame) + elif isinstance(frame, LLMFullResponseStartFrame) and self._params.bot_llm_enabled: + await self.send_rtvi_message(RTVI.BotLLMStartedMessage()) + elif isinstance(frame, LLMFullResponseEndFrame) and self._params.bot_llm_enabled: + await self.send_rtvi_message(RTVI.BotLLMStoppedMessage()) + elif isinstance(frame, LLMTextFrame) and self._params.bot_llm_enabled: + await self._handle_llm_text_frame(frame) + elif isinstance(frame, TTSStartedFrame) and self._params.bot_tts_enabled: + await self.send_rtvi_message(RTVI.BotTTSStartedMessage()) + elif isinstance(frame, TTSStoppedFrame) and self._params.bot_tts_enabled: + await self.send_rtvi_message(RTVI.BotTTSStoppedMessage()) + elif isinstance(frame, AggregatedTextFrame) and ( + self._params.bot_output_enabled or self._params.bot_tts_enabled + ): + if isinstance(frame, TTSTextFrame) and not isinstance(src, BaseOutputTransport): + # This check is to make sure we handle the frame when it has gone + # through the transport and has correct timing. + mark_as_seen = False + else: + await self._handle_aggregated_llm_text(frame) + elif isinstance(frame, MetricsFrame) and self._params.metrics_enabled: + await self._handle_metrics(frame) + elif isinstance(frame, FunctionCallsStartedFrame): + for function_call in frame.function_calls: + report_level = self._get_function_call_report_level(function_call.function_name) + if report_level == RTVIFunctionCallReportLevel.DISABLED: + continue + data = RTVI.LLMFunctionCallStartMessageData() + if report_level in ( + RTVIFunctionCallReportLevel.NAME, + RTVIFunctionCallReportLevel.FULL, + ): + data.function_name = function_call.function_name + message = RTVI.LLMFunctionCallStartMessage(data=data) + await self.send_rtvi_message(message) + elif isinstance(frame, FunctionCallInProgressFrame): + report_level = self._get_function_call_report_level(frame.function_name) + if report_level != RTVIFunctionCallReportLevel.DISABLED: + data = RTVI.LLMFunctionCallInProgressMessageData(tool_call_id=frame.tool_call_id) + if report_level in ( + RTVIFunctionCallReportLevel.NAME, + RTVIFunctionCallReportLevel.FULL, + ): + data.function_name = frame.function_name + if report_level == RTVIFunctionCallReportLevel.FULL: + data.arguments = frame.arguments + message = RTVI.LLMFunctionCallInProgressMessage(data=data) + await self.send_rtvi_message(message) + elif isinstance(frame, FunctionCallCancelFrame): + report_level = self._get_function_call_report_level(frame.function_name) + if report_level != RTVIFunctionCallReportLevel.DISABLED: + data = RTVI.LLMFunctionCallStoppedMessageData( + tool_call_id=frame.tool_call_id, + cancelled=True, + ) + if report_level in ( + RTVIFunctionCallReportLevel.NAME, + RTVIFunctionCallReportLevel.FULL, + ): + data.function_name = frame.function_name + message = RTVI.LLMFunctionCallStoppedMessage(data=data) + await self.send_rtvi_message(message) + elif isinstance(frame, FunctionCallResultFrame): + report_level = self._get_function_call_report_level(frame.function_name) + if report_level != RTVIFunctionCallReportLevel.DISABLED: + data = RTVI.LLMFunctionCallStoppedMessageData( + tool_call_id=frame.tool_call_id, + cancelled=False, + ) + if report_level in ( + RTVIFunctionCallReportLevel.NAME, + RTVIFunctionCallReportLevel.FULL, + ): + data.function_name = frame.function_name + if report_level == RTVIFunctionCallReportLevel.FULL: + data.result = frame.result if frame.result else None + message = RTVI.LLMFunctionCallStoppedMessage(data=data) + await self.send_rtvi_message(message) + elif isinstance(frame, RTVIServerMessageFrame): + message = RTVI.ServerMessage(data=frame.data) + await self.send_rtvi_message(message) + elif isinstance(frame, RTVIServerResponseFrame): + if frame.error is not None: + await self._send_error_response(frame) + else: + await self._send_server_response(frame) + elif isinstance(frame, InputAudioRawFrame) and self._params.user_audio_level_enabled: + curr_time = time.time() + diff_time = curr_time - self._last_user_audio_level + if diff_time > self._params.audio_level_period_secs: + level = calculate_audio_volume(frame.audio, frame.sample_rate) + message = RTVI.UserAudioLevelMessage(data=RTVI.AudioLevelMessageData(value=level)) + await self.send_rtvi_message(message) + self._last_user_audio_level = curr_time + elif isinstance(frame, TTSAudioRawFrame) and self._params.bot_audio_level_enabled: + curr_time = time.time() + diff_time = curr_time - self._last_bot_audio_level + if diff_time > self._params.audio_level_period_secs: + level = calculate_audio_volume(frame.audio, frame.sample_rate) + message = RTVI.BotAudioLevelMessage(data=RTVI.AudioLevelMessageData(value=level)) + await self.send_rtvi_message(message) + self._last_bot_audio_level = curr_time + + if mark_as_seen: + self._frames_seen.add(frame.id) + + async def _handle_interruptions(self, frame: Frame): + """Handle user speaking interruption frames.""" + message = None + if isinstance(frame, UserStartedSpeakingFrame): + message = RTVI.UserStartedSpeakingMessage() + elif isinstance(frame, UserStoppedSpeakingFrame): + message = RTVI.UserStoppedSpeakingMessage() + + if message: + await self.send_rtvi_message(message) + + async def _handle_user_mute(self, frame: Frame): + """Handle user mute/unmute frames.""" + message = None + if isinstance(frame, UserMuteStartedFrame): + message = RTVI.UserMuteStartedMessage() + elif isinstance(frame, UserMuteStoppedFrame): + message = RTVI.UserMuteStoppedMessage() + + if message: + await self.send_rtvi_message(message) + + async def _handle_bot_speaking(self, frame: Frame): + """Handle bot speaking event frames.""" + if isinstance(frame, BotStartedSpeakingFrame): + message = RTVI.BotStartedSpeakingMessage() + await self.send_rtvi_message(message) + # Flush any queued aggregated text frames + for queued_frame in self._queued_aggregated_text_frames: + await self._send_aggregated_llm_text(queued_frame) + self._queued_aggregated_text_frames.clear() + self._bot_is_speaking = True + elif isinstance(frame, BotStoppedSpeakingFrame): + message = RTVI.BotStoppedSpeakingMessage() + await self.send_rtvi_message(message) + self._bot_is_speaking = False + + async def _handle_aggregated_llm_text(self, frame: AggregatedTextFrame): + """Handle aggregated LLM text output frames.""" + if self._bot_is_speaking: + # Bot has already started speaking, send directly + await self._send_aggregated_llm_text(frame) + else: + # Bot hasn't started speaking yet, queue the frame + self._queued_aggregated_text_frames.append(frame) + + async def _send_aggregated_llm_text(self, frame: AggregatedTextFrame): + """Send aggregated LLM text messages.""" + # Skip certain aggregator types if configured to do so. + if ( + self._params.skip_aggregator_types + and frame.aggregated_by in self._params.skip_aggregator_types + ): + return + + text = frame.text + type = frame.aggregated_by + for aggregation_type, transform in self._aggregation_transforms: + if aggregation_type == type or aggregation_type == "*": + text = await transform(text, type) + + isTTS = isinstance(frame, TTSTextFrame) + if self._params.bot_output_enabled: + message = RTVI.BotOutputMessage( + data=RTVI.BotOutputMessageData(text=text, spoken=isTTS, aggregated_by=type) + ) + await self.send_rtvi_message(message) + + if isTTS and self._params.bot_tts_enabled: + tts_message = RTVI.BotTTSTextMessage(data=RTVI.TextMessageData(text=text)) + await self.send_rtvi_message(tts_message) + + async def _handle_llm_text_frame(self, frame: LLMTextFrame): + """Handle LLM text output frames.""" + message = RTVI.BotLLMTextMessage(data=RTVI.TextMessageData(text=frame.text)) + await self.send_rtvi_message(message) + + # TODO (mrkb): Remove all this logic when we fully deprecate bot-transcription messages. + self._bot_transcription += frame.text + + if match_endofsentence(self._bot_transcription) and len(self._bot_transcription) > 0: + await self.send_rtvi_message( + RTVI.BotTranscriptionMessage( + data=RTVI.TextMessageData(text=self._bot_transcription) + ) + ) + self._bot_transcription = "" + + async def _handle_user_transcriptions(self, frame: Frame): + """Handle user transcription frames.""" + message = None + if isinstance(frame, TranscriptionFrame): + message = RTVI.UserTranscriptionMessage( + data=RTVI.UserTranscriptionMessageData( + text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp, final=True + ) + ) + elif isinstance(frame, InterimTranscriptionFrame): + message = RTVI.UserTranscriptionMessage( + data=RTVI.UserTranscriptionMessageData( + text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp, final=False + ) + ) + + if message: + await self.send_rtvi_message(message) + + async def _handle_context(self, frame: OpenAILLMContextFrame | LLMContextFrame): + """Process LLM context frames to extract user messages for the RTVI client.""" + try: + if isinstance(frame, OpenAILLMContextFrame): + messages = frame.context.messages + else: + messages = frame.context.get_messages() + if not messages: + return + + message = messages[-1] + + # Handle Google LLM format (protobuf objects with attributes) + # Note: not possible if frame is a universal LLMContextFrame + if hasattr(message, "role") and message.role == "user" and hasattr(message, "parts"): + text = "".join(part.text for part in message.parts if hasattr(part, "text")) + if text: + rtvi_message = RTVI.UserLLMTextMessage(data=RTVI.TextMessageData(text=text)) + await self.send_rtvi_message(rtvi_message) + + # Handle OpenAI format (original implementation) + elif isinstance(message, dict): + if message["role"] == "user": + content = message["content"] + if isinstance(content, list): + text = " ".join(item["text"] for item in content if "text" in item) + else: + text = content + rtvi_message = RTVI.UserLLMTextMessage(data=RTVI.TextMessageData(text=text)) + await self.send_rtvi_message(rtvi_message) + + except Exception as e: + logger.warning(f"Caught an error while trying to handle context: {e}") + + async def _handle_metrics(self, frame: MetricsFrame): + """Handle metrics frames and convert to RTVI metrics messages.""" + metrics = {} + for d in frame.data: + if isinstance(d, TTFBMetricsData): + if "ttfb" not in metrics: + metrics["ttfb"] = [] + metrics["ttfb"].append(d.model_dump(exclude_none=True)) + elif isinstance(d, ProcessingMetricsData): + if "processing" not in metrics: + metrics["processing"] = [] + metrics["processing"].append(d.model_dump(exclude_none=True)) + elif isinstance(d, LLMUsageMetricsData): + if "tokens" not in metrics: + metrics["tokens"] = [] + metrics["tokens"].append(d.value.model_dump(exclude_none=True)) + elif isinstance(d, TTSUsageMetricsData): + if "characters" not in metrics: + metrics["characters"] = [] + metrics["characters"].append(d.model_dump(exclude_none=True)) + + message = RTVI.MetricsMessage(data=metrics) + await self.send_rtvi_message(message) + + async def _send_server_response(self, frame: RTVIServerResponseFrame): + """Send a response to the client for a specific request.""" + message = RTVI.ServerResponse( + id=str(frame.client_msg.msg_id), + data=RTVI.RawServerResponseData(t=frame.client_msg.type, d=frame.data), + ) + await self.send_rtvi_message(message) + + async def _send_error_response(self, frame: RTVIServerResponseFrame): + """Send a response to the client for a specific request.""" + message = RTVI.ErrorResponse( + id=str(frame.client_msg.msg_id), data=RTVI.ErrorResponseData(error=frame.error) + ) + await self.send_rtvi_message(message) diff --git a/src/pipecat/processors/frameworks/rtvi/processor.py b/src/pipecat/processors/frameworks/rtvi/processor.py new file mode 100644 index 000000000..345dcb8c4 --- /dev/null +++ b/src/pipecat/processors/frameworks/rtvi/processor.py @@ -0,0 +1,653 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""RTVIProcessor: main RTVI protocol processor.""" + +import asyncio +import base64 +from typing import Any, Dict, Mapping, Optional + +from loguru import logger +from pydantic import BaseModel, ValidationError + +import pipecat.processors.frameworks.rtvi.models_v1 as RTVI +from pipecat import version as pipecat_version +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + EndTaskFrame, + ErrorFrame, + Frame, + FunctionCallResultFrame, + InputAudioRawFrame, + InputTransportMessageFrame, + LLMConfigureOutputFrame, + LLMMessagesAppendFrame, + OutputTransportMessageUrgentFrame, + StartFrame, + SystemFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.frameworks.rtvi.frames import RTVIActionFrame, RTVIClientMessageFrame +from pipecat.processors.frameworks.rtvi.models_v0 import ( + RTVIAction, + RTVIActionResponse, + RTVIActionResponseData, + RTVIActionRun, + RTVIBotReadyDataDeprecated, + RTVIConfig, + RTVIConfigResponse, + RTVIDescribeActions, + RTVIDescribeActionsData, + RTVIDescribeConfig, + RTVIDescribeConfigData, + RTVIService, + RTVIServiceConfig, + RTVIServiceOptionConfig, + RTVIUpdateConfig, +) +from pipecat.processors.frameworks.rtvi.observer import RTVIObserver, RTVIObserverParams +from pipecat.services.llm_service import ( + FunctionCallParams, # TODO(aleix): we shouldn't import `services` from `processors` +) +from pipecat.transports.base_input import BaseInputTransport +from pipecat.transports.base_transport import BaseTransport + + +class RTVIProcessor(FrameProcessor): + """Main processor for handling RTVI protocol messages and actions. + + This processor manages the RTVI protocol communication including client-server + handshaking, configuration management, action execution, and message routing. + It serves as the central hub for RTVI protocol operations. + """ + + def __init__( + self, + *, + config: Optional[RTVIConfig] = None, + transport: Optional[BaseTransport] = None, + **kwargs, + ): + """Initialize the RTVI processor. + + Args: + config: Initial RTVI configuration. + transport: Transport layer for communication. + **kwargs: Additional arguments passed to parent class. + """ + super().__init__(**kwargs) + self._config = config or RTVIConfig(config=[]) + + self._bot_ready = False + self._client_ready = False + self._client_ready_id = "" + # Default to 0.3.0 which is the last version before actually having a + # "client-version". + self._client_version = [0, 3, 0] + self._llm_skip_tts: bool = False # Keep in sync with llm_service.py's configuration. + + self._registered_actions: Dict[str, RTVIAction] = {} + self._registered_services: Dict[str, RTVIService] = {} + + # A task to process incoming action frames. + self._action_task: Optional[asyncio.Task] = None + + # A task to process incoming transport messages. + self._message_task: Optional[asyncio.Task] = None + + self._register_event_handler("on_bot_started") + self._register_event_handler("on_client_ready") + self._register_event_handler("on_client_message") + + self._input_transport = None + self._transport = transport + if self._transport: + input_transport = self._transport.input() + if isinstance(input_transport, BaseInputTransport): + self._input_transport = input_transport + self._input_transport.enable_audio_in_stream_on_start(False) + + def register_action(self, action: RTVIAction): + """Register an action that can be executed via RTVI. + + Args: + action: The action to register. + """ + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "The actions API is deprecated, use server and client messages instead.", + DeprecationWarning, + ) + + id = self._action_id(action.service, action.action) + self._registered_actions[id] = action + + def register_service(self, service: RTVIService): + """Register a service that can be configured via RTVI. + + Args: + service: The service to register. + """ + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "The actions API is deprecated, use server and client messages instead.", + DeprecationWarning, + ) + + self._registered_services[service.name] = service + + def create_rtvi_observer(self, *, params: Optional[RTVIObserverParams] = None, **kwargs): + """Creates a new RTVI Observer. + + Args: + params: Settings to enable/disable specific messages. + **kwargs: Additional arguments passed to the observer. + + Returns: + A new RTVI observer. + """ + return RTVIObserver(self, params=params, **kwargs) + + async def set_client_ready(self): + """Mark the client as ready and trigger the ready event.""" + self._client_ready = True + await self._call_event_handler("on_client_ready") + + async def set_bot_ready(self, about: Mapping[str, Any] = None): + """Mark the bot as ready and send the bot-ready message. + + Args: + about: Optional information about the bot to include in the ready message. + If left as None, the Pipecat library and version will be used. + """ + self._bot_ready = True + # Only call the (deprecated) _update_config method if the we're using a + # config (which is deprecated). Otherwise we'd always print an + # unnecessary deprecation warning. + if self._config.config: + await self._update_config(self._config, False) + await self._send_bot_ready(about=about) + + async def interrupt_bot(self): + """Send a bot interruption frame upstream.""" + await self.broadcast_interruption() + + async def send_server_message(self, data: Any): + """Send a server message to the client.""" + message = RTVI.ServerMessage(data=data) + await self._send_server_message(message) + + async def send_server_response(self, client_msg: RTVI.ClientMessage, data: Any): + """Send a server response for a given client message.""" + message = RTVI.ServerResponse( + id=client_msg.msg_id, data=RTVI.RawServerResponseData(t=client_msg.type, d=data) + ) + await self._send_server_message(message) + + async def send_error_response(self, client_msg: RTVI.ClientMessage, error: str): + """Send an error response for a given client message.""" + await self._send_error_response(id=client_msg.msg_id, error=error) + + async def send_error(self, error: str): + """Send an error message to the client. + + Args: + error: The error message to send. + """ + await self._send_error_frame(ErrorFrame(error=error)) + + async def push_transport_message(self, model: BaseModel, exclude_none: bool = True): + """Push a transport message frame.""" + frame = OutputTransportMessageUrgentFrame( + message=model.model_dump(exclude_none=exclude_none) + ) + await self.push_frame(frame) + + async def handle_message(self, message: RTVI.Message): + """Handle an incoming RTVI message. + + Args: + message: The RTVI message to handle. + """ + await self._message_queue.put(message) + + async def handle_function_call(self, params: FunctionCallParams): + """Handle a function call from the LLM. + + Args: + params: The function call parameters. + + .. deprecated:: 0.0.102 + This method is deprecated. Function call events are now automatically + sent by ``RTVIObserver`` using the ``llm-function-call-in-progress`` event. + Configure reporting level via ``RTVIObserverParams.function_call_report_level``. + """ + import warnings + + warnings.warn( + "handle_function_call is deprecated. Function call events are now " + "automatically sent by RTVIObserver using llm-function-call-in-progress.", + DeprecationWarning, + stacklevel=2, + ) + fn = RTVI.LLMFunctionCallMessageData( + function_name=params.function_name, + tool_call_id=params.tool_call_id, + args=params.arguments, + ) + message = RTVI.LLMFunctionCallMessage(data=fn) + await self.push_transport_message(message, exclude_none=False) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process incoming frames through the RTVI processor. + + Args: + frame: The frame to process. + direction: The direction of frame flow. + """ + await super().process_frame(frame, direction) + + # Specific system frames + if isinstance(frame, StartFrame): + # Push StartFrame before start(), because we want StartFrame to be + # processed by every processor before any other frame is processed. + await self.push_frame(frame, direction) + await self._start(frame) + elif isinstance(frame, CancelFrame): + await self._cancel(frame) + await self.push_frame(frame, direction) + elif isinstance(frame, ErrorFrame): + await self._send_error_frame(frame) + await self.push_frame(frame, direction) + elif isinstance(frame, InputTransportMessageFrame): + await self._handle_transport_message(frame) + # All other system frames + elif isinstance(frame, SystemFrame): + await self.push_frame(frame, direction) + # Control frames + elif isinstance(frame, EndFrame): + # Push EndFrame before stop(), because stop() waits on the task to + # finish and the task finishes when EndFrame is processed. + await self.push_frame(frame, direction) + await self._stop(frame) + # Data frames + elif isinstance(frame, RTVIActionFrame): + await self._action_queue.put(frame) + elif isinstance(frame, LLMConfigureOutputFrame): + self._llm_skip_tts = frame.skip_tts + await self.push_frame(frame, direction) + # Other frames + else: + await self.push_frame(frame, direction) + + async def _start(self, frame: StartFrame): + """Start the RTVI processor tasks.""" + if not self._action_task: + self._action_queue = asyncio.Queue() + self._action_task = self.create_task(self._action_task_handler()) + if not self._message_task: + self._message_queue = asyncio.Queue() + self._message_task = self.create_task(self._message_task_handler()) + await self._call_event_handler("on_bot_started") + + async def _stop(self, frame: EndFrame): + """Stop the RTVI processor tasks.""" + await self._cancel_tasks() + + async def _cancel(self, frame: CancelFrame): + """Cancel the RTVI processor tasks.""" + await self._cancel_tasks() + + async def _cancel_tasks(self): + """Cancel all running tasks.""" + if self._action_task: + await self.cancel_task(self._action_task) + self._action_task = None + + if self._message_task: + await self.cancel_task(self._message_task) + self._message_task = None + + async def _action_task_handler(self): + """Handle incoming action frames.""" + while True: + frame = await self._action_queue.get() + await self._handle_action(frame.message_id, frame.rtvi_action_run) + self._action_queue.task_done() + + async def _message_task_handler(self): + """Handle incoming transport messages.""" + while True: + message = await self._message_queue.get() + await self._handle_message(message) + self._message_queue.task_done() + + async def _handle_transport_message(self, frame: InputTransportMessageFrame): + """Handle an incoming transport message frame.""" + try: + transport_message = frame.message + if transport_message.get("label") != RTVI.MESSAGE_LABEL: + logger.warning(f"Ignoring not RTVI message: {transport_message}") + return + message = RTVI.Message.model_validate(transport_message) + await self._message_queue.put(message) + except ValidationError as e: + await self.send_error(f"Invalid RTVI transport message: {e}") + logger.warning(f"Invalid RTVI transport message: {e}") + + async def _handle_message(self, message: RTVI.Message): + """Handle a parsed RTVI message.""" + try: + match message.type: + case "client-ready": + data = None + try: + data = RTVI.ClientReadyData.model_validate(message.data) + except ValidationError: + # Not all clients have been updated to RTVI 1.0.0. + # For now, that's okay, we just log their info as unknown. + data = None + pass + await self._handle_client_ready(message.id, data) + case "describe-actions": + await self._handle_describe_actions(message.id) + case "describe-config": + await self._handle_describe_config(message.id) + case "get-config": + await self._handle_get_config(message.id) + case "update-config": + update_config = RTVIUpdateConfig.model_validate(message.data) + await self._handle_update_config(message.id, update_config) + case "disconnect-bot": + await self.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM) + case "client-message": + data = RTVI.RawClientMessageData.model_validate(message.data) + await self._handle_client_message(message.id, data) + case "action": + action = RTVIActionRun.model_validate(message.data) + action_frame = RTVIActionFrame(message_id=message.id, rtvi_action_run=action) + await self._action_queue.put(action_frame) + case "llm-function-call-result": + data = RTVI.LLMFunctionCallResultData.model_validate(message.data) + await self._handle_function_call_result(data) + case "send-text": + data = RTVI.SendTextData.model_validate(message.data) + await self._handle_send_text(data) + case "append-to-context": + logger.warning( + f"The append-to-context message is deprecated, use send-text instead." + ) + data = RTVI.AppendToContextData.model_validate(message.data) + await self._handle_update_context(data) + case "raw-audio" | "raw-audio-batch": + await self._handle_audio_buffer(message.data) + + case _: + await self._send_error_response(message.id, f"Unsupported type {message.type}") + + except ValidationError as e: + await self._send_error_response(message.id, f"Invalid message: {e}") + logger.warning(f"Invalid message: {e}") + except Exception as e: + await self._send_error_response(message.id, f"Exception processing message: {e}") + logger.warning(f"Exception processing message: {e}") + + async def _handle_client_ready(self, request_id: str, data: RTVI.ClientReadyData | None): + """Handle the client-ready message from the client.""" + version = data.version if data else None + logger.debug(f"Received client-ready: version {version}") + if version: + try: + self._client_version = [int(v) for v in version.split(".")] + except ValueError: + logger.warning(f"Invalid client version format: {version}") + about = data.about if data else {"library": "unknown"} + logger.debug(f"Client Details: {about}") + if self._input_transport: + await self._input_transport.start_audio_in_streaming() + + self._client_ready_id = request_id + await self.set_client_ready() + + async def _handle_audio_buffer(self, data): + """Handle incoming audio buffer data.""" + if not self._input_transport: + return + + # Extract audio batch ensuring it's a list + audio_list = data.get("base64AudioBatch") or [data.get("base64Audio")] + + try: + for base64_audio in filter(None, audio_list): # Filter out None values + pcm_bytes = base64.b64decode(base64_audio) + frame = InputAudioRawFrame( + audio=pcm_bytes, + sample_rate=data["sampleRate"], + num_channels=data["numChannels"], + ) + await self._input_transport.push_audio_frame(frame) + + except (KeyError, TypeError, ValueError) as e: + # Handle missing keys, decoding errors, and invalid types + logger.error(f"Error processing audio buffer: {e}") + + async def _handle_describe_config(self, request_id: str): + """Handle a describe-config request.""" + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.", + DeprecationWarning, + ) + + services = list(self._registered_services.values()) + message = RTVIDescribeConfig(id=request_id, data=RTVIDescribeConfigData(config=services)) + await self.push_transport_message(message) + + async def _handle_describe_actions(self, request_id: str): + """Handle a describe-actions request.""" + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "The Actions API is deprecated, use custom server and client messages instead.", + DeprecationWarning, + ) + + actions = list(self._registered_actions.values()) + message = RTVIDescribeActions(id=request_id, data=RTVIDescribeActionsData(actions=actions)) + await self.push_transport_message(message) + + async def _handle_get_config(self, request_id: str): + """Handle a get-config request.""" + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.", + DeprecationWarning, + ) + + message = RTVIConfigResponse(id=request_id, data=self._config) + await self.push_transport_message(message) + + def _update_config_option(self, service: str, config: RTVIServiceOptionConfig): + """Update a specific configuration option.""" + for service_config in self._config.config: + if service_config.service == service: + for option_config in service_config.options: + if option_config.name == config.name: + option_config.value = config.value + return + # If we couldn't find a value for this config, we simply need to + # add it. + service_config.options.append(config) + + async def _update_service_config(self, config: RTVIServiceConfig): + """Update configuration for a specific service.""" + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.", + DeprecationWarning, + ) + + service = self._registered_services[config.service] + for option in config.options: + handler = service._options_dict[option.name].handler + await handler(self, service.name, option) + self._update_config_option(service.name, option) + + async def _update_config(self, data: RTVIConfig, interrupt: bool): + """Update the RTVI configuration.""" + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.", + DeprecationWarning, + ) + + if interrupt: + await self.interrupt_bot() + for service_config in data.config: + await self._update_service_config(service_config) + + async def _handle_update_config(self, request_id: str, data: RTVIUpdateConfig): + """Handle an update-config request.""" + await self._update_config(RTVIConfig(config=data.config), data.interrupt) + await self._handle_get_config(request_id) + + async def _handle_send_text(self, data: RTVI.SendTextData): + """Handle a send-text message from the client.""" + opts = data.options if data.options is not None else RTVI.SendTextOptions() + if opts.run_immediately: + await self.interrupt_bot() + cur_llm_skip_tts = self._llm_skip_tts + should_skip_tts = not opts.audio_response + toggle_skip_tts = cur_llm_skip_tts != should_skip_tts + if toggle_skip_tts: + output_frame = LLMConfigureOutputFrame(skip_tts=should_skip_tts) + await self.push_frame(output_frame) + text_frame = LLMMessagesAppendFrame( + messages=[{"role": "user", "content": data.content}], + run_llm=opts.run_immediately, + ) + await self.push_frame(text_frame) + if toggle_skip_tts: + output_frame = LLMConfigureOutputFrame(skip_tts=cur_llm_skip_tts) + await self.push_frame(output_frame) + + async def _handle_update_context(self, data: RTVI.AppendToContextData): + if data.run_immediately: + await self.interrupt_bot() + frame = LLMMessagesAppendFrame( + messages=[{"role": data.role, "content": data.content}], + run_llm=data.run_immediately, + ) + await self.push_frame(frame) + + async def _handle_client_message(self, msg_id: str, data: RTVI.RawClientMessageData): + """Handle a client message frame.""" + if not data: + await self._send_error_response(msg_id, "Malformed client message") + return + + # Create a RTVIClientMessageFrame to push the message + frame = RTVIClientMessageFrame(msg_id=msg_id, type=data.t, data=data.d) + await self.push_frame(frame) + await self._call_event_handler( + "on_client_message", + RTVI.ClientMessage( + msg_id=msg_id, + type=data.t, + data=data.d, + ), + ) + + async def _handle_function_call_result(self, data): + """Handle a function call result from the client.""" + frame = FunctionCallResultFrame( + function_name=data.function_name, + tool_call_id=data.tool_call_id, + arguments=data.arguments, + result=data.result, + ) + await self.push_frame(frame) + + async def _handle_action(self, request_id: Optional[str], data: RTVIActionRun): + """Handle an action execution request.""" + action_id = self._action_id(data.service, data.action) + if action_id not in self._registered_actions: + await self._send_error_response(request_id, f"Action {action_id} not registered") + return + action = self._registered_actions[action_id] + arguments = {} + if data.arguments: + for arg in data.arguments: + arguments[arg.name] = arg.value + result = await action.handler(self, action.service, arguments) + # Only send a response if request_id is present. Things that don't care about + # action responses (such as webhooks) don't set a request_id + if request_id: + message = RTVIActionResponse(id=request_id, data=RTVIActionResponseData(result=result)) + await self.push_transport_message(message) + + async def _send_bot_ready(self, about: Mapping[str, Any] = None): + """Send the bot-ready message to the client. + + Args: + about: Optional information about the bot to include in the ready message. + If left as None, the pipecat library and version will be used. + """ + if not about: + about = {"library": "pipecat-ai", "library_version": f"{pipecat_version()}"} + if self._client_version and self._client_version[0] < 1: + config = self._config.config + message = RTVI.BotReady( + id=self._client_ready_id, + data=RTVIBotReadyDataDeprecated( + version=RTVI.PROTOCOL_VERSION, about=about, config=config + ), + ) + else: + message = RTVI.BotReady( + id=self._client_ready_id, + data=RTVI.BotReadyData(version=RTVI.PROTOCOL_VERSION, about=about), + ) + await self.push_transport_message(message) + + async def _send_server_message(self, message: RTVI.ServerMessage | RTVI.ServerResponse): + """Send a message or response to the client.""" + await self.push_transport_message(message) + + async def _send_error_frame(self, frame: ErrorFrame): + """Send an error frame as an RTVI error message.""" + message = RTVI.Error(data=RTVI.ErrorData(error=frame.error, fatal=frame.fatal)) + await self.push_transport_message(message) + + async def _send_error_response(self, id: str, error: str): + """Send an error response message.""" + message = RTVI.ErrorResponse(id=id, data=RTVI.ErrorResponseData(error=error)) + await self.push_transport_message(message) + + def _action_id(self, service: str, action: str) -> str: + """Generate an action ID from service and action names.""" + return f"{service}:{action}" diff --git a/src/pipecat/serializers/base_serializer.py b/src/pipecat/serializers/base_serializer.py index edf005a81..4807f8606 100644 --- a/src/pipecat/serializers/base_serializer.py +++ b/src/pipecat/serializers/base_serializer.py @@ -11,13 +11,13 @@ from typing import Optional from pydantic import BaseModel +import pipecat.processors.frameworks.rtvi.models_v1 as RTVI 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 @@ -64,7 +64,7 @@ class FrameSerializer(BaseObject): if ( self._params.ignore_rtvi_messages and isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)) - and frame.message.get("label") == RTVI_MESSAGE_LABEL + and frame.message.get("label") == RTVI.MESSAGE_LABEL ): return True return False