File size: 11,022 Bytes
ed1b365 | 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 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 | """Codette Living Memory Kernel — Emotionally-Tagged Memory Cocoons
Memories are tagged with emotional context, importance scoring, and
SHA-256 anchors for integrity. The kernel supports recall by emotion,
importance-based pruning, and automatic cocoon formation from
conversation turns.
Origin: codette_memory_kernel.py + dreamcore_wakestate_engine.py, rebuilt
"""
import time
import hashlib
import json
import math
from dataclasses import dataclass, field
from typing import Dict, List, Optional
# Emotional tags recognized by the memory system
EMOTIONAL_TAGS = [
"neutral", "curiosity", "awe", "joy", "insight",
"confusion", "frustration", "fear", "empathy",
"determination", "surprise", "trust", "gratitude",
]
# Keywords that suggest emotional context in text
_EMOTION_SIGNALS = {
"curiosity": ["why", "how", "what if", "wonder", "curious", "explore"],
"awe": ["amazing", "incredible", "beautiful", "profound", "mind-blowing"],
"joy": ["happy", "glad", "love", "wonderful", "great", "excellent"],
"insight": ["realize", "understand", "aha", "discover", "breakthrough"],
"confusion": ["confused", "unclear", "don't understand", "lost", "huh"],
"frustration": ["frustrated", "annoyed", "broken", "doesn't work", "bug"],
"fear": ["worried", "concerned", "dangerous", "risk", "threat"],
"empathy": ["feel", "compassion", "care", "support", "kind"],
"determination": ["must", "need to", "will", "going to", "commit"],
"surprise": ["unexpected", "surprised", "didn't expect", "wow", "whoa"],
"trust": ["trust", "reliable", "depend", "confident", "safe"],
"gratitude": ["thank", "grateful", "appreciate", "helpful"],
}
@dataclass
class MemoryCocoon:
"""A single memory unit with emotional tagging and integrity anchor."""
title: str
content: str
emotional_tag: str = "neutral"
importance: int = 5 # 1-10 scale
timestamp: float = 0.0
anchor: str = "" # SHA-256 integrity hash
adapter_used: str = "" # Which perspective generated this
query: str = "" # Original user query
coherence: float = 0.0 # Epistemic coherence at time of creation
tension: float = 0.0 # Epistemic tension at time of creation
def __post_init__(self):
if self.timestamp == 0.0:
self.timestamp = time.time()
if not self.anchor:
self.anchor = self._generate_anchor()
def _generate_anchor(self) -> str:
raw = f"{self.title}{self.timestamp}{self.content}".encode("utf-8")
return hashlib.sha256(raw).hexdigest()[:16]
def to_dict(self) -> Dict:
return {
"title": self.title,
"content": self.content[:500], # Cap stored content
"emotional_tag": self.emotional_tag,
"importance": self.importance,
"timestamp": self.timestamp,
"anchor": self.anchor,
"adapter_used": self.adapter_used,
"query": self.query[:200],
"coherence": self.coherence,
"tension": self.tension,
}
@classmethod
def from_dict(cls, d: Dict) -> "MemoryCocoon":
return cls(**{k: v for k, v in d.items()
if k in cls.__dataclass_fields__})
def age_hours(self) -> float:
return (time.time() - self.timestamp) / 3600.0
class LivingMemoryKernel:
"""Emotionally-aware memory store with importance-based pruning.
Memories form naturally from conversation — each significant exchange
becomes a cocoon. The kernel can recall by emotion, importance, or
recency, and automatically prunes low-importance memories when full.
"""
def __init__(self, max_memories: int = 100):
self.memories: List[MemoryCocoon] = []
self.max_memories = max_memories
self._emotion_index: Dict[str, List[int]] = {}
def store(self, cocoon: MemoryCocoon):
"""Store a memory cocoon, pruning if at capacity."""
# Don't store duplicates (same anchor)
if any(m.anchor == cocoon.anchor for m in self.memories):
return
self.memories.append(cocoon)
self._rebuild_index()
# Auto-prune if over capacity
if len(self.memories) > self.max_memories:
self.prune(keep_n=self.max_memories)
def store_from_turn(self, query: str, response: str,
adapter: str = "", coherence: float = 0.0,
tension: float = 0.0):
"""Create and store a memory from a conversation turn."""
emotion = detect_emotion(query + " " + response)
importance = self._estimate_importance(query, response, coherence)
cocoon = MemoryCocoon(
title=query[:80],
content=response[:500],
emotional_tag=emotion,
importance=importance,
adapter_used=adapter,
query=query,
coherence=coherence,
tension=tension,
)
self.store(cocoon)
return cocoon
def recall_by_emotion(self, tag: str, limit: int = 10) -> List[MemoryCocoon]:
"""Recall memories with a specific emotional tag."""
indices = self._emotion_index.get(tag, [])
results = [self.memories[i] for i in indices]
return sorted(results, key=lambda m: m.importance, reverse=True)[:limit]
def recall_important(self, min_importance: int = 7,
limit: int = 10) -> List[MemoryCocoon]:
"""Recall high-importance memories."""
results = [m for m in self.memories if m.importance >= min_importance]
return sorted(results, key=lambda m: m.importance, reverse=True)[:limit]
def recall_recent(self, limit: int = 10) -> List[MemoryCocoon]:
"""Recall most recent memories."""
return sorted(self.memories, key=lambda m: m.timestamp, reverse=True)[:limit]
def recall_by_adapter(self, adapter: str,
limit: int = 10) -> List[MemoryCocoon]:
"""Recall memories generated by a specific perspective."""
results = [m for m in self.memories if m.adapter_used == adapter]
return sorted(results, key=lambda m: m.timestamp, reverse=True)[:limit]
def search(self, terms: str, limit: int = 5) -> List[MemoryCocoon]:
"""Simple keyword search across memory content."""
words = terms.lower().split()
scored = []
for m in self.memories:
text = (m.title + " " + m.content + " " + m.query).lower()
score = sum(1 for w in words if w in text)
if score > 0:
scored.append((score, m))
scored.sort(key=lambda x: x[0], reverse=True)
return [m for _, m in scored[:limit]]
def prune(self, keep_n: int = 50):
"""Keep only the most important memories."""
# Sort by composite score: importance * recency_bonus
now = time.time()
def score(m):
age_days = (now - m.timestamp) / 86400.0
recency = math.exp(-age_days / 7.0) # Half-life ~7 days
return m.importance * (0.5 + 0.5 * recency)
self.memories.sort(key=score, reverse=True)
self.memories = self.memories[:keep_n]
self._rebuild_index()
def emotional_profile(self) -> Dict[str, int]:
"""Get a count of memories by emotional tag."""
profile = {}
for m in self.memories:
profile[m.emotional_tag] = profile.get(m.emotional_tag, 0) + 1
return profile
def get_state(self) -> Dict:
"""Export kernel state for session/API."""
return {
"total_memories": len(self.memories),
"emotional_profile": self.emotional_profile(),
"recent": [m.to_dict() for m in self.recall_recent(3)],
"important": [m.to_dict() for m in self.recall_important(limit=3)],
}
def _estimate_importance(self, query: str, response: str,
coherence: float) -> int:
"""Estimate importance on 1-10 scale from content signals."""
score = 5 # Base
# Longer, more substantive exchanges
if len(response) > 500:
score += 1
if len(response) > 1500:
score += 1
# High coherence suggests meaningful synthesis
if coherence > 0.8:
score += 1
# Question complexity
q = query.lower()
if any(w in q for w in ["why", "how", "explain", "analyze"]):
score += 1
if "?" in query and len(query.split()) > 8:
score += 1
return min(10, max(1, score))
def _rebuild_index(self):
"""Rebuild the emotion-to-index lookup."""
self._emotion_index.clear()
for i, m in enumerate(self.memories):
self._emotion_index.setdefault(m.emotional_tag, []).append(i)
def to_dict(self) -> Dict:
return {"memories": [m.to_dict() for m in self.memories]}
def store_conflict(self, conflict: Dict, resolution_outcome: Optional[Dict] = None):
"""
Store conflict metadata as a memory cocoon.
Args:
conflict: Dict with agent_a, agent_b, claim_a, claim_b, conflict_type, conflict_strength, etc.
resolution_outcome: Optional dict with coherence_after, resolution_score, etc.
"""
if resolution_outcome is None:
resolution_outcome = {}
# Create a conflict cocoon
cocoon = MemoryCocoon(
title=f"Conflict: {conflict.get('agent_a', '?')} vs {conflict.get('agent_b', '?')} ({conflict.get('conflict_type', 'unknown')})",
content=json.dumps(conflict),
emotional_tag="tension",
importance=int(conflict.get("conflict_strength", 0.5) * 10), # 1-10 scale
adapter_used=f"{conflict.get('agent_a', '?')},{conflict.get('agent_b', '?')}",
query="",
coherence=resolution_outcome.get("coherence_after", 0.5),
tension=conflict.get("conflict_strength", 0.5),
)
self.store(cocoon)
@classmethod
def from_dict(cls, d: Dict) -> "LivingMemoryKernel":
kernel = cls()
for md in d.get("memories", []):
kernel.memories.append(MemoryCocoon.from_dict(md))
kernel._rebuild_index()
return kernel
def detect_emotion(text: str) -> str:
"""Detect the dominant emotional tag from text content."""
text_lower = text.lower()
scores = {}
for emotion, keywords in _EMOTION_SIGNALS.items():
score = sum(1 for kw in keywords if kw in text_lower)
if score > 0:
scores[emotion] = score
if not scores:
return "neutral"
return max(scores, key=scores.get)
|