ant_ai.memory.protocol

Memory pydantic-model

Bases: BaseModel

Pluggable memory interface for agents.

Implement retrieve and update to connect any memory backend. Both methods accept **kwargs so callers can pass backend-specific parameters (e.g. mem0's user_id, filters, run_id) without changing the interface.

Show JSON schema:

{
  "description": "Pluggable memory interface for agents.\n\nImplement `retrieve` and `update` to connect any memory backend.\nBoth methods accept `**kwargs` so callers can pass backend-specific\nparameters (e.g. mem0's `user_id`, `filters`, `run_id`) without\nchanging the interface.",
  "properties": {},
  "title": "Memory",
  "type": "object"
}

Source code in src/ant_ai/memory/protocol.py

class Memory(BaseModel):
    """
    Pluggable memory interface for agents.

    Implement `retrieve` and `update` to connect any memory backend.
    Both methods accept `**kwargs` so callers can pass backend-specific
    parameters (e.g. mem0's `user_id`, `filters`, `run_id`) without
    changing the interface.
    """

    async def retrieve(
        self, query: str, *, top_k: int = 5, **kwargs: Any
    ) -> list[Message]:
        """Return relevant memories as Messages, ready to inject into agent state."""
        raise NotImplementedError

    async def update(self, messages: list[Message], **kwargs: Any) -> None:
        """Persist new knowledge from the current session."""
        raise NotImplementedError

retrieve async

retrieve(
    query: str, *, top_k: int = 5, **kwargs: Any
) -> list[Message]

Return relevant memories as Messages, ready to inject into agent state.

Source code in src/ant_ai/memory/protocol.py

async def retrieve(
    self, query: str, *, top_k: int = 5, **kwargs: Any
) -> list[Message]:
    """Return relevant memories as Messages, ready to inject into agent state."""
    raise NotImplementedError

update async

update(messages: list[Message], **kwargs: Any) -> None

Persist new knowledge from the current session.

Source code in src/ant_ai/memory/protocol.py

async def update(self, messages: list[Message], **kwargs: Any) -> None:
    """Persist new knowledge from the current session."""
    raise NotImplementedError