Add error handling for unsupported files

src/pipecat/adapters/services/anthropic_adapter.py

@@ -96,6 +96,8 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
                             item["source"]["data"] = "..."
                         if item["type"] == "thinking" and item.get("signature"):
                             item["signature"] = "..."
+                        if item["type"] == "file":
+                            item["file"]["file_data"] = "data:..."
             messages_for_logging.append(msg)
         return messages_for_logging
 

src/pipecat/adapters/services/bedrock_adapter.py

@@ -91,6 +91,8 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
                     for item in msg["content"]:
                         if item.get("image"):
                             item["image"]["source"]["bytes"] = "..."
+                        if item.get("type") == "file":
+                            item["file"]["file_data"] = "data:..."
             messages_for_logging.append(msg)
         return messages_for_logging
 

src/pipecat/adapters/services/grok_realtime_adapter.py

@@ -87,6 +87,9 @@ class GrokRealtimeLLMAdapter(BaseLLMAdapter):
                             item["audio"] = "..."
                         if item.get("type") == "audio":
                             item["audio"] = "..."
+                        if item.get("type") == "file":
+                            if item["file"]["file_data"].startswith("data:"):
+                                item["file"]["file_data"] = "data:..."
             msgs.append(msg)
         return msgs
 

src/pipecat/adapters/services/open_ai_adapter.py

@@ -105,6 +105,9 @@ class OpenAILLMAdapter(BaseLLMAdapter[OpenAILLMInvocationParams]):
                                 item["image_url"]["url"] = "data:image/..."
                         if item["type"] == "input_audio":
                             item["input_audio"]["data"] = "..."
+                        if item["type"] == "file":
+                            if item["file"]["file_data"].startswith("data:"):
+                                item["file"]["file_data"] = "data:..."
             if "mime_type" in msg and msg["mime_type"].startswith("image/"):
                 msg["data"] = "..."
             msgs.append(msg)

src/pipecat/adapters/services/open_ai_realtime_adapter.py

@@ -90,6 +90,9 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
                                 item["image_url"]["url"] = "data:image/..."
                         if item["type"] == "input_audio":
                             item["input_audio"]["data"] = "..."
+                        if item["type"] == "file":
+                            if item["file"]["file_data"].startswith("data:"):
+                                item["file"]["file_data"] = "data:..."
             if "mime_type" in msg and msg["mime_type"].startswith("image/"):
                 msg["data"] = "..."
             msgs.append(msg)

src/pipecat/frames/frames.py

@@ -244,7 +244,7 @@ class ImageRawFrame:
     format: Optional[str]
 
 
-FileSourceType = Literal["bytes", "url"]
+FileSourceType = Literal["bytes", "url", "id"]
 
 
 @dataclass
@@ -253,7 +253,7 @@ class FileRawFrame:
 
     Parameters:
         file: Raw file bytes.
-        type: Type of the file ('bytes' or 'url'),
+        type: Type of the file ('bytes', 'url', or 'id'),
         name: Optional name of the file.
         format: File format (expected in Mime Format).
     """

src/pipecat/processors/aggregators/llm_context.py

@@ -414,9 +414,8 @@ class LLMContext:
         bytes = file
         is_image = format.startswith("image/")
         if not is_image and format != "application/pdf":
-            # TODO how to send error back?
             logger.warning(f"Unsupported file format for LLM context: {format}")
-            return
+            raise ValueError(f"Unsupported file format for LLM context: {format}")
         if type == "url":
             async with aiohttp.ClientSession() as session:
                 async with session.get(file) as response:
@@ -467,6 +466,33 @@ class LLMContext:
         message = await LLMContext.create_audio_message(audio_frames=audio_frames, text=text)
         self.add_message(message)
 
+    def maybe_remove_invalid_message(self, exception: Exception):
+        """Remove messages from context if an exception indicates they were invalid.
+
+        This is a best-effort method to handle cases where an LLM service returns
+        an error indicating that a particular message in the context was invalid
+        (e.g., due to unsupported content). The method inspects the exception,
+        and if it can identify a specific message that caused the issue, it removes
+        that message from the context.
+
+        Args:
+            exception: The exception raised during LLM processing, which may contain
+                information about invalid messages.
+        """
+        # Check for OpenAI-specific error indicating an invalid file message, and remove it if found.
+        if exception.type == "invalid_request_error" and exception.param == "file_id":
+            for i in range(len(self._messages) - 1, -1, -1):
+                message = self._messages[i]
+                content = message.get("content", [])
+                if isinstance(content, list) and any(
+                    item.get("type") == "file" for item in content
+                ):
+                    logger.warning(
+                        "Removing message with file content from context due to invalid response."
+                    )
+                    self._messages.pop(i)
+                    break
+
     @staticmethod
     def _normalize_and_validate_tools(tools: ToolsSchema | NotGiven) -> ToolsSchema | NotGiven:
         """Normalize and validate the given tools.

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -233,6 +233,9 @@ class OpenAILLMContext:
                         if item["type"] == "image_url":
                             if item["image_url"]["url"].startswith("data:image/"):
                                 item["image_url"]["url"] = "data:image/..."
+                        if item["type"] == "file":
+                            if item["file"]["file_data"].startswith("data:"):
+                                item["file"]["file_data"] = "data:..."
             if "mime_type" in msg and msg["mime_type"].startswith("image/"):
                 msg["data"] = "..."
             msgs.append(msg)

src/pipecat/processors/frameworks/rtvi/models.py

@@ -30,7 +30,7 @@ from pipecat.frames.frames import (
 )
 
 # -- Constants --
-RTVI_PROTOCOL_VERSION = "1.3.0"
+PROTOCOL_VERSION = "1.3.0"
 
 MESSAGE_LABEL = "rtvi-ai"
 MessageLiteral = Literal["rtvi-ai"]

src/pipecat/processors/frameworks/rtvi/processor.py

@@ -607,13 +607,12 @@ class RTVIProcessor(FrameProcessor):
                 file_path = os.path.join(self._folder, file.source.id.removeprefix("pipecat:"))
                 with open(file_path, "rb") as f:
                     raw_bytes = f.read()
-                    encoded_image = base64.b64encode(raw_bytes).decode("utf-8")
+                    encoded_file = base64.b64encode(raw_bytes).decode("utf-8")
-                    source = f"data:{file.format};base64,{encoded_image}"
+                    source = f"data:{file.format};base64,{encoded_file}"
                 try:
                     os.remove(file_path)
                 except OSError as e:
                     logger.warning(f"Failed to remove uploaded file {file_path}: {e}")
-                source = file.source.id
             case _:
                 logger.warning(f"Unsupported file source type: {type}")
                 return

src/pipecat/runner/run.py

@@ -217,6 +217,9 @@ def _create_server_app(args: argparse.Namespace):
     else:
         logger.warning(f"Unknown transport type: {args.transport}")
 
+    if args.uploads_folder:
+        _setup_file_uploads_route(app, args)
+
     return app
 
 
@@ -284,37 +287,6 @@ def _setup_webrtc_routes(app: FastAPI, args: argparse.Namespace):
 
         return FileResponse(path=file_path, media_type=media_type, filename=safe_name)
 
-    @app.post("/files")
-    async def upload_file(file: UploadFile = File(...)):
-        """Handle file uploads from clients. Requires --uploads-folder to be set."""
-        if not args.uploads_folder:
-            raise HTTPException(
-                503,
-                "File upload is disabled: start the runner with -u/--uploads-folder to set the uploads directory.",
-            )
-        folder = Path(args.uploads_folder)
-        folder.mkdir(parents=True, exist_ok=True)
-        # Always save as UUID so uploads are not discoverable by guessing filenames
-        safe_name = uuid.uuid4().hex
-        file_path = folder / safe_name
-        try:
-            contents = await file.read()
-            file_path.write_bytes(contents)
-            logger.debug(f"Uploaded file to {file_path}")
-        except OSError as e:
-            logger.error(f"Failed to save upload: {e}")
-            raise HTTPException(500, "Failed to save file") from e
-        _trim_uploads_folder(folder, args.uploads_folder_max_files)
-        # Use original filename only for format/mime in response
-        original_name = Path(file.filename or "").name
-        media_type, _ = mimetypes.guess_type(original_name, strict=False)
-        # Return the file as a URL that can be used in the client, matching the format of RTVIFile
-        return {
-            "name": original_name,
-            "source": {"type": "id", "id": f"pipecat:{safe_name}"},
-            "format": media_type,
-        }
-
     # Initialize the SmallWebRTC request handler
     small_webrtc_handler: SmallWebRTCRequestHandler = SmallWebRTCRequestHandler(
         esp32_mode=args.esp32, host=args.host
@@ -869,6 +841,39 @@ def _setup_telephony_routes(app: FastAPI, args: argparse.Namespace):
         return {"status": f"Bot started with {args.transport}"}
 
 
+def _setup_file_uploads_route(app: FastAPI, args: argparse.Namespace):
+    @app.post("/files")
+    async def upload_file(file: UploadFile = File(...)):
+        """Handle file uploads from clients. Requires --uploads-folder to be set."""
+        if not args.uploads_folder:
+            raise HTTPException(
+                503,
+                "File upload is disabled: start the runner with -u/--uploads-folder to set the uploads directory.",
+            )
+        folder = Path(args.uploads_folder)
+        folder.mkdir(parents=True, exist_ok=True)
+        # Always save as UUID so uploads are not discoverable by guessing filenames
+        safe_name = uuid.uuid4().hex
+        file_path = folder / safe_name
+        try:
+            contents = await file.read()
+            file_path.write_bytes(contents)
+            logger.debug(f"Uploaded file to {file_path}")
+        except OSError as e:
+            logger.error(f"Failed to save upload: {e}")
+            raise HTTPException(500, "Failed to save file") from e
+        _trim_uploads_folder(folder, args.uploads_folder_max_files)
+        # Use original filename only for format/mime in response
+        original_name = Path(file.filename or "").name
+        media_type, _ = mimetypes.guess_type(original_name, strict=False)
+        # Return the file as a URL that can be used in the client, matching the format of RTVIFile
+        return {
+            "name": original_name,
+            "source": {"type": "id", "id": f"pipecat:{safe_name}"},
+            "format": media_type,
+        }
+
+
 async def _run_daily_direct(args: argparse.Namespace):
     """Run Daily bot with direct connection (no FastAPI server)."""
     try:

src/pipecat/services/anthropic/llm.py

@@ -1162,6 +1162,8 @@ class AnthropicLLMContext(OpenAILLMContext):
                     for item in msg["content"]:
                         if item["type"] == "image":
                             item["source"]["data"] = "..."
+                        if item["type"] == "file":
+                            item["file"]["file_data"] = "data:..."
             msgs.append(msg)
         return msgs
 

src/pipecat/services/aws/llm.py

@@ -607,6 +607,8 @@ class AWSBedrockLLMContext(OpenAILLMContext):
                     for item in msg["content"]:
                         if item.get("image"):
                             item["image"]["source"]["bytes"] = "..."
+                        if item.get("type") == "file":
+                            item["file"]["file_data"] = "data:..."
             msgs.append(msg)
         return msgs
 

src/pipecat/services/openai/base_llm.py

@@ -618,6 +618,7 @@ class BaseOpenAILLMService(LLMService):
                 await self.push_error(error_msg="LLM completion timeout", exception=e)
             except Exception as e:
                 await self.push_error(error_msg=f"Error during completion: {e}", exception=e)
+                context.maybe_remove_invalid_message(e)
             finally:
                 await self.stop_processing_metrics()
                 await self.push_frame(LLMFullResponseEndFrame())