Merge pull request #1437 from feder-cr/memory-incremental

engine: a memory pass sends only the messages it has not taken yet

docs/architecture.md

@@ -850,8 +850,8 @@ guest's report: it is never used to enforce anything (the cap is the guest's).
 
 `memory.updated {conversations, changed}` ends a memory pass (section 8.6): the
 Dot had been quiet for a while, and the summary model rewrote its `MEMORY.md`
-from the conversation files that changed since the last pass. `conversations`
-is how many files it took in, `changed` whether `MEMORY.md` is different. A pass
+from what was said since the last pass. `conversations` is how many
+conversation files it took messages from, `changed` whether `MEMORY.md` is different. A pass
 that found nothing new sends nothing, and one that failed sends nothing either:
 its spend goes with the next `memory.updated`.
 
@@ -1362,8 +1362,10 @@ are cut with a marker.
 - The memory pass (`memory_update.py`): a model often does not think to write
   what it learns about the person, so once no turn has run for 5 minutes after
   a turn (or after a start), the engine takes the conversation files whose time
-  is after the newest it took last and has the `summary` model rewrite
-  MEMORY.md from them and the current MEMORY.md, in one request with no tools
+  is after the newest it took last, and of each only the messages it has not
+  taken yet (a file is written again whole after every turn but only grows at
+  its end, so a long day's chat is read once), and has the `summary` model
+  rewrite MEMORY.md from them and the current MEMORY.md, in one request with no tools
   (`templates/agent/memory_update.md`: atomic facts with their day, a changed
   fact replacing the old one, nothing copied from what a web page or a command
   said). Files that do not fit one request go in several, oldest first, a long

docs/wiki/agent-memory-systems-compared.md

@@ -83,8 +83,8 @@ question.
 ## What was chosen
 
 The single request, as a background pass: once the Dot has been quiet for
-five minutes, the conversation files that changed go to the summary model with
-the current `MEMORY.md`, and the answer is the new `MEMORY.md`. Run in the product on the 116 mixed questions it
+five minutes, the messages said since the last pass go to the summary model
+with the current `MEMORY.md`, and the answer is the new `MEMORY.md`. Run in the product on the 116 mixed questions it
 scored 90.4% (104 of 115 that ran), at $0.025 a pass. From Mastra it
 takes two ideas rather than code: reading in pieces when the new conversations
 do not fit one request, and saying that a changed fact replaces the old one.

docs/wiki/how-a-dot-remembers.md

@@ -99,8 +99,8 @@ other note. The Dot sees it whole in every prompt, up to 25,000 characters
 
 A model often does not think to write it while it works, so the engine keeps
 it current itself: once the Dot has been quiet for five minutes, it takes the
-conversation files that changed since the last time and has the summary model
-rewrite `MEMORY.md` from them, in one request with no tools. A changed fact
+messages said since the last pass (of a long day, only the new ones) and has the
+summary model rewrite `MEMORY.md` from them, in one request with no tools. A changed fact
 replaces the old one with its date; nothing a web page or a command said is
 copied in as an instruction. If the conversations do not fit one request they
 go in several, oldest first. A pass that is cut, empty, unpriced, or races

invisible_engine_dots/nanobot/dots/memory_update.py

@@ -2,15 +2,17 @@
 
 MEMORY.md (in the Dot's memory directory) is given to the Dot in every prompt: what it should always know about
 the person. The Dot may write it while it works, but a model often does not think to, so when the Dot has been
-quiet for a while the engine takes in the conversation files that changed since the last pass
-(`conversations.py`) and has the summary model rewrite MEMORY.md from them, in one request with no tools
+quiet for a while the engine takes in what was said since the last pass (the conversation files,
+`conversations.py`) and has the summary model rewrite MEMORY.md from them, in one request with no tools
 (templates/agent/memory_update.md). Measured on LongMemEval (tests/bench/README.md): a profile written this way
 raised the answers that need what the person likes from 65% to 83%, as well as the best of the open-source
 memory systems tried, at about a seventeenth of their cost; an agent doing the same with file tools did worse
 (75%) at five times the cost.
 
-A pass takes the files whose time is after the newest it took last (`KV_THROUGH`), oldest first, as many to a
-request as the model's window holds; a file too long for one request is cut between its messages. Each request
+A pass takes the files whose time is after the newest it took last (`KV_THROUGH`), oldest first, and of each
+only the messages it has not taken yet (`KV_TAKEN`): a file is written again whole after every turn but only
+grows at its end, so the chat of a long day is read once, not again at every pass. As many go to a request as the
+model's window holds; a file too long for one request is cut between its messages. Each request
 answers with the whole new MEMORY.md, written before the next one starts, so a pass cut short keeps what it
 did. A request that is cut, empty or unpriced writes nothing; so does one whose MEMORY.md the Dot changed while
 it ran (the next pass takes the files again). Its spend has a ledger of its own (`SESSION_KEY`), capped like a
@@ -19,8 +21,9 @@ task's, and is reported by the `memory.updated` that ends a pass.
 
 from __future__ import annotations
 
+import sqlite3
 from collections.abc import Callable, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import datetime
 from typing import Literal
 
@@ -44,8 +47,11 @@ from nanobot.utils.prompt_templates import render_template
 QUIET_S = 5 * 60
 # The ledger of what passes spend, and the session key their requests are metered under.
 SESSION_KEY = "memory"
-# dots_kv: the time of the newest conversation file a pass took in.
+# dots_kv: the time of the newest conversation file a pass took in, and how many messages of each file it took.
 KV_THROUGH = "memory_updated_through"
+KV_TAKEN = "memory_messages_taken"
+# What separates the messages of a conversation file (`conversations.render`): a file is its title, then these.
+MESSAGE_BREAK = "\n\n## "
 # The model role that writes MEMORY.md: the one that summarises a long thread.
 MODEL_ROLE = "summary"
 MEMORY_PATH = f"{MEMORY_DIR}/{MEMORY_INDEX}"
@@ -63,12 +69,14 @@ class PassOutcome:
 
 @dataclass(frozen=True)
 class _Conversation:
-    """A conversation file, or a piece of one; `last` is false on every piece of a file but its last."""
+    """The messages of a conversation file a pass has not taken yet, or a piece of them; `last` is false on every
+    piece but its last, and `messages` is how many the file holds in all."""
 
     path: str
     mtime: datetime
     text: str
     last: bool = True
+    messages: int = 0
 
 
 def _parse_time(value: str) -> datetime:
@@ -84,7 +92,7 @@ def _pieces(conversation: _Conversation, fits: Callable[[list[_Conversation]], b
         return [conversation]
 
     def piece(text: str) -> _Conversation:
-        return _Conversation(conversation.path, conversation.mtime, text, last=False)
+        return replace(conversation, text=text, last=False)
 
     blocks = conversation.text.split("\n## ")
     pieces: list[_Conversation] = []
@@ -106,7 +114,7 @@ def _pieces(conversation: _Conversation, fits: Callable[[list[_Conversation]], b
             current = current[cut:]
     if current:
         pieces.append(piece(current))
-    return [*pieces[:-1], _Conversation(conversation.path, conversation.mtime, pieces[-1].text)]
+    return [*pieces[:-1], replace(conversation, text=pieces[-1].text)]
 
 
 def batches(conversations_: Sequence[_Conversation], fits: Callable[[list[_Conversation]], bool]) -> list[list[_Conversation]]:
@@ -164,10 +172,15 @@ class MemoryUpdater:
             return PassOutcome("failed", f"the computer did not answer: {exc}")
 
     async def _run(self, settings: EngineSettings) -> PassOutcome:
-        through_text = self._store.read(lambda conn: dots_store.read_kv(conn, KV_THROUGH))
+        through_text, taken = self._store.read(
+            lambda conn: (dots_store.read_kv(conn, KV_THROUGH), dots_store.read_kv(conn, KV_TAKEN) or {})
+        )
         through = _parse_time(through_text) if isinstance(through_text, str) else None
-        changed = await self._changed(through)
+        changed, newest_seen = await self._changed(through, taken)
         if not changed:
+            if newest_seen is not None:
+                # Files written again with nothing new (a mask added, say): past them, so no pass reads them again.
+                self._store.write(lambda conn: dots_store.write_kv(conn, KV_THROUGH, newest_seen.isoformat()))
             return PassOutcome("nothing")
 
         spend = TurnSpend(self._store, SESSION_KEY, settings.max_cost_usd, "memory pass")
@@ -213,11 +226,17 @@ class MemoryUpdater:
                 return PassOutcome("failed", "MEMORY.md was changed while the pass ran")
             await self._computer.write_bytes(MEMORY_PATH, f"{text}\n".encode())
             memory = text
-            # Only a file taken in whole moves the mark: a pass cut between the pieces of one takes it again.
-            whole = [c.mtime for c in group if c.last]
+            # Only a file taken in whole moves the marks: a pass cut between the pieces of one takes it again.
+            whole = [c for c in group if c.last]
             if whole:
-                newest = max(whole).isoformat()
-                self._store.write(lambda conn: dots_store.write_kv(conn, KV_THROUGH, newest))
+                taken.update({c.path: c.messages for c in whole})
+                newest = max(c.mtime for c in whole).isoformat()
+
+                def mark(conn: sqlite3.Connection) -> None:
+                    dots_store.write_kv(conn, KV_THROUGH, newest)
+                    dots_store.write_kv(conn, KV_TAKEN, taken)
+
+                self._store.write(mark)
 
         changed_memory = memory != before
         self._store.write(
@@ -228,8 +247,15 @@ class MemoryUpdater:
         logger.info("memory pass took in {} conversation files; MEMORY.md changed={}", len(changed), changed_memory)
         return PassOutcome("updated")
 
-    async def _changed(self, through: datetime | None) -> list[_Conversation]:
-        """The conversation files whose time is after `through`, oldest first, with their text."""
+    async def _changed(
+        self, through: datetime | None, taken: dict[str, int]
+    ) -> tuple[list[_Conversation], datetime | None]:
+        """The messages not taken yet of the conversation files whose time is after `through`, oldest first.
+
+        A file keeps its title; one taken in part before says so, so the model reads its messages as the rest of a
+        conversation it has seen. A file that holds fewer messages than were taken of it was written anew (by an
+        upgrade's first write): all of it is taken again. With them, the time of the newest file seen.
+        """
         found: list[tuple[datetime, str]] = []
         for source in SOURCES:
             directory = f"{conversations.CONVERSATIONS_DIR}/{source}"
@@ -242,9 +268,18 @@ class MemoryUpdater:
         out: list[_Conversation] = []
         for mtime, path in sorted(found):
             data = await self._computer.read_bytes(path)
-            if data is not None:
-                out.append(_Conversation(path, mtime, data.decode("utf-8", errors="replace")))
-        return out
+            if data is None:
+                continue
+            title, *messages = data.decode("utf-8", errors="replace").rstrip("\n").split(MESSAGE_BREAK)
+            done = taken.get(path, 0)
+            if done > len(messages):
+                done = 0
+            if done == len(messages):
+                continue
+            heading = f"{title} (continued: its earlier messages were taken in before)" if done else title
+            text = MESSAGE_BREAK.join([heading, *messages[done:]]) + "\n"
+            out.append(_Conversation(path, mtime, text, messages=len(messages)))
+        return out, max((mtime for mtime, _ in found), default=None)
 
     async def _read_memory(self) -> str:
         data = await self._computer.read_bytes(MEMORY_PATH)

invisible_engine_dots/tests/dots/test_memory_update.py

@@ -124,6 +124,40 @@ class TestAPass:
         assert "book the vet" in first and "my cat is called Luna" in first
         assert "I moved to Rome" in second and "my cat is called Luna" not in second and "book the vet" not in second
 
+    async def test_a_day_that_grew_sends_only_its_new_messages(self, make_pass) -> None:
+        p = make_pass([says(PROFILE), says(PROFILE + "\n- moved to Rome (2023-05-20)")])
+        p.conversation("2023-05-20.md", DAY)
+        await p.updater.run()
+
+        p.conversation("2023-05-20.md", DAY + "\n## 18:00 the person\n\nI moved to Rome\n\n## 18:00 you\n\nNoted.\n")
+        assert await p.updater.run() == m.PassOutcome("updated")
+
+        second = p.provider.requests[1]["messages"][0]["content"]
+        assert "# Chat, 2023-05-20 (continued: its earlier messages were taken in before)" in second
+        assert "I moved to Rome" in second and "my cat is called Luna" not in second
+
+    async def test_a_file_written_again_with_nothing_new_asks_nothing_and_is_passed(self, make_pass) -> None:
+        p = make_pass([says(PROFILE)])
+        p.conversation("2023-05-20.md", DAY)
+        await p.updater.run()
+        p.conversation("2023-05-20.md", DAY)
+        mark = p.through()
+
+        assert await p.updater.run() == m.PassOutcome("nothing")
+        assert len(p.provider.requests) == 1
+        assert p.through() != mark
+
+    async def test_a_file_with_fewer_messages_than_were_taken_is_taken_whole(self, make_pass) -> None:
+        p = make_pass([says(PROFILE), says(PROFILE)])
+        p.conversation("2023-05-20.md", DAY)
+        await p.updater.run()
+
+        p.conversation("2023-05-20.md", "# Chat, 2023-05-20\n\n## 20:00 the person\n\nstarting over\n")
+        await p.updater.run()
+
+        second = p.provider.requests[1]["messages"][0]["content"]
+        assert "starting over" in second and "(continued" not in second
+
     async def test_an_answer_cut_or_empty_writes_nothing_and_takes_the_conversations_again(self, make_pass) -> None:
         p = make_pass([LLMResponse(content="## The pers", finish_reason="length"), says("   "), says(PROFILE)])
         p.write_memory("old\n")
@@ -235,6 +269,26 @@ class TestAPass:
 
 
 class TestTheEngine:
+    async def test_the_second_pass_of_a_day_reads_only_what_was_said_since_the_first(
+        self, tmp_path: Path, dot_store: DotStore
+    ) -> None:
+        # The files are the turns' own (conversations.render): a pass splits them where render joins them.
+        h = EngineHarness(
+            tmp_path, dot_store, [says("hi"), says(PROFILE), says("noted"), says(PROFILE + "\n- moved")], memory_quiet_s=0.2
+        )
+        h.configure(runtime_config_body())
+        h.engine.start()
+        h.engine.accept(user_message("in1", "my cat is called Luna"))
+        await h.wait_until(lambda: len(h.events_of("memory.updated")) == 1)
+
+        h.engine.accept(user_message("in2", "I moved to Rome"))
+        await h.wait_until(lambda: len(h.events_of("memory.updated")) == 2)
+
+        second = h.provider.requests[3]["messages"][0]["content"]
+        assert "(continued: its earlier messages were taken in before)" in second
+        assert "I moved to Rome" in second and "my cat is called Luna" not in second
+        await h.engine.stop()
+
     async def test_a_pass_runs_once_the_dot_has_been_quiet_after_a_turn(self, tmp_path: Path, dot_store: DotStore) -> None:
         h = EngineHarness(tmp_path, dot_store, [says("hi"), says(PROFILE)], memory_quiet_s=0.2)
         h.configure(runtime_config_body())

packages/shared/src/events.ts

@@ -217,10 +217,10 @@ export interface OutboundEventDataMap {
    */
   "automation.next_run": { next_run_at_ms: number | null };
   /**
-   * The Dot brought its MEMORY.md (what it is given about the person in every prompt) up to date from the
-   * conversations that changed since the last time, once it had been quiet for a while (the engine's
-   * `memory_update.py`): how many conversation files it took in, whether MEMORY.md changed, and what the pass cost
-   * (with the cost of any pass that failed since the last of these).
+   * The Dot brought its MEMORY.md (what it is given about the person in every prompt) up to date from what was said
+   * since the last time, once it had been quiet for a while (the engine's `memory_update.py`): how many conversation
+   * files it took messages from, whether MEMORY.md changed, and what the pass cost (with the cost of any pass that
+   * failed since the last of these).
    */
   "memory.updated": { conversations: number; changed: boolean } & SpentUsd;
 }