Skip to content

ant_ai.memory.backends.mem0

Mem0Memory pydantic-model

Bases: MemoryTool

mem0 cloud backend for AgentMemory.

Requires MEM0_API_KEY in the environment, or pass api_key explicitly.

Example::

memory = Mem0Memory(api_key="m0-...")
msgs = await memory.retrieve("user preferences", user_id="alice")
await memory.update(conversation, user_id="alice")

Fields:

Validators:

  • _set_defaults
Source code in src/ant_ai/memory/backends/mem0.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class Mem0Memory(MemoryTool):
    """
    mem0 cloud backend for AgentMemory.

    Requires MEM0_API_KEY in the environment, or pass ``api_key`` explicitly.

    Example::

        memory = Mem0Memory(api_key="m0-...")
        msgs = await memory.retrieve("user preferences", user_id="alice")
        await memory.update(conversation, user_id="alice")
    """

    api_key: str | None = Field(default=None)
    _client: Any = PrivateAttr()

    @model_validator(mode="after")
    def _init_client(self) -> Mem0Memory:
        self._client: AsyncMemoryClient = (
            AsyncMemoryClient(api_key=self.api_key)
            if self.api_key
            else AsyncMemoryClient()
        )
        return self

    async def retrieve(
        self,
        query: str,
        *,
        top_k: int = 5,
        ctx: InvocationContext | None = None,
        **kwargs: Any,
    ) -> list[Message]:
        filters = _resolve_scope(ctx, kwargs)
        options = SearchMemoryOptions(filters=filters or None, top_k=top_k)
        results: Any = await self._client.search(query, options=options)
        return [
            Message(role="system", content=r["memory"])
            for r in results.get("results", [])
        ]

    async def update(
        self,
        messages: list[Message],
        *,
        ctx: InvocationContext | None = None,
        **kwargs: Any,
    ) -> None:
        filters = _resolve_scope(ctx, kwargs)
        msg_dicts: list[dict[str, str]] = [
            {"role": m.role, "content": m.content} for m in messages if m.content
        ]
        if not msg_dicts:
            return
        await self._client.add(msg_dicts, **filters)