feat(File): add File type for handling file data

dspy/__init__.py

@@ -6,7 +6,7 @@
 
 from dspy.evaluate import Evaluate  # isort: skip
 from dspy.clients import *  # isort: skip
-from dspy.adapters import Adapter, ChatAdapter, JSONAdapter, XMLAdapter, TwoStepAdapter, Image, Audio, History, Type, Tool, ToolCalls, Code  # isort: skip
+from dspy.adapters import Adapter, ChatAdapter, JSONAdapter, XMLAdapter, TwoStepAdapter, Image, Audio, File, History, Type, Tool, ToolCalls, Code  # isort: skip
 from dspy.utils.logging_utils import configure_dspy_loggers, disable_logging, enable_logging
 from dspy.utils.asyncify import asyncify
 from dspy.utils.syncify import syncify

dspy/adapters/__init__.py

@@ -2,7 +2,7 @@
 from dspy.adapters.chat_adapter import ChatAdapter
 from dspy.adapters.json_adapter import JSONAdapter
 from dspy.adapters.two_step_adapter import TwoStepAdapter
-from dspy.adapters.types import Audio, Code, History, Image, Tool, ToolCalls, Type
+from dspy.adapters.types import Audio, Code, File, History, Image, Tool, ToolCalls, Type
 from dspy.adapters.xml_adapter import XMLAdapter
 
 __all__ = [
@@ -12,6 +12,7 @@
     "History",
     "Image",
     "Audio",
+    "File",
     "Code",
     "JSONAdapter",
     "XMLAdapter",

dspy/adapters/types/__init__.py

@@ -1,8 +1,9 @@
 from dspy.adapters.types.audio import Audio
 from dspy.adapters.types.base_type import Type
 from dspy.adapters.types.code import Code
+from dspy.adapters.types.file import File
 from dspy.adapters.types.history import History
 from dspy.adapters.types.image import Image
 from dspy.adapters.types.tool import Tool, ToolCalls
 
-__all__ = ["History", "Image", "Audio", "Type", "Tool", "ToolCalls", "Code"]
+__all__ = ["History", "Image", "Audio", "File", "Type", "Tool", "ToolCalls", "Code"]

dspy/adapters/types/file.py

@@ -0,0 +1,156 @@
+import base64
+import mimetypes
+import os
+from typing import Any
+
+import pydantic
+
+from dspy.adapters.types.base_type import Type
+
+
+class File(Type):
+    file_data: str | None = None
+    file_id: str | None = None
+    filename: str | None = None
+
+    model_config = pydantic.ConfigDict(
+        frozen=True,
+        str_strip_whitespace=True,
+        validate_assignment=True,
+        extra="forbid",
+    )
+
+    @pydantic.model_validator(mode="before")
+    @classmethod
+    def validate_input(cls, values: Any) -> Any:
+        if isinstance(values, cls):
+            return {
+                "file_data": values.file_data,
+                "file_id": values.file_id,
+                "filename": values.filename,
+            }
+
+        if isinstance(values, dict):
+            if "file_data" in values or "file_id" in values or "filename" in values:
+                return values
+            raise ValueError("Dict must contain at least one of: file_data, file_id, or filename")
+
+        return encode_file_to_dict(values)
+
+    def format(self) -> list[dict[str, Any]]:
+        try:
+            file_dict = {}
+            if self.file_data is not None:
+                file_dict["file_data"] = self.file_data
+            if self.file_id is not None:
+                file_dict["file_id"] = self.file_id
+            if self.filename is not None:
+                file_dict["filename"] = self.filename
+
+            return [{"type": "file", "file": file_dict}]
+        except Exception as e:
+            raise ValueError(f"Failed to format file for DSPy: {e}")
+
+    def __str__(self):
+        return self.serialize_model()
+
+    def __repr__(self):
+        parts = []
+        if self.file_data is not None:
+            if self.file_data.startswith("data:"):
+                mime_type = self.file_data.split(";")[0].split(":")[1]
+                len_data = len(self.file_data.split("base64,")[1]) if "base64," in self.file_data else len(self.file_data)
+                parts.append(f"file_data=<DATA_URI({mime_type}, {len_data} chars)>")
+            else:
+                len_data = len(self.file_data)
+                parts.append(f"file_data=<DATA({len_data} chars)>")
+        if self.file_id is not None:
+            parts.append(f"file_id='{self.file_id}'")
+        if self.filename is not None:
+            parts.append(f"filename='{self.filename}'")
+        return f"File({', '.join(parts)})"
+
+    @classmethod
+    def from_path(cls, file_path: str, filename: str | None = None, mime_type: str | None = None) -> "File":
+        """Create a File from a local file path.
+
+        Args:
+            file_path: Path to the file to read
+            filename: Optional filename to use (defaults to basename of path)
+            mime_type: Optional MIME type (defaults to auto-detection from file extension)
+        """
+        if not os.path.isfile(file_path):
+            raise ValueError(f"File not found: {file_path}")
+
+        with open(file_path, "rb") as f:
+            file_bytes = f.read()
+
+        if filename is None:
+            filename = os.path.basename(file_path)
+
+        if mime_type is None:
+            mime_type, _ = mimetypes.guess_type(file_path)
+            if mime_type is None:
+                mime_type = "application/octet-stream"
+
+        encoded_data = base64.b64encode(file_bytes).decode("utf-8")
+        file_data = f"data:{mime_type};base64,{encoded_data}"
+
+        return cls(file_data=file_data, filename=filename)
+
+    @classmethod
+    def from_bytes(cls, file_bytes: bytes, filename: str | None = None, mime_type: str = "application/octet-stream") -> "File":
+        """Create a File from raw bytes.
+
+        Args:
+            file_bytes: Raw bytes of the file
+            filename: Optional filename
+            mime_type: MIME type (defaults to 'application/octet-stream')
+        """
+        encoded_data = base64.b64encode(file_bytes).decode("utf-8")
+        file_data = f"data:{mime_type};base64,{encoded_data}"
+        return cls(file_data=file_data, filename=filename)
+
+    @classmethod
+    def from_file_id(cls, file_id: str, filename: str | None = None) -> "File":
+        """Create a File from an uploaded file ID."""
+        return cls(file_id=file_id, filename=filename)
+
+
+def encode_file_to_dict(file_input: Any) -> dict:
+    """
+    Encode various file inputs to a dict with file_data, file_id, and/or filename.
+
+    Args:
+        file_input: Can be a file path (str), bytes, or File instance.
+
+    Returns:
+        dict: A dictionary with file_data, file_id, and/or filename keys.
+    """
+    if isinstance(file_input, File):
+        result = {}
+        if file_input.file_data is not None:
+            result["file_data"] = file_input.file_data
+        if file_input.file_id is not None:
+            result["file_id"] = file_input.file_id
+        if file_input.filename is not None:
+            result["filename"] = file_input.filename
+        return result
+
+    elif isinstance(file_input, str):
+        if os.path.isfile(file_input):
+            file_obj = File.from_path(file_input)
+        else:
+            raise ValueError(f"Unrecognized file string: {file_input}; must be a valid file path")
+
+        return {
+            "file_data": file_obj.file_data,
+            "filename": file_obj.filename,
+        }
+
+    elif isinstance(file_input, bytes):
+        file_obj = File.from_bytes(file_input)
+        return {"file_data": file_obj.file_data}
+
+    else:
+        raise ValueError(f"Unsupported file input type: {type(file_input)}")