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

Initial commit: Introducing RTVI support for files

Browse Source 
This commit introduces the types for all RTVI file messaging and full
support for sending images as byte strings
This commit is contained in:
mattie ruth backman
2026-02-10 15:32:05 -05:00
parent 53388e0426
commit 4f290be834
6 changed files with 217 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
src/pipecat/frames/frames.py
View File

@@ -25,6 +25,7 @@ from typing import (
Optional,
Sequence,
Tuple,
get_args,
)
from pipecat.adapters.schemas.tools_schema import ToolsSchema
@@ -234,7 +235,7 @@ class ImageRawFrame:
"""A frame containing a raw image.
Parameters:
image: Raw image bytes.
image: Raw image bytes or a base64-encoded string.
size: Image dimensions as (width, height) tuple.
format: Image format (e.g., 'RGB', 'RGBA').
"""
@@ -244,6 +245,35 @@ class ImageRawFrame:
format: Optional[str]
ImageFileFormat = Literal["png", "jpg", "jpeg", "webp", "gif", "heic", "hief"]
DocFileFormat = Literal[
"pdf", "csv", "txt", "md", "doc", "docx", "xls", "xlsx", "json", "html", "css", "javascript"
]
MediaFileFormat = Literal["mp3", "wav", "ogg", "aac", "mp4", "webm", "ogg", "avi"]
_all_formats = list(
set(get_args(ImageFileFormat) + get_args(DocFileFormat) + get_args(MediaFileFormat))
)
FileFormat = Literal[tuple(_all_formats)]
FileSourceType = Literal["bytes", "url"] # TODO: Add support for "id"
@dataclass
class FileRawFrame:
"""A frame containing a raw file.
Parameters:
file: Raw file bytes.
type: Type of the file ('bytes' or 'url'),
format: File format (e.g., 'pdf', 'docx').
"""
file: bytes | str
type: FileSourceType
format: Optional[FileFormat]
#
# Data frames.
#
@@ -1584,6 +1614,18 @@ class InputImageRawFrame(SystemFrame, ImageRawFrame):
return f"{self.name}(pts: {pts}, source: {self.transport_source}, size: {self.size}, format: {self.format})"
@dataclass
class InputFileRawFrame(SystemFrame, FileRawFrame):
"""Raw file input frame.
A file usually coming from RTVI.
"""
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, type: {self.type})"
@dataclass
class InputTextRawFrame(SystemFrame, TextFrame):
"""Raw text input frame from transport.
@@ -1638,6 +1680,28 @@ class UserImageRawFrame(InputImageRawFrame):
return f"{self.name}(pts: {pts}, user: {self.user_id}, source: {self.transport_source}, size: {self.size}, format: {self.format}, text: {self.text}, append_to_context: {self.append_to_context})"
@dataclass
class UserFileRawFrame(InputFileRawFrame):
"""Raw file input frame associated with a specific user.
A file associated to a user.
Parameters:
user_id: Identifier of the user who provided this file.
text: Text associated to this file.
append_to_context: Whether the requested file should be appended to the LLM context.
"""
user_id: str = ""
text: str = ""
append_to_context: Optional[bool] = None
custom_options: Optional[dict] = None
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, user: {self.user_id}, format: {self.format}, type: {self.type}, text: {self.text}, append_to_context: {self.append_to_context})"
@dataclass
class AssistantImageRawFrame(OutputImageRawFrame):
"""Frame containing an image generated by the assistant.

29
src/pipecat/processors/aggregators/llm_context.py
View File

@@ -173,6 +173,11 @@ class LLMContext:
image: Raw image bytes.
text: Optional text to include with the image.
"""
# Format is a data URL: data:<mime type>;base64,<data> already provided
if format.startswith("url/"):
url = image
return LLMContext.create_image_url_message(role=role, url=url, text=text)
# Format is a mime type: image is already encoded
image_already_encoded = format.startswith("image/")
@@ -357,6 +362,30 @@ class LLMContext:
"""
self._tool_choice = tool_choice
async def add_file_frame_message(
self,
*,
format: str,
size: tuple[int, int],
image: bytes,
text: Optional[str] = None,
role: str = "user",
):
"""Add a message containing a file frame.
Args:
format: File format (e.g., 'RGB', 'RGBA', or, if already encoded,
the MIME type like 'image/jpeg').
size: File dimensions as (width, height) tuple.
image: Raw image bytes.
text: Optional text to include with the image.
role: The role of this message (defaults to "user").
"""
message = await LLMContext.create_image_message(
role=role, format=format, size=size, image=image, text=text
)
self.add_message(message)
async def add_image_frame_message(
self,
*,

19
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -55,6 +55,7 @@ from pipecat.frames.frames import (
TextFrame,
TranscriptionFrame,
TranslationFrame,
UserFileRawFrame,
UserImageRawFrame,
UserMuteStartedFrame,
UserMuteStoppedFrame,
@@ -957,6 +958,8 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self._handle_function_call_cancel(frame)
elif isinstance(frame, UserImageRawFrame):
await self._handle_user_image_frame(frame)
elif isinstance(frame, UserFileRawFrame):
await self._handle_user_file_frame(frame)
elif isinstance(frame, AssistantImageRawFrame):
await self._handle_assistant_image_frame(frame)
else:
@@ -1135,6 +1138,22 @@ class LLMAssistantAggregator(LLMContextAggregator):
if image_appended:
await self.push_context_frame(FrameDirection.UPSTREAM)
async def _handle_user_file_frame(self, frame: UserFileRawFrame):
if not frame.append_to_context:
return
logger.debug(f"{self} Appending UserFileRawFrame to LLM context (format: {frame.format})")
await self._context.add_file_frame_message(
format=frame.format,
text=frame.text,
type=frame.type,
file=frame.file,
options=frame.custom_options,
)
await self.push_aggregation()
await self.push_context_frame(FrameDirection.UPSTREAM)
async def _handle_assistant_image_frame(self, frame: AssistantImageRawFrame):
logger.debug(f"{self} Appending AssistantImageRawFrame to LLM context (size: {frame.size})")

45
src/pipecat/processors/frameworks/rtvi/models.py
View File

@@ -26,10 +26,12 @@ from pydantic import BaseModel
from pipecat.frames.frames import (
AggregationType,
FileFormat,
FileSourceType,
)
# -- Constants --
PROTOCOL_VERSION = "1.2.0"
RTVI_PROTOCOL_VERSION = "1.3.0"
MESSAGE_LABEL = "rtvi-ai"
MessageLiteral = Literal["rtvi-ai"]
@@ -229,6 +231,47 @@ class SendTextData(BaseModel):
options: Optional[SendTextOptions] = None
class FileSource(BaseModel):
"""Base class for RTVI file sources."""
type: FileSourceType
class FileBytes(FileSource):
"""File source as base64-encoded bytes."""
type: FileSourceType = "bytes"
bytes: str # base64-encoded string
width: Optional[int] = None
height: Optional[int] = None
class FileUrl(FileSource):
"""File source as a URL."""
type: FileSourceType = "url"
url: str
class File(BaseModel):
"""File data structure for RTVI file sending."""
format: FileFormat
source: FileBytes | FileUrl
customOpts: Optional[dict] = None # ex. 'detail' in openAI or 'citations' in Bedrock
class SendFileData(BaseModel):
"""Data format for sending a file to the LLM.
Contains the information of the file to send and any options for how the pipeline should process it.
"""
content: str # Text to accompany the file
file: File
options: Optional[SendTextOptions] = None
class AppendToContextData(BaseModel):
"""Data format for appending messages to the context.

60
src/pipecat/processors/frameworks/rtvi/processor.py
View File

@@ -8,7 +8,7 @@
import asyncio
import base64
from typing import Any, Dict, Mapping, Optional
from typing import Any, Dict, Mapping, Optional, get_args
from loguru import logger
from pydantic import BaseModel, ValidationError
@@ -20,8 +20,10 @@ from pipecat.frames.frames import (
EndFrame,
EndTaskFrame,
ErrorFrame,
FileFormat,
Frame,
FunctionCallResultFrame,
ImageFileFormat,
InputAudioRawFrame,
InputTransportMessageFrame,
LLMConfigureOutputFrame,
@@ -29,6 +31,8 @@ from pipecat.frames.frames import (
OutputTransportMessageUrgentFrame,
StartFrame,
SystemFrame,
UserFileRawFrame,
UserImageRawFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.frameworks.rtvi.frames import RTVIActionFrame, RTVIClientMessageFrame
@@ -383,6 +387,9 @@ class RTVIProcessor(FrameProcessor):
case "send-text":
data = RTVI.SendTextData.model_validate(message.data)
await self._handle_send_text(data)
case "send-file":
data = RTVI.SendFileData.model_validate(message.data)
await self._handle_send_file(data)
case "append-to-context":
logger.warning(
f"The append-to-context message is deprecated, use send-text instead."
@@ -393,6 +400,7 @@ class RTVIProcessor(FrameProcessor):
await self._handle_audio_buffer(message.data)
case _:
logger.warning(f"Unsupported RTVI message type: {message.type}")
await self._send_error_response(message.id, f"Unsupported type {message.type}")
except ValidationError as e:
@@ -555,6 +563,56 @@ class RTVIProcessor(FrameProcessor):
output_frame = LLMConfigureOutputFrame(skip_tts=cur_llm_skip_tts)
await self.push_frame(output_frame)
async def _handle_send_file(self, data: RTVI.SendFileData):
"""Handle a send-file message from the client."""
file = data.file
if file.format not in get_args(FileFormat):
logger.warning(f"Unsupported file format: {file.format}")
return
source = None
if file.source.type == "bytes":
source = file.source.bytes
elif file.source.type == "url":
source = file.source.url
elif file.source.type == "id":
logger.warning("File source type 'id' is not supported yet.")
return
else:
logger.warning(f"Unsupported file source type: {file.source.type}")
return
if file.source.type == "bytes" and file.format in get_args(ImageFileFormat):
size = [file.source.width or 0, file.source.height or 0]
file_frame = UserImageRawFrame(
text=data.content,
image=source,
size=size,
format=f"url/{file.format}",
append_to_context=True,
)
else:
file_frame = UserFileRawFrame(
text=data.content,
file=source,
type=file.source.type,
format=file.format,
custom_options=file.customOpts,
)
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)
await self.push_frame(file_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()

1
src/pipecat/runner/run.py
View File

@@ -576,6 +576,7 @@ def _setup_daily_routes(app: FastAPI, args: argparse.Namespace):
result = None
print(f"create_daily_room: {create_daily_room}, existing_room_url: {existing_room_url}")
# Configure room if:
# 1. Explicitly requested via createDailyRoom in payload
# 2. Using pre-configured room from DAILY_ROOM_URL env var