RTVI Observer

The RTVIObserver converts internal pipeline frames into RTVI protocol messages that can be consumed by clients. It acts as a bridge between Pipecat’s internal frame system and external Pipecat client SDKs.

​

Purpose

The RTVIObserver serves two main purposes:

1. Converting internal frames to RTVI protocol messages
2. Managing state for aggregated messages (like bot transcriptions)

​

Frame Types Handled

The observer processes several types of frames:

​

Speech Events

• UserStartedSpeakingFrame → RTVIUserStartedSpeakingMessage
• UserStoppedSpeakingFrame → RTVIUserStoppedSpeakingMessage
• BotStartedSpeakingFrame → RTVIBotStartedSpeakingMessage
• BotStoppedSpeakingFrame → RTVIBotStoppedSpeakingMessage

​

Transcription Events

• TranscriptionFrame → RTVIUserTranscriptionMessage(final=true)
• InterimTranscriptionFrame → RTVIUserTranscriptionMessage(final=false)

​

LLM Events

• LLMFullResponseStartFrame → RTVIBotLLMStartedMessage
• LLMFullResponseEndFrame → RTVIBotLLMStoppedMessage
• LLMTextFrame → RTVIBotLLMTextMessage

​

TTS Events

• TTSStartedFrame → RTVIBotTTSStartedMessage
• TTSStoppedFrame → RTVIBotTTSStoppedMessage
• TTSTextFrame → RTVIBotTTSTextMessage

​

Other Events

• OpenAILLMContextFrame → RTVIUserLLMTextMessage
• MetricsFrame → RTVIMetricsMessage

​

Frame Deduplication

The observer uses a set to track processed frames, ensuring each frame is only processed once:

if frame.id in self._frames_seen:
    return
self._frames_seen.add(frame.id)

​

Usage

The RTVIObserver is created and attached through the RTVI processor:

rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
task = PipelineTask(
    pipeline,
    PipelineParams(
        observers=[rtvi.observer()],
    ),
)

​

Key Features

• Frame Deduplication: Ensures each frame is processed only once
• Message Conversion: Transforms internal frames to RTVI protocol messages
• Protocol Compliance: Ensures all messages follow the RTVI protocol