diff --git a/app/nullius/catalogue/capabilities/mem0/capability.py b/app/nullius/catalogue/capabilities/mem0/capability.py index afd1e5d..7e4c44c 100644 --- a/app/nullius/catalogue/capabilities/mem0/capability.py +++ b/app/nullius/catalogue/capabilities/mem0/capability.py @@ -12,7 +12,8 @@ ``add {session}`` - the session's turns as Mem0 messages (a role other than user or assistant is a user turn naming its speaker), - added for one user, the session's id and date as metadata + added for one user, the session's id and date as metadata, + the date also as ``timestamp`` ``recall {question}`` - ``{memory}``: what Mem0's search returns, best first, each memory headed by its session's date @@ -111,8 +112,10 @@ def add(self, s): self.dates[sid] = str(s.get("date") or "") self.warnings.take() t0 = time.time() - got = self.mem.add(messages, user_id=USER, metadata={"session": sid, "date": self.dates[sid]}, - infer=setting("MEM0_INFER", "on") != "off") + meta = {"session": sid, "date": self.dates[sid]} + if self.dates[sid]: + meta["timestamp"] = self.dates[sid] + got = self.mem.add(messages, user_id=USER, metadata=meta, infer=setting("MEM0_INFER", "on") != "off") items = got.get("results") if isinstance(got, dict) else got return {"trace": {"messages": len(messages), "seconds": round(time.time() - t0, 3), "written": [{k: it.get(k) for k in ("id", "event", "memory", "previous_memory") if it.get(k) is not None} diff --git a/app/tests/test_mem0_capability.py b/app/tests/test_mem0_capability.py new file mode 100644 index 0000000..2ed54a7 --- /dev/null +++ b/app/tests/test_mem0_capability.py @@ -0,0 +1,53 @@ +"""The mem0 capability's add: what it hands Mem0, with a stand-in for Mem0.""" +import importlib.util +import os +import sys +import types +import unittest + +HERE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "nullius", "catalogue", "capabilities", "mem0", "capability.py") + + +class FakeMem0: + def __init__(self): + self.added = [] + + def add(self, messages, **kw): + self.added.append(kw) + return {"results": []} + + def get_all(self, **kw): + return {"results": []} + + +def load(): + out = sys.stdout # the module sends stdout to stderr when it loads + try: + spec = importlib.util.spec_from_file_location("mem0_capability_under_test", HERE) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + finally: + sys.stdout = out + return mod + + +class Add(unittest.TestCase): + def setUp(self): + self.mod = load() + self.m = self.mod.Memory() + self.m.mem = FakeMem0() + + def test_the_session_date_is_also_the_timestamp_mem0_reads(self): + self.m.add({"id": "s1", "date": "2023/05/21 (Sun) 16:22", "turns": [{"role": "user", "text": "hi"}]}) + self.assertEqual(self.m.mem.added[0]["metadata"], + {"session": "s1", "date": "2023/05/21 (Sun) 16:22", "timestamp": "2023/05/21 (Sun) 16:22"}) + + def test_an_undated_session_has_no_timestamp(self): + # an empty timestamp would be an Observation Date of nothing; without one Mem0 keeps its default + self.m.add({"id": "s1", "turns": [{"role": "user", "text": "hi"}]}) + self.assertNotIn("timestamp", self.m.mem.added[0]["metadata"]) + + +if __name__ == "__main__": + unittest.main()