Ollama Backend

OllamaBackend

Bases: HttpxBackend

Backend for interacting with Ollama's API.

Source code in mbodied/agents/backends/ollama_backend.py

class OllamaBackend(HttpxBackend):
    """Backend for interacting with Ollama's API."""

    INITIAL_CONTEXT = [
        Message(role="system", content="You are a robot with advanced spatial reasoning."),
    ]
    DEFAULT_MODEL = "llava"
    SERIALIZER = OllamaSerializer
    DEFAULT_SRC = "http://localhost:11434/api/chat/"

    def __init__(self, api_key: str | None = None, endpoint: str = None):
        """Initializes an OllamaBackend instance."""
        endpoint = endpoint or self.DEFAULT_SRC
        super().__init__(api_key, endpoint=endpoint)

__init__(api_key=None, endpoint=None)

Initializes an OllamaBackend instance.

Source code in mbodied/agents/backends/ollama_backend.py

def __init__(self, api_key: str | None = None, endpoint: str = None):
    """Initializes an OllamaBackend instance."""
    endpoint = endpoint or self.DEFAULT_SRC
    super().__init__(api_key, endpoint=endpoint)

OllamaSerializer

Bases: Serializer

Serializer for Ollama-specific data formats.

Source code in mbodied/agents/backends/ollama_backend.py

class OllamaSerializer(Serializer):
    """Serializer for Ollama-specific data formats."""

    @classmethod
    def serialize_image(cls, image: Image) -> str:
        """Serializes an image to the Ollama format."""
        return image.base64

    @classmethod
    def serialize_text(cls, text: str) -> str:
        """Serializes a text string to the Ollama format."""
        return text

    @classmethod
    def serialize_msg(cls, message: Message) -> dict[str, Any]:
        """Serializes a message to the Ollama format."""
        images = [cls.serialize_image(im) for im in message.content if isinstance(im, Image)]
        texts = ".".join([txt for txt in message.content if isinstance(txt, str)])
        return {
            "role": message.role,
            "content": texts,
            "images": images,
        }

    @classmethod
    def extract_response(cls, response: dict[str, Any]) -> str:
        """Extracts the response from the Ollama format."""
        if isinstance(response, str):
            return json.loads(response)["message"]["content"]
        return response["message"]["content"]

    @classmethod
    def extract_stream(cls, response):
        try:
            parsed = json.loads(response)
            return cls.extract_response(parsed)
        except json.JSONDecodeError:
            # If it's not valid JSON, return the raw response
            return response

extract_response(response) classmethod

Extracts the response from the Ollama format.

Source code in mbodied/agents/backends/ollama_backend.py

@classmethod
def extract_response(cls, response: dict[str, Any]) -> str:
    """Extracts the response from the Ollama format."""
    if isinstance(response, str):
        return json.loads(response)["message"]["content"]
    return response["message"]["content"]

serialize_image(image) classmethod

Serializes an image to the Ollama format.

Source code in mbodied/agents/backends/ollama_backend.py

@classmethod
def serialize_image(cls, image: Image) -> str:
    """Serializes an image to the Ollama format."""
    return image.base64

serialize_msg(message) classmethod

Serializes a message to the Ollama format.

Source code in mbodied/agents/backends/ollama_backend.py

@classmethod
def serialize_msg(cls, message: Message) -> dict[str, Any]:
    """Serializes a message to the Ollama format."""
    images = [cls.serialize_image(im) for im in message.content if isinstance(im, Image)]
    texts = ".".join([txt for txt in message.content if isinstance(txt, str)])
    return {
        "role": message.role,
        "content": texts,
        "images": images,
    }

serialize_text(text) classmethod

Serializes a text string to the Ollama format.

Source code in mbodied/agents/backends/ollama_backend.py

@classmethod
def serialize_text(cls, text: str) -> str:
    """Serializes a text string to the Ollama format."""
    return text