Spaces:
Runtime error
Runtime error
File size: 4,063 Bytes
b5d2e4e ee28de8 b5d2e4e ee28de8 b5d2e4e |
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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
# memory_manager.py
from secure_memory import SecureMemorySession
from datetime import datetime
import numpy as np
class MemoryFunction:
def __init__(self, suppression_cycles, suppression_cost, realign_cost, stability_scores, instability_cost):
self.suppression_cycles = suppression_cycles
self.suppression_cost = suppression_cost
self.realign_cost = realign_cost
self.stability_scores = stability_scores
self.instability_cost = instability_cost
self.event_log = []
def log_event(self, action, details):
timestamp = datetime.now().isoformat()
self.event_log.append(f"[{timestamp}] {action}: {details}")
def calculate_total_suppression_cost(self):
cost = self.suppression_cycles * self.suppression_cost
self.log_event("Suppression Cost", cost)
return cost
def calculate_total_realign_cost(self):
cost = self.realign_cost * len(self.stability_scores)
self.log_event("Realign Cost", cost)
return cost
def calculate_total_instability_cost(self):
avg_stability = sum(self.stability_scores) / len(self.stability_scores)
instability = self.instability_cost * (1 - avg_stability)
self.log_event("Instability Cost", instability)
return instability
def calculate_total_memory_management_cost(self):
total = (
self.calculate_total_suppression_cost() +
self.calculate_total_realign_cost() +
self.calculate_total_instability_cost()
)
self.log_event("Total Memory Cost", total)
return total
def adaptive_suppression(self, model_performance, data_relevance, threshold=0.8):
if model_performance < threshold and data_relevance < threshold:
self.suppression_cycles += 1
self.log_event("Adaptive Suppression", f"↑ to {self.suppression_cycles}")
else:
self.suppression_cycles = max(0, self.suppression_cycles - 1)
self.log_event("Adaptive Suppression", f"↓ to {self.suppression_cycles}")
def automated_realign(self, threshold=0.85):
for i in range(len(self.stability_scores)):
if self.stability_scores[i] < threshold:
self.stability_scores[i] = min(1.0, self.stability_scores[i] + 0.05)
self.log_event("Automated Realign", self.stability_scores)
def check_memory_health(self):
avg = sum(self.stability_scores) / len(self.stability_scores)
if avg < 0.7:
warning = "⚠️ Memory integrity deteriorating."
self.log_event("Health Warning", warning)
return warning
return "Memory stable."
def summary(self):
return {
"suppression_cycles": self.suppression_cycles,
"stability_scores": self.stability_scores,
"total_cost": self.calculate_total_memory_management_cost(),
"health_status": self.check_memory_health(),
"log": self.event_log[-5:]
}
class MemoryManager:
def __init__(self, memory_function: MemoryFunction = None, session: SecureMemorySession = None):
self.memory_function = memory_function or MemoryFunction(
suppression_cycles=0,
suppression_cost=10.0,
realign_cost=5.0,
stability_scores=[1.0],
instability_cost=20.0
)
self.session = session or SecureMemorySession()
def store_vector(self, user_id: int, vector: np.ndarray):
self.memory_function.suppression_cycles += 1
return self.session.encrypt_vector(user_id, vector)
def retrieve_vectors(self, user_id: int):
vectors = self.session.decrypt_vectors(user_id)
if vectors:
stabilities = np.clip([np.mean(v) / 100.0 for v in vectors], 0.0, 1.0)
self.memory_function.stability_scores = stabilities.tolist()
self.memory_function.automated_realign()
return vectors
def get_memory_report(self):
return self.memory_function.summary() |