From 4f290be83488dcac592d36ad212dac90bd92ac11 Mon Sep 17 00:00:00 2001 From: mattie ruth backman Date: Tue, 10 Feb 2026 15:32:05 -0500 Subject: [PATCH] Initial commit: Introducing RTVI support for files This commit introduces the types for all RTVI file messaging and full support for sending images as byte strings --- src/pipecat/frames/frames.py | 66 ++++++++++++++++++- .../processors/aggregators/llm_context.py | 29 ++++++++ .../aggregators/llm_response_universal.py | 19 ++++++ .../processors/frameworks/rtvi/models.py | 45 ++++++++++++- .../processors/frameworks/rtvi/processor.py | 60 ++++++++++++++++- src/pipecat/runner/run.py | 1 + 6 files changed, 217 insertions(+), 3 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index fbc01c488..bbab49cb2 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -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. diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index 1375b8297..819674ba0 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -173,6 +173,11 @@ class LLMContext: image: Raw image bytes. text: Optional text to include with the image. """ + # Format is a data URL: data:;base64, 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, *, diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index b8a7061d8..867f65764 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -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})") diff --git a/src/pipecat/processors/frameworks/rtvi/models.py b/src/pipecat/processors/frameworks/rtvi/models.py index 7a9d4e633..5255fef48 100644 --- a/src/pipecat/processors/frameworks/rtvi/models.py +++ b/src/pipecat/processors/frameworks/rtvi/models.py @@ -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. diff --git a/src/pipecat/processors/frameworks/rtvi/processor.py b/src/pipecat/processors/frameworks/rtvi/processor.py index b750bd70b..28e309c8c 100644 --- a/src/pipecat/processors/frameworks/rtvi/processor.py +++ b/src/pipecat/processors/frameworks/rtvi/processor.py @@ -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() diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 76c870468..8714fc338 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -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