Skip to content

anthropic_image_query_driver

AnthropicImageQueryDriver

Bases: BaseImageQueryDriver

Anthropic Image Query Driver.

Attributes:

Name Type Description
api_key Optional[str]

Anthropic API key.
model str

Anthropic model name.
client Anthropic

Custom Anthropic client.
Source code in griptape/drivers/image_query/anthropic_image_query_driver.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@define
class AnthropicImageQueryDriver(BaseImageQueryDriver):
    """Anthropic Image Query Driver.

    Attributes:
        api_key: Anthropic API key.
        model: Anthropic model name.
        client: Custom `Anthropic` client.
    """

    api_key: Optional[str] = field(default=None, kw_only=True, metadata={"serializable": False})
    model: str = field(kw_only=True, metadata={"serializable": True})
    _client: Anthropic = field(default=None, kw_only=True, alias="client", metadata={"serializable": False})

    @lazy_property()
    def client(self) -> Anthropic:
        return import_optional_dependency("anthropic").Anthropic(api_key=self.api_key)

    def try_query(self, query: str, images: list[ImageArtifact]) -> TextArtifact:
        if self.max_tokens is None:
            raise TypeError("max_output_tokens can't be empty")

        response = self.client.messages.create(**self._base_params(query, images))
        content_blocks = response.content

        if len(content_blocks) < 1:
            raise ValueError("Response content is empty")

        text_content = content_blocks[0].text

        return TextArtifact(text_content)

    def _base_params(self, text_query: str, images: list[ImageArtifact]) -> dict:
        content = [self._construct_image_message(image) for image in images]
        content.append(self._construct_text_message(text_query))
        messages = self._construct_messages(content)
        return {"model": self.model, "messages": messages, "max_tokens": self.max_tokens}

    def _construct_image_message(self, image_data: ImageArtifact) -> dict:
        data = image_data.base64
        media_type = image_data.mime_type

        return {"source": {"data": data, "media_type": media_type, "type": "base64"}, "type": "image"}

    def _construct_text_message(self, query: str) -> dict:
        return {"text": query, "type": "text"}

    def _construct_messages(self, content: list) -> list:
        return [{"content": content, "role": "user"}]

api_key: Optional[str] = field(default=None, kw_only=True, metadata={'serializable': False}) class-attribute instance-attribute

model: str = field(kw_only=True, metadata={'serializable': True}) class-attribute instance-attribute

client()

Source code in griptape/drivers/image_query/anthropic_image_query_driver.py 
30
31
32
@lazy_property()
def client(self) -> Anthropic:
    return import_optional_dependency("anthropic").Anthropic(api_key=self.api_key)

try_query(query, images)

Source code in griptape/drivers/image_query/anthropic_image_query_driver.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
def try_query(self, query: str, images: list[ImageArtifact]) -> TextArtifact:
    if self.max_tokens is None:
        raise TypeError("max_output_tokens can't be empty")

    response = self.client.messages.create(**self._base_params(query, images))
    content_blocks = response.content

    if len(content_blocks) < 1:
        raise ValueError("Response content is empty")

    text_content = content_blocks[0].text

    return TextArtifact(text_content)