File size: 6,947 Bytes
d997548 | 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 | """
Cognitive Evolution Log for Berna R7.
Records every event in the model's cognitive lifecycle:
- birth: new cell created when new pattern discovered
- split: existing cell divides due to saturation
- pattern: new knowledge signature detected
- milestone: notable thresholds crossed
Persistent (JSONL), queryable, narrative-generating, feedable to model.
"""
import json
import hashlib
from pathlib import Path
from datetime import datetime, timezone
from typing import List, Optional
class CognitiveLog:
def __init__(self, log_dir, model_uuid="mother-001"):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(parents=True, exist_ok=True)
self.model_uuid = model_uuid
self.jsonl_path = self.log_dir / f"{model_uuid}_events.jsonl"
self.narrative_path = self.log_dir / f"{model_uuid}_narrative.txt"
self.events = []
self.MAX_EVENTS = 10000
self._load_existing()
def _load_existing(self):
if self.jsonl_path.exists():
with open(self.jsonl_path, "r", encoding="utf-8") as f:
for line in f:
try:
self.events.append(json.loads(line))
except Exception:
pass
def record(self, event_type, **fields):
event = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"step": fields.pop("step", None),
"type": event_type,
"model_uuid": self.model_uuid,
}
event.update(fields)
self.events.append(event)
if len(self.events) > self.MAX_EVENTS:
self.events = self.events[-self.MAX_EVENTS:]
with open(self.jsonl_path, "a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False) + "\n")
return event
def record_birth(self, cell_id, parent_id, pattern_sig, step, layer_idx):
return self.record("birth", step=step, cell_id=cell_id,
parent_id=parent_id, pattern_sig=pattern_sig,
layer_idx=layer_idx)
def record_split(self, parent_id, child_a, child_b, S, step, layer_idx):
return self.record("split", step=step, parent_id=parent_id,
child_a=child_a, child_b=child_b,
S=round(float(S), 4), layer_idx=layer_idx)
def record_pattern(self, pattern_sig, cell_ids, step, count):
return self.record("pattern", step=step, pattern_sig=pattern_sig,
cell_ids=cell_ids, seen_count=count)
def record_milestone(self, name, value, step):
return self.record("milestone", step=step, name=name, value=value)
def recent(self, n=20, event_type=None):
if event_type is None:
return self.events[-n:]
return [e for e in self.events if e["type"] == event_type][-n:]
def count_by_type(self):
counts = {}
for e in self.events:
counts[e["type"]] = counts.get(e["type"], 0) + 1
return counts
def to_narrative(self, max_events=50):
events = self.events[-max_events:]
lines = []
lines.append(f"=== Berna {self.model_uuid} — Cognitive Chronicle ===")
lines.append(f"Total events: {len(self.events)}")
lines.append(f"Breakdown: {self.count_by_type()}")
lines.append("")
for e in events:
t = e["type"]
if t == "birth":
lines.append(f"[step {e['step']}] Born: {e['cell_id']} "
f"(from {e.get('parent_id','seed')}) — "
f"knows pattern {str(e.get('pattern_sig','?'))[:8]}")
elif t == "split":
lines.append(f"[step {e['step']}] Split: {e['parent_id']} → "
f"{e['child_a']} + {e['child_b']} (S={e['S']})")
elif t == "pattern":
lines.append(f"[step {e['step']}] New pattern: "
f"{str(e['pattern_sig'])[:8]} "
f"(seen {e['seen_count']}×, matched "
f"{len(e.get('cell_ids', []))} cells)")
elif t == "milestone":
lines.append(f"[step {e['step']}] Milestone: "
f"{e['name']} = {e['value']}")
return "\n".join(lines)
def write_narrative(self, path=None):
text = self.to_narrative()
p = Path(path) if path else self.narrative_path
p.write_text(text, encoding="utf-8")
return p
def sample_for_training(self, n=2):
"""Short self-narrative snippet to prepend to training samples."""
births = self.recent(n=n, event_type="birth")
splits = self.recent(n=n, event_type="split")
lines = ["I am Berna R7, a self-growing language model."]
if births:
ids = ", ".join(e["cell_id"] for e in births)
lines.append(f"Recently I gave birth to new cells: {ids}.")
if splits:
ids = ", ".join(e["parent_id"] for e in splits)
lines.append(f"I also split {ids} to refine my knowledge.")
if not births and not splits:
lines.append("I am learning language, mathematics, and code.")
return " ".join(lines)
def pattern_signature(activations, top_k=50):
"""Binary fingerprint of top-K activated neurons, hashed."""
import numpy as np
a = np.abs(np.asarray(activations, dtype=np.float32))
mean_a = a.mean(axis=0)
if mean_a.size == 0:
return "empty"
k = min(top_k, mean_a.size)
top_idx = np.argsort(mean_a)[-k:]
fp = np.zeros(mean_a.size, dtype=np.uint8)
fp[top_idx] = 1
return hashlib.sha256(fp.tobytes()).hexdigest()[:16]
def hamming_similarity(fp_a, fp_b):
"""Similarity ∈ [0,1] between two fingerprints (hex strings)."""
import numpy as np
try:
a = np.frombuffer(bytes.fromhex(fp_a), dtype=np.uint8)
b = np.frombuffer(bytes.fromhex(fp_b), dtype=np.uint8)
except Exception:
return 0.0
n = min(len(a), len(b))
if n == 0:
return 0.0
matches = np.sum(a[:n] == b[:n])
return float(matches / n)
if __name__ == "__main__":
import numpy as np
log = CognitiveLog("/tmp/berna_test_log")
log.record_birth("c-001", "seed", "abc123", 1, 0)
log.record_split("c-000", "c-000-a", "c-000-b", 0.87, 100, 0)
log.record_milestone("first_split", 1, 100)
print(log.to_narrative())
print()
a = np.random.randn(32, 128).astype(np.float32)
b = np.random.randn(32, 128).astype(np.float32)
print("fp_a:", pattern_signature(a))
print("fp_b:", pattern_signature(b))
print("sim:", hamming_similarity(pattern_signature(a), pattern_signature(b)))
|