wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Convert RTVI framework into a structured package

Browse Source 
Replace the monolithic rtvi.py with a proper package split by concern
protocol version:
  - models_v0.py: deprecated pre-1.0 Pydantic models
  - models_v1.py: current RTVI protocol v1 message models
  - frames.py: RTVI pipeline frame dataclasses
  - observer.py: RTVIObserver and RTVIObserverParams
  - processor.py: RTVIProcessor (now lean, imports from submodules)
  - __init__.py: re-exports full public API for backward compatability
This commit is contained in:
mattie ruth backman
2025-12-22 15:39:47 -05:00
committed by Mattie Ruth
parent ac80b787bf
commit 158424aa28
8 changed files with 2381 additions and 2171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2169
src/pipecat/processors/frameworks/rtvi.py
View File

File diff suppressed because it is too large Load Diff

73
src/pipecat/processors/frameworks/rtvi/__init__.py Normal file
View File

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

74
src/pipecat/processors/frameworks/rtvi/frames.py Normal file
View File

@@ -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

334
src/pipecat/processors/frameworks/rtvi/models_v0.py Normal file
View File

@@ -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

581
src/pipecat/processors/frameworks/rtvi/models_v1.py Normal file
View File

@@ -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

664
src/pipecat/processors/frameworks/rtvi/observer.py Normal file
View File

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

653
src/pipecat/processors/frameworks/rtvi/processor.py Normal file
View File

@@ -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}"

4
src/pipecat/serializers/base_serializer.py
View File

@@ -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