File size: 1,588 Bytes
28a08e7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
41
42
43
44
45
"""
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}