Skip to content

Anthropic Backend

AnthropicBackend

Bases: OpenAIBackendMixin

Backend for interacting with Anthropic's API.

Attributes:

Name Type Description
api_key

The API key for the Anthropic service.
client

The client for the Anthropic service.
serialized

The serializer for Anthropic-specific data formats.
Source code in mbodied/agents/backends/anthropic_backend.py 
 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
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class AnthropicBackend(OpenAIBackendMixin):
    """Backend for interacting with Anthropic's API.

    Attributes:
        api_key: The API key for the Anthropic service.
        client: The client for the Anthropic service.
        serialized: The serializer for Anthropic-specific data formats.
    """

    DEFAULT_MODEL = "claude-3-5-sonnet-20240620"
    INITIAL_CONTEXT = [
        Message(
            role="user", content="Imagine you are a robot with advanced spatial reasoning."),
        Message(role="assistant", content="Got it!"),
    ]

    def __init__(self, api_key: str | None, client: anthropic.Anthropic | None = None, **kwargs):
        """Initializes the AnthropicBackend with the given API key and client.

        Args:
            api_key: The API key for the Anthropic service.
            client: An optional client for the Anthropic service.
            kwargs: Additional keyword arguments.
        """
        self.api_key = api_key
        self.client = client

        self.model = kwargs.pop("model", self.DEFAULT_MODEL)
        if self.client is None:
            self.client = anthropic.Anthropic(api_key=self.api_key)
        self.serialized = AnthropicSerializer

    def predict(
        self,
        message: Message,
        context: List[Message] | None = None,
        model: str | None = "claude-3-5-sonnet-20240620",
        **kwargs,
    ) -> str:
        """Creates a completion for the given messages using the Anthropic API.

        Args:
            message: the message to be sent to the completion API.
            context: The context for the completion.
            model: The model to be used for the completion.
            **kwargs: Additional keyword arguments.

        Returns:
            str: The content of the completion response.
        """
        if model is None:
            model = self.DEFAULT_MODEL
        serialized_messages = [self.serialized(
            msg).serialize() for msg in context + [message]]
        completion = self.client.messages.create(
            model=model,
            messages=serialized_messages,
            max_tokens=1024,
            **kwargs,
        )
        return completion.content[0].text

    async def async_predict(
        self, message: Message, context: List[Message] | None = None, model: Any | None = None
    ) -> str:
        """Asynchronously predict the next message in the conversation."""
        # For now, we'll use the synchronous method since Anthropic doesn't provide an async API
        return self.predict(message, context, model)

    def stream(self, message: Message, context: List[Message] = None, model: str = "claude-3-5-sonnet-20240620", **kwargs):
        """Streams a completion for the given messages using the Anthropic API standard.

        Args:
            message: Message to be sent to the completion API.
            context: The context of the messages.
            model: The model to be used for the completion.
            **kwargs: Additional keyword arguments.
        """
        model = model or self.DEFAULT_MODEL
        serialized_messages = [self.serialized(
            msg).serialize() for msg in context + [message]]
        with self.client.messages.stream(
            max_tokens=1024,
            messages=serialized_messages,
            model=model,
            **kwargs,
        ) as stream:
            for text in stream.text_stream:
                yield text or ""

__init__(api_key, client=None, **kwargs)

Initializes the AnthropicBackend with the given API key and client.

Parameters:

Name Type Description Default
api_key str | None

The API key for the Anthropic service.

 required
client Anthropic | None

An optional client for the Anthropic service.

 None
kwargs

Additional keyword arguments.

 {}
Source code in mbodied/agents/backends/anthropic_backend.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def __init__(self, api_key: str | None, client: anthropic.Anthropic | None = None, **kwargs):
    """Initializes the AnthropicBackend with the given API key and client.

    Args:
        api_key: The API key for the Anthropic service.
        client: An optional client for the Anthropic service.
        kwargs: Additional keyword arguments.
    """
    self.api_key = api_key
    self.client = client

    self.model = kwargs.pop("model", self.DEFAULT_MODEL)
    if self.client is None:
        self.client = anthropic.Anthropic(api_key=self.api_key)
    self.serialized = AnthropicSerializer

async_predict(message, context=None, model=None) async

Asynchronously predict the next message in the conversation.

Source code in mbodied/agents/backends/anthropic_backend.py 
122
123
124
125
126
127
async def async_predict(
    self, message: Message, context: List[Message] | None = None, model: Any | None = None
) -> str:
    """Asynchronously predict the next message in the conversation."""
    # For now, we'll use the synchronous method since Anthropic doesn't provide an async API
    return self.predict(message, context, model)

predict(message, context=None, model='claude-3-5-sonnet-20240620', **kwargs)

Creates a completion for the given messages using the Anthropic API.

Parameters:

Name Type Description Default
message Message

the message to be sent to the completion API.

 required
context List[Message] | None

The context for the completion.

 None
model str | None

The model to be used for the completion.

 'claude-3-5-sonnet-20240620'
**kwargs

Additional keyword arguments.

 {}

Returns:

Name Type Description
str str

The content of the completion response.
Source code in mbodied/agents/backends/anthropic_backend.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def predict(
    self,
    message: Message,
    context: List[Message] | None = None,
    model: str | None = "claude-3-5-sonnet-20240620",
    **kwargs,
) -> str:
    """Creates a completion for the given messages using the Anthropic API.

    Args:
        message: the message to be sent to the completion API.
        context: The context for the completion.
        model: The model to be used for the completion.
        **kwargs: Additional keyword arguments.

    Returns:
        str: The content of the completion response.
    """
    if model is None:
        model = self.DEFAULT_MODEL
    serialized_messages = [self.serialized(
        msg).serialize() for msg in context + [message]]
    completion = self.client.messages.create(
        model=model,
        messages=serialized_messages,
        max_tokens=1024,
        **kwargs,
    )
    return completion.content[0].text

stream(message, context=None, model='claude-3-5-sonnet-20240620', **kwargs)

Streams a completion for the given messages using the Anthropic API standard.

Parameters:

Name Type Description Default
message Message

Message to be sent to the completion API.

 required
context List[Message]

The context of the messages.

 None
model str

The model to be used for the completion.

 'claude-3-5-sonnet-20240620'
**kwargs

Additional keyword arguments.

 {}
Source code in mbodied/agents/backends/anthropic_backend.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def stream(self, message: Message, context: List[Message] = None, model: str = "claude-3-5-sonnet-20240620", **kwargs):
    """Streams a completion for the given messages using the Anthropic API standard.

    Args:
        message: Message to be sent to the completion API.
        context: The context of the messages.
        model: The model to be used for the completion.
        **kwargs: Additional keyword arguments.
    """
    model = model or self.DEFAULT_MODEL
    serialized_messages = [self.serialized(
        msg).serialize() for msg in context + [message]]
    with self.client.messages.stream(
        max_tokens=1024,
        messages=serialized_messages,
        model=model,
        **kwargs,
    ) as stream:
        for text in stream.text_stream:
            yield text or ""

AnthropicSerializer

Bases: Serializer

Serializer for Anthropic-specific data formats.

Source code in mbodied/agents/backends/anthropic_backend.py 
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
class AnthropicSerializer(Serializer):
    """Serializer for Anthropic-specific data formats."""

    @classmethod
    def serialize_image(cls, image: Image) -> dict[str, Any]:
        """Serializes an image to the Anthropic format.

        Args:
            image: The image to be serialized.

        Returns:
            A dictionary representing the serialized image.
        """
        return {
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": f"image/{image.encoding}",
                "data": image.base64,
            },
        }

    @classmethod
    def serialize_text(cls, text: str) -> dict[str, Any]:
        """Serializes a text string to the Anthropic format.

        Args:
            text: The text to be serialized.

        Returns:
            A dictionary representing the serialized text.
        """
        return {"type": "text", "text": text}

serialize_image(image) classmethod

Serializes an image to the Anthropic format.

Parameters:

Name Type Description Default
image Image

The image to be serialized.

 required

Returns:

Type Description
dict[str, Any]

A dictionary representing the serialized image.
Source code in mbodied/agents/backends/anthropic_backend.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@classmethod
def serialize_image(cls, image: Image) -> dict[str, Any]:
    """Serializes an image to the Anthropic format.

    Args:
        image: The image to be serialized.

    Returns:
        A dictionary representing the serialized image.
    """
    return {
        "type": "image",
        "source": {
            "type": "base64",
            "media_type": f"image/{image.encoding}",
            "data": image.base64,
        },
    }

serialize_text(text) classmethod

Serializes a text string to the Anthropic format.

Parameters:

Name Type Description Default
text str

The text to be serialized.

 required

Returns:

Type Description
dict[str, Any]

A dictionary representing the serialized text.
Source code in mbodied/agents/backends/anthropic_backend.py 
47
48
49
50
51
52
53
54
55
56
57
@classmethod
def serialize_text(cls, text: str) -> dict[str, Any]:
    """Serializes a text string to the Anthropic format.

    Args:
        text: The text to be serialized.

    Returns:
        A dictionary representing the serialized text.
    """
    return {"type": "text", "text": text}