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, 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) -> str:
        entry_id = make_entry_id()
        vector = await self._embedder.embed(text)
        self._index.upsert(vectors=[{
            "id": entry_id,
            "values": vector,
            "metadata": {"text": text, "type": "text"},
        }])
        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)
            for m in results.matches
            if m.score >= similarity_threshold
        ]

    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, memory):
    tools.register(tool)

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

Key rules

• store(), retrieve(), delete(), save_text(), search_text() are all async
• count() is synchronous
• 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)