Terminal / memory /working.py
Pulka
feat(telegram): bot UX 100% — WebApp button, /logs, /ping, inline query, _BACK_KB
4cbd6c4 verified
Raw
History Blame
1.59 kB
"""
working.py — Working Memory (short-term, in-RAM)
Ultime N conversazioni per contesto immediato.
"""
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class WorkingEntry:
role: str
content: str
ts: float = field(default_factory=lambda: datetime.now().timestamp())
class WorkingMemory:
def __init__(self, max_entries: int = 40):
self._buf: deque[WorkingEntry] = deque(maxlen=max_entries)
def add(self, role: str, content: str):
self._buf.append(WorkingEntry(role=role, content=content))
def get_recent(self, n: int = 10) -> list[dict]:
import time as _t
_ttl = _t.time() - 3600 # D7: 1h TTL — entry più vecchie non distorcono il contesto
entries = [e for e in list(self._buf)[-n:] if e.ts >= _ttl]
return [{"role": e.role, "content": e.content} for e in entries]
def get_context_string(self, n: int = 6) -> str:
recent = self.get_recent(n)
if not recent:
return ""
lines = []
for m in recent:
prefix = "Utente" if m["role"] == "user" else "AI"
# S571: 200→400 — evita di tagliare risposte con codice corto
# S600: 400→600 — working memory snippet più lungo per risposte con codice
lines.append(f"{prefix}: {m['content'][:600]}")
return "Conversazione recente:\n" + "\n".join(lines)
def clear(self):
self._buf.clear()
def stats(self) -> dict:
return {"entries": len(self._buf), "max": self._buf.maxlen}