MemoryService

from mango.memory import MemoryService

Abstract base class for all memory backends. Implement this to use any vector store.

Interface

class MemoryService(ABC):

    # Tool-usage memory
    @abstractmethod
    async def store(self, entry: MemoryEntry) -> None: ...

    @abstractmethod
    async def retrieve(
        self,
        question: str,
        top_k: int = 3,
        similarity_threshold: float = 0.6,
    ) -> list[MemoryEntry]: ...

    @abstractmethod
    async def delete(self, entry_id: str) -> None: ...

    # Text memory
    @abstractmethod
    async def save_text(self, text: str) -> str: ...

    @abstractmethod
    async def search_text(
        self,
        query: str,
        top_k: int = 3,
        similarity_threshold: float = 0.6,
    ) -> list[TextMemoryEntry]: ...

    # Sync
    @abstractmethod
    def count(self) -> int: ...

Data types

MemoryEntry

@dataclass
class MemoryEntry:
    id: str
    question: str        # original natural language question
    tool_name: str       # tool that was called
    tool_args: dict      # exact arguments that worked
    result_summary: str  # first 300 chars of the result
    similarity: float    # filled by retrieve(), cosine similarity
    timestamp: str       # ISO timestamp

TextMemoryEntry

@dataclass
class TextMemoryEntry:
    id: str
    text: str
    similarity: float    # filled by search_text()

make_entry_id

from mango.memory import make_entry_id

entry_id = make_entry_id()   # str(uuid.uuid4())

TrainingEntry

@dataclass
class TrainingEntry:
    id: str
    question: str        # natural language question
    tool_name: str       # tool to call
    tool_args: dict      # verified arguments
    result_summary: str  # optional description of the expected result
    similarity: float    # filled by get_training_entries()
    timestamp: str       # ISO timestamp

Training methods

These are implemented by ChromaAgentMemory on top of the base interface.

# Store a verified training entry (upserts — safe to call twice with same id)
await memory.train(entry: TrainingEntry) -> None

# Retrieve top_k training entries similar to a question
entries = await memory.get_training_entries(
    question: str,
    top_k: int = 5,
    similarity_threshold: float = 0.5,
) -> list[TrainingEntry]

# Count total training entries
memory.training_count() -> int

Export / Import

# Export all entries across all layers (tool-usage, text, training)
entries = await memory.export_all() -> list[dict]

# Import entries in export_all() format — skips unknown types with a warning
imported = await memory.import_all(entries: list[dict]) -> int

Each dict in the export has a "type" field: "tool", "text", or "training".

Built-in implementation

→ See Custom Memory Backend to implement your own.