dspy.Image¶

@classmethod
def adapt_to_native_lm_feature(
    cls,
    signature: type["Signature"],
    field_name: str,
    lm: BaseLM,
    lm_kwargs: dict[str, Any],
) -> type["Signature"]:
    """Adapt the custom type to the native LM feature if possible.

    When the LM and configuration supports the related native LM feature, e.g., native tool calling, native
    reasoning, etc., we adapt the signature and `lm_kwargs` to enable the native LM feature.

    Args:
        signature: The DSPy signature for the LM call.
        field_name: The name of the field in the signature to adapt to the native LM feature.
        lm: The LM instance.
        lm_kwargs: The keyword arguments for the LM call, subject to in-place updates if adaptation if required.

    Returns:
        The adapted signature. If the custom type is not natively supported by the LM, return the original
        signature.
    """
    return signature

@classmethod
def description(cls) -> str:
    """Description of the custom type"""
    return ""

@classmethod
def extract_custom_type_from_annotation(cls, annotation):
    """Extract all custom types from the annotation.

    This is used to extract all custom types from the annotation of a field, while the annotation can
    have arbitrary level of nesting. For example, we detect `Tool` is in `list[dict[str, Tool]]`.
    """
    # Direct match. Nested type like `list[dict[str, Event]]` passes `isinstance(annotation, type)` in python 3.10
    # while fails in python 3.11. To accommodate users using python 3.10, we need to capture the error and ignore it.
    try:
        if isinstance(annotation, type) and issubclass(annotation, cls):
            return [annotation]
    except TypeError:
        pass

    origin = get_origin(annotation)
    if origin is None:
        return []

    result = []
    # Recurse into all type args
    for arg in get_args(annotation):
        result.extend(cls.extract_custom_type_from_annotation(arg))

    return result

@lru_cache(maxsize=32)
def format(self) -> list[dict[str, Any]] | str:
    try:
        image_url = encode_image(self.url)
    except Exception as e:
        raise ValueError(f"Failed to format image for DSPy: {e}")
    return [{"type": "image_url", "image_url": {"url": image_url}}]

@classmethod
def from_PIL(cls, pil_image):  # noqa: N802
    warnings.warn(
        "Image.from_PIL is deprecated and will be removed in 3.4; use Image(pil_image) instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return cls(pil_image)

@classmethod
def from_file(cls, file_path: str) -> "Image":
    """Deprecated alias for :meth:`from_path`."""
    warnings.warn(
        "Image.from_file is deprecated and will be removed in 3.4; use Image.from_path instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return cls.from_path(file_path)

@classmethod
def from_path(cls, file_path: str) -> "Image":
    """Read a local file and encode it as a data URI."""
    if not os.path.isfile(file_path):
        raise ValueError(f"File not found: {file_path}")
    return cls(_encode_image_from_file(file_path))

@classmethod
def from_url(cls, url: str, verify: bool = True, timeout: float | None = 30.0) -> "Image":
    """Download an HTTP(S) resource and encode it as a data URI.

    Security: this performs an explicit, caller-initiated fetch and applies no
    SSRF protection beyond requiring an HTTP(S) scheme. Like ``requests.get``, it
    will reach private, loopback, or cloud-metadata hosts and follow redirects to
    them. When ``url`` is derived from untrusted input, the caller is responsible
    for validating the host against an allowlist before calling this method.
    """
    if not _is_http_url(url):
        raise ValueError(f"Image.from_url requires an HTTP(S) URL, received: {url}")
    return cls(_encode_image_from_url(url, verify=verify, timeout=timeout))

@classmethod
def is_streamable(cls) -> bool:
    """Whether the custom type is streamable."""
    return False

@classmethod
def parse_lm_response(cls, response: str | dict[str, Any]) -> Optional["Type"]:
    """Parse a LM response into the custom type.

    Args:
        response: A LM response.

    Returns:
        A custom type object.
    """
    return None

@classmethod
def parse_stream_chunk(cls, chunk: "ModelResponseStream") -> Optional["Type"]:
    """
    Parse a stream chunk into the custom type.

    Args:
        chunk: A stream chunk.

    Returns:
        A custom type object or None if the chunk is not for this custom type.
    """
    return None

@pydantic.model_serializer()
def serialize_model(self):
    formatted = self.format()
    if isinstance(formatted, list):
        return (
            f"{CUSTOM_TYPE_START_IDENTIFIER}{json.dumps(formatted, ensure_ascii=False)}{CUSTOM_TYPE_END_IDENTIFIER}"
        )
    return formatted