Skip to content

anthropic_factory

Module for creating an Anthropic factory.

AnthropicFactory

Bases: ClientFactory

An Anthropic factory.

Source code in spark_instructor/factory/anthropic_factory.py 
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
class AnthropicFactory(ClientFactory):
    """An Anthropic factory."""

    @classmethod
    def from_config(
        cls,
        mode: Optional[instructor.Mode] = None,
        base_url: Optional[str] = None,
        api_key: Optional[str] = None,
        **kwargs,
    ) -> "AnthropicFactory":
        """Get an anthropic client."""
        return cls(
            get_anthropic_aclient(
                mode or instructor.Mode.ANTHROPIC_TOOLS, base_url, api_key, kwargs.get("enable_caching", False)
            )
        )

    def format_completion(self, completion: Message) -> ChatCompletion:
        """Format an Anthropic ``Message`` as an OpenAI ``ChatCompletion``."""
        return transform_message_to_chat_completion(completion, enable_created_at=True)

    def format_messages(
        self, messages: SparkChatCompletionMessages, ignore_system: bool = False
    ) -> List[ChatCompletionMessageParam]:
        """Format messages so that images serialize properly."""
        messages_unpacked = (
            [message() for message in messages.root if message.role != "system"]
            if ignore_system
            else [message() for message in messages.root]
        )
        for j, message in enumerate(messages_unpacked):
            if message["role"] == "user":
                casted_message = cast(ChatCompletionUserMessageParam, message)
                if isinstance(casted_message["content"], list):
                    for i, content in enumerate(casted_message["content"]):
                        casted_content = cast(ChatCompletionContentPartParam, content)
                        if casted_content["type"] == "image_url":
                            casted_content_image = cast(ChatCompletionContentPartImageParam, casted_content)
                            casted_message["content"][i] = convert_image_url_to_image_block_param(
                                casted_content_image["image_url"]
                            )
                    if messages.root[j].cache_control:
                        casted_message["content"][-1]["cache_control"] = {"type": "ephemeral"}
            elif message["role"] == "system" and messages.root[j].cache_control and isinstance(message["content"], str):
                message["content"] = [
                    {"type": "text", "text": message["content"], "cache_control": {"type": "ephemeral"}}  # type: ignore
                ]  # type: ignore
        return messages_unpacked

format_completion(completion)

Format an Anthropic Message as an OpenAI ChatCompletion.

Source code in spark_instructor/factory/anthropic_factory.py 
48
49
50
def format_completion(self, completion: Message) -> ChatCompletion:
    """Format an Anthropic ``Message`` as an OpenAI ``ChatCompletion``."""
    return transform_message_to_chat_completion(completion, enable_created_at=True)

format_messages(messages, ignore_system=False)

Format messages so that images serialize properly.

Source code in spark_instructor/factory/anthropic_factory.py 
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
def format_messages(
    self, messages: SparkChatCompletionMessages, ignore_system: bool = False
) -> List[ChatCompletionMessageParam]:
    """Format messages so that images serialize properly."""
    messages_unpacked = (
        [message() for message in messages.root if message.role != "system"]
        if ignore_system
        else [message() for message in messages.root]
    )
    for j, message in enumerate(messages_unpacked):
        if message["role"] == "user":
            casted_message = cast(ChatCompletionUserMessageParam, message)
            if isinstance(casted_message["content"], list):
                for i, content in enumerate(casted_message["content"]):
                    casted_content = cast(ChatCompletionContentPartParam, content)
                    if casted_content["type"] == "image_url":
                        casted_content_image = cast(ChatCompletionContentPartImageParam, casted_content)
                        casted_message["content"][i] = convert_image_url_to_image_block_param(
                            casted_content_image["image_url"]
                        )
                if messages.root[j].cache_control:
                    casted_message["content"][-1]["cache_control"] = {"type": "ephemeral"}
        elif message["role"] == "system" and messages.root[j].cache_control and isinstance(message["content"], str):
            message["content"] = [
                {"type": "text", "text": message["content"], "cache_control": {"type": "ephemeral"}}  # type: ignore
            ]  # type: ignore
    return messages_unpacked

from_config(mode=None, base_url=None, api_key=None, **kwargs) classmethod

Get an anthropic client.

Source code in spark_instructor/factory/anthropic_factory.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@classmethod
def from_config(
    cls,
    mode: Optional[instructor.Mode] = None,
    base_url: Optional[str] = None,
    api_key: Optional[str] = None,
    **kwargs,
) -> "AnthropicFactory":
    """Get an anthropic client."""
    return cls(
        get_anthropic_aclient(
            mode or instructor.Mode.ANTHROPIC_TOOLS, base_url, api_key, kwargs.get("enable_caching", False)
        )
    )