mbodied.agents.backends package

Submodules

mbodied.agents.backends.anthropic_backend module

class mbodied.agents.backends.anthropic_backend.AnthropicBackend(api_key: str | None, client: Anthropic | None = None, **kwargs)

Bases: OpenAIBackendMixin

Backend for interacting with Anthropic’s API.

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!'])]

class mbodied.agents.backends.anthropic_backend.AnthropicSerializer(wrapped: Message | Sample | list[Message] | None = None, *, message: Message | None = None, sample: Sample | None = None, **data)

Bases: Serializer

Serializer for Anthropic-specific data formats.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ConfigDict = {'arbitrary_types_allowed': True, 'extra': 'allow', 'from_attributes': True, 'use_enum_values': False, 'validate_assignment': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'wrapped': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod serialize_image(image: Image) → dict[str, Any]

Serializes an image to the Anthropic format.

Parameters:

image – The image to be serialized.

Returns:

A dictionary representing the serialized image.

classmethod serialize_text(text: str) → dict[str, Any]

Serializes a text string to the Anthropic format.

Parameters:

text – The text to be serialized.

Returns:

A dictionary representing the serialized text.

mbodied.agents.backends.gradio_backend module

class mbodied.agents.backends.gradio_backend.GradioBackend(model_src: str = None, **kwargs)

Bases: Backend

Gradio backend that handles connections to gradio servers.

predict(*args, **kwargs) → str

Forward queries to the gradio api endpoint predict.

Parameters:

• *args – The arguments to pass to the gradio server.

• **kwargs – The keywrod arguments to pass to the gradio server.

submit(*args, api_name='/predict', result_callbacks=None, **kwargs) → Job

Submit queries asynchronously without need of asyncio.

Parameters:

• *args – The arguments to pass to the gradio server.

• api_name – The name of the api endpoint to submit the job.

• result_callbacks – The callbacks to apply to the result.

• **kwargs – The keywrod arguments to pass to the gradio server.

Returns:

Gradio job object.

Return type:

Job

mbodied.agents.backends.ollama_backend module

class mbodied.agents.backends.ollama_backend.OllamaBackend(api_key: str | None = None, model_src: str = None)

Bases: HttpxBackend

Backend for interacting with Ollama’s API.

DEFAULT_MODEL = 'llava'

DEFAULT_SRC = 'http://localhost:11434/api/chat/'

INITIAL_CONTEXT = [Message(role='system', content=['You are a robot with advanced spatial reasoning.'])]

SERIALIZER

alias of OllamaSerializer

class mbodied.agents.backends.ollama_backend.OllamaSerializer(wrapped: Message | Sample | list[Message] | None = None, *, message: Message | None = None, sample: Sample | None = None, **data)

Bases: Serializer

Serializer for Ollama-specific data formats.

classmethod extract_response(response: dict[str, Any]) → str

Extracts the response from the Ollama format.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ConfigDict = {'arbitrary_types_allowed': True, 'extra': 'allow', 'from_attributes': True, 'use_enum_values': False, 'validate_assignment': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'wrapped': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod serialize_image(image: Image) → str

Serializes an image to the Ollama format.

classmethod serialize_msg(message: Message) → dict[str, Any]

Serializes a message to the Ollama format.

classmethod serialize_text(text: str) → str

Serializes a text string to the Ollama format.

mbodied.agents.backends.openai_backend module

class mbodied.agents.backends.openai_backend.OpenAIBackendMixin(api_key: str | None = None, client: Any | None = None, response_format: str = None, **kwargs)

Bases: object

Backend for interacting with OpenAI’s API.

api_key

The API key for the OpenAI service.

client

The client for the OpenAI service.

serialized

The serializer for the OpenAI backend.

response_format

The format for the response.

DEFAULT_MODEL = 'gpt-4o'

INITIAL_CONTEXT = [Message(role='system', content=['You are a robot with advanced spatial reasoning.'])]

predict(message: Message, context: List[Message] | None = None, model: Any | None = None, **kwargs) → str

Create a completion based on the given message and context.

Parameters:

• message (Message) – The message to process.

• context (Optional[List[Message]]) – The context of messages.

• model (Optional[Any]) – The model used for processing the messages.

• **kwargs – Additional keyword arguments.

Returns:

The result of the completion.

Return type:

str

class mbodied.agents.backends.openai_backend.OpenAISerializer(wrapped: Message | Sample | list[Message] | None = None, *, message: Message | None = None, sample: Sample | None = None, **data)

Bases: Serializer

Serializer for OpenAI-specific data formats.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ConfigDict = {'arbitrary_types_allowed': True, 'extra': 'allow', 'from_attributes': True, 'use_enum_values': False, 'validate_assignment': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'wrapped': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod serialize_image(image: Image) → dict[str, Any]

Serializes an image to the OpenAI format.

Parameters:

image – The image to be serialized.

Returns:

A dictionary representing the serialized image.

classmethod serialize_text(text: str) → dict[str, Any]

Serializes a text string to the OpenAI format.

Parameters:

text – The text to be serialized.

Returns:

A dictionary representing the serialized text.

mbodied.agents.backends.openvla_backend module

mbodied.agents.backends.serializer module

class mbodied.agents.backends.serializer.Serializer(wrapped: Message | Sample | list[Message] | None = None, *, message: Message | None = None, sample: Sample | None = None, **data)

Bases: Sample

A class to serialize messages and samples.

This class provides a mechanism to serialize messages and samples into a dictionary format used by i.e. OpenAI, Anthropic, or other APIs.

wrapped

The message or sample to be serialized.

Type:

Any | None

model_config

The Pydantic configuration for the Serializer model.

Type:

pydantic.config.ConfigDict

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ConfigDict = {'arbitrary_types_allowed': True, 'extra': 'allow', 'from_attributes': True, 'use_enum_values': False, 'validate_assignment': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'wrapped': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

serialize() → dict[str, Any] | list[Any]

Serializes the wrapped content of the Serializer instance.

Returns:

A dictionary representing the serialized wrapped content.

classmethod serialize_image(image: Image) → dict[str, Any]

Serializes an Image instance.

Parameters:

image – The Image to be serialized.

Returns:

A dictionary representing the serialized Image.

serialize_msg(message: Message) → dict[str, Any]

Serializes a Message instance.

Parameters:

message – The Message to be serialized.

Returns:

A dictionary representing the serialized Message.

serialize_sample(sample: Any) → dict[str, Any]

Serializes a given sample.

Parameters:

sample – The sample to be serialized.

Returns:

A dictionary representing the serialized sample.

classmethod serialize_text(text: str) → dict[str, Any]

Serializes a text string.

Parameters:

text – The text to be serialized.

Returns:

A dictionary representing the serialized text.

classmethod validate_model(values: dict[str, Any]) → dict[str, Any] | list[Any]

Validates the ‘wrapped’ field of the model.

Parameters:

values – A dictionary of field values to validate.

Returns:

The validated values dictionary.

Raises:

ValueError – If the ‘wrapped’ field contains an invalid type.

wrapped: Any | None

Module contents

class mbodied.agents.backends.AnthropicBackend(api_key: str | None, client: Anthropic | None = None, **kwargs)

Bases: OpenAIBackendMixin

Backend for interacting with Anthropic’s API.

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!'])]

class mbodied.agents.backends.AnthropicSerializer(wrapped: Message | Sample | list[Message] | None = None, *, message: Message | None = None, sample: Sample | None = None, **data)

Bases: Serializer

Serializer for Anthropic-specific data formats.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ConfigDict = {'arbitrary_types_allowed': True, 'extra': 'allow', 'from_attributes': True, 'use_enum_values': False, 'validate_assignment': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'wrapped': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod serialize_image(image: Image) → dict[str, Any]

Serializes an image to the Anthropic format.

Parameters:

image – The image to be serialized.

Returns:

A dictionary representing the serialized image.

classmethod serialize_text(text: str) → dict[str, Any]

Serializes a text string to the Anthropic format.

Parameters:

text – The text to be serialized.

Returns:

A dictionary representing the serialized text.

wrapped: Any | None

class mbodied.agents.backends.GradioBackend(model_src: str = None, **kwargs)

Bases: Backend

Gradio backend that handles connections to gradio servers.

predict(*args, **kwargs) → str

Forward queries to the gradio api endpoint predict.

Parameters:

• *args – The arguments to pass to the gradio server.

• **kwargs – The keywrod arguments to pass to the gradio server.

submit(*args, api_name='/predict', result_callbacks=None, **kwargs) → Job

Submit queries asynchronously without need of asyncio.

Parameters:

• *args – The arguments to pass to the gradio server.

• api_name – The name of the api endpoint to submit the job.

• result_callbacks – The callbacks to apply to the result.

• **kwargs – The keywrod arguments to pass to the gradio server.

Returns:

Gradio job object.

Return type:

Job

class mbodied.agents.backends.HttpxBackend(api_key=None, model_src: str | None = None, serializer: Serializer | None = None, **kwargs)

Bases: OpenAIBackendMixin

DEFAULT_MODEL = 'reka-core'

DEFAULT_SRC = 'https://api.reka.ai/v1/chat'

SERIALIZER

alias of HttpxSerializer

class mbodied.agents.backends.HttpxSerializer(wrapped: Message | Sample | list[Message] | None = None, *, message: Message | None = None, sample: Sample | None = None, **data)

Bases: Serializer

classmethod extract_response(response)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ConfigDict = {'arbitrary_types_allowed': True, 'extra': 'allow', 'from_attributes': True, 'use_enum_values': False, 'validate_assignment': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'wrapped': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod serialize_image(image: Image)

Serializes an Image instance.

Parameters:

image – The Image to be serialized.

Returns:

A dictionary representing the serialized Image.

classmethod serialize_text(text)

Serializes a text string.

Parameters:

text – The text to be serialized.

Returns:

A dictionary representing the serialized text.

class mbodied.agents.backends.OllamaBackend(api_key: str | None = None, model_src: str = None)

Bases: HttpxBackend

Backend for interacting with Ollama’s API.

DEFAULT_MODEL = 'llava'

DEFAULT_SRC = 'http://localhost:11434/api/chat/'

INITIAL_CONTEXT = [Message(role='system', content=['You are a robot with advanced spatial reasoning.'])]

SERIALIZER

alias of OllamaSerializer

class mbodied.agents.backends.OllamaSerializer(wrapped: Message | Sample | list[Message] | None = None, *, message: Message | None = None, sample: Sample | None = None, **data)

Bases: Serializer

Serializer for Ollama-specific data formats.

classmethod extract_response(response: dict[str, Any]) → str

Extracts the response from the Ollama format.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ConfigDict = {'arbitrary_types_allowed': True, 'extra': 'allow', 'from_attributes': True, 'use_enum_values': False, 'validate_assignment': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'wrapped': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod serialize_image(image: Image) → str

Serializes an image to the Ollama format.

classmethod serialize_msg(message: Message) → dict[str, Any]

Serializes a message to the Ollama format.

classmethod serialize_text(text: str) → str

Serializes a text string to the Ollama format.

wrapped: Any | None

mbodied.agents.backends.OpenAIBackend

alias of OpenAIBackendMixin

class mbodied.agents.backends.OpenAISerializer(wrapped: Message | Sample | list[Message] | None = None, *, message: Message | None = None, sample: Sample | None = None, **data)

Bases: Serializer

Serializer for OpenAI-specific data formats.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ConfigDict = {'arbitrary_types_allowed': True, 'extra': 'allow', 'from_attributes': True, 'use_enum_values': False, 'validate_assignment': False}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'wrapped': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod serialize_image(image: Image) → dict[str, Any]

Serializes an image to the OpenAI format.

Parameters:

image – The image to be serialized.

Returns:

A dictionary representing the serialized image.

classmethod serialize_text(text: str) → dict[str, Any]

Serializes a text string to the OpenAI format.

Parameters:

text – The text to be serialized.

Returns:

A dictionary representing the serialized text.

wrapped: Any | None