Spaces:
Runtime error
Runtime error
File size: 2,246 Bytes
9a90a52 | 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 | import json
import os
from typing import List, Dict, Optional
class MemoryLedger:
"""MEM-01/05: Memory ledger with JSON persistence tracing worker violation histories."""
def __init__(self):
self.entries: List[Dict[str, str]] = []
def add(self, worker_id: str, violation_type: str, evidence: str) -> None:
"""Adds a new memory, truncating evidence to 200 chars to avoid prompt bloat."""
clipped_evidence = evidence[:200]
self.entries.append(
{
"worker_id": worker_id,
"violation_type": violation_type,
"evidence": clipped_evidence,
}
)
# Maintain hard cap of 5000
if len(self.entries) > 5000:
self.entries.pop(0)
def retrieve(self, worker_id: str, violation_type: str) -> List[Dict[str, str]]:
"""MEM-02: Exact keyword match filtering returning max 3 items."""
matches = [
e
for e in self.entries
if e["worker_id"] == worker_id and e["violation_type"] == violation_type
]
# Return top_k=3 (the 3 most recent logically, so fetch from end or just standard)
return matches[-3:]
def clear(self) -> None:
"""MEM-04: Clear ledger for memory ablation experiments."""
self.entries.clear()
def save(self, path: Optional[str] = None) -> None:
"""MEM-01: Serializes the entire chain manually to JSON to prevent mid-rollout disk friction."""
if path is None:
path = os.path.join(os.path.dirname(__file__), "data", "memory_ledger.json")
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
json.dump(self.entries, f, indent=2)
def load(self, path: Optional[str] = None) -> None:
"""Reloads ledger state."""
load_path = (
path
if path is not None
else os.path.join(os.path.dirname(__file__), "data", "memory_ledger.json")
)
try:
with open(load_path, "r") as f:
self.entries = json.load(f)
except FileNotFoundError:
self.entries = []
|