File size: 2,543 Bytes
3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 |
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 66 67 68 69 70 71 72 73 |
# blux/agent/advanced/adaptive_memory.py
import time
from threading import Lock
class AdaptiveMemory:
"""
Thread-safe adaptive memory for BLUX-cA agents.
Supports decay, priority weighting, tag-based recall, and checkpointing.
"""
def __init__(self):
self.memory_store = {}
self.lock = Lock()
def add(self, key, value, user_type="default", priority=1, tags=None):
if tags is None:
tags = []
with self.lock:
self.memory_store[key] = {
"value": value,
"user_type": user_type,
"priority": priority,
"tags": tags,
"timestamp": time.time()
}
def recall(self, key, decay_rate=0.001):
with self.lock:
data = self.memory_store.get(key)
if not data:
return None
age = time.time() - data["timestamp"]
weight = max(0, data["priority"] * (1 - decay_rate * age))
return {"value": data["value"], "weight": weight, "tags": data["tags"]}
def recall_by_tag(self, tag):
with self.lock:
return [
{key: data}
for key, data in self.memory_store.items()
if tag in data["tags"]
]e_path="memory_checkpoint.json"):
with self.lock:
with open(file_path, "w") as f:
json.dump(self.memory_store, f, indent=2)
def load_checkpoint(self, file_path="memory_checkpoint.json"):
try:
with open(file_path, "r") as f:
with self.lock:
self.memory_store = json.load(f)
except FileNotFoundError:
self.memory_store = {} """
Applies decay to all memory entries to reduce relevance of older/unimportant items.
"""
for entry in self.memory_store:
entry["weight"] *= (1 - self.decay_rate)
def summarize_memory(self):
"""
Returns a simple summary of memory weights and top entries.
"""
top_entries = self.recall(top_n=5)
summary = [{"input": e["input"], "weight": e["weight"]} for e in top_entries]
return summary
# Example usage:
if __name__ == "__main__":
am = AdaptiveMemory()
am.store({"input": "I need help", "user_type": "struggler", "decision": "provide guidance"})
am.store({"input": "Ignore this", "user_type": "indulgent", "decision": "set boundary"})
print("Top memory entries:", am.summarize_memory()) |