File size: 6,775 Bytes
dbb04e4 | 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 | """
Learning Journal
================
Tracks what works and what doesn't. Meta-learning layer for HAIM.
"""
import json
import os
from datetime import datetime, timezone
from typing import Dict, List, Optional
from dataclasses import dataclass, field, asdict
JOURNAL_PATH = "./data/learning_journal.json"
@dataclass
class LearningEntry:
"""A single learning."""
id: str
lesson: str
context: str
outcome: str # "success" | "failure" | "mixed"
confidence: float # 0.0 - 1.0
applications: int = 0 # Times this learning was applied
created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
tags: List[str] = field(default_factory=list)
class LearningJournal:
"""Meta-learning storage."""
def __init__(self, path: str = JOURNAL_PATH):
self.path = path
self.entries: Dict[str, LearningEntry] = {}
self.predictions: Dict[str, dict] = {} # In-memory prediction buffer
self._load()
def _load(self):
if os.path.exists(self.path):
with open(self.path, "r") as f:
data = json.load(f)
for eid, entry_data in data.items():
self.entries[eid] = LearningEntry(**entry_data)
def _save(self):
os.makedirs(os.path.dirname(self.path), exist_ok=True)
with open(self.path, "w") as f:
json.dump({k: asdict(v) for k, v in self.entries.items()}, f, indent=2)
def record(
self,
lesson: str,
context: str,
outcome: str = "success",
confidence: float = 0.7,
tags: List[str] = None,
surprise: float = 0.0
) -> str:
"""Record a new learning."""
entry_id = f"learn_{len(self.entries)}"
# Boost confidence if high surprise (flashbulb learning)
if surprise > 0.5:
confidence = min(1.0, confidence * (1.0 + surprise))
entry = LearningEntry(
id=entry_id,
lesson=lesson,
context=context,
outcome=outcome,
confidence=confidence,
tags=tags or []
)
if surprise > 0:
entry.tags.append(f"surprise_{surprise:.2f}")
if surprise > 0.7:
entry.tags.append("flashbulb_memory")
self.entries[entry_id] = entry
self._save()
return entry_id
def register_prediction(self, context: str, expectation: str) -> str:
"""Start a prediction cycle (v1.6)."""
pred_id = f"pred_{datetime.now(timezone.utc).timestamp()}"
self.predictions[pred_id] = {
"context": context,
"expectation": expectation,
"timestamp": datetime.now(timezone.utc)
}
return pred_id
def evaluate_surprise(self, expectation: str, actual: str) -> float:
"""
Calculate surprise metric (0.0 to 1.0).
Simple heuristic: Length difference + keyword overlap.
In v1.8 this should use semantic embedding distance.
"""
if expectation == actual:
return 0.0
# Jaccard similarity of words
exp_words = set(expectation.lower().split())
act_words = set(actual.lower().split())
if not exp_words or not act_words:
return 1.0
intersection = len(exp_words.intersection(act_words))
union = len(exp_words.union(act_words))
similarity = intersection / union
surprise = 1.0 - similarity
return surprise
def resolve_prediction(self, pred_id: str, actual_result: str) -> Optional[str]:
"""
Close the loop: Compare expectation vs reality, record learning if surprised.
"""
pred = self.predictions.pop(pred_id, None)
if not pred:
return None
surprise = self.evaluate_surprise(pred['expectation'], actual_result)
# Auto-record if significant surprise
if surprise > 0.3:
return self.record(
lesson=f"Expectation '{pred['expectation']}' differed from '{actual_result}'",
context=pred['context'],
outcome="mixed" if surprise < 0.8 else "failure",
confidence=0.8,
surprise=surprise,
tags=["prediction_error", "auto_generated"]
)
return None
def apply(self, entry_id: str):
"""Mark a learning as applied (reinforcement)."""
if entry_id in self.entries:
self.entries[entry_id].applications += 1
self.entries[entry_id].confidence = min(1.0, self.entries[entry_id].confidence * 1.05)
self._save()
def contradict(self, entry_id: str):
"""Mark a learning as contradicted (weakening)."""
if entry_id in self.entries:
self.entries[entry_id].confidence *= 0.8
self._save()
def query(self, context: str, top_k: int = 5) -> List[LearningEntry]:
"""Find relevant learnings for a context."""
# Simple keyword matching for now
context_lower = context.lower()
scored = []
for entry in self.entries.values():
score = 0
for word in context_lower.split():
if word in entry.context.lower() or word in entry.lesson.lower():
score += 1
if word in entry.tags:
score += 2
score *= entry.confidence
if score > 0:
scored.append((score, entry))
scored.sort(key=lambda x: x[0], reverse=True)
return [e for _, e in scored[:top_k]]
def get_top_learnings(self, n: int = 10) -> List[LearningEntry]:
"""Get most confident/applied learnings."""
sorted_entries = sorted(
self.entries.values(),
key=lambda e: e.confidence * (1 + e.applications * 0.1),
reverse=True
)
return sorted_entries[:n]
def stats(self) -> Dict:
return {
"total_learnings": len(self.entries),
"successes": sum(1 for e in self.entries.values() if e.outcome == "success"),
"failures": sum(1 for e in self.entries.values() if e.outcome == "failure"),
"avg_confidence": sum(e.confidence for e in self.entries.values()) / max(1, len(self.entries)),
"total_applications": sum(e.applications for e in self.entries.values())
}
|