mime_type and resolution parameters aded to the dspy.File type

dspy/adapters/types/file.py

@@ -36,6 +36,8 @@ class QA(dspy.Signature):
     file_data: str | None = None
     file_id: str | None = None
     filename: str | None = None
+    mime_type: str | None = None
+    resolution: str | None = None
 
     model_config = pydantic.ConfigDict(
         frozen=True,
@@ -66,6 +68,8 @@ def validate_input(cls, values: Any) -> Any:
                 "file_data": values.file_data,
                 "file_id": values.file_id,
                 "filename": values.filename,
+                "mime_type": values.mime_type,
+                "resolution": values.resolution
             }
 
         if isinstance(values, dict):
@@ -84,6 +88,10 @@ def format(self) -> list[dict[str, Any]]:
                 file_dict["file_id"] = self.file_id
             if self.filename:
                 file_dict["filename"] = self.filename
+            if self.mime_type:
+                file_dict["format"] = self.mime_type
+            if self.resolution:
+                file_dict["resolution"] = self.resolution
 
             return [{"type": "file", "file": file_dict}]
         except Exception as e:
@@ -98,9 +106,7 @@ def __repr__(self):
             if self.file_data.startswith("data:"):
                 # file data has "data:text/plain;base64,..." format
                 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)
-                )
+                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)
@@ -109,16 +115,21 @@ def __repr__(self):
             parts.append(f"file_id='{self.file_id}'")
         if self.filename is not None:
             parts.append(f"filename='{self.filename}'")
+        if self.mime_type is not None:
+            parts.append(f"mime_type='{self.mime_type}'")
+        if self.resolution is not None:
+            parts.append(f"resolution='{self.resolution}'")
         return f"File({', '.join(parts)})"
 
     @classmethod
-    def from_path(cls, file_path: str, filename: str | None = None, mime_type: str | None = None) -> "File":
+    def from_path(cls, file_path: str, filename: str | None = None, mime_type: str | None = None, resolution: 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)
+            resolution: Optional resolution used to rescale the file by some providers
         """
         if not os.path.isfile(file_path):
             raise ValueError(f"File not found: {file_path}")
@@ -137,11 +148,13 @@ def from_path(cls, file_path: str, filename: str | None = None, mime_type: str |
         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)
+        return cls(file_data=file_data, filename=filename, resolution=resolution)
 
     @classmethod
     def from_bytes(
-        cls, file_bytes: bytes, filename: str | None = None, mime_type: str = "application/octet-stream"
+        cls, file_bytes: bytes, filename: str | None = None,
+        mime_type: str = "application/octet-stream",
+        resolution:str = "medium"
     ) -> "File":
         """Create a File from raw bytes.
 
@@ -152,12 +165,22 @@ def from_bytes(
         """
         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)
+        return cls(file_data=file_data, filename=filename, resolution=resolution)
 
     @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 from_file_id(cls, file_id: str, filename: str | None = None,
+                mime_type: str | None = None,
+                resolution: str | None = None
+    ) -> "File":
+        """Create a File from an uploaded file ID.
+
+        Args:
+            file_id: The uploaded file ID
+            filename: Optional filename
+            mime_type: Optional MIME type (passed as 'format' to the API)
+            resolution: Optional file resolution (e.g., low, medium, high)
+        """
+        return cls(file_id=file_id, filename=filename, mime_type=mime_type, resolution=resolution)
 
 
 def encode_file_to_dict(file_input: Any) -> dict:
@@ -178,6 +201,10 @@ def encode_file_to_dict(file_input: Any) -> dict:
             result["file_id"] = file_input.file_id
         if file_input.filename is not None:
             result["filename"] = file_input.filename
+        if file_input.mime_type is not None:
+            result["format"] = file_input.mime_type
+        if file_input.resolution is not None:
+            result["resolution"] = file_input.resolution
         return result
 
     elif isinstance(file_input, dict):

dspy/clients/base_lm.py

@@ -54,10 +54,14 @@ def _import_lm_class(class_path: str) -> type:
             continue
 
         if not isinstance(obj, type):
-            raise TypeError(f"Serialized LM class `{class_path}` did not resolve to a class.")
+            raise TypeError(
+                f"Serialized LM class `{class_path}` did not resolve to a class."
+            )
         return obj
 
-    raise ImportError(f"Could not import serialized LM class `{class_path}`.") from last_error
+    raise ImportError(
+        f"Could not import serialized LM class `{class_path}`."
+    ) from last_error
 
 
 class BaseLM:
@@ -113,11 +117,15 @@ def __init__(
         self.cache = cache
         self.callbacks = list(callbacks or [])
         self.num_retries = num_retries
-        self.kwargs = self._get_initial_kwargs(temperature=temperature, max_tokens=max_tokens, **kwargs)
+        self.kwargs = self._get_initial_kwargs(
+            temperature=temperature, max_tokens=max_tokens, **kwargs
+        )
         self.history = []
         self._warned_zero_temp_rollout = False
 
-    def _get_initial_kwargs(self, *, temperature, max_tokens, **kwargs) -> dict[str, Any]:
+    def _get_initial_kwargs(
+        self, *, temperature, max_tokens, **kwargs
+    ) -> dict[str, Any]:
         return dict(temperature=temperature, max_tokens=max_tokens, **kwargs)
 
     @property
@@ -152,7 +160,9 @@ def _process_lm_response(self, response, prompt, messages, **kwargs):
 
     def _record_response(self, response, prompt, messages, outputs, kwargs, request=None):
         if not getattr(response, "cache_hit", False) and settings.usage_tracker:
-            settings.usage_tracker.add_usage(self.model, dict(getattr(response, "usage", {}) or {}))
+            settings.usage_tracker.add_usage(
+                self.model, dict(getattr(response, "usage", {}) or {})
+            )
 
         if settings.disable_history:
             return outputs
@@ -225,7 +235,11 @@ def dump_state(self) -> dict[str, Any]:
             A dictionary that can be passed to `BaseLM.load_state`. The state
             excludes API keys.
         """
-        filtered_kwargs = {key: value for key, value in self.kwargs.items() if key not in ("api_key", LM_CLASS_STATE_KEY)}
+        filtered_kwargs = {
+            key: value
+            for key, value in self.kwargs.items()
+            if key not in ("api_key", LM_CLASS_STATE_KEY)
+        }
         return {
             LM_CLASS_STATE_KEY: f"{type(self).__module__}.{type(self).__qualname__}",
             "model": self.model,
@@ -236,7 +250,9 @@ def dump_state(self) -> dict[str, Any]:
         }
 
     @classmethod
-    def load_state(cls, state: dict[str, Any], *, allow_custom_lm_class: bool = False) -> "BaseLM":
+    def load_state(
+        cls, state: dict[str, Any], *, allow_custom_lm_class: bool = False
+    ) -> "BaseLM":
         """Reconstruct an LM from `dump_state` output.
 
         Legacy states without a class marker load as `dspy.LM`. Custom LM
@@ -280,9 +296,16 @@ def load_state(cls, state: dict[str, Any], *, allow_custom_lm_class: bool = Fals
 
             lm_cls = _import_lm_class(class_path)
             if not issubclass(lm_cls, BaseLM):
-                raise TypeError(f"Serialized LM class `{class_path}` must be a subclass of dspy.BaseLM.")
-            if "allow_custom_lm_class" in inspect.signature(lm_cls.load_state).parameters:
-                return lm_cls.load_state(state, allow_custom_lm_class=allow_custom_lm_class)
+                raise TypeError(
+                    f"Serialized LM class `{class_path}` must be a subclass of dspy.BaseLM."
+                )
+            if (
+                "allow_custom_lm_class"
+                in inspect.signature(lm_cls.load_state).parameters
+            ):
+                return lm_cls.load_state(
+                    state, allow_custom_lm_class=allow_custom_lm_class
+                )
             return lm_cls.load_state(state)
 
         return cls(**state)