Skip to content

Anthropic

Anthropic

Bases: BaseLLM

Anthropic LLM node.

This class provides an implementation for the Anthropic Language Model node.

Attributes:

Name Type Description
connection Anthropic | None

The connection to use for the Anthropic LLM.
cache_control AnthropicCacheControl | None

The cache control configuration.
Source code in dynamiq/nodes/llms/anthropic.py 
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class Anthropic(BaseLLM):
    """Anthropic LLM node.

    This class provides an implementation for the Anthropic Language Model node.

    Attributes:
        connection (AnthropicConnection | None): The connection to use for the Anthropic LLM.
        cache_control (AnthropicCacheControl | None): The cache control configuration.
    """
    connection: AnthropicConnection | None = None
    MODEL_PREFIX = "anthropic/"
    cache_control: AnthropicCacheControl | None = None

    def __init__(self, **kwargs):
        """Initialize the Anthropic LLM node.

        Args:
            **kwargs: Additional keyword arguments.
        """
        if kwargs.get("client") is None and kwargs.get("connection") is None:
            kwargs["connection"] = AnthropicConnection()
        super().__init__(**kwargs)

    @staticmethod
    def _convert_non_image_to_file_content(messages: list[dict]) -> list[dict]:
        for message in messages:
            content = message.get("content")
            if not isinstance(content, list):
                continue

            new_content = []
            for item in content:
                if isinstance(item, dict) and item.get("type") == VisionMessageType.IMAGE_URL and "image_url" in item:
                    url = item["image_url"].get("url", "")
                    if url.startswith("data:") and not url.startswith("data:image/"):
                        logger.debug("Anthropic: converting non-image image_url to file content format")
                        new_content.append(
                            {
                                "type": VisionMessageType.FILE,
                                "file": {"file_data": url},
                            }
                        )
                    else:
                        new_content.append(item)
                else:
                    new_content.append(item)

            message["content"] = new_content

        return messages

    def get_messages(self, prompt, input_data) -> list[dict]:
        """
        Format messages and convert non-image files to Anthropic file content format.
        """
        messages = super().get_messages(prompt, input_data)
        return self._convert_non_image_to_file_content(messages)

    def update_completion_params(self, params: dict[str, Any]) -> dict[str, Any]:
        """Attach Anthropic prompt caching configuration to completion params."""
        params = super().update_completion_params(params)
        if self.cache_control:
            params.setdefault("cache_control_injection_points", []).append(
                {
                    "location": "message",
                    "index": self.cache_control.cache_injection_point_index,
                    "control": self.cache_control.model_dump(
                        exclude_none=True,
                        exclude={"cache_injection_point_index"},
                    ),
                }
            )
        return params

__init__(**kwargs)

Initialize the Anthropic LLM node.

Parameters:

Name Type Description Default
**kwargs

Additional keyword arguments.

 {}
Source code in dynamiq/nodes/llms/anthropic.py 
32
33
34
35
36
37
38
39
40
def __init__(self, **kwargs):
    """Initialize the Anthropic LLM node.

    Args:
        **kwargs: Additional keyword arguments.
    """
    if kwargs.get("client") is None and kwargs.get("connection") is None:
        kwargs["connection"] = AnthropicConnection()
    super().__init__(**kwargs)

get_messages(prompt, input_data)

Format messages and convert non-image files to Anthropic file content format.

Source code in dynamiq/nodes/llms/anthropic.py 
70
71
72
73
74
75
def get_messages(self, prompt, input_data) -> list[dict]:
    """
    Format messages and convert non-image files to Anthropic file content format.
    """
    messages = super().get_messages(prompt, input_data)
    return self._convert_non_image_to_file_content(messages)

update_completion_params(params)

Attach Anthropic prompt caching configuration to completion params.

Source code in dynamiq/nodes/llms/anthropic.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def update_completion_params(self, params: dict[str, Any]) -> dict[str, Any]:
    """Attach Anthropic prompt caching configuration to completion params."""
    params = super().update_completion_params(params)
    if self.cache_control:
        params.setdefault("cache_control_injection_points", []).append(
            {
                "location": "message",
                "index": self.cache_control.cache_injection_point_index,
                "control": self.cache_control.model_dump(
                    exclude_none=True,
                    exclude={"cache_injection_point_index"},
                ),
            }
        )
    return params

AnthropicCacheControl

Bases: BaseModel

Anthropic prompt caching configuration.

Source code in dynamiq/nodes/llms/anthropic.py 
11
12
13
14
15
16
class AnthropicCacheControl(BaseModel):
    """Anthropic prompt caching configuration."""

    type: Literal["ephemeral"] = "ephemeral"
    ttl: Literal["5m", "1h"] | None = "5m"
    cache_injection_point_index: int = -2