File size: 2,139 Bytes
3d46076
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
import time
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any

@dataclass
class WorkingMemoryItem:
    key: str
    content: Any
    salience: float = 1.0
    created_at: float = field(default_factory=time.time)
    access_count: int = 0

class WorkingMemory:
    """
    Capacity-limited working memory (Miller's Law: 7 +/- 2 slots).
    Decays over time unless refreshed by attention or access.
    """
    def __init__(self, capacity: int = 7, decay_rate: float = 0.05):
        self.capacity = capacity
        self.decay_rate = decay_rate
        self.items: Dict[str, WorkingMemoryItem] = {}

    def put(self, key: str, content: Any, salience: float = 1.0):
        if key in self.items:
            item = self.items[key]
            item.content = content
            item.salience = min(2.0, item.salience + salience)
            item.access_count += 1
            return

        # If full, evict lowest salience item
        if len(self.items) >= self.capacity:
            lowest_key = min(self.items.keys(), key=lambda k: self.items[k].salience)
            del self.items[lowest_key]

        self.items[key] = WorkingMemoryItem(key=key, content=content, salience=salience)

    def get(self, key: str) -> Optional[Any]:
        if key in self.items:
            item = self.items[key]
            item.salience = min(2.0, item.salience + 0.2)
            item.access_count += 1
            return item.content
        return None

    def decay_step(self):
        to_delete = []
        for key, item in self.items.items():
            item.salience -= self.decay_rate
            if item.salience <= 0.0:
                to_delete.append(key)
        for k in to_delete:
            del self.items[k]

    def get_all_active(self) -> List[Dict[str, Any]]:
        return [
            {
                "key": item.key,
                "content": str(item.content),
                "salience": round(item.salience, 3),
                "access_count": item.access_count
            }
            for item in sorted(self.items.values(), key=lambda x: x.salience, reverse=True)
        ]