Accept images in decisions

examples/models/openai/decisions_images.py

@@ -0,0 +1,87 @@
+"""Ask decision questions about images with OpenAI ``gpt-6-luna``.
+
+Set ``OPENAI_API_KEY`` and run:
+
+    uv run python examples/models/openai/decisions_images.py
+
+OpenAI decisions take images in two places:
+
+* in the state, as a list of text and file parts (one user message), and
+* in a question's instructions, to give one question its own image.
+
+OpenAI does not accept images nested inside JSON state or in criteria.
+"""
+
+import asyncio
+import pathlib
+
+import pydantic
+
+import ai
+
+
+class DeliveryQuestions(pydantic.BaseModel):
+    damaged: ai.ops.experimental.PredicateQuestion
+    condition: ai.ops.experimental.ScoreQuestion
+    matches_reference: ai.ops.experimental.PredicateQuestion
+
+
+class DeliveryAnswers(pydantic.BaseModel):
+    damaged: ai.ops.experimental.PredicateAnswer
+    condition: ai.ops.experimental.ScoreAnswer
+    matches_reference: ai.ops.experimental.PredicateAnswer
+
+
+async def main() -> None:
+    model = ai.get_model("openai:gpt-6-luna")
+    if not model.provider.is_configured():
+        print("Set OPENAI_API_KEY to run this example.")
+        return
+
+    image_path = (
+        pathlib.Path(__file__).parents[2] / "media" / "sample_image.jpg"
+    )
+
+    # Shared evidence for all questions: text and an image, like the content
+    # of a user message.
+    state: ai.ops.experimental.DecisionInput = [
+        "A customer uploaded this photo of a delivered item. This is photo A.",
+        ai.file_part(image_path.read_bytes()),
+    ]
+
+    result = await ai.ops.experimental.decide(
+        model,
+        state,
+        DeliveryQuestions(
+            damaged=ai.ops.experimental.PredicateQuestion(
+                instructions=(
+                    "Does the item in photo A have visible damage, such as a "
+                    "crack, tear, or dent? Ignore shadows."
+                ),
+            ),
+            condition=ai.ops.experimental.ScoreQuestion(
+                instructions="What is the condition of the item in photo A?",
+                criteria=["Broken", "Worn", "Like new"],
+            ),
+            # This question has its own image. Here it is the same file, so
+            # the answer should be close to 100%.
+            matches_reference=ai.ops.experimental.PredicateQuestion(
+                instructions=[
+                    "Is candidate photo B a copy or variant of photo A, "
+                    "including recompression or resizing?",
+                    ai.file_part(image_path.read_bytes()),
+                ],
+            ),
+        ),
+        output_type=DeliveryAnswers,
+    )
+
+    answers = result.value
+    print("damaged:", f"{answers.damaged.probability:.0%}")
+    print("condition:", answers.condition.score)
+    print("matches reference:", f"{answers.matches_reference.probability:.0%}")
+    print("usage:", result.usage)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/models/typesafe/decisions_images.py

@@ -0,0 +1,97 @@
+"""Ask decision questions about images with TypeSafe.
+
+Set ``TYPESAFE_API_KEY`` and run:
+
+    uv run python examples/models/typesafe/decisions_images.py
+
+Images are file parts. TypeSafe accepts them anywhere text can go: inside
+JSON state, in a question's instructions, and in its criteria. Questions
+refer to state fields by name in backticks.
+"""
+
+import asyncio
+import pathlib
+
+import pydantic
+
+import ai
+from ai.providers.typesafe import TypeSafeProvider
+
+
+class ListingQuestions(pydantic.BaseModel):
+    matches_listing: ai.ops.experimental.PredicateQuestion
+    lighting: ai.ops.experimental.ChoiceQuestion
+    photo_quality: ai.ops.experimental.ScoreQuestion
+
+
+class ListingAnswers(pydantic.BaseModel):
+    matches_listing: ai.ops.experimental.PredicateAnswer
+    lighting: ai.ops.experimental.ChoiceAnswer
+    photo_quality: ai.ops.experimental.ScoreAnswer
+
+
+async def main() -> None:
+    model = ai.Model(id="jev-latest", provider=TypeSafeProvider())
+    if not model.provider.is_configured():
+        print("Set TYPESAFE_API_KEY to run this example.")
+        return
+
+    image_path = (
+        pathlib.Path(__file__).parents[2] / "media" / "sample_image.jpg"
+    )
+    photo = ai.file_part(image_path.read_bytes(), media_type="image/jpeg")
+
+    # The image sits inside structured state, next to the data it belongs to.
+    listing: ai.ops.experimental.DecisionInput = {
+        "listing": {
+            "title": "Scenic print, framed",
+            "price_usd": 120,
+            "photo": photo,
+        },
+        "buyer_report": {
+            "message": "This is what arrived.",
+            "photo": photo,
+        },
+    }
+
+    result = await ai.ops.experimental.decide(
+        model,
+        listing,
+        ListingQuestions(
+            matches_listing=ai.ops.experimental.PredicateQuestion(
+                instructions=(
+                    "Does `buyer_report.photo` show the same item as "
+                    "`listing.photo`?"
+                ),
+            ),
+            lighting=ai.ops.experimental.ChoiceQuestion(
+                instructions="What is the lighting in `listing.photo` like?",
+                criteria={
+                    "daylight": "Natural daylight",
+                    "artificial": "Lamps or other artificial light",
+                    "dark": "Little or no light",
+                },
+            ),
+            # A criterion can be an image: here, the listing photo is the
+            # example of the top quality level.
+            photo_quality=ai.ops.experimental.ScoreQuestion(
+                instructions="How good is `buyer_report.photo` as evidence?",
+                criteria=[
+                    "Unusable: blurry, dark, or the item is not visible",
+                    "Usable: the item is visible but details are unclear",
+                    ["As clear as this reference photo:", photo],
+                ],
+            ),
+        ),
+        output_type=ListingAnswers,
+    )
+
+    answers = result.value
+    print("matches listing:", f"{answers.matches_listing.probability:.0%}")
+    print("lighting:", answers.lighting.choice)
+    print("photo quality:", answers.photo_quality.score)
+    print("usage:", result.usage)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -45,7 +45,7 @@ vercel = [
     "vercel-oidc>=0.8.0",
 ]
 typesafe = [
-    "typesafe-sdk>=0.7.2,<1",
+    "typesafe-sdk>=0.7.5,<1",
 ]
 
 [project.urls]
@@ -82,7 +82,7 @@ dev = [
     "vercel-oidc>=0.8.0",
     "opentelemetry-sdk>=1.43.0",
     "opentelemetry-exporter-otlp-proto-http>=1.43.0",
-    "typesafe-sdk>=0.7.2",
+    "typesafe-sdk>=0.7.5",
 ]
 
 [tool.uv]

src/ai/ops/experimental/decisions.py

@@ -22,28 +22,116 @@ class Answers(pydantic.BaseModel):
         output_type=Answers,
     )
     result.value.requests_refund
+
+Images go anywhere in the state or question text as file parts::
+
+    state = {"photo": ai.file_part(photo_bytes), "note": "Delivered today."}
 """
 
 from __future__ import annotations
 
 import dataclasses
 import warnings
-from collections.abc import Iterable, Mapping
+from collections.abc import Iterable, Mapping, Sequence
 from typing import TYPE_CHECKING, Annotated, Any, Literal, TypedDict, overload
 
 import pydantic
 
 from ... import experimental_telemetry as telemetry
+from ...types import messages
 from .. import items
 
 if TYPE_CHECKING:
     from ...models.core import model as model_
 
 
-type DecisionInput = (
-    str | dict[str, pydantic.JsonValue] | list[pydantic.JsonValue]
+_JSON_TYPES = frozenset((str, int, float, bool, list, dict))
+
+
+def _decision_value_tag(value: Any) -> str:
+    # Like pydantic.JsonValue: pick one union member by type, so bad input
+    # gives one error at its path. Exact types take the fast path. Subclasses
+    # validate as their base type; bool goes before int because bool
+    # subclasses int. Other sequences and mappings validate as list and dict,
+    # except binary data, which is never a list of ints.
+    exact = type(value)
+    if exact in _JSON_TYPES:
+        return exact.__name__
+    for type_ in (bool, int, float, str, messages.FilePart):
+        if isinstance(value, type_):
+            return type_.__name__
+    if isinstance(value, Mapping):
+        return "dict"
+    if isinstance(value, Sequence) and not isinstance(
+        value, bytes | bytearray | memoryview
+    ):
+        return "list"
+    return "None" if value is None else "invalid"
+
+
+_VALUE_DISCRIMINATOR = pydantic.Discriminator(
+    _decision_value_tag,
+    custom_error_type="invalid-decision-value",
+    custom_error_message="input was not text, JSON, or a FilePart",
+)
+_INPUT_DISCRIMINATOR = pydantic.Discriminator(
+    _decision_value_tag,
+    custom_error_type="invalid-decision-input",
+    custom_error_message=(
+        "input was not text, a FilePart, a mapping, or a sequence"
+    ),
 )
 
+if TYPE_CHECKING:
+    # Read-only containers are covariant, so a ``list[str]`` or a
+    # ``dict[str, int]`` is accepted as is.
+    type _DecisionValue = (
+        str
+        | int
+        | float
+        | bool
+        | messages.FilePart
+        | Sequence[_DecisionValue]
+        | Mapping[str, _DecisionValue]
+        | None
+    )
+    type DecisionInput = (
+        str
+        | messages.FilePart
+        | Mapping[str, _DecisionValue]
+        | Sequence[_DecisionValue]
+    )
+    """Text, a file, or JSON data with files at any position.
+
+    Files are :class:`~ai.types.messages.FilePart` objects, e.g. from
+    :func:`ai.file_part`. A plain ``str`` is always text, never a URL. Any
+    sequence becomes a list and any mapping a dict.
+    """
+else:
+    # Validation stores every sequence as a list and every mapping as a dict.
+    _AS_LIST = pydantic.BeforeValidator(list)
+    _AS_DICT = pydantic.BeforeValidator(dict)
+
+    type _DecisionValue = Annotated[
+        Annotated[str, pydantic.Tag("str")]
+        | Annotated[int, pydantic.Tag("int")]
+        | Annotated[float, pydantic.Tag("float")]
+        | Annotated[bool, pydantic.Tag("bool")]
+        | Annotated[messages.FilePart, pydantic.Tag("FilePart")]
+        | Annotated[list[_DecisionValue], _AS_LIST, pydantic.Tag("list")]
+        | Annotated[dict[str, _DecisionValue], _AS_DICT, pydantic.Tag("dict")]
+        | Annotated[None, pydantic.Tag("None")],
+        _VALUE_DISCRIMINATOR,
+    ]
+
+    type DecisionInput = Annotated[
+        Annotated[str, pydantic.Tag("str")]
+        | Annotated[messages.FilePart, pydantic.Tag("FilePart")]
+        | Annotated[dict[str, _DecisionValue], _AS_DICT, pydantic.Tag("dict")]
+        | Annotated[list[_DecisionValue], _AS_LIST, pydantic.Tag("list")],
+        _INPUT_DISCRIMINATOR,
+    ]
+
 
 class PredicateCriteria(TypedDict, total=False):
     """Optional descriptions of when a predicate answer is true or false."""
@@ -241,6 +329,11 @@ async def decide(
     value, or pass a question mapping to receive an answer mapping. Each choice,
     score, or predicate question is validated against its corresponding answer.
 
+    ``state``, question ``instructions``, and ``criteria`` accept files
+    (:class:`~ai.types.messages.FilePart`) alongside text and JSON, e.g.
+    ``{"photo": ai.file_part(data), "note": "..."}``. Providers reject files
+    they cannot send.
+
     Experimental: not part of the stable API, may change or be removed.
     """
     # Validate the shared state before dispatching it to a provider.

src/ai/providers/ai_gateway/protocol/v4.py

@@ -546,6 +546,22 @@ async def decide(
     params: ops.experimental.DecideParams,
 ) -> ops.items.Item[dict[str, Any]]:
     """Hit ``/evaluation-model`` and return raw decision answers."""
+
+    # The v4 spec does not define files in decision requests yet.
+    def has_file(value: Any) -> bool:
+        if isinstance(value, types.messages.FilePart):
+            return True
+        if isinstance(value, dict):
+            return any(has_file(item) for item in value.values())
+        if isinstance(value, list):
+            return any(has_file(item) for item in value)
+        return False
+
+    if has_file(state) or any(
+        has_file(dict(question)) for question in questions.values()
+    ):
+        raise ValueError("AI Gateway decisions do not support files")
+
     wire_questions: dict[str, dict[str, Any]] = {}
     for question_id, question in questions.items():
         wire_question = question.model_dump(mode="json", by_alias=True)

src/ai/providers/typesafe/_sdk.py

@@ -12,6 +12,7 @@
 
 class TypeSafeSDK(Protocol):
     AsyncTypeSafeClient: type[typesafe_sdk.AsyncTypeSafeClient]
+    Image: type[typesafe_sdk.Image]
     TypeSafeError: type[typesafe_sdk.TypeSafeError]
     TypeSafeAPIError: type[typesafe_sdk.TypeSafeAPIError]
     TypeSafeAPIConnectionError: type[typesafe_sdk.TypeSafeAPIConnectionError]