Bases: Memory
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")
Source code in src/ant_ai/memory/backends/mem0.py
33
34
35
36
37
38
39
40
41
42
43
44
45
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 | class Mem0Memory(Memory):
"""
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(api_key=self.api_key)
if self.api_key
else AsyncMemoryClient()
)
return self
async def retrieve(
self, query: str, *, top_k: int = 5, **kwargs: Any
) -> list[Message]:
filters = _resolve_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], **kwargs: Any) -> None:
filters = _resolve_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)
|