Add file frames and universal LLM context support for files

changelog/3415.changed.2.md

@@ -0,0 +1 @@
+- The minimum `aiohttp` version is now 3.14.0 (required by `google-genai` >= 2.8.0). Applications pinning `aiohttp<3.14` will need to update their pin.

pyproject.toml

@@ -21,7 +21,7 @@ classifiers = [
 ]
 dependencies = [
     "aiofiles>=24.1.0,<27",
-    "aiohttp>=3.11.12,<4",
+    "aiohttp>=3.14.0,<4",
     "audioop-lts~=0.2.1; python_version>='3.13'",
     "docstring_parser>=0.16,<1",
     "loguru~=0.7.3",

src/pipecat/adapters/services/anthropic_adapter.py

@@ -159,6 +159,10 @@ def get_messages_for_logging(self, context: LLMContext) -> list[dict[str, Any]]:
                             source["data"] = "..."
                     if item.get("type") == "thinking" and item.get("signature"):
                         item["signature"] = "..."
+                    if item.get("type") == "document":
+                        source = item.get("source")
+                        if isinstance(source, dict) and "data" in source:
+                            source["data"] = "..."
             messages_for_logging.append(msg)
         return messages_for_logging
 
@@ -383,6 +387,7 @@ def _from_standard_message(self, message: LLMStandardMessage) -> MessageParam:
             if content == "":
                 content = "(empty)"
         elif isinstance(content, list):
+            new_content = []
             for item in content:
                 # fix empty text
                 if item["type"] == "text" and item["text"] == "":
@@ -408,22 +413,64 @@ def _from_standard_message(self, message: LLMStandardMessage) -> MessageParam:
                         }
                         del item["image_url"]
                     else:
-                        url = item["image_url"]["url"]
-                        logger.warning(f"Unsupported 'image_url': {url}")
-
-            # In the case where there's a single image in the list (like what
-            # would result from a UserImageRawFrame), ensure that the image
-            # comes before text, as recommended by Anthropic docs
+                        logger.warning(f"Unsupported 'image_url': {item['image_url']['url']}")
+                        continue
+                if item["type"] == "file_url":
+                    f_data = item["file"]
+                    if f_data["mime_type"].startswith("image/"):
+                        item["type"] = "image"
+                        item["source"] = {
+                            "type": "url",
+                            "url": f_data["url"],
+                        }
+                        del item["file"]
+                    elif f_data["mime_type"] == "application/pdf":
+                        item["type"] = "document"
+                        item["source"] = {
+                            "type": "url",
+                            "url": f_data["url"],
+                        }
+                        del item["file"]
+                    else:
+                        # Wrapped as LLMContextConversionError by the caller in
+                        # _from_universal_context_messages.
+                        raise ValueError(
+                            f"Unsupported 'file_url' MIME type: {f_data['mime_type']} "
+                            f"for URL: {f_data['url']}"
+                        )
+                if item["type"] == "file_base64":
+                    f_data = item["file"]
+                    if f_data["mime_type"] != "application/pdf":
+                        # Wrapped as LLMContextConversionError by the caller in
+                        # _from_universal_context_messages.
+                        raise ValueError(f"Unsupported 'file' MIME type: {f_data['mime_type']}")
+
+                    item["type"] = "document"
+                    item["source"] = {
+                        "type": "base64",
+                        "media_type": f_data["mime_type"],
+                        "data": f_data["file_data"].split(",")[1],
+                    }
+                    del item["file"]
+                new_content.append(item)
+            content = new_content
+            msg["content"] = content
+
+            # In the case where there's a single image or document in the list (like
+            # what would result from a UserImageRawFrame or UserFileRawFrame), ensure
+            # it comes before text, as recommended by Anthropic docs
             # (https://docs.anthropic.com/en/docs/build-with-claude/vision#example-one-image)
-            image_indices = [i for i, item in enumerate(content) if item["type"] == "image"]
+            media_indices = [
+                i for i, item in enumerate(content) if item["type"] in ("image", "document")
+            ]
             text_indices = [i for i, item in enumerate(content) if item["type"] == "text"]
-            if len(image_indices) == 1 and text_indices:
-                img_idx = image_indices[0]
+            if len(media_indices) == 1 and text_indices:
+                media_idx = media_indices[0]
                 first_txt_idx = text_indices[0]
-                if img_idx > first_txt_idx:
-                    # Move image before the first text
-                    image_item = content.pop(img_idx)
-                    content.insert(first_txt_idx, image_item)
+                if media_idx > first_txt_idx:
+                    # Move the image/document before the first text
+                    media_item = content.pop(media_idx)
+                    content.insert(first_txt_idx, media_item)
 
         return cast(MessageParam, msg)
 

src/pipecat/adapters/services/bedrock_adapter.py

@@ -9,6 +9,7 @@
 import base64
 import copy
 import json
+import re
 from dataclasses import dataclass
 from typing import Any, TypedDict, cast
 
@@ -25,6 +26,72 @@
     LLMStandardMessage,
 )
 
+_MIME_TO_BEDROCK_FORMAT: dict[str, str] = {
+    "application/pdf": "pdf",
+    "text/csv": "csv",
+    "application/msword": "doc",
+    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": "docx",
+    "application/vnd.ms-excel": "xls",
+    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": "xlsx",
+    "text/html": "html",
+    "text/plain": "txt",
+    "text/markdown": "md",
+}
+
+_MIME_TO_BEDROCK_IMAGE_FORMAT: dict[str, str] = {
+    "image/png": "png",
+    "image/jpeg": "jpeg",
+    "image/gif": "gif",
+    "image/webp": "webp",
+}
+
+_MIME_TO_BEDROCK_VIDEO_FORMAT: dict[str, str] = {
+    "video/x-flv": "flv",
+    "video/x-matroska": "mkv",
+    "video/quicktime": "mov",
+    "video/mpeg": "mpeg",
+    "video/mp4": "mp4",
+    "video/3gpp": "three_gp",
+    "video/webm": "webm",
+    "video/x-ms-wmv": "wmv",
+}
+
+
+def _bedrock_video_format(mime_type: str, filename: str | None) -> str | None:
+    """Map a video MIME type to Bedrock's video format.
+
+    ``video/mpeg`` covers both the ``mpeg`` and ``mpg`` Bedrock formats, which
+    aren't distinguishable by MIME type alone, so fall back to the filename
+    extension to disambiguate.
+    """
+    if mime_type == "video/mpeg" and filename and filename.lower().endswith(".mpg"):
+        return "mpg"
+    return _MIME_TO_BEDROCK_VIDEO_FORMAT.get(mime_type)
+
+
+# Bedrock's API reference documents an AudioBlock content type
+# (https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_AudioBlock.html),
+# but in practice ConverseStream rejects it outright ("ContentBlock object
+# ... must set one of the following keys: text, image, toolUse, toolResult,
+# document, video, cachePoint, searchResult") regardless of model, unlike
+# unsupported image/video/document blocks, which are rejected per-model with
+# a specific "this model doesn't support the X content block" error. So
+# audio files aren't handled as a distinct content block below; treat this
+# as unsupported until the live API actually accepts it.
+
+# Bedrock's document name only allows alphanumerics, whitespace, hyphens,
+# parentheses, and square brackets, with no consecutive whitespace — notably
+# no ".", so an ordinary filename like "notes.pdf" is rejected as-is.
+_BEDROCK_DOCUMENT_NAME_DISALLOWED = re.compile(r"[^A-Za-z0-9\s\-()\[\]]")
+_BEDROCK_DOCUMENT_NAME_WHITESPACE = re.compile(r"\s+")
+
+
+def _sanitize_bedrock_document_name(name: str) -> str:
+    """Rewrite `name` to satisfy Bedrock's document name restrictions."""
+    sanitized = _BEDROCK_DOCUMENT_NAME_DISALLOWED.sub(" ", name)
+    sanitized = _BEDROCK_DOCUMENT_NAME_WHITESPACE.sub(" ", sanitized).strip()
+    return sanitized or "document"
+
 
 class AWSBedrockLLMInvocationParams(TypedDict):
     """Context-based parameters for invoking AWS Bedrock's LLM API."""
@@ -115,6 +182,10 @@ def get_messages_for_logging(self, context) -> list[dict[str, Any]]:
                     for item in msg["content"]:
                         if item.get("image"):
                             item["image"]["source"]["bytes"] = "..."
+                        if item.get("document"):
+                            item["document"]["source"]["bytes"] = "..."
+                        if item.get("video"):
+                            item["video"]["source"]["bytes"] = "..."
             messages_for_logging.append(msg)
         return messages_for_logging
 
@@ -321,19 +392,112 @@ def _from_standard_message(self, message: LLMStandardMessage) -> dict[str, Any]:
                     else:
                         url = item["image_url"]["url"]
                         logger.warning(f"Unsupported 'image_url': {url}")
-
-            # In the case where there's a single image in the list (like what
-            # would result from a UserImageRawFrame), ensure that the image
-            # comes before text
-            image_indices = [i for i, item in enumerate(new_content) if "image" in item]
-            text_indices = [i for i, item in enumerate(new_content) if "text" in item]
-            if len(image_indices) == 1 and text_indices:
-                img_idx = image_indices[0]
-                first_txt_idx = text_indices[0]
-                if img_idx > first_txt_idx:
-                    # Move image before the first text
-                    image_item = new_content.pop(img_idx)
-                    new_content.insert(first_txt_idx, image_item)
+                elif item["type"] == "file_base64":
+                    f_data = item["file"]
+                    mime_type = f_data["mime_type"]
+                    file_data_url = f_data["file_data"]
+                    raw_bytes = base64.b64decode(file_data_url.split(",")[1])
+                    image_format = _MIME_TO_BEDROCK_IMAGE_FORMAT.get(mime_type)
+                    if image_format is not None:
+                        new_content.append(
+                            {
+                                "image": {
+                                    "format": image_format,
+                                    "source": {"bytes": raw_bytes},
+                                }
+                            }
+                        )
+                        continue
+                    video_format = _bedrock_video_format(mime_type, f_data["filename"])
+                    if video_format is not None:
+                        new_content.append(
+                            {
+                                "video": {
+                                    "format": video_format,
+                                    "source": {"bytes": raw_bytes},
+                                }
+                            }
+                        )
+                        continue
+                    bedrock_format = _MIME_TO_BEDROCK_FORMAT.get(mime_type)
+                    if bedrock_format is None:
+                        # Wrapped as LLMContextConversionError by the caller in
+                        # _from_universal_context_messages.
+                        raise ValueError(f"Unsupported 'file' MIME type for Bedrock: {mime_type}")
+                    new_content.append(
+                        {
+                            "document": {
+                                "name": _sanitize_bedrock_document_name(
+                                    f_data["filename"] or "document"
+                                ),
+                                "format": bedrock_format,
+                                "source": {"bytes": raw_bytes},
+                            }
+                        }
+                    )
+                elif item["type"] == "file_url":
+                    f_data = item["file"]
+                    url = f_data["url"]
+                    if url.startswith("s3://"):
+                        mime_type = f_data["mime_type"]
+                        image_format = _MIME_TO_BEDROCK_IMAGE_FORMAT.get(mime_type)
+                        if image_format is not None:
+                            new_content.append(
+                                {
+                                    "image": {
+                                        "format": image_format,
+                                        "source": {"s3Location": {"uri": url}},
+                                    }
+                                }
+                            )
+                            continue
+                        video_format = _bedrock_video_format(mime_type, f_data["filename"])
+                        if video_format is not None:
+                            new_content.append(
+                                {
+                                    "video": {
+                                        "format": video_format,
+                                        "source": {"s3Location": {"uri": url}},
+                                    }
+                                }
+                            )
+                            continue
+                        bedrock_format = _MIME_TO_BEDROCK_FORMAT.get(mime_type)
+                        if bedrock_format is None:
+                            # Wrapped as LLMContextConversionError by the caller in
+                            # _from_universal_context_messages.
+                            raise ValueError(
+                                f"Unsupported MIME type for Bedrock S3 URI: {mime_type}"
+                            )
+                        new_content.append(
+                            {
+                                "document": {
+                                    "name": _sanitize_bedrock_document_name(
+                                        f_data["filename"] or "document"
+                                    ),
+                                    "format": bedrock_format,
+                                    "source": {"s3Location": {"uri": url}},
+                                }
+                            }
+                        )
+                    else:
+                        # Wrapped as LLMContextConversionError by the caller in
+                        # _from_universal_context_messages.
+                        raise ValueError(f"Bedrock only supports S3 URLs for file sources: {url}")
+
+            # In the case where there's a single image, document, or video in
+            # the list (like what would result from a UserImageRawFrame),
+            # ensure it comes before text
+            for key in ("image", "document", "video"):
+                media_indices = [i for i, item in enumerate(new_content) if key in item]
+                text_indices = [i for i, item in enumerate(new_content) if "text" in item]
+                if len(media_indices) == 1 and text_indices:
+                    media_idx = media_indices[0]
+                    first_txt_idx = text_indices[0]
+                    if media_idx > first_txt_idx:
+                        # Move the media item before the first text
+                        media_item = new_content.pop(media_idx)
+                        new_content.insert(first_txt_idx, media_item)
             return {"role": msg["role"], "content": new_content}
 
         return msg

src/pipecat/adapters/services/gemini_adapter.py

@@ -545,6 +545,23 @@ def _from_standard_message(
                     input_audio = c["input_audio"]
                     audio_bytes = base64.b64decode(input_audio["data"])
                     parts.append(Part(inline_data=Blob(mime_type="audio/wav", data=audio_bytes)))
+                elif c["type"] == "file_base64":
+                    f_data = c["file"]
+                    mime_type = f_data["mime_type"]
+                    file_data_url = f_data["file_data"]
+                    parts.append(
+                        Part(
+                            inline_data=Blob(
+                                mime_type=mime_type,
+                                data=base64.b64decode(file_data_url.split(",")[1]),
+                            )
+                        )
+                    )
+                elif c["type"] == "file_url":
+                    f_data = c["file"]
+                    parts.append(
+                        Part.from_uri(file_uri=f_data["url"], mime_type=f_data["mime_type"])
+                    )
                 elif c["type"] == "file_data":
                     file_data = c["file_data"]
                     parts.append(