Summarized memory

A hard cap keeps memory small, but it drops the earliest context. Users notice: "I told you my preferences ten minutes ago". The summarization pattern keeps a short rolling summary alongside the recent turns. When the deque is full, the summary absorbs the evicted turn before it disappears.

layers/memory.py (summarized)
python
class MemoryLayer:
    def __init__(self, summarizer=None) -> None:
        self._threads: Dict[str, Deque[dict]] = defaultdict(
            lambda: deque(maxlen=MAX_TURNS_PER_THREAD)
        )
        self._summary: Dict[str, str] = defaultdict(str)
        self._summarizer = summarizer  # async callable(turns) -> str

    async def append(self, thread_id: str, role: str, content: str) -> None:
        thread = self._threads[thread_id]
        if len(thread) == thread.maxlen:
            # About to evict the oldest turn; fold it into the summary first.
            evicting = thread[0]
            if self._summarizer is not None:
                current = self._summary[thread_id]
                self._summary[thread_id] = await self._summarizer(current, evicting)
        thread.append({"role": role, "content": content})

    def context(self, thread_id: str) -> dict:
        return {
            "summary": self._summary[thread_id],
            "recent": list(self._threads[thread_id]),
        }

The summarizer is pluggable. In tests you pass a deterministic fake. In production you pass a small LLM call. The layer does not care which, as long as it returns a short string.

Ordering exercise: Order the summarized write path

Loading practice…

Quiz: Quiz

Loading practice…