File size: 1,501 Bytes
729a1f7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c287012
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# ────────────────────────────── memo/memory.py ──────────────────────────────
from collections import deque, defaultdict
from typing import List, Dict

class MemoryLRU:
    """
    Per-user LRU-like memory of the last N (default 20) summarized chat sessions.
    Each item is a single string in the format: "q: ...\na: ..."
    """
    def __init__(self, capacity: int = 20):
        self.capacity = capacity
        self._store: Dict[str, deque] = defaultdict(lambda: deque(maxlen=self.capacity))

    def add(self, user_id: str, qa_summary: str):
        self._store[user_id].append(qa_summary)

    def recent(self, user_id: str, n: int = 3) -> List[str]:
        d = self._store[user_id]
        if not d:
            return []
        # Return last n in recency order (most recent first)
        return list(d)[-n:][::-1]

    def rest(self, user_id: str, skip_n: int = 3) -> List[str]:
        d = self._store[user_id]
        if not d:
            return []
        # Everything except the most recent `skip_n`, oldest first
        return list(d)[:-skip_n] if len(d) > skip_n else []

    def all(self, user_id: str) -> List[str]:
        return list(self._store[user_id])

    def clear(self, user_id: str) -> None:
        """
        Clear all cached summaries for the given user.
        """
        if user_id in self._store:
            self._store[user_id].clear()