From 7742d1a83b1181e9b297bc3ed3a1c78b09fbed2a Mon Sep 17 00:00:00 2001 From: mattie ruth backman Date: Wed, 18 Mar 2026 15:42:35 -0400 Subject: [PATCH] Add error handling for unsupported files --- .../adapters/services/anthropic_adapter.py | 2 + .../adapters/services/bedrock_adapter.py | 2 + .../services/grok_realtime_adapter.py | 3 + .../adapters/services/open_ai_adapter.py | 3 + .../services/open_ai_realtime_adapter.py | 3 + src/pipecat/frames/frames.py | 4 +- .../processors/aggregators/llm_context.py | 30 ++++++++- .../aggregators/openai_llm_context.py | 3 + .../processors/frameworks/rtvi/models.py | 2 +- .../processors/frameworks/rtvi/processor.py | 5 +- src/pipecat/runner/run.py | 67 ++++++++++--------- src/pipecat/services/anthropic/llm.py | 2 + src/pipecat/services/aws/llm.py | 2 + src/pipecat/services/openai/base_llm.py | 1 + 14 files changed, 90 insertions(+), 39 deletions(-) diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py index ecc87154c..366393412 100644 --- a/src/pipecat/adapters/services/anthropic_adapter.py +++ b/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 diff --git a/src/pipecat/adapters/services/bedrock_adapter.py b/src/pipecat/adapters/services/bedrock_adapter.py index d63c5cf0f..66e769972 100644 --- a/src/pipecat/adapters/services/bedrock_adapter.py +++ b/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 diff --git a/src/pipecat/adapters/services/grok_realtime_adapter.py b/src/pipecat/adapters/services/grok_realtime_adapter.py index 03a3ed475..01cabac7b 100644 --- a/src/pipecat/adapters/services/grok_realtime_adapter.py +++ b/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 diff --git a/src/pipecat/adapters/services/open_ai_adapter.py b/src/pipecat/adapters/services/open_ai_adapter.py index f4b534f2c..bbfdfa6b4 100644 --- a/src/pipecat/adapters/services/open_ai_adapter.py +++ b/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) diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 136dc1a2f..6929152a5 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/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) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index ed6e83b6e..76afe9dde 100644 --- a/src/pipecat/frames/frames.py +++ b/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). """ diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index b45c20e1e..0908f179d 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/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. diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index f75625156..d9707ed5f 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/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) diff --git a/src/pipecat/processors/frameworks/rtvi/models.py b/src/pipecat/processors/frameworks/rtvi/models.py index e9eb52041..9524012ca 100644 --- a/src/pipecat/processors/frameworks/rtvi/models.py +++ b/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"] diff --git a/src/pipecat/processors/frameworks/rtvi/processor.py b/src/pipecat/processors/frameworks/rtvi/processor.py index d0443366e..5a4f6b0af 100644 --- a/src/pipecat/processors/frameworks/rtvi/processor.py +++ b/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") - source = f"data:{file.format};base64,{encoded_image}" + encoded_file = base64.b64encode(raw_bytes).decode("utf-8") + 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 diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 786f3a8e6..ed74a1420 100644 --- a/src/pipecat/runner/run.py +++ b/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: diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 2f375df82..fdae862b9 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/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 diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index 92049dffb..cbaffbb97 100644 --- a/src/pipecat/services/aws/llm.py +++ b/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 diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index eb8ce3cc6..0fb8486b4 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/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())