Custom Memory Backend

Swap ChromaDB for any vector store by implementing MemoryService.

Full example — Pinecone

from pinecone import Pinecone
from mango.memory import MemoryService, MemoryEntry, TextMemoryEntry, TrainingEntry, make_entry_id
from mango.core.types import MangoError

class PineconeMemory(MemoryService):

    def __init__(self, api_key: str, index_name: str):
        pc = Pinecone(api_key=api_key)
        self._index = pc.Index(index_name)
        self._embedder = MyEmbedder()   # bring your own embeddings

    async def store(self, entry: MemoryEntry) -> None:
        vector = await self._embedder.embed(entry.question)
        self._index.upsert(vectors=[{
            "id": entry.id,
            "values": vector,
            "metadata": {
                "question": entry.question,
                "tool_name": entry.tool_name,
                "tool_args": json.dumps(entry.tool_args),
                "result_summary": entry.result_summary,
                "timestamp": entry.timestamp,
                "type": "tool",
            },
        }])

    async def retrieve(
        self,
        question: str,
        top_k: int = 3,
        similarity_threshold: float = 0.6,
    ) -> list[MemoryEntry]:
        vector = await self._embedder.embed(question)
        results = self._index.query(
            vector=vector,
            top_k=top_k,
            filter={"type": "tool"},
            include_metadata=True,
        )
        entries = []
        for match in results.matches:
            if match.score < similarity_threshold:
                continue
            m = match.metadata
            entries.append(MemoryEntry(
                id=match.id,
                question=m["question"],
                tool_name=m["tool_name"],
                tool_args=json.loads(m["tool_args"]),
                result_summary=m["result_summary"],
                similarity=match.score,
                timestamp=m["timestamp"],
            ))
        return entries

    async def delete(self, entry_id: str) -> None:
        self._index.delete(ids=[entry_id])

    async def save_text(
        self,
        text: str,
        *,
        source: str = "manual",
        verified: bool | None = None,
    ) -> str:
        entry_id = make_entry_id()
        if verified is None:
            verified = source in ("manual", "import")
        vector = await self._embedder.embed(text)
        self._index.upsert(vectors=[{
            "id": entry_id,
            "values": vector,
            "metadata": {"text": text, "type": "text", "source": source, "verified": verified},
        }])
        return entry_id

    async def search_text(
        self,
        query: str,
        top_k: int = 3,
        similarity_threshold: float = 0.6,
    ) -> list[TextMemoryEntry]:
        vector = await self._embedder.embed(query)
        results = self._index.query(
            vector=vector,
            top_k=top_k,
            filter={"type": "text"},
            include_metadata=True,
        )
        return [
            TextMemoryEntry(
                id=m.id,
                text=m.metadata["text"],
                similarity=m.score,
                source=m.metadata.get("source", "legacy"),
                verified=bool(m.metadata.get("verified", False)),
            )
            for m in results.matches
            if m.score >= similarity_threshold
        ]

    # Training layer — same pattern with metadata type="training"
    async def train(self, entry: TrainingEntry) -> None: ...
    async def get_training_entries(
        self, question: str, top_k: int = 5, similarity_threshold: float = 0.5
    ) -> list[TrainingEntry]: ...
    def training_count(self) -> int: ...

    # Export / import — list of dicts with a "type" key: "tool" | "text" | "training"
    async def export_all(self) -> list[dict]: ...
    async def import_all(self, entries: list[dict]) -> int: ...

    def count(self) -> int:
        stats = self._index.describe_index_stats()
        return stats.total_vector_count

Use it

memory = PineconeMemory(api_key="...", index_name="mango-memory")

tools = ToolRegistry()
for tool in build_mongo_tools(db):
    tools.register(tool)

agent = MangoAgent(..., agent_memory=memory)

Key rules

• All eleven methods are abstract: store(), retrieve(), delete(), save_text(), search_text(), train(), get_training_entries(), export_all(), import_all() are async
• count() and training_count() are synchronous
• save_text() takes keyword-only source / verified; search_text() results should carry them back
• retrieve() must populate entry.similarity on returned entries
• save_text() must return the generated entry ID
• Separate tool-usage memories from text memories (use metadata type field or separate indexes)